video_translation/run.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
VideoLingo 非图形界面版本
支持通过参数赋值的方式处理视频
"""

import os
import sys
import shutil
from pathlib import Path

# 设置路径
current_dir = os.path.dirname(os.path.abspath(__file__))
os.environ['PATH'] += os.pathsep + current_dir
sys.path.append(current_dir)

# 导入核心模块
from core.utils.config_utils import load_key, update_key
from core.utils.onekeycleanup import cleanup
from core.utils.delete_retry_dubbing import delete_dubbing_files
from translations.translations import translate as t
from core import (
    _2_asr, _3_1_split_nlp, _3_2_split_meaning,
    _4_1_summarize, _4_2_translate, _5_split_sub,
    _6_gen_sub, _7_sub_into_vid, _8_1_audio_task,
    _8_2_dub_chunks, _9_refer_audio, _10_gen_audio,
    _11_merge_audio, _12_dub_to_vid
)


class VideoLingoProcessor:
    """VideoLingo 视频处理器"""
    
    def __init__(self, input_path, output_dir="output"):
        """
        初始化处理器
        
        Args:
            input_path (str): 输入视频文件路径
            output_dir (str): 输出目录路径，默认为 "output"
        """
        self.input_path = Path(input_path)
        self.output_dir = Path(output_dir)
        
        # 验证输入文件
        if not self.input_path.exists():
            raise FileNotFoundError(f"输入文件不存在: {self.input_path}")
        
        # 创建输出目录
        self.output_dir.mkdir(exist_ok=True)
        
        # 设置输出文件路径
        self.sub_video = self.output_dir / "output_sub.mp4"
        self.dub_video = self.output_dir / "output_dub.mp4"
        
        print(f"📹 输入视频: {self.input_path}")
        print(f"📁 输出目录: {self.output_dir}")
    
    def setup_video_file(self):
        """设置视频文件到指定位置"""
        # 将输入视频复制到 "output" 目录
        output_dir = Path("output")
        output_dir.mkdir(exist_ok=True)
        
        target_video = output_dir / self.input_path.name
        if not target_video.exists():
            print(f"📋 复制视频文件到: {target_video}")
            shutil.copy2(self.input_path, target_video)
        
        return target_video
    
    def process_subtitles(self):
        """处理字幕生成"""
        print("\n🎬 开始字幕处理流程...")
        
        try:
            print("🎤 使用 Whisper 进行语音转录...")
            _2_asr.transcribe()
            
            print("✂️  拆分长句子...")
            _3_1_split_nlp.split_by_spacy()
            _3_2_split_meaning.split_sentences_by_meaning()
            
            print("📝 总结和翻译...")
            _4_1_summarize.get_summary()
            
            # 检查是否需要暂停编辑术语
            if load_key("pause_before_translate"):
                input("⚠️  暂停以便编辑术语。请前往 'output/log/terminology.json' 编辑术语表，然后按回车继续...")
            
            _4_2_translate.translate_all()
            
            print("⏱️  处理和对齐字幕...")
            _5_split_sub.split_for_sub_main()
            _6_gen_sub.align_timestamp_main()
            
            print("🎞️  将字幕合并到视频...")
            _7_sub_into_vid.merge_subtitles_to_video()
            
            print("✅ 字幕处理完成! 🎉")
            return True
            
        except Exception as e:
            print(f"❌ 字幕处理失败: {str(e)}")
            return False
    
    def process_dubbing(self):
        """处理配音生成"""
        print("\n🎙️  开始配音处理流程...")
        
        try:
            print("📋 生成音频任务...")
            _8_1_audio_task.gen_audio_task_main()
            _8_2_dub_chunks.gen_dub_chunks()
            
            print("🎵 提取参考音频...")
            _9_refer_audio.extract_refer_audio_main()
            
            print("🔊 生成所有音频...")
            _10_gen_audio.gen_audio()
            
            print("🎶 合并完整音频...")
            _11_merge_audio.merge_full_audio()
            
            print("🎬 将配音合并到视频...")
            _12_dub_to_vid.merge_video_audio()
            
            print("✅ 配音处理完成! 🎇")
            return True
            
        except Exception as e:
            print(f"❌ 配音处理失败: {str(e)}")
            return False
    
    def process_all(self, include_dubbing=True):
        """
        执行完整的处理流程
        
        Args:
            include_dubbing (bool): 是否包含配音处理，默认为 True
        """
        print("🚀 开始 VideoLingo 处理流程")
        print("=" * 50)
        
        # 设置视频文件
        self.setup_video_file()
        
        # 处理字幕
        subtitle_success = self.process_subtitles()
        if not subtitle_success:
            print("❌ 字幕处理失败，停止执行")
            return False
        
        # 如果字幕成功且需要配音，则处理配音
        if include_dubbing:
            dubbing_success = self.process_dubbing()
            if not dubbing_success:
                print("❌ 配音处理失败")
                return False
        
        print("\n🎊 所有处理完成!")
        print(f"📁 输出文件位于: {self.output_dir.absolute()}")
        
        # 显示输出文件
        if self.sub_video.exists():
            print(f"  🎬 字幕视频: {self.sub_video}")
        if include_dubbing and self.dub_video.exists():
            print(f"  🎙️  配音视频: {self.dub_video}")
        
        return True
    
    def cleanup_files(self):
        """清理临时文件"""
        print("🧹 清理临时文件...")
        cleanup()
        print("✅ 清理完成")
    
    def delete_dubbing_files(self):
        """删除配音相关文件"""
        print("🗑️  删除配音文件...")
        delete_dubbing_files()
        print("✅ 配音文件删除完成")


def main():
    """主函数 - 示例用法"""
    
    # ==================== 配置参数 ====================
    # 请在这里修改您的参数
    
    # 输入视频路径（必需）
    INPUT_VIDEO_PATH = "/root/autodl-tmp/output/Dajing_Gate/video_gen_20250721_145356/videos/composed_video_20250721_145724.mp4"
    
    # 输出目录（可选，默认为 "output"）
    OUTPUT_DIR = "output"
    
    # 处理选项
    INCLUDE_DUBBING = True  # 是否包含配音处理
    
    # ==================== 可选配置覆盖 ====================
    # 您可以在这里覆盖 config.yaml 中的设置
    
    # 配置中文转英文翻译
    update_key("whisper.language", "zh")  # 原始语言为中文
    update_key("target_language", "English")  # 目标语言为英文
    update_key("tts_method", "edge_tts")  # 使用 Edge TTS
    update_key("burn_subtitles", True)  # 烧录字幕到视频
    update_key("reflect_translate", True)  # 启用 Expressiveness 阶段进行翻译润色
    
    # ===================================================
    
    try:
        # 创建处理器实例
        processor = VideoLingoProcessor(
            input_path=INPUT_VIDEO_PATH,
            output_dir=OUTPUT_DIR
        )
        
        # 执行处理
        success = processor.process_all(include_dubbing=INCLUDE_DUBBING)
        
        if success:
            print("\n🎉 处理成功完成!")
        else:
            print("\n❌ 处理过程中出现错误")
            
    except FileNotFoundError as e:
        print(f"❌ 文件错误: {e}")
        print("请检查输入视频路径是否正确")
    except Exception as e:
        print(f"❌ 处理失败: {e}")


if __name__ == "__main__":
    main()