Luma API 快速开始

import requests
import time

class LumaAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = 'https://api.kie.ai/api/v1/modify'
        self.headers = {
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        }
    
    def modify_video(self, prompt, video_url, **options):
        data = {
            'prompt': prompt,
            'videoUrl': video_url,
            'callBackUrl': options.get('callBackUrl'),
            'watermark': options.get('watermark'),
            **options
        }
        
        # 移除 None 值
        data = {k: v for k, v in data.items() if v is not None}
        
        response = requests.post(f'{self.base_url}/generate', 
                               headers=self.headers, json=data)
        result = response.json()
        
        if not (response.ok and result.get('code') == 200):
            raise Exception(f"生成失败: {result.get('msg', '未知错误')}")
        
        return result['data']['taskId']
    
    def wait_for_completion(self, task_id, max_wait_time=900):
        start_time = time.time()
        
        while time.time() - start_time < max_wait_time:
            status = self.get_task_status(task_id)
            
            if status['successFlag'] == 1:
                return status['response']
            elif status['successFlag'] in [2, 3]:
                error_msg = status.get('errorMessage', '未知错误')
                raise Exception(f"生成失败: {error_msg}")
            
            time.sleep(10)  # 等待 10 秒
        
        raise Exception('生成超时')
    
    def get_task_status(self, task_id):
        response = requests.get(f'{self.base_url}/record-info?taskId={task_id}',
                              headers={'Authorization': f'Bearer {self.api_key}'})
        result = response.json()
        
        if not (response.ok and result.get('code') == 200):
            raise Exception(f"查询失败: {result.get('msg', '未知错误')}")
        
        return result['data']

# 使用示例
def main():
    api = LumaAPI('YOUR_API_KEY')
    
    try:
        # 视频修改
        print('开始视频修改...')
        task_id = api.modify_video(
            'A futuristic cityscape at night with towering glass spires reaching into a starry sky. Neon lights in blue and purple illuminate the buildings while flying vehicles glide silently between the structures. Holographic advertisements flicker and change on building facades.',
            'https://example.com/input-video.mp4',
            callBackUrl='https://your-callback-url.com/luma-callback',
            watermark='your-watermark-id'
        )
        
        # 等待完成
        print(f'任务 ID: {task_id}。等待完成...')
        result = api.wait_for_completion(task_id)
        
        print('视频修改成功！')
        print(f'结果视频 URL: {result["resultUrls"]}')
        print(f'原始视频 URL: {result["originUrls"]}')
        
        # 下载视频
        for i, url in enumerate(result['resultUrls']):
            filename = f'modified_video_{i+1}.mp4'
            download_video(url, filename)
        
    except Exception as error:
        print(f'错误: {error}')

def download_video(url, filename):
    response = requests.get(url, stream=True)
    response.raise_for_status()
    
    with open(filename, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    print(f'已下载: {filename}')

if __name__ == '__main__':
    main()