# -*- coding:utf-8 -*- # POST /api/submit - 提交KTV视频合成任务 import json import uuid from ahserver.serverenv import ServerEnv method = request.method if method == 'POST': # 必需参数 video_clips = params_kw.get('video_clips', []) audio_path = params_kw.get('audio_path', '') subtitle_path = params_kw.get('subtitle_path', '') if not video_clips or not isinstance(video_clips, list): return json.dumps({'error': 'video_clips (list) is required'}, ensure_ascii=False) if not audio_path: return json.dumps({'error': 'audio_path is required'}, ensure_ascii=False) if not subtitle_path: return json.dumps({'error': 'subtitle_path (ASS file) is required'}, ensure_ascii=False) task_id = params_kw.get('task_id', str(uuid.uuid4()).replace("-", "")[:12]) output_format = params_kw.get('output_format', 'mp4') resolution = params_kw.get('resolution', '1920x1080') payload = { 'task_type': 'synthesize', 'task_id': task_id, 'video_clips': video_clips, 'audio_path': audio_path, 'subtitle_path': subtitle_path, 'output_format': output_format, 'resolution': resolution } env = ServerEnv() longtasks = env.longtasks if longtasks is None: return json.dumps({'error': 'service not ready'}, ensure_ascii=False) result = await longtasks.submit_task(payload) real_task_id = result.get('task_id', str(result)) if isinstance(result, dict) else str(result) return json.dumps({ 'task_id': real_task_id, 'status': 'queued', 'video_clips_count': len(video_clips), 'audio_path': audio_path, 'subtitle_path': subtitle_path, 'message': 'task submitted', 'check_url': f'/api/task?task_id={real_task_id}' }, ensure_ascii=False) else: return json.dumps({ 'usage': 'POST with JSON body', 'params': { 'video_clips': 'list (required, server paths to video segments)', 'audio_path': 'string (required, server path to audio file)', 'subtitle_path': 'string (required, server path to ASS subtitle file)', 'output_format': 'string (default mp4)', 'resolution': 'string (default 1920x1080)', 'task_id': 'string (optional, auto-generated)', } }, ensure_ascii=False)