import argparse import gradio as gr import json from textwrap import dedent from task_plan import task_plan_list import subprocess from datetime import datetime def read_specific_line(file_path, line_number): try: with open(file_path, 'r') as file: for current_line, line in enumerate(file, start=1): if current_line == line_number: return line.strip() return None # 如果文件不足指定的行数，返回None except FileNotFoundError: return None # 如果文件不存在，返回None def get_folder_paths(file_path): source_folder = read_specific_line(file_path, 2) target_folder = read_specific_line(file_path, 3) if source_folder is None or target_folder is None: return "错误信息：文件不存在。", None, None elif source_folder == "" or target_folder == "": return "错误信息：文件中的第2行或第3行为空。", None, None else: return None, source_folder, target_folder def read_specific_line(file_path, line_number): with open(file_path, "r") as file: for i, line in enumerate(file, 1): if i == line_number: return line.strip() def repo_pull(url): # 从远程仓库拉取代码的操作 subprocess.check_output(["python3", "repo_pull.py", "--repo_url", str(url).replace(".git", "").replace("git", "").replace("clone", "").replace(" ", "")]) yield # 将代码推送到远程仓库的操作 def repo_push(hf_repo_id, tmp1, tmp2, tmp3, tmp4, select_task_plan): task_plan_index = list(task_plan_list.keys()).index(select_task_plan) if task_plan_index >= 0 and task_plan_index <= 5: if task_plan_index == 0: # 克隆目标仓库到本地 subprocess.check_output(["python3", "repo_pull.py", "--repo_url", str(hf_repo_id).replace(".git", "").replace("git", "").replace("clone", "").replace(" ", "")]) # # 打开文件并读取第2行数据 file_path = ".folderinfo" # # 读取指定行数的数据 # source_folder = read_specific_line(file_path, 1) # target_folder = read_specific_line(file_path, 2) error, source_folder, target_folder = get_folder_paths(file_path) if error: print(error) # 这将打印出错误信息 else: print(f"源文件夹: {source_folder}") print(f"目标文件夹: {target_folder}") # 获取当前日期和时间 current_datetime = datetime.now() # 格式化日期和时间 formatted_datetime = current_datetime.strftime("%Y-%m-%d %H:%M:%S") commit_message = f"魏然 首次上传于 {formatted_datetime}" # subprocess.check_output( # f"cd {target_folder} && cp -rf ../{str(source_folder).replace('./', '')}/* . && git add -A . && git commit -m \"{commit_message}\"", # shell=True) elif task_plan_index == 1: print("选中了 1") elif task_plan_index == 2: print("选中了 2") elif task_plan_index == 3: print("选中了 3") elif task_plan_index == 4: print("选中了 4") else: print("选中了 5") else: print("task_plan_index 不在范围内") yield # 更新进度 def load_set(wisemodel_link, hf_repo_id, tmp1, tmp2, tmp3, tmp4, select_task_plan, progress=gr.Progress()): tasks = [repo_pull(wisemodel_link), repo_push(hf_repo_id, tmp1, tmp2, tmp3, tmp4, select_task_plan)] # 将 repo_pull 和 repo_push 添加到任务列表中 total_tasks = len(tasks) for i, task in enumerate(tasks, start=1): progress(i / total_tasks, desc=f"{i} / {total_tasks} ") for _ in task: pass return "下载完成！" # 创建解析命令行参数的函数 def parse_args(): parser = argparse.ArgumentParser() parser.add_argument("--language", type=str, default="zh-CN", help="Language option, e.g., 'en-US' or 'zh-CN' or ja-JP or ko-KR or th-TH support now!") return parser.parse_args() # 加载语言对应的JSON文件 def load_language_json(language): filename = f"locales/{language}.json" try: with open(filename, "r", encoding="utf-8") as file: return json.load(file) except FileNotFoundError: print(f"Error: File '{filename}' not found.") return {} # 加载JSON以支持国际化 args = parse_args() language_json = load_language_json(args.language) # # 获取对应语言的文本信息 info = language_json.get("info", {}) _title = info.get("page_title", {}) _about = info.get("about", {}) _wisemodel_lable_title = info.get("wisemodel_lable_title", {}) _hf_repo_id_title = info.get("hf_repo_id_title", {}) _button = info.get("button", {}) _clear = info.get("clear", {}) _setting = info.get("setting", {}) _HuggingFace_Token_title = info.get("HuggingFace_Token_title", {}) _HuggingFace_Token_placeholder = info.get("HuggingFace_Token_placeholder", {}) _WiseModel_Token_title = info.get("WiseModel_Token_title", {}) _WiseModel_Token_placeholder = info.get("WiseModel_Token_placeholder", {}) _Logs = info.get("Logs", {}) _task_plan = info.get("task_plan", {}) _source_user_id = info.get("source_user_id", {}) _target_user_id = info.get("target_user_id", {}) _source_user_placeholder = info.get("source_user_placeholder", {}) _target_user_placeholder = info.get("target_user_placeholder", {}) # with gr.Blocks(title=_title, theme=gr.themes.Soft(text_size="sm")) as demo: with gr.Accordion("📒 " + _about, open=False): _ = f""" 这个空间适用于将 WiseModel 仓库同步转移到 Hugging Face 仓库。\n **请首先确保你是仓库的所有者或者拥有者同意你这样做！**\n # 如何使用这个空间?\n - 复制这个空间并提供 WiseModel token（可选）和你的 读/写 HF token（必填）\n - 在Hugging Face上创建你的目标模型仓库。这一步需要手动完成，空间不会为你创建一个空的仓库。\n - 在你自己的私有空间中填写以下信息。\n - 点击提交按钮，然后观察日志中的输出进展。\n - 重新创建修改自己的 README.md 文件（因为元数据与 HF 不兼容） """ gr.Markdown(dedent(_)) wisemodel_link = gr.Textbox(label=_wisemodel_lable_title) hf_repo_id = gr.Textbox( label=_hf_repo_id_title) with gr.Accordion("🔧 " + _setting, open=False): with gr.Column(scale=12): select_task_plan = gr.Dropdown(choices=list(task_plan_list.keys()), allow_custom_value=True, value="", label=_task_plan) with gr.Row(): tmp1 = gr.Textbox( label=_HuggingFace_Token_title, placeholder=_HuggingFace_Token_placeholder, type="password", # 设置为密码类型，不显示明文 ) tmp2 = gr.Textbox( label=_WiseModel_Token_title, placeholder=_WiseModel_Token_placeholder, type="password", # 设置为密码类型，不显示明文 ) with gr.Row(): tmp3 = gr.Textbox( label=_source_user_id, placeholder=_source_user_placeholder, ) tmp4 = gr.Textbox( label=_target_user_id, placeholder=_target_user_placeholder, ) with gr.Row(): with gr.Column(scale=100): button = gr.Button(_button, variant="primary") with gr.Column(scale=1): clear = gr.Button(_clear) output = gr.Textbox(label=_Logs) button.click(load_set, [wisemodel_link, hf_repo_id, tmp1, tmp2, tmp3, tmp4, select_task_plan], [output]) if __name__ == "__main__": demo.launch(show_error=True, share=True, server_name="0.0.0.0", server_port=7860, debug=True)