video_translation/run.py

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

import os
import sys
import shutil
import time
from datetime import datetime, timedelta
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_split_semantic, _4_1_summarize, _4_2_translate,
    _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"
        
        # 初始化计时器
        self.start_time = None
        self.step_times = {}
        self.total_time = 0
        
        print(f"📹 输入视频: {self.input_path}")
        print(f"📁 输出目录: {self.output_dir}")
    
    @staticmethod
    def format_time(seconds):
        """
        格式化时间显示
        
        Args:
            seconds (float): 秒数
        
        Returns:
            str: 格式化的时间字符串
        """
        if seconds < 60:
            return f"{seconds:.2f}秒"
        elif seconds < 3600:
            minutes = int(seconds // 60)
            secs = seconds % 60
            return f"{minutes}分{secs:.2f}秒"
        else:
            hours = int(seconds // 3600)
            minutes = int((seconds % 3600) // 60)
            secs = seconds % 60
            return f"{hours}小时{minutes}分{secs:.2f}秒"
    
    def start_timer(self, step_name):
        """
        开始计时
        
        Args:
            step_name (str): 步骤名称
        """
        current_time = time.time()
        if self.start_time is None:
            self.start_time = current_time
        
        self.step_times[step_name] = {'start': current_time}
        print(f"⏱️  开始 {step_name} - {datetime.now().strftime('%H:%M:%S')}")
    
    def end_timer(self, step_name):
        """
        结束计时
        
        Args:
            step_name (str): 步骤名称
        """
        if step_name in self.step_times:
            end_time = time.time()
            self.step_times[step_name]['end'] = end_time
            duration = end_time - self.step_times[step_name]['start']
            self.step_times[step_name]['duration'] = duration
            print(f"✅ 完成 {step_name} - 耗时: {self.format_time(duration)}")
    
    def print_timing_summary(self):
        """打印计时总结"""
        if self.start_time is None:
            return
        
        self.total_time = time.time() - self.start_time
        
        print("\n" + "="*60)
        print("📊 处理时间统计")
        print("="*60)
        
        for step_name, times in self.step_times.items():
            if 'duration' in times:
                print(f"🔹 {step_name:<20} : {self.format_time(times['duration'])}")
        
        print("-"*60)
        print(f"🕐 总处理时间: {self.format_time(self.total_time)}")
        print(f"🕐 开始时间: {datetime.fromtimestamp(self.start_time).strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"🕐 结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("="*60)
    
    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_all(self):
        """处理完整的视频翻译和配音流程"""
        print("\n🎬 开始VideoLingo重构版处理流程...")
        
        try:
            # 设置视频文件
            self.start_timer("视频文件设置")
            self.setup_video_file()
            self.end_timer("视频文件设置")
            
            # 语音转录
            self.start_timer("Whisper语音转录")
            _2_asr.transcribe()
            self.end_timer("Whisper语音转录")
            
            # 语义切割
            self.start_timer("语义切割")
            _3_split_semantic.split_semantic()
            self.end_timer("语义切割")
            
            # 总结
            self.start_timer("内容总结")
            _4_1_summarize.get_summary()
            self.end_timer("内容总结")
            
            # 检查是否需要暂停编辑术语
            if load_key("pause_before_translate"):
                print("⚠️  暂停以便编辑术语。请前往 'output/log/terminology.json' 编辑术语表...")
                input("按回车继续...")
            
            # 翻译
            self.start_timer("文本翻译")
            _4_2_translate.translate_all()
            self.end_timer("文本翻译")
            
            # 音频任务生成
            self.start_timer("音频任务生成")
            _8_1_audio_task.gen_audio_task_main()
            self.end_timer("音频任务生成")
            
            # 生成配音片段
            self.start_timer("配音片段生成")
            _8_2_dub_chunks.gen_dub_chunks()
            self.end_timer("配音片段生成")
            
            # 提取参考音频
            self.start_timer("参考音频提取")
            _9_refer_audio.extract_refer_audio_main()
            self.end_timer("参考音频提取")
            
            # 生成音频
            self.start_timer("音频生成")
            _10_gen_audio.gen_audio()
            self.end_timer("音频生成")
            
            # 合并音频
            self.start_timer("音频合并")
            _11_merge_audio.merge_full_audio()
            self.end_timer("音频合并")
            
            # 字幕对齐
            self.start_timer("字幕时间对齐")
            _6_gen_sub.align_timestamp_main()
            self.end_timer("字幕时间对齐")
            
            # 字幕合并到视频
            self.start_timer("字幕合并到视频")
            _7_sub_into_vid.merge_subtitles_to_video()
            self.end_timer("字幕合并到视频")
            
            # 配音合并到视频
            self.start_timer("配音合并到视频")
            _12_dub_to_vid.merge_video_audio()
            self.end_timer("配音合并到视频")
            
            # 打印计时总结
            self.print_timing_summary()
            
            print("\n🎉 处理完成! 🎉")
            return True
            
        except Exception as e:
            print(f"❌ 处理失败: {str(e)}")
            # 即使失败也显示已完成步骤的计时信息
            self.print_timing_summary()
            return False
    

    
    def cleanup_files(self):
        """清理临时文件"""
        self.start_timer("清理临时文件")
        cleanup()
        self.end_timer("清理临时文件")
    
    def delete_dubbing_files(self):
        """删除配音相关文件"""
        self.start_timer("删除配音文件")
        delete_dubbing_files()
        self.end_timer("删除配音文件")


def main():
    """主函数 - 示例用法"""
    
    # ==================== 配置参数 ====================
    # 请在这里修改您的参数
    
    # 输入视频路径（必需）
    INPUT_VIDEO_PATH = "composed_video_20250729_183303_no_sub.mp4"
    
    # 输出目录（可选，默认为 "output"）
    OUTPUT_DIR = "output"
    
    # 重构版本已统一处理流程，不再需要分离字幕和配音处理
    
    # ==================== 可选配置覆盖 ====================
    # 您可以在这里覆盖 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 阶段进行翻译润色
    
    # ===================================================
    
    # 记录整体开始时间
    overall_start_time = time.time()
    print(f"🚀 VideoLingo 处理开始 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    try:
        # 创建处理器实例
        processor = VideoLingoProcessor(
            input_path=INPUT_VIDEO_PATH,
            output_dir=OUTPUT_DIR
        )
        
        # 执行处理
        success = processor.process_all()
        
        # 计算总体运行时间
        overall_end_time = time.time()
        overall_duration = overall_end_time - overall_start_time
        
        if success:
            print(f"\n🎉 VideoLingo 处理成功完成!")
            print(f"🎯 总体运行时间: {VideoLingoProcessor.format_time(overall_duration)}")
        else:
            print(f"\n❌ 处理过程中出现错误")
            print(f"⏱️  运行时间: {VideoLingoProcessor.format_time(overall_duration)}")
            
    except FileNotFoundError as e:
        overall_duration = time.time() - overall_start_time
        print(f"❌ 文件错误: {e}")
        print("请检查输入视频路径是否正确")
        print(f"⏱️  运行时间: {VideoLingoProcessor.format_time(overall_duration)}")
    except Exception as e:
        overall_duration = time.time() - overall_start_time
        print(f"❌ 处理失败: {e}")
        print(f"⏱️  运行时间: {VideoLingoProcessor.format_time(overall_duration)}")


if __name__ == "__main__":
    main()