804 lines
30 KiB
Python
804 lines
30 KiB
Python
# -*- coding:utf-8 -*-
|
||
"""
|
||
KTV Pipeline Worker v2 - 标准化歌曲制作流程
|
||
|
||
支持3种输入模式：
|
||
1. audio_lyrics: 原唱音频 + 歌词 → 快速生成KTV
|
||
2. video_lyrics: 原唱视频 + 歌词 → 使用原视频画面
|
||
3. lyrics_only: 仅歌词/大纲 → 完整AI生成（音乐+MV）
|
||
|
||
输出产物：
|
||
1. KTV视频（双轨音频：伴奏默认播放 + 原唱可切换，卡拉OK字幕）
|
||
2. 单轨字幕视频（原声混合 + 烧录字幕）
|
||
|
||
状态机：
|
||
Mode A (audio_lyrics):
|
||
submitted → audio_preparing → demucs_separating → lyric_calibrating → subtitle_rendering → ktv_synthesizing → completed
|
||
|
||
Mode B (video_lyrics):
|
||
submitted → video_preparing → demucs_separating → lyric_calibrating → subtitle_rendering → ktv_synthesizing → completed
|
||
|
||
Mode C (lyrics_only):
|
||
submitted → lyric_generating → lyric_evaluating → music_generating → music_polling →
|
||
demucs_separating → lyric_calibrating → subtitle_rendering → ktv_synthesizing → completed
|
||
|
||
任何状态均可转换到 'failed'（错误时）
|
||
lyric_evaluating 可循环回 lyric_generating（阈值未达标时，最多3次）
|
||
"""
|
||
import json
|
||
import os
|
||
import asyncio
|
||
import time
|
||
import aiohttp
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
SERVICES = {
|
||
'demucs': 'http://127.0.0.1:9080/api/demucs',
|
||
'lyric_calibrate': 'http://127.0.0.1:9080/api/lyric_calibrate',
|
||
'songrate': 'http://127.0.0.1:8900/api/evaluate',
|
||
}
|
||
|
||
LLM_API = os.environ.get('LLM_API_BASE', 'https://token.opencomputing.cn/llmage/v1')
|
||
WORK_DIR = '/tmp/ktv_pipelines'
|
||
|
||
|
||
async def update_state(redis, pipeline_id, state, **kwargs):
|
||
"""更新Pipeline状态到Redis"""
|
||
data = await redis.get(f'pipeline:{pipeline_id}')
|
||
if data:
|
||
pipeline = json.loads(data)
|
||
pipeline['state'] = state
|
||
pipeline['updated_at'] = time.time()
|
||
if 'artifacts' in kwargs:
|
||
pipeline['artifacts'].update(kwargs['artifacts'])
|
||
if 'error' in kwargs:
|
||
pipeline['errors'].append({'state': state, 'error': kwargs['error'], 'time': time.time()})
|
||
pipeline.update({k: v for k, v in kwargs.items() if k not in ('artifacts', 'error')})
|
||
await redis.set(f'pipeline:{pipeline_id}', json.dumps(pipeline, ensure_ascii=False), ex=86400)
|
||
|
||
|
||
async def call_llm(session, prompt, model='qwen3-235b-a22b', temperature=0.7, max_tokens=4096):
|
||
"""调用LLM API"""
|
||
api_key = os.environ.get('LLM_API_KEY', '')
|
||
if not api_key:
|
||
try:
|
||
from ahserver.serverenv import ServerEnv
|
||
env = ServerEnv()
|
||
api_key = getattr(env, 'llm_api_key', '') or ''
|
||
except:
|
||
pass
|
||
|
||
headers = {
|
||
'Authorization': f'Bearer {api_key}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
payload = {
|
||
'model': model,
|
||
'catelogid': 't2t',
|
||
'messages': [{'role': 'user', 'content': prompt}],
|
||
'temperature': temperature,
|
||
'max_tokens': max_tokens
|
||
}
|
||
async with session.post(f'{LLM_API}/chat/completions', json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=120)) as resp:
|
||
data = await resp.json()
|
||
return data.get('choices', [{}])[0].get('message', {}).get('content', '')
|
||
|
||
|
||
async def call_service(session, url, data, timeout=300):
|
||
"""调用内部服务"""
|
||
async with session.post(url, json=data, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
|
||
return await resp.json()
|
||
|
||
|
||
async def call_service_multipart(session, url, audio_data, audio_filename, lyrics, timeout=600):
|
||
"""调用lyric_calibrate服务（multipart/form-data）"""
|
||
boundary = f'Boundary{int(time.time()*1000)}'
|
||
|
||
body_parts = []
|
||
# audio_file field
|
||
body_parts.append(('--' + boundary + '\r\n').encode())
|
||
body_parts.append(f'Content-Disposition: form-data; name="audio_file"; filename="{audio_filename}"\r\n'.encode())
|
||
body_parts.append(b'Content-Type: audio/wav\r\n\r\n')
|
||
body_parts.append(audio_data)
|
||
body_parts.append(b'\r\n')
|
||
|
||
# lyrics field
|
||
body_parts.append(('--' + boundary + '\r\n').encode())
|
||
body_parts.append(b'Content-Disposition: form-data; name="lyrics"\r\n\r\n')
|
||
body_parts.append(lyrics.encode('utf-8'))
|
||
body_parts.append(b'\r\n')
|
||
|
||
# closing boundary
|
||
body_parts.append(('--' + boundary + '--\r\n').encode())
|
||
|
||
body = b''.join(body_parts)
|
||
|
||
headers = {
|
||
'Content-Type': f'multipart/form-data; boundary={boundary}'
|
||
}
|
||
|
||
async with session.post(url, data=body, headers=headers, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
|
||
return await resp.json()
|
||
|
||
|
||
# ============================================================================
|
||
# Mode A: audio_lyrics - 原唱音频 + 歌词
|
||
# ============================================================================
|
||
|
||
async def step_audio_preparing(pipeline, session):
|
||
"""准备原唱音频：提取音频并确保格式正确"""
|
||
audio_path = pipeline.get('input_audio', '')
|
||
pipeline_dir = os.path.join(WORK_DIR, pipeline['id'])
|
||
os.makedirs(pipeline_dir, exist_ok=True)
|
||
|
||
# 复制输入音频到工作目录
|
||
input_audio_local = os.path.join(pipeline_dir, 'input_audio.mp3')
|
||
|
||
if audio_path.startswith('http'):
|
||
# 下载远程音频
|
||
async with session.get(audio_path, timeout=aiohttp.ClientTimeout(total=120)) as resp:
|
||
with open(input_audio_local, 'wb') as f:
|
||
f.write(await resp.read())
|
||
else:
|
||
# 本地文件，复制
|
||
if not os.path.exists(audio_path):
|
||
raise ValueError(f'Input audio not found: {audio_path}')
|
||
subprocess.run(['cp', audio_path, input_audio_local], check=True)
|
||
|
||
# 验证音频文件
|
||
probe = subprocess.run(
|
||
['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
|
||
'-of', 'default=noprint_wrappers=1:nokey=1', input_audio_local],
|
||
capture_output=True, text=True
|
||
)
|
||
duration = float(probe.stdout.strip())
|
||
|
||
return {
|
||
'audio_path': input_audio_local,
|
||
'audio_duration': duration,
|
||
'mode': 'audio_lyrics'
|
||
}
|
||
|
||
|
||
# ============================================================================
|
||
# Mode B: video_lyrics - 原唱视频 + 歌词
|
||
# ============================================================================
|
||
|
||
async def step_video_preparing(pipeline, session):
|
||
"""准备原唱视频：提取音频并确保格式正确"""
|
||
video_path = pipeline.get('input_video', '')
|
||
pipeline_dir = os.path.join(WORK_DIR, pipeline['id'])
|
||
os.makedirs(pipeline_dir, exist_ok=True)
|
||
|
||
# 复制输入视频到工作目录
|
||
input_video_local = os.path.join(pipeline_dir, 'input_video.mp4')
|
||
input_audio_local = os.path.join(pipeline_dir, 'input_audio.mp3')
|
||
|
||
if video_path.startswith('http'):
|
||
# 下载远程视频
|
||
async with session.get(video_path, timeout=aiohttp.ClientTimeout(total=300)) as resp:
|
||
with open(input_video_local, 'wb') as f:
|
||
f.write(await resp.read())
|
||
else:
|
||
# 本地文件，复制
|
||
if not os.path.exists(video_path):
|
||
raise ValueError(f'Input video not found: {video_path}')
|
||
subprocess.run(['cp', video_path, input_video_local], check=True)
|
||
|
||
# 从视频提取音频
|
||
subprocess.run([
|
||
'ffmpeg', '-y', '-i', input_video_local,
|
||
'-vn', '-acodec', 'libmp3lame', '-ab', '192k',
|
||
input_audio_local
|
||
], check=True, capture_output=True)
|
||
|
||
# 验证视频文件
|
||
probe = subprocess.run(
|
||
['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
|
||
'-of', 'default=noprint_wrappers=1:nokey=1', input_video_local],
|
||
capture_output=True, text=True
|
||
)
|
||
duration = float(probe.stdout.strip())
|
||
|
||
return {
|
||
'video_path': input_video_local,
|
||
'audio_path': input_audio_local,
|
||
'audio_duration': duration,
|
||
'mode': 'video_lyrics'
|
||
}
|
||
|
||
|
||
# ============================================================================
|
||
# Mode C: lyrics_only - 仅歌词（完整AI生成）
|
||
# ============================================================================
|
||
|
||
async def step_lyric_generate(pipeline, session):
|
||
"""生成歌词（如果用户只提供了大纲）"""
|
||
outline = pipeline.get('outline', '')
|
||
if outline:
|
||
# 用户提供了大纲，生成完整歌词
|
||
prompt = f"""你是一位专业的华语歌词创作者。
|
||
|
||
请根据以下大纲创作一首完整的歌词：
|
||
|
||
大纲：{outline}
|
||
风格：{pipeline.get('scene', 'pop')}
|
||
|
||
要求：
|
||
1. 歌词结构完整：包含主歌(verse)、副歌(chorus)、桥段(bridge)
|
||
2. 每行歌词节奏感强，适合演唱
|
||
3. 标注段落类型，如 [Verse1], [Chorus], [Bridge] 等
|
||
4. 总共16-24行歌词
|
||
5. 注意押韵和情感表达
|
||
|
||
请直接输出歌词（不要markdown标记）：
|
||
"""
|
||
lyrics = await call_llm(session, prompt)
|
||
return {'lyrics': lyrics.strip()}
|
||
else:
|
||
# 用户已提供歌词，跳过生成
|
||
return {'lyrics': pipeline.get('lyrics', '')}
|
||
|
||
|
||
async def step_lyric_evaluate(pipeline, session):
|
||
"""评估歌词质量"""
|
||
lyrics = pipeline['artifacts'].get('lyrics', '')
|
||
prompt = f"""你是一位严格的歌词评审专家。请评估以下歌词的质量，满分10分。
|
||
|
||
评分维度：
|
||
1. 结构与韵律 (2分)
|
||
2. 情感表达 (2分)
|
||
3. 意象与画面感 (2分)
|
||
4. 语言质量 (2分)
|
||
5. 可唱性 (2分)
|
||
|
||
歌词：
|
||
{lyrics}
|
||
|
||
请严格按JSON格式返回：
|
||
{{"total_score": 7.5, "dimensions": {{"structure": 1.5, "emotion": 1.5, "imagery": 1.5, "language": 1.5, "singability": 1.5}}, "comment": "简短评语"}}
|
||
"""
|
||
result = await call_llm(session, prompt, temperature=0.3)
|
||
try:
|
||
result = result.strip()
|
||
if result.startswith('```'):
|
||
result = result.split('```')[1]
|
||
if result.startswith('json'):
|
||
result = result[4:]
|
||
score_data = json.loads(result.strip())
|
||
except:
|
||
score_data = {'total_score': 7.0, 'dimensions': {}, 'comment': result[:200]}
|
||
return {'lyric_score': score_data}
|
||
|
||
|
||
async def step_music_generate(pipeline, session):
|
||
"""生成音乐（Suno API）"""
|
||
lyrics = pipeline['artifacts'].get('lyrics', '')
|
||
scene = pipeline.get('scene', 'pop')
|
||
|
||
api_key = os.environ.get('LLM_API_KEY', '')
|
||
if not api_key:
|
||
try:
|
||
from ahserver.serverenv import ServerEnv
|
||
env = ServerEnv()
|
||
api_key = getattr(env, 'llm_api_key', '') or ''
|
||
except:
|
||
pass
|
||
|
||
headers = {
|
||
'Authorization': f'Bearer {api_key}',
|
||
'Content-Type': 'application/json'
|
||
}
|
||
payload = {
|
||
'model': 'suno-v4',
|
||
'catelogid': 'music_gen',
|
||
'prompt': f'{scene} style music',
|
||
'tags': scene,
|
||
'title': f"AI Song - {pipeline['id']}",
|
||
}
|
||
async with session.post(f'{LLM_API}/audio/generations', json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=60)) as resp:
|
||
data = await resp.json()
|
||
taskid = data.get('taskid', data.get('id', ''))
|
||
return {'music_task_id': taskid, 'music_submit_response': data}
|
||
|
||
|
||
async def step_music_poll(pipeline, session):
|
||
"""轮询音乐生成结果"""
|
||
task_id = pipeline['artifacts'].get('music_task_id', '')
|
||
if not task_id:
|
||
raise ValueError('No music_task_id found')
|
||
|
||
api_key = os.environ.get('LLM_API_KEY', '')
|
||
headers = {'Authorization': f'Bearer {api_key}'}
|
||
|
||
# 最多等待10分钟
|
||
for attempt in range(60):
|
||
async with session.get(f'{LLM_API}/tasks?taskid={task_id}', headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp:
|
||
data = await resp.json()
|
||
status = data.get('status', '')
|
||
if status == 'SUCCEEDED':
|
||
return {'music_url': data.get('result_url', ''), 'music_status': 'SUCCEEDED'}
|
||
elif status == 'FAILED':
|
||
raise ValueError(f'Music generation failed: {data}')
|
||
await asyncio.sleep(10)
|
||
|
||
raise ValueError('Music generation timed out')
|
||
|
||
|
||
async def step_demucs_separate(pipeline, session):
|
||
"""Demucs人声分离（通用步骤，适用于所有模式）"""
|
||
# 确定输入音频路径
|
||
audio_path = pipeline['artifacts'].get('audio_path', '')
|
||
music_url = pipeline['artifacts'].get('music_url', '')
|
||
|
||
pipeline_dir = os.path.join(WORK_DIR, pipeline['id'])
|
||
os.makedirs(pipeline_dir, exist_ok=True)
|
||
|
||
# Mode C: 需要先下载音乐
|
||
if not audio_path and music_url:
|
||
audio_path = os.path.join(pipeline_dir, 'music.mp3')
|
||
async with session.get(music_url, timeout=aiohttp.ClientTimeout(total=120)) as resp:
|
||
with open(audio_path, 'wb') as f:
|
||
f.write(await resp.read())
|
||
|
||
if not audio_path:
|
||
raise ValueError('No audio source for demucs separation')
|
||
if not os.path.exists(audio_path):
|
||
raise ValueError(f'Audio file not found: {audio_path}')
|
||
|
||
# 调用demucs分离
|
||
result = await call_service(session, SERVICES['demucs'], {'filepath': audio_path})
|
||
|
||
if result.get('status') != 'success':
|
||
raise ValueError(f"Demucs failed: {result.get('error', result)}")
|
||
|
||
# demucs API 返回 vocals_url/no_vocals_url (格式: /idfile?path=...)
|
||
# 需要转换为本地文件路径
|
||
import re
|
||
|
||
vocals_url = result.get('vocals_url', '')
|
||
no_vocals_url = result.get('no_vocals_url', '')
|
||
|
||
def url_to_path(url):
|
||
"""Convert /idfile?path=xxx to /tmp/xxx"""
|
||
match = re.search(r'path=(.+)', url)
|
||
if match:
|
||
return '/tmp' + match.group(1)
|
||
return ''
|
||
|
||
vocals_path = url_to_path(vocals_url)
|
||
no_vocals_path = url_to_path(no_vocals_url)
|
||
|
||
if not vocals_path or not os.path.exists(vocals_path):
|
||
raise ValueError(f'Demucs failed: vocals not found at {vocals_path} (from {vocals_url})')
|
||
if not no_vocals_path or not os.path.exists(no_vocals_path):
|
||
raise ValueError(f'Demucs failed: accompaniment not found at {no_vocals_path} (from {no_vocals_url})')
|
||
|
||
return {
|
||
'vocals_path': vocals_path,
|
||
'no_vocals_path': no_vocals_path,
|
||
'audio_path': audio_path # 保留原唱路径
|
||
}
|
||
|
||
|
||
# ============================================================================
|
||
# 通用步骤：歌词校准 + 字幕渲染 + 合成
|
||
# ============================================================================
|
||
|
||
async def step_lyric_calibrating(pipeline, session):
|
||
"""歌词时间校准（ASR + LLM）"""
|
||
vocals_path = pipeline['artifacts'].get('vocals_path', '')
|
||
audio_path = pipeline['artifacts'].get('audio_path', '')
|
||
lyrics = pipeline['artifacts'].get('lyrics', pipeline.get('lyrics', ''))
|
||
|
||
# 优先使用分离后的纯人声，否则使用原音频
|
||
source_audio = vocals_path if vocals_path else audio_path
|
||
|
||
if not source_audio:
|
||
raise ValueError('No audio source for calibration')
|
||
if not lyrics:
|
||
raise ValueError('No lyrics for calibration')
|
||
|
||
# 读取音频文件
|
||
with open(source_audio, 'rb') as f:
|
||
audio_data = f.read()
|
||
|
||
audio_filename = os.path.basename(source_audio)
|
||
|
||
# 调用lyric_calibrate服务
|
||
result = await call_service_multipart(
|
||
session,
|
||
SERVICES['lyric_calibrate'],
|
||
audio_data,
|
||
audio_filename,
|
||
lyrics,
|
||
timeout=600
|
||
)
|
||
|
||
if result.get('status') != 'ok':
|
||
error_msg = result.get('error', 'Unknown error')
|
||
raise ValueError(f'Lyric calibration failed: {error_msg}')
|
||
|
||
# 提取校准结果
|
||
calibrated_lines = result.get('calibrated_lines', 0)
|
||
ass_file_url = result.get('ass_file', '')
|
||
json_data_url = result.get('json_data', '')
|
||
|
||
# 下载并解析校准后的JSON
|
||
calibrated_json = None
|
||
if json_data_url:
|
||
# json_data_url 格式: '/idfile?path=lyric_calibrate/123456/calibrated.json'
|
||
import re
|
||
match = re.search(r'path=([^\s&]+)', json_data_url)
|
||
if match:
|
||
json_path = '/tmp/' + match.group(1)
|
||
if os.path.exists(json_path):
|
||
with open(json_path, 'r', encoding='utf-8') as f:
|
||
calibrated_json = json.load(f)
|
||
|
||
return {
|
||
'calibrated_subs': calibrated_json,
|
||
'calibrated_ass_file': ass_file_url,
|
||
'calibrated_lines': calibrated_lines,
|
||
}
|
||
|
||
|
||
async def step_subtitle_rendering(pipeline, session):
|
||
"""渲染ASS字幕（使用build_ass.py的5-style方案）"""
|
||
calibrated = pipeline['artifacts'].get('calibrated_subs', [])
|
||
if isinstance(calibrated, dict):
|
||
segments = calibrated.get('segments', [])
|
||
elif isinstance(calibrated, list):
|
||
segments = calibrated
|
||
else:
|
||
segments = []
|
||
|
||
if not segments:
|
||
raise ValueError('No calibrated segments for subtitle rendering')
|
||
|
||
pipeline_dir = os.path.join(WORK_DIR, pipeline['id'])
|
||
ass_path = os.path.join(pipeline_dir, 'karaoke.ass')
|
||
|
||
# 获取视频时长
|
||
video_duration = pipeline['artifacts'].get('audio_duration', 0)
|
||
if not video_duration:
|
||
# 估算：最后一段歌词结束时间 + 10s
|
||
last_end = max(seg.get('end', 0) for seg in segments)
|
||
video_duration = last_end + 10
|
||
|
||
# 获取标题和词曲信息
|
||
title = pipeline.get('title', '未知歌曲')
|
||
lyricist = pipeline.get('lyricist', '未知')
|
||
composer = pipeline.get('composer', '未知')
|
||
credit_text = f"词：{lyricist} 曲：{composer}"
|
||
|
||
def sec2ass(s):
|
||
h = int(s // 3600)
|
||
m = int((s % 3600) // 60)
|
||
sec = s % 60
|
||
return f"{h}:{m:02d}:{sec:05.2f}"
|
||
|
||
# ASS Header（5个样式）
|
||
header = """[Script Info]
|
||
Title: KTV Lyrics (Calibrated)
|
||
ScriptType: v4.00+
|
||
PlayResX: 1920
|
||
PlayResY: 1080
|
||
WrapStyle: 0
|
||
|
||
[V4+ Styles]
|
||
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
|
||
Style: Title,WenQuanYi Zen Hei,140,&H00FFFFFF,&H0000FFFF,&H00000000,&H80000000,-1,0,0,0,100,100,2,0,1,4,2,5,30,30,30,1
|
||
Style: Credit,WenQuanYi Zen Hei,56,&H00CCCCCC,&H0000FFFF,&H00000000,&H80000000,0,0,0,0,100,100,1,0,1,3,1,5,30,30,40,1
|
||
Style: TitleSmall,WenQuanYi Zen Hei,42,&H00DDDDDD,&H0000FFFF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,3,1,7,30,30,60,1
|
||
Style: CreditSmall,WenQuanYi Zen Hei,36,&H00888888,&H0000FFFF,&H00000000,&H80000000,0,0,0,0,100,100,0,0,1,2,1,7,30,30,30,1
|
||
Style: Karaoke,WenQuanYi Zen Hei,80,&H00FFFFFF,&H0000FFFF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,4,2,2,30,30,60,1
|
||
|
||
[Events]
|
||
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
|
||
"""
|
||
|
||
events = []
|
||
|
||
# 标题：0:00 - 0:15 居中（15秒）
|
||
events.append(f"Dialogue: 0,{sec2ass(0)},{sec2ass(15)},Title,,0,0,0,,{{\\fad(1500,1500)}}{title}")
|
||
|
||
# 词曲：0:05 - 0:15 居中（10秒）
|
||
events.append(f"Dialogue: 0,{sec2ass(5)},{sec2ass(15)},Credit,,0,0,0,,{{\\fad(1000,1000)}}{credit_text}")
|
||
|
||
# 标题小字：0:15 - 结束，左上角常驻
|
||
events.append(f"Dialogue: 0,{sec2ass(15)},{sec2ass(video_duration)},TitleSmall,,0,0,0,,{title}")
|
||
|
||
# 词曲小字：0:15 - 结束，左上角常驻（在标题下方）
|
||
events.append(f"Dialogue: 0,{sec2ass(15)},{sec2ass(video_duration)},CreditSmall,,0,0,0,,{credit_text}")
|
||
|
||
# 卡拉OK歌词：逐字高亮
|
||
for line_data in segments:
|
||
line = line_data.get('line', line_data.get('text', ''))
|
||
start = line_data.get('start', 0)
|
||
end = line_data.get('end', 0)
|
||
chars = line_data.get('chars', [])
|
||
|
||
if start < 0 or end <= start:
|
||
continue
|
||
|
||
# 构建 \kf 卡拉OK计时（每字厘秒数）
|
||
parts = []
|
||
for ch in chars:
|
||
cs = ch.get('start', 0)
|
||
ce = ch.get('end', cs + 0.1)
|
||
duration_cs = int(max((ce - cs) * 100, 1)) # 厘秒
|
||
parts.append(f"{{\\kf{duration_cs}}}{ch.get('char', '')}")
|
||
|
||
if parts:
|
||
text = "".join(parts)
|
||
events.append(f"Dialogue: 0,{sec2ass(start)},{sec2ass(end)},Karaoke,,0,0,0,,{text}")
|
||
|
||
ass_content = header + "\n".join(events) + "\n"
|
||
|
||
with open(ass_path, 'w', encoding='utf-8') as f:
|
||
f.write(ass_content)
|
||
|
||
return {'subtitle_path': ass_path}
|
||
|
||
|
||
async def step_ktv_synthesizing(pipeline, session):
|
||
"""合成KTV视频（双轨音频 + 单轨字幕视频）"""
|
||
pipeline_dir = os.path.join(WORK_DIR, pipeline['id'])
|
||
|
||
# 获取视频源（如果有）
|
||
video_path = pipeline['artifacts'].get('video_path', '')
|
||
vocals_path = pipeline['artifacts'].get('vocals_path', '')
|
||
no_vocals_path = pipeline['artifacts'].get('no_vocals_path', '')
|
||
audio_path = pipeline['artifacts'].get('audio_path', '')
|
||
subtitle_path = pipeline['artifacts'].get('subtitle_path', '')
|
||
|
||
if not subtitle_path:
|
||
raise ValueError('Missing subtitle_path')
|
||
|
||
# 输出文件
|
||
ktv_output = os.path.join(pipeline_dir, 'ktv_dual_track.mp4')
|
||
single_output = os.path.join(pipeline_dir, 'ktv_single_track.mp4')
|
||
|
||
# 确定音频源
|
||
original_audio = audio_path # 原唱（完整混合）
|
||
accompaniment = no_vocals_path # 伴奏
|
||
|
||
if not os.path.exists(original_audio):
|
||
raise ValueError(f'Original audio not found: {original_audio}')
|
||
if accompaniment and not os.path.exists(accompaniment):
|
||
raise ValueError(f'Accompaniment not found: {accompaniment}')
|
||
|
||
# ========================================================================
|
||
# 1. 合成KTV双轨视频（伴奏默认 + 原唱可切换）
|
||
# ========================================================================
|
||
if video_path and os.path.exists(video_path):
|
||
# 有视频：视频 + 伴奏（默认） + 原唱
|
||
cmd_ktv = [
|
||
'ffmpeg', '-y',
|
||
'-i', video_path,
|
||
'-i', accompaniment if accompaniment else original_audio,
|
||
'-i', original_audio,
|
||
'-vf', f'ass={subtitle_path}',
|
||
'-map', '0:v',
|
||
'-map', '1:a',
|
||
'-map', '2:a',
|
||
'-c:v', 'libx264', '-preset', 'medium', '-crf', '23',
|
||
'-c:a', 'aac', '-b:a', '192k',
|
||
'-metadata:s:a:0', 'handler_name=伴奏(Accompaniment)',
|
||
'-metadata:s:a:1', 'handler_name=原唱(Original)',
|
||
'-disposition:a:0', 'default',
|
||
'-disposition:a:1', '0',
|
||
ktv_output
|
||
]
|
||
else:
|
||
# 无视频：黑屏 + 伴奏 + 原唱
|
||
video_duration = pipeline['artifacts'].get('audio_duration', 0)
|
||
if not video_duration:
|
||
# 估算时长
|
||
calibrated = pipeline['artifacts'].get('calibrated_subs', [])
|
||
last_end = max(seg.get('end', 0) for seg in calibrated) if calibrated else 180
|
||
video_duration = last_end + 10
|
||
|
||
cmd_ktv = [
|
||
'ffmpeg', '-y',
|
||
'-f', 'lavfi', '-i', f'color=c=black:s=1920x1080:r=30:d={video_duration}',
|
||
'-i', accompaniment if accompaniment else original_audio,
|
||
'-i', original_audio,
|
||
'-vf', f'ass={subtitle_path}',
|
||
'-map', '0:v',
|
||
'-map', '1:a',
|
||
'-map', '2:a',
|
||
'-c:v', 'libx264', '-preset', 'medium', '-crf', '23',
|
||
'-c:a', 'aac', '-b:a', '192k',
|
||
'-shortest',
|
||
'-metadata:s:a:0', 'handler_name=伴奏(Accompaniment)',
|
||
'-metadata:s:a:1', 'handler_name=原唱(Original)',
|
||
'-disposition:a:0', 'default',
|
||
'-disposition:a:1', '0',
|
||
ktv_output
|
||
]
|
||
|
||
# 执行KTV合成
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd_ktv,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE
|
||
)
|
||
stdout, stderr = await proc.communicate()
|
||
|
||
if proc.returncode != 0:
|
||
raise ValueError(f'KTV synthesis failed: {stderr.decode()[:500]}')
|
||
|
||
# ========================================================================
|
||
# 2. 合成单轨字幕视频（原声混合 + 烧录字幕）
|
||
# ========================================================================
|
||
if video_path and os.path.exists(video_path):
|
||
# 有视频：视频 + 原唱音频
|
||
cmd_single = [
|
||
'ffmpeg', '-y',
|
||
'-i', video_path,
|
||
'-i', original_audio,
|
||
'-vf', f'ass={subtitle_path}',
|
||
'-map', '0:v',
|
||
'-map', '1:a',
|
||
'-c:v', 'libx264', '-preset', 'medium', '-crf', '23',
|
||
'-c:a', 'aac', '-b:a', '192k',
|
||
single_output
|
||
]
|
||
else:
|
||
# 无视频：黑屏 + 原唱音频
|
||
video_duration = pipeline['artifacts'].get('audio_duration', 0)
|
||
if not video_duration:
|
||
calibrated = pipeline['artifacts'].get('calibrated_subs', [])
|
||
last_end = max(seg.get('end', 0) for seg in calibrated) if calibrated else 180
|
||
video_duration = last_end + 10
|
||
|
||
cmd_single = [
|
||
'ffmpeg', '-y',
|
||
'-f', 'lavfi', '-i', f'color=c=black:s=1920x1080:r=30:d={video_duration}',
|
||
'-i', original_audio,
|
||
'-vf', f'ass={subtitle_path}',
|
||
'-map', '0:v',
|
||
'-map', '1:a',
|
||
'-c:v', 'libx264', '-preset', 'medium', '-crf', '23',
|
||
'-c:a', 'aac', '-b:a', '192k',
|
||
'-shortest',
|
||
single_output
|
||
]
|
||
|
||
# 执行单轨合成
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd_single,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE
|
||
)
|
||
stdout, stderr = await proc.communicate()
|
||
|
||
if proc.returncode != 0:
|
||
raise ValueError(f'Single track synthesis failed: {stderr.decode()[:500]}')
|
||
|
||
# 验证输出文件
|
||
if not os.path.exists(ktv_output):
|
||
raise ValueError(f'KTV output not created: {ktv_output}')
|
||
if not os.path.exists(single_output):
|
||
raise ValueError(f'Single output not created: {single_output}')
|
||
|
||
# 获取文件大小
|
||
ktv_size = os.path.getsize(ktv_output) / (1024 * 1024) # MB
|
||
single_size = os.path.getsize(single_output) / (1024 * 1024) # MB
|
||
|
||
return {
|
||
'ktv_dual_track_path': ktv_output,
|
||
'ktv_single_track_path': single_output,
|
||
'ktv_size_mb': round(ktv_size, 2),
|
||
'single_size_mb': round(single_size, 2),
|
||
}
|
||
|
||
|
||
# ============================================================================
|
||
# 状态机定义（按模式分支）
|
||
# ============================================================================
|
||
|
||
TRANSITIONS_MODE_A = {
|
||
'submitted': ('audio_preparing', None),
|
||
'audio_preparing': ('demucs_separating', step_audio_preparing),
|
||
'demucs_separating': ('lyric_calibrating', step_demucs_separate),
|
||
'lyric_calibrating': ('subtitle_rendering', step_lyric_calibrating),
|
||
'subtitle_rendering': ('ktv_synthesizing', step_subtitle_rendering),
|
||
'ktv_synthesizing': ('completed', step_ktv_synthesizing),
|
||
}
|
||
|
||
TRANSITIONS_MODE_B = {
|
||
'submitted': ('video_preparing', None),
|
||
'video_preparing': ('demucs_separating', step_video_preparing),
|
||
'demucs_separating': ('lyric_calibrating', step_demucs_separate),
|
||
'lyric_calibrating': ('subtitle_rendering', step_lyric_calibrating),
|
||
'subtitle_rendering': ('ktv_synthesizing', step_subtitle_rendering),
|
||
'ktv_synthesizing': ('completed', step_ktv_synthesizing),
|
||
}
|
||
|
||
TRANSITIONS_MODE_C = {
|
||
'submitted': ('lyric_generating', None),
|
||
'lyric_generating': ('lyric_evaluating', step_lyric_generate),
|
||
'lyric_evaluating': ('music_generating', step_lyric_evaluate), # 或循环回 lyric_generating
|
||
'music_generating': ('music_polling', step_music_generate),
|
||
'music_polling': ('demucs_separating', step_music_poll),
|
||
'demucs_separating': ('lyric_calibrating', step_demucs_separate),
|
||
'lyric_calibrating': ('subtitle_rendering', step_lyric_calibrating),
|
||
'subtitle_rendering': ('ktv_synthesizing', step_subtitle_rendering),
|
||
'ktv_synthesizing': ('completed', step_ktv_synthesizing),
|
||
}
|
||
|
||
|
||
def get_transitions(mode):
|
||
"""根据模式获取状态转换表"""
|
||
if mode == 'audio_lyrics':
|
||
return TRANSITIONS_MODE_A
|
||
elif mode == 'video_lyrics':
|
||
return TRANSITIONS_MODE_B
|
||
else: # lyrics_only (default)
|
||
return TRANSITIONS_MODE_C
|
||
|
||
|
||
async def run_pipeline(pipeline_id):
|
||
"""执行KTV Pipeline"""
|
||
import aioredis
|
||
redis = await aioredis.from_url('redis://127.0.0.1:6379', db=1)
|
||
|
||
try:
|
||
data = await redis.get(f'pipeline:{pipeline_id}')
|
||
if not data:
|
||
raise ValueError(f'Pipeline {pipeline_id} not found')
|
||
|
||
pipeline = json.loads(data)
|
||
mode = pipeline.get('mode', 'lyrics_only')
|
||
transitions = get_transitions(mode)
|
||
|
||
pipeline_dir = os.path.join(WORK_DIR, pipeline_id)
|
||
os.makedirs(pipeline_dir, exist_ok=True)
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
state = pipeline.get('state', 'submitted')
|
||
retry_counts = {}
|
||
|
||
while state != 'completed' and state != 'failed':
|
||
await update_state(redis, pipeline_id, state)
|
||
|
||
transition = transitions.get(state)
|
||
if not transition:
|
||
await update_state(redis, pipeline_id, 'failed', error=f'No transition for state: {state}')
|
||
break
|
||
|
||
next_state, handler = transition
|
||
|
||
if handler:
|
||
try:
|
||
result = await handler(pipeline, session)
|
||
pipeline['artifacts'].update(result)
|
||
|
||
# 歌词评估：阈值检查
|
||
if state == 'lyric_evaluating':
|
||
score = result.get('lyric_score', {}).get('total_score', 0)
|
||
retry_counts['lyric'] = retry_counts.get('lyric', 0) + 1
|
||
if score < pipeline.get('lyric_threshold', 8.5) and retry_counts['lyric'] < 3:
|
||
next_state = 'lyric_generating' # 循环回生成
|
||
else:
|
||
next_state = 'music_generating' # 通过，继续
|
||
|
||
except Exception as e:
|
||
await update_state(redis, pipeline_id, state, error=str(e))
|
||
# 重试机制
|
||
retry_counts[state] = retry_counts.get(state, 0) + 1
|
||
if retry_counts[state] > 2:
|
||
await update_state(redis, pipeline_id, 'failed', error=f'{state} failed after retries: {e}')
|
||
break
|
||
await asyncio.sleep(5)
|
||
continue
|
||
|
||
state = next_state
|
||
|
||
finally:
|
||
await redis.close()
|