app.py

import os
import subprocess
import gradio as gr
import requests
import py7zr
import stat

inited = False
app_dir = ''

language_map = {
        "自动检测": "-",
        "中文": "zh",
        "英文": "en",
        "日文": "ja",
        "韩文": "ko",
        "泰文": "th",
        "马来西亚": "ms",
        "越南": "vi",
        "印尼": "id"
    }

def initialize() :
    global inited,app_dir
    if(inited): 
        return
    # 克隆GitHub仓库
    token = os.environ['github_token']
    repo_url = f"https://{token}@github.com/chacha-ai-lab/AutoVideoTrans.git"

    print(repo_url)
    repo_dir = "videotrans"
    if not os.path.exists(repo_dir):
        subprocess.run(["git", "clone", repo_url, repo_dir])
    
    # 切换到仓库目录
    os.chdir(repo_dir)
    # 记录一下 cwd
    app_dir = os.getcwd()

    downloadModel()

    inited = True
    
def downloadModel():
    download_uvr()
    # download_whisper()

def download_uvr():
    url = "https://github.com/jianchang512/stt/releases/download/0.0/uvr5-model.7z"
    local_filename = "uvr5-model.7z"
    if not os.path.exists(local_filename):
        # 下载文件
        print(f"Downloading the file {local_filename} ...")
        with requests.get(url, stream=True) as r:
            r.raise_for_status()
            with open(local_filename, 'wb') as f:
                for chunk in r.iter_content(chunk_size=8192): 
                    f.write(chunk)
        print("Download complete.")

    # 解压缩文件
    print(f"Extracting the file {local_filename} ...")
    with py7zr.SevenZipFile(local_filename, mode='r') as archive:
        archive.extractall()  
    print("Extraction complete.")

def download_whisper():
    url = "https://huggingface.co/spaces/mortimerme/s4/resolve/main/faster-large-v1.7z?download=true"
    local_filename = "faster-large-v1.7z"
    # 下载文件
    print(f"Downloading the file {local_filename} ...")
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        with open(local_filename, 'wb') as f:
            for chunk in r.iter_content(chunk_size=8192): 
                f.write(chunk)
    print("Download complete.")

    # 解压缩文件
    print(f"Extracting the file {local_filename} ...")
    with py7zr.SevenZipFile(local_filename, mode='r') as archive:
        archive.extractall(path='models')  
    print("Extraction complete.")

# ‘’‘
# 参数解释：
# file_obj 文件路径
# source_language 原始视频的语言
# target_language 希望翻译成的语言
# role_number  视频中说话人的数量，默认是单人，多人模式会走另外的逻辑
# ’‘’
def process_file(file_obj, source_language=None, target_language=None, role_number=1):
    """
    接收文件路径作为输入，并返回视频文件的路径。
    """

    if file_obj is None:
        print("Error: 文件还在上传中")
        return "请等待文件上传完成"

    # 获取文件的临时路径
    file_path = file_obj.name
    # 确保路径正确地转换为字符串，尽管在大多数情况下这一步可能是多余的
    file_path_str = os.fspath(file_path)
    print(file_path_str)

    # 重置 cwd
    os.chdir(app_dir)
    
    print(f"当前进程ID: {os.getpid()}")
    print('cwd:'+os.getcwd())

    # 动态设定classpath
    import sys
    sys.path.append(os.getcwd())

    import cli

    cfg_file = os.path.join(os.getcwd(), 'cli.ini')
    print('cfg_file:'+cfg_file)

    multi_speaker = role_number > 1
    enableCuda = False
    result = cli.process(file_path_str, cfg_file, source_language, target_language, None, multi_speaker, enableCuda)

    # result = subprocess.run(["python", "cli.py", "-m", file_path_str])
    
    target_mp4 = None
    if isinstance(result, tuple) and len(result) == 2:
        return_message, target_mp4 = result
        print(f"处理结果：{return_message}")
        print(f"输出文件：{target_mp4}")
    else:
        return_message = result
        target_mp4 = None
        print(f"处理结果：{return_message}")

    return target_mp4

def cancel_task():
    # 暴露给外面的接口函数，取消任务
    import cli
    cli.cancel_task()
    print("cancel_task")
    
initialize()

if __name__ == "__main__":
    # 创建Gradio接口
    # iface = gr.Interface(fn=predict, inputs="text", outputs="json", title="demo", description="输入一点什么")
    # iface.launch()

    # 创建一个列表，包含显示值和实际值的元组
    language_choices = list(language_map.items())
        
    with gr.Blocks() as demo:
        gr.Markdown("## 视频翻译工具")

        with gr.Row():
            file_input = gr.File(label="上传视频文件")  # 上传视频按钮
        
        with gr.Row():
            source_language_input = gr.Dropdown(
                choices=language_choices[:3],  # 只使用前三个选项
                label="源语言",
                value="zh"  # 设置默认值为 "zh"（中文的实际值）
            )
            target_language_input = gr.Dropdown(
                choices=language_choices[1:],  # 排除第一个选项（自动检测）
                label="目标语言",
                value="en"  # 设置默认值为 "en"（英文的实际值）
            )
            role_number_input = gr.Slider(1, 5, value=1, step=1, label="说话人数")

        with gr.Row():
            with gr.Column(scale=100):  # 左侧原视频
                original_video = gr.Video(label="原视频", elem_id="original-video")

            with gr.Column(scale=1, elem_id="button-column"):
                submit_button = gr.Button(value="一键翻译")  # 翻译按钮
                video_path_input = gr.Textbox(label="视频文件路径", placeholder="请输入视频文件路径")  # 新增的文本输入框
                read_video_button = gr.Button(value="检查文件是否存在")  
                # 新增一个取消按钮
                cancel_button = gr.Button(value="取消任务")  # 取消按钮
                

            with gr.Column(scale=100):  # 右侧处理后的视频
                output_video = gr.Video(label="处理后的视频", elem_id="output-video")

        # 当点击“读取视频”按钮时，调用 read_video 函数
        def read_video(video_path):
            if video_path:
                exists = os.path.exists(video_path)
                print(f"文件存在？{exists}")
                if exists:
                    file_size = os.path.getsize(video_path)
                    file_info = f"文件大小为: {file_size} 字节"
                    print(file_info)

            else:
                return "请输入有效的视频文件路径"
            return video_path
    
        read_video_button.click(read_video, inputs=video_path_input, outputs=output_video)
        cancel_button.click(cancel_task, None, None)

        # 设置交互逻辑
        file_input.change(lambda file: file, inputs=file_input, outputs=original_video)

        # def inner_process_file(file_obj, source_language, target_language, role_number):
        #     print(f"file_obj: {file_obj}")
        #     print(f"source_language: {language_map[source_language]}")
        #     print(f"target_language: {language_map[target_language]}")
        #     print(f"role_number: {role_number}")
        #     return process_file(file_obj, language_map[source_language], language_map[target_language], role_number)

        submit_button.click(process_file, inputs=[file_input, source_language_input, target_language_input, role_number_input], outputs=output_video)

    # 自定义CSS
    demo.css = """
        #original-video, #output-video {
            width: 100%;
            height: 400px;  # 设置视频组件的高度
        }
        #button-column {
            display: flex;
            justify-content: center;
            align-items: center;
        }
    """

    # 启动界面
    demo.launch(server_port=7860, server_name="0.0.0.0", show_api=True)