67 lines
2.3 KiB
Plaintext

# -*- 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)