# models_manager.py import json import os import sys import time import threading import traceback import tkinter as tk from tkinter import ttk, messagebox, filedialog from tkinter.scrolledtext import ScrolledText import urllib.request import urllib.error from pathlib import Path import subprocess import tempfile import pandas as pd from datasets import Dataset, Features, Image as HFImage, Value from huggingface_hub import HfApi, DatasetCard from PIL import Image import os import re for k in ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy", "ALL_PROXY", "all_proxy"]: os.environ.pop(k, None) # ================= 配置区 ================= SITE_CONFIG_FILE = "site_config.json" SPACE_ID = "waveshaker/loras_gallery" MODEL_REPO_ID = "waveshaker/loras" DATASET_REPO_ID = "waveshaker/dataset_collections" DATASET_BASE = "https://huggingface.co/datasets/waveshaker/dataset_collections/tree/main" OUTPUT_FILE = "models.json" TOKEN_FILE = "hf_token.txt" LOG_FILE = "models_manager.log" HTTP_TIMEOUT = 20 # ========================================== TOP_ORDER = {"krea2": 1, "anima": 2, "flux1_dev": 3, "pony": 4, "sd15": 5} SUB_ORDER = {"CN": 1, "KR": 2, "JP": 3, "Fictional": 4, "WE": 5} IMG_EXTS = {".jpg", ".jpeg", ".png", ".webp"} def custom_sort_key(file_path): parts = file_path.split('/') top_dir = parts[0] if len(parts) > 0 else "" sub_dir = parts[1] if len(parts) > 1 else "" return (TOP_ORDER.get(top_dir, 999), SUB_ORDER.get(sub_dir, 999), file_path) # ---------- 可选拖拽 ---------- try: from tkinterdnd2 import DND_FILES, TkinterDnD _HAS_DND = True except Exception: DND_FILES = None TkinterDnD = None _HAS_DND = False class ModelsManager(tk.Tk): def __init__(self): super().__init__() self.title("LoRA 仓库管理工具") self.geometry("1250x850") # 全局 Token self.token_var = tk.StringVar(value=self._load_token()) # 🌟 新增：保存 site_config 数据 self.site_config_data = {"models": [], "categories": []} # JSON 相关状态 self.items = {} self.remote_files = [] self.new_paths = set() self.stale_paths = set() self._search_job = None self._loading = False self.api = None # 🌟 Model 上传相关状态：自动获取当前脚本路径，并指向同级的 models 文件夹 if getattr(sys, 'frozen', False): current_dir = os.path.dirname(sys.executable) else: current_dir = os.path.dirname(os.path.abspath(__file__)) default_models_dir = os.path.join(current_dir, "models") if not os.path.exists(default_models_dir): os.makedirs(default_models_dir, exist_ok=True) self.local_folder_var = tk.StringVar(value=default_models_dir) self.remote_dir_var = tk.StringVar() # 全局 Token self.token_var = tk.StringVar(value=self._load_token()) # ============ 初始化 UI ============ self._build_global_ui() self._build_json_tab() self._build_model_tab() self._build_dataset_tab() # <--- 新增这一行 self._build_showcase_tab() # 🌟 加上这一行 self._refresh_api() self._setup_dnd() self.log_json("程序启动") # ========================================================== # ===================== 全局基础函数 ======================= # ========================================================== def log_json(self, msg): self._log_to_tab(msg, getattr(self, 'log_json_text', None)) def log_model(self, msg): self._log_to_tab(msg, getattr(self, 'log_model_text', None)) def _log_to_tab(self, msg, target_text_widget): if not target_text_widget: return ts = time.strftime("%H:%M:%S") line = f"[{ts}] {msg}" try: with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception: pass def _append(): try: target_text_widget.configure(state="normal") target_text_widget.insert("end", line + "\n") target_text_widget.see("end") target_text_widget.configure(state="disabled") except Exception: pass if threading.current_thread() is threading.main_thread(): _append() else: self.after(0, _append) def log_exc(self, prefix="", is_model=False): tb = traceback.format_exc() msg = f"{prefix}异常:\n{tb}" if is_model: self.log_model(msg) else: self.log_json(msg) def _load_token(self): env = os.environ.get("HF_TOKEN", "").strip() if env: return env try: with open(TOKEN_FILE, "r", encoding="utf-8") as f: return f.read().strip() except FileNotFoundError: return "" def _save_token(self, token): if token: with open(TOKEN_FILE, "w", encoding="utf-8") as f: f.write(token) else: try: os.remove(TOKEN_FILE) except FileNotFoundError: pass def _refresh_api(self): token = self.token_var.get().strip() self.api = HfApi(token=token) if token else HfApi() def _ensure_token(self): token = self.token_var.get().strip() if not token: messagebox.showwarning("需要 Token", "请先在上方填写 HF Token 再操作。") self.log_json("操作中止：未填写 Token") return None self._refresh_api() return token def _build_global_ui(self): # 1. 顶部 Token 行 token_row = ttk.Frame(self) token_row.pack(fill="x", padx=8, pady=(8, 2)) ttk.Label(token_row, text="HF Token:").pack(side="left") ttk.Entry(token_row, textvariable=self.token_var, width=60, show="*").pack(side="left", padx=6) ttk.Button(token_row, text="📥 保存 Token", command=self._on_save_token).pack(side="left") ttk.Button(token_row, text="❌ 清除", command=self._on_clear_token).pack(side="left", padx=4) ttk.Button(token_row, text="📄 打开日志", command=self.open_log_file).pack(side="right", padx=6) # 1. 恢复 Windows 默认主题（看起来最舒服） style = ttk.Style() # 如果你在 Windows 11 上，可以用 'vista' 或 'winnative'，如果想要更现代可以试试 'xpnative' if sys.platform.startswith("win"): style.theme_use('vista') # 2. 强制扁平化 Tab 的渲染结构（屏蔽掉系统的边框和虚线焦点） style.layout("TNotebook.Tab", [ ('Notebook.tab', {'sticky': 'nswe', 'children': [ ('Notebook.padding', {'side': 'top', 'sticky': 'nswe', 'children': [ ('Notebook.label', {'side': 'top', 'sticky': ''}) ]}) ]}) ]) # 3. 基础设置 style.configure("TNotebook.Tab", padding=[12, 6], font=("Microsoft YaHei", 9), borderwidth=0, background="#f0f0f0", foreground="#333333" ) # 4. 锁定所有状态，防止任何像素级跳动 style.map("TNotebook.Tab", padding=[("selected", [12, 6]), ("active", [12, 6]), ("!selected", [12, 6])], borderwidth=[("selected", 0), ("active", 0), ("!selected", 0)], # 焦点框设为空，去掉点击时的虚线框 focuscolor=[("selected", "none"), ("active", "none"), ("!selected", "none")], # 选中和悬停时的颜色 background=[("selected", "#e1f5fe"), ("active", "#f5f5f5")], foreground=[("selected", "#0277bd")] ) # 3. 创建 Notebook 和 Tab 页面 self.notebook = ttk.Notebook(self) self.notebook.pack(fill="both", expand=True, padx=8, pady=4) self.tab_json = ttk.Frame(self.notebook) self.tab_model = ttk.Frame(self.notebook) self.tab_dataset = ttk.Frame(self.notebook) # 4. 🌟 添加 Tab（Emoji + 文字） self.notebook.add(self.tab_json, text=" 🚀 Space - JSON 管理 ") self.notebook.add(self.tab_model, text=" 🧊 Models - 模型上传 ") self.notebook.add(self.tab_dataset, text=" 📚 Dataset - 数据集上传 ") self.tab_showcase = ttk.Frame(self.notebook) self.notebook.add(self.tab_showcase, text=" ✨ Showcase - 案例上传 ") def _on_save_token(self): self._save_token(self.token_var.get().strip()) self._refresh_api() self.log_json("Token 已保存") messagebox.showinfo("完成", "Token 已保存到本地") def _on_clear_token(self): self.token_var.set("") self._save_token("") self._refresh_api() self.log_json("Token 已清除") messagebox.showinfo("完成", "Token 已清除") def open_log_file(self): try: os.startfile(LOG_FILE) except Exception: try: import subprocess subprocess.Popen(["notepad", LOG_FILE]) except Exception as e: messagebox.showerror("错误", f"无法打开日志: {e}") def _setup_dnd(self): if not _HAS_DND: return try: TkinterDnD._require(self) self.tree.drop_target_register(DND_FILES) self.tree.dnd_bind("<>", self._on_drop_json) except Exception as e: self.log_json(f"DnD 初始化失败: {e}") # ========================================================== # ==================== 区域 1: JSON 管理 =================== # ========================================================== def _build_json_tab(self): parent = self.tab_json top = ttk.Frame(parent) top.pack(fill="x", padx=8, pady=4) self.btn_load_remote = ttk.Button(top, text="🔄 拉取 models.json", command=self.load_remote) self.btn_load_remote.pack(side="left") ttk.Button(top, text="📂 加载本地 models.json", command=self.load_local).pack(side="left", padx=6) ttk.Button(top, text="💾 保存 models.json 到本地", command=self.save_local).pack(side="left") ttk.Button(top, text="🢁 上传 models.json", command=self.upload_models_json).pack(side="left", padx=6) # 🌟 新增：配置同步按钮 ttk.Button(top, text="🌐 打开 Dataset Repo", compound="left", command=self.open_dataset_repo).pack(side="right", padx=6) ttk.Button(top, text="⚙️ 编辑 site_config", command=self.open_site_config_editor).pack(side="right", padx=12) filt = ttk.Frame(parent) filt.pack(fill="x", padx=8) ttk.Label(filt, text="🔍 搜索:").pack(side="left") self.search_var = tk.StringVar() self.search_var.trace_add("write", lambda *_: self._debounced_refresh()) ttk.Entry(filt, textvariable=self.search_var, width=30).pack(side="left", padx=4) self.only_new = tk.BooleanVar(value=False) ttk.Checkbutton(filt, text="只看新增", variable=self.only_new, command=self.refresh_tree).pack(side="left", padx=6) self.only_empty_trigger = tk.BooleanVar(value=False) ttk.Checkbutton(filt, text="只看 trigger 为空", variable=self.only_empty_trigger, command=self.refresh_tree).pack(side="left") body = ttk.Panedwindow(parent, orient="vertical") body.pack(fill="both", expand=True, padx=8, pady=6) tree_frame = ttk.Frame(body) cols = ("title", "trigger", "dataset", "showcase") # 🌟 新增 showcase 列 self.tree = ttk.Treeview(tree_frame, columns=cols, show="tree headings", height=20) self.tree.heading("#0", text="路径") self.tree.heading("title", text="title") self.tree.heading("trigger", text="trigger") self.tree.heading("dataset", text="dataset") self.tree.heading("showcase", text="showcase") # 🌟 新增 self.tree.column("#0", width=380, stretch=True) self.tree.column("title", width=180) self.tree.column("trigger", width=260) self.tree.column("dataset", width=240) self.tree.column("showcase", width=200) # 🌟 加入 showcase vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.tree.yview) self.tree.configure(yscrollcommand=vsb.set) self.tree.pack(side="left", fill="both", expand=True) vsb.pack(side="left", fill="y") self.tree.tag_configure("new", background="#d4f5d4") self.tree.tag_configure("empty_trigger", background="#fff3c4") self.tree.tag_configure("stale", foreground="#999999") self.tree.bind("", self.on_double_click) log_frame = ttk.LabelFrame(body, text="JSON 操作日志") self.log_json_text = ScrolledText(log_frame, height=10, state="disabled", wrap="word") self.log_json_text.pack(fill="both", expand=True) body.add(tree_frame, weight=3) body.add(log_frame, weight=1) def open_dataset_repo(self): import webbrowser webbrowser.open(f"https://huggingface.co/datasets/{DATASET_REPO_ID}/tree/main") def _debounced_refresh(self): if self._search_job: self.after_cancel(self._search_job) self._search_job = self.after(250, self.refresh_tree) def load_local(self): self.log_json(f"加载本地 {OUTPUT_FILE}") try: with open(OUTPUT_FILE, "r", encoding="utf-8") as f: data = json.load(f) except FileNotFoundError: messagebox.showwarning("提示", f"本地没有 {OUTPUT_FILE}") return except Exception as e: self.log_json(f"读取失败: {e}") return self.items = data.get("items", {}) if not self.items and "titles" in data: self.items = {k: {"title": v, "trigger": "", "dataset": ""} for k, v in data["titles"].items()} self.new_paths = set() self.stale_paths = set() self.refresh_tree() self.log_json(f"本地加载完成，{len(self.items)} 条") def load_remote(self): if self._loading: return self._loading = True self.btn_load_remote.configure(state="disabled") token = self.token_var.get().strip() self.log_json(f"开始拉取远端…") threading.Thread(target=self._load_remote_worker, args=(token,), daemon=True).start() def _load_remote_worker(self, token): try: api = HfApi(token=token) if token else HfApi() t0 = time.time() all_files = list(api.list_repo_files(repo_id=MODEL_REPO_ID, repo_type="model")) remote_files = [f for f in all_files if f.endswith(".safetensors")] self.log_json(f"步骤1: 远端文件 {len(all_files)}，safetensors {len(remote_files)}，{time.time()-t0:.2f}s") space_subdomain = SPACE_ID.replace("/", "-").replace("_", "-").lower() url = f"https://{space_subdomain}.static.hf.space/models.json" remote_data = None try: req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as response: remote_data = json.loads(response.read().decode()) self.log_json(f"步骤2: JSON 获取成功") except Exception as e: self.log_json(f"步骤2 失败: {type(e).__name__}: {e}") existing_items = {} if remote_data: existing_items = remote_data.get("items", {}) if not existing_items and "titles" in remote_data: existing_items = {k: {"title": v, "trigger": "", "dataset": ""} for k, v in remote_data["titles"].items()} merged, new_paths = {}, set() remote_set = set(remote_files) for path in remote_files: if path in existing_items: merged[path] = existing_items[path] else: merged[path] = {"title": "", "trigger": "", "dataset": ""} new_paths.add(path) stale_paths = set(existing_items.keys()) - remote_set for p in stale_paths: merged[p] = existing_items[p] items = {k: merged[k] for k in sorted(merged.keys(), key=custom_sort_key)} self.log_json(f"步骤3: 合并 {len(items)} 条，新增 {len(new_paths)}，失效 {len(stale_paths)}") self.after(0, lambda: self._apply_remote(items, remote_files, new_paths, stale_paths)) except Exception: self.log_exc("拉取失败") self.after(0, lambda: messagebox.showerror("错误", "拉取失败，详见日志")) finally: self.after(0, self._finish_loading) def _apply_remote(self, items, remote_files, new_paths, stale_paths): self.items = items self.remote_files = remote_files self.new_paths = new_paths self.stale_paths = stale_paths self.refresh_tree() def _finish_loading(self): self._loading = False self.btn_load_remote.configure(state="normal") def refresh_tree(self): self.tree.delete(*self.tree.get_children()) kw = self.search_var.get().strip().lower() only_new = self.only_new.get() only_empty = self.only_empty_trigger.get() for path, meta in self.items.items(): title = meta.get("title", "") trigger = meta.get("trigger", "") if kw and kw not in path.lower() and kw not in str(title).lower(): continue if only_new and path not in self.new_paths: continue if only_empty and trigger not in ("", [], None): continue tags = [] if path in self.new_paths: tags.append("new") elif path in self.stale_paths: tags.append("stale") if trigger in ("", [], None): tags.append("empty_trigger") trig_display = trigger if isinstance(trigger, str) else " | ".join(map(str, trigger)) self.tree.insert("", "end", iid=path, text=path, values=(title, trig_display, meta.get("dataset", ""), meta.get("showcase", "")), tags=tags) def on_double_click(self, event): sel = self.tree.focus() if sel and sel in self.items: self.open_manager(sel) def open_manager(self, path): meta = self.items[path] win = tk.Toplevel(self) win.title(f"编辑: {path}") win.geometry("720x560") win.transient(self); win.grab_set() ttk.Label(win, text="路径").pack(anchor="w", padx=10, pady=(10, 0)) p_entry = ttk.Entry(win, width=90) p_entry.insert(0, path); p_entry.configure(state="readonly"); p_entry.pack(padx=10, fill="x") ttk.Label(win, text="title").pack(anchor="w", padx=10, pady=(8, 0)) title_var = tk.StringVar(value=meta.get("title", "")) ttk.Entry(win, textvariable=title_var, width=90).pack(padx=10, fill="x") ttk.Label(win, text="trigger").pack(anchor="w", padx=10, pady=(8, 0)) trig_text = tk.Text(win, height=8, wrap="word"); trig_text.pack(padx=10, fill="both") trig = meta.get("trigger", "") trig_text.insert("1.0", "\n".join(map(str, trig)) if isinstance(trig, list) else trig) ttk.Label(win, text="dataset").pack(anchor="w", padx=10, pady=(8, 0)) ds_var = tk.StringVar(value=meta.get("dataset", "")) ttk.Entry(win, textvariable=ds_var, width=90).pack(padx=10, fill="x") # 🌟 新增：showcase 输入框 ttk.Label(win, text="showcase (URL)").pack(anchor="w", padx=10, pady=(8, 0)) sc_var = tk.StringVar(value=meta.get("showcase", "")) ttk.Entry(win, textvariable=sc_var, width=90).pack(padx=10, fill="x") def auto_dataset(): parts = path.split("/") if len(parts) >= 3: top, sub = parts[0], parts[1] base = Path(parts[2]).stem for noise in ("_onetrainer", "_Flux_512", "_Flux", "_512"): base = base.replace(noise, "") ds_var.set(f"{DATASET_BASE}/{top}/{sub}/{base}") ttk.Button(win, text="自动生成 dataset", command=auto_dataset).pack(anchor="w", padx=10, pady=4) def save_and_close(): title = title_var.get().strip() raw = trig_text.get("1.0", "end").strip() trigger_val = ([l.strip() for l in raw.splitlines() if l.strip()] if "\n" in raw else raw) self.items[path] = {"title": title, "trigger": trigger_val, "dataset": ds_var.get().strip(),"showcase": sc_var.get().strip()} self.refresh_tree() win.destroy() btn = ttk.Frame(win); btn.pack(fill="x", pady=10, padx=10) ttk.Button(btn, text="保存", command=save_and_close).pack(side="right") ttk.Button(btn, text="取消", command=win.destroy).pack(side="right", padx=6) def save_local(self): data_items = {k: v for k, v in self.items.items() if k not in self.stale_paths} sorted_items = {k: data_items[k] for k in sorted(data_items.keys(), key=custom_sort_key)} with open(OUTPUT_FILE, "w", encoding="utf-8") as f: json.dump({"items": sorted_items}, f, indent=2, ensure_ascii=False) self.log_json(f"已保存本地 {OUTPUT_FILE}，{len(sorted_items)} 条") messagebox.showinfo("完成", f"已保存到 {OUTPUT_FILE}") def upload_models_json(self): if not self._ensure_token(): return if not messagebox.askyesno("确认", f"上传 models.json 到 {SPACE_ID}？"): return data_items = {k: v for k, v in self.items.items() if k not in self.stale_paths} sorted_items = {k: data_items[k] for k in sorted(data_items.keys(), key=custom_sort_key)} tmp = "models.upload.json" with open(tmp, "w", encoding="utf-8") as f: json.dump({"items": sorted_items}, f, indent=2, ensure_ascii=False) self.log_json(f"开始上传 JSON ({len(sorted_items)} 条)") api = self.api def work(): try: api.upload_file(path_or_fileobj=tmp, path_in_repo="models.json", repo_id=SPACE_ID, repo_type="space") self.log_json("models.json 上传成功") self.after(0, lambda: messagebox.showinfo("完成", "JSON 上传成功")) except Exception: self.log_exc("JSON 上传失败") self.after(0, lambda: messagebox.showerror("失败", "详见日志")) finally: try: os.remove(tmp) except OSError: pass threading.Thread(target=work, daemon=True).start() # ========================================================== # ================= 新增：site_config 管理 ================= # ========================================================== def open_site_config_editor(self): if not self._ensure_token(): return self.log_json("正在从 Space 拉取 site_config.json...") def fetch_worker(): try: # 直接从 Space 文件仓库下载 file_path = self.api.hf_hub_download( repo_id=SPACE_ID, filename=SITE_CONFIG_FILE, repo_type="space" ) with open(file_path, "r", encoding="utf-8") as f: config_data = json.load(f) self.after(0, lambda: self._show_config_window(config_data)) except Exception as e: self.log_json(f"拉取 site_config 失败: {e}") # 如果远端没有，就初始化一个空的 self.after(0, lambda: self._show_config_window({"models": [], "categories": []})) threading.Thread(target=fetch_worker, daemon=True).start() def _show_config_window(self, config_data): win = tk.Toplevel(self) win.title("编辑 site_config.json (保存后自动上传)") win.geometry("600x550") win.transient(self); win.grab_set() ttk.Label(win, text="💡 提示：修改顺序即可改变前端左侧导航栏的展示顺序。", foreground="gray").pack(pady=5) # 分成两个框：Models 和 Categories frame_models = ttk.LabelFrame(win, text="Base Models (按顺序)") frame_models.pack(fill="both", expand=True, padx=10, pady=5) text_models = tk.Text(frame_models, height=10, wrap="none") text_models.pack(fill="both", expand=True, padx=5, pady=5) # 将 JSON 格式化为易读的字符串填入 text_models.insert("1.0", json.dumps(config_data.get("models", []), indent=2, ensure_ascii=False)) frame_cats = ttk.LabelFrame(win, text="Categories (按顺序)") frame_cats.pack(fill="both", expand=True, padx=10, pady=5) text_cats = tk.Text(frame_cats, height=10, wrap="none") text_cats.pack(fill="both", expand=True, padx=5, pady=5) text_cats.insert("1.0", json.dumps(config_data.get("categories", []), indent=2, ensure_ascii=False)) def save_and_upload(): try: # 解析用户修改后的文本 new_models = json.loads(text_models.get("1.0", "end").strip()) new_cats = json.loads(text_cats.get("1.0", "end").strip()) new_config = { "models": new_models, "categories": new_cats } # 保存到本地临时文件 with open(SITE_CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(new_config, f, indent=2, ensure_ascii=False) self.log_json("正在上传 site_config.json 到 Space...") self.api.upload_file( path_or_fileobj=SITE_CONFIG_FILE, path_in_repo=SITE_CONFIG_FILE, repo_id=SPACE_ID, repo_type="space" ) self.log_json("🎉 site_config.json 上传成功！") messagebox.showinfo("成功", "配置已更新并上传到 Space！") win.destroy() except json.JSONDecodeError as e: messagebox.showerror("JSON 格式错误", f"请检查括号、逗号是否漏了。\n报错信息: {e}") except Exception as e: self.log_json(f"上传失败: {e}") messagebox.showerror("错误", f"上传失败: {e}") btn_frame = ttk.Frame(win) btn_frame.pack(fill="x", pady=10, padx=10) ttk.Button(btn_frame, text="保存并上传", command=save_and_upload).pack(side="right") ttk.Button(btn_frame, text="取消", command=win.destroy).pack(side="right", padx=6) def _on_drop_json(self, event): raw = event.data.strip("{}") p = Path(raw) if p.is_dir(): self.notebook.select(self.tab_model) self.local_folder_var.set(str(p)) # ========================================================== # ==================== 区域 2: 模型上传 ==================== # ========================================================== def _build_model_tab(self): parent = self.tab_model # 1. 输入区 frame = ttk.Frame(parent) frame.pack(fill="x", padx=10, pady=10) # 本地文件夹 ttk.Label(frame, text="本地文件夹:").grid(row=0, column=0, sticky="w", pady=4) ttk.Entry(frame, textvariable=self.local_folder_var, width=70).grid(row=0, column=1, padx=6, sticky="we") ttk.Button(frame, text="📁 浏览...", width=18, command=self._browse_model_local).grid(row=0, column=2) # 远端目录 (Treeview 改造) ttk.Label(frame, text="上传到 HF 目录:").grid(row=1, column=0, sticky="nw", pady=4) tree_frame = ttk.Frame(frame) tree_frame.grid(row=1, column=1, padx=6, sticky="nsew") # 滚动条 vsb = ttk.Scrollbar(tree_frame, orient="vertical") vsb.pack(side="right", fill="y") # 树形列表 self.remote_tree = ttk.Treeview(tree_frame, height=6, show="tree", yscrollcommand=vsb.set, selectmode="browse") self.remote_tree.pack(side="left", fill="both", expand=True) vsb.config(command=self.remote_tree.yview) # 绑定树形点击和双击事件 self.remote_tree.bind("<>", self._on_remote_tree_select) self.remote_tree.bind("", lambda e: self._start_model_upload()) # 刷新按钮 ttk.Button(frame, text="🔄 拉取 Models 目录", width=18, command=self._fetch_hf_dirs).grid(row=1, column=2, sticky="n") # 按钮区 btn_row = ttk.Frame(frame) btn_row.grid(row=2, column=0, columnspan=3, pady=15, sticky="w") # 🌟 开始上传按钮 (绑定 self.btn_upload，方便后面禁用/恢复) self.btn_upload = ttk.Button(btn_row, text="🢁 开始上传", command=self._start_model_upload) self.btn_upload.pack(side="left") # 状态文字 self.status_model_var = tk.StringVar(value="就绪") ttk.Label(btn_row, textvariable=self.status_model_var).pack(side="left", padx=10) # 🌟 进度条 self.progress_var = tk.DoubleVar() self.progress_bar = ttk.Progressbar(btn_row, variable=self.progress_var, maximum=100, length=200, mode='determinate') self.progress_bar.pack(side="left", padx=10) frame.columnconfigure(1, weight=1) # 让 Treeview 高度自适应 frame.rowconfigure(1, weight=1) # 2. 日志区 log_frame = ttk.LabelFrame(parent, text="模型上传日志") log_frame.pack(fill="both", expand=True, padx=10, pady=6) self.log_model_text = ScrolledText(log_frame, height=10, state="disabled", wrap="word") self.log_model_text.pack(fill="both", expand=True) def _browse_model_local(self): # 🌟 强制从当前脚本所在目录开始（即 zz_huggingface 这一层） if getattr(sys, 'frozen', False): current_dir = os.path.dirname(sys.executable) else: current_dir = os.path.dirname(os.path.abspath(__file__)) folder = filedialog.askdirectory(title="选择包含 .safetensors 的文件夹", initialdir=current_dir) if folder: self.local_folder_var.set(folder) def _fetch_hf_dirs(self): if not self._ensure_token(): return self.status_model_var.set("正在读取远端目录...") self.log_model("正在读取远端仓库目录列表...") def work(): try: api = self.api all_files = list(api.list_repo_files(repo_id=MODEL_REPO_ID, repo_type="model")) dirs = set() for f in all_files: d = str(Path(f).parent) if d != ".": dirs.add(d) sorted_dirs = sorted(list(dirs)) self.after(0, lambda: self._update_remote_tree(sorted_dirs)) except Exception: self.log_exc("读取远端目录失败", is_model=True) finally: self.after(0, lambda: self.status_model_var.set("就绪")) threading.Thread(target=work, daemon=True).start() def _update_remote_tree(self, dirs): # 清空旧的树 for item in self.remote_tree.get_children(): self.remote_tree.delete(item) tree_nodes = {} # 用于记录路径对应的 treeview item id for d in dirs: # 把 Windows 反斜杠统一替换成正斜杠，方便拆分 safe_d = d.replace("\\", "/") parts = safe_d.split("/") parent_id = "" current_path = "" for i, part in enumerate(parts): current_path = f"{current_path}/{part}" if current_path else part # 如果这个节点还没创建，就创建它 if current_path not in tree_nodes: # 判断是不是最后一级目录（如果是最后一级，加上图标和展开符） is_leaf = (i == len(parts) - 1) display_text = f"📁 {part}" if is_leaf else part node_id = self.remote_tree.insert( parent_id, "end", text=display_text, values=(current_path,) # 把真实路径藏在 values 里 ) tree_nodes[current_path] = node_id # 下一级的父节点变成了当前节点 parent_id = tree_nodes[current_path] # 默认展开所有根节点 for root_node in self.remote_tree.get_children(""): self.remote_tree.item(root_node, open=True) self.log_model(f"已加载 {len(dirs)} 个远端目录结构") def _on_remote_tree_select(self, event): # 当用户在树上点击时，提取路径存到 self.remote_dir_var selection = self.remote_tree.selection() if selection: item_id = selection[0] # 从 values 里拿我们存的真实路径 values = self.remote_tree.item(item_id, "values") if values: self.remote_dir_var.set(values[0]) def _start_model_upload(self): local_dir = self.local_folder_var.get().strip() # 如果用户没点选树，默认上传到根目录；如果点了，就是选中的路径 remote_dir = self.remote_dir_var.get().strip().strip("/") if not local_dir or not os.path.exists(local_dir): messagebox.showwarning("错误", "请选择有效的本地文件夹。") return if not remote_dir: if not messagebox.askyesno("确认", "远端目录为空，将上传到仓库根目录，继续？"): return if not self._ensure_token(): return self.status_model_var.set("正在扫描并比对文件...") self.log_model(f"开始扫描: {local_dir} -> {MODEL_REPO_ID}/{remote_dir}") def scan_worker(): try: api = self.api # 1. 清洗本地图片 (png/webp -> jpg) local_files = [f for f in Path(local_dir).iterdir() if f.is_file()] for f in local_files: if f.suffix.lower() in [".png", ".webp"]: jpg_path = f.with_suffix(".jpg") try: self.log_model(f"清洗图片并转码: {f.name} -> {jpg_path.name}") with Image.open(f) as img: if img.mode in ("RGBA", "P"): img = img.convert("RGB") img.save(jpg_path, format="JPEG", quality=95) os.remove(f) # 清理原图 except Exception as e: self.log_model(f"转码失败 {f.name}: {e}") # 2. 重新读取清洗后的文件列表 local_files = [f for f in Path(local_dir).iterdir() if f.is_file()] if not local_files: self.after(0, lambda: messagebox.showwarning("提示", "本地文件夹是空的。")) return # 3. 拉取远端文件用于比对 remote_all = set(api.list_repo_files(repo_id=MODEL_REPO_ID, repo_type="model")) to_upload = [] for f in local_files: if f.suffix.lower() not in [".safetensors"] + list(IMG_EXTS): continue remote_path = f"{remote_dir}/{f.name}" if remote_dir else f.name if remote_path not in remote_all: to_upload.append((f, remote_path)) if not to_upload: self.after(0, lambda: messagebox.showinfo("完成", "远端已存在所有文件，无需上传。")) self.after(0, lambda: self.status_model_var.set("无需上传")) self.log_model("扫描完毕，无新增文件。") return msg = f"本地清理完毕。找到 {len(to_upload)} 个新文件待上传。\n\n目标目录: {remote_dir}\n\n继续？" self.after(0, lambda: self._do_model_upload_confirm(msg, to_upload, api)) except Exception: self.log_exc("扫描失败", is_model=True) self.after(0, lambda: self.status_model_var.set("扫描失败")) threading.Thread(target=scan_worker, daemon=True).start() def _do_model_upload_confirm(self, msg, to_upload, api): if not messagebox.askyesno("确认上传", msg): self.status_model_var.set("已取消") return def upload_worker(): try: from huggingface_hub import CommitOperationAdd # 🌟 开始上传，禁用按钮，重置进度条 self.after(0, lambda: self.btn_upload.configure(state="disabled")) self.after(0, lambda: self.progress_var.set(0)) self.after(0, lambda: self.status_model_var.set("正在打包并上传...")) # 1. 准备操作列表 operations = [] for local_file, remote_path in to_upload: # 极度关键：防止 Windows 反斜杠导致远端多建文件夹 safe_remote_path = remote_path.replace("\\", "/") operations.append( CommitOperationAdd( path_or_fileobj=str(local_file), path_in_repo=safe_remote_path ) ) # 2. 准备 commit message remote_dir = self.remote_dir_var.get().strip().replace("\\", "/").strip("/") commit_msg = f"Upload files to {remote_dir}" if remote_dir else "Upload files to root" self.log_model(f"开始打包 {len(operations)} 个文件，一次性 Commit...") self.log_model(f"Commit message: {commit_msg}") # 🌟 3. 假进度条控制变量 is_uploading = [True] def update_progress(): if not is_uploading[0]: return # 如果上传结束，停止动画 current = self.progress_var.get() # 慢慢爬升到 95%，最后留 5% 给真正完成时冲刺 if current < 95: self.progress_var.set(current + 2) self.after(300, update_progress) self.after(100, update_progress) # 启动动画 # 4. 一次性提交多个文件 (同步阻塞操作) api.create_commit( repo_id=MODEL_REPO_ID, repo_type="model", operations=operations, commit_message=commit_msg ) # 🌟 5. 上传成功：停止动画，进度条瞬间满格 is_uploading[0] = False self.after(0, lambda: self.progress_var.set(100)) self.after(0, lambda: self.status_model_var.set("上传完成 ✅")) self.after(0, lambda: self.btn_upload.configure(state="normal")) self.log_model("打包上传成功！") self.after(0, lambda: messagebox.showinfo("完成", "上传成功！")) self.after(0, self.load_remote) # 自动刷新 JSON except Exception as e: # 🌟 失败时：停止动画，重置进度条，恢复按钮 is_uploading[0] = False self.after(0, lambda: self.progress_var.set(0)) self.after(0, lambda: self.btn_upload.configure(state="normal")) self.after(0, lambda: self.status_model_var.set("上传失败 ❌")) self.log_exc("打包上传失败", is_model=True) self.after(0, lambda: messagebox.showerror("失败", "详见日志")) threading.Thread(target=upload_worker, daemon=True).start() # ========================================================== # ==================== 区域 3: Dataset 上传 ================= # ========================================================== def _build_dataset_tab(self): parent = self.tab_dataset frame = ttk.Frame(parent) frame.pack(fill="x", padx=10, pady=10) # 1. 本地母文件夹 ttk.Label(frame, text="本地母文件夹:").grid(row=0, column=0, sticky="w", pady=4) self.ds_local_folder_var = tk.StringVar() ttk.Entry(frame, textvariable=self.ds_local_folder_var, width=70).grid(row=0, column=1, columnspan=2, padx=6, sticky="we") ttk.Button(frame, text="📁 浏览...", width=18, command=self._browse_ds_local).grid(row=0, column=3) # 2. 地区/分类前缀 (动态读取改造) ttk.Label(frame, text="Region 前缀:").grid(row=1, column=0, sticky="w", pady=8) prefix_frame = ttk.Frame(frame) prefix_frame.grid(row=1, column=1, columnspan=3, sticky="w", padx=6, pady=8) # 核心变量初始化 self.ds_region_var = tk.StringVar(value="KR") self.ds_custom_region_var = tk.StringVar() # 必须保留这个变量用于输入框 # 按钮：刷新 Dataset 目录 ttk.Button(prefix_frame, text="🔄 拉取 datasets 目录", command=self._fetch_dataset_dirs).pack(side="left", padx=(0, 10)) # 用子 Frame 专门放单选框，刷新时只清空这里的控件 self.ds_radio_container = ttk.Frame(prefix_frame) self.ds_radio_container.pack(side="left") # 初始默认放几个，防止还没拉取时界面太难看 self._render_dataset_radios(["CN", "KR", "JP", "Fictional", "WE"]) # 3. 按钮区 btn_row = ttk.Frame(frame) btn_row.grid(row=2, column=0, columnspan=4, pady=15, sticky="w") self.btn_ds_upload = ttk.Button(btn_row, text="🢁 开始 5合1 处理并上传", command=self._start_dataset_pipeline) self.btn_ds_upload.pack(side="left") self.status_ds_var = tk.StringVar(value="就绪") ttk.Label(btn_row, textvariable=self.status_ds_var).pack(side="left", padx=10) frame.columnconfigure(1, weight=1) frame.columnconfigure(2, weight=1) # 4. 日志区 log_frame = ttk.LabelFrame(parent, text="Dataset 自动化流水线日志") log_frame.pack(fill="both", expand=True, padx=10, pady=6) self.log_dataset_text = ScrolledText(log_frame, height=15, state="disabled", wrap="word") self.log_dataset_text.pack(fill="both", expand=True) def _fetch_dataset_dirs(self): if not self._ensure_token(): return self.log_dataset("正在读取 Dataset 仓库根目录...") def work(): try: files = self.api.list_repo_files(repo_id=DATASET_REPO_ID, repo_type="dataset") dirs = set() for f in files: parts = f.split("/") # 过滤掉 tool 和根目录文件，只保留文件夹 if len(parts) > 1 and parts[0] not in ["tool", ".git"]: dirs.add(parts[0]) sorted_dirs = sorted(list(dirs)) # 🌟 新增：在主线程更新 UI 和打印日志 def update_ui(): self._render_dataset_radios(sorted_dirs) self.log_dataset(f"✅ 远端目录刷新成功！共发现 {len(sorted_dirs)} 个分类: {', '.join(sorted_dirs)}") self.after(0, update_ui) except Exception as e: self.log_exc("读取 Dataset 目录失败", is_model=True) # 🌟 新增：报错时更新状态栏 self.after(0, lambda: self.status_ds_var.set("读取目录失败")) threading.Thread(target=work, daemon=True).start() def _render_dataset_radios(self, dirs): # 清空旧的单选框和输入框 for widget in self.ds_radio_container.winfo_children(): widget.destroy() # 保持原本选中的值，如果不存在了，默认选第一个 current_val = self.ds_region_var.get() if current_val not in dirs and current_val != "Custom" and dirs: self.ds_region_var.set(dirs[0]) # 动态生成分类单选框 for region in dirs: ttk.Radiobutton(self.ds_radio_container, text=region, variable=self.ds_region_var, value=region).pack(side="left", padx=(0, 10)) # 加上自定义部分 (注意：父级是 self.ds_radio_container) ttk.Radiobutton(self.ds_radio_container, text="自定义:", variable=self.ds_region_var, value="Custom").pack(side="left", padx=(10, 2)) self.ds_custom_entry = ttk.Entry(self.ds_radio_container, textvariable=self.ds_custom_region_var, width=15, state="disabled") self.ds_custom_entry.pack(side="left") # 🌟 重新绑定输入框激活逻辑 (因为输入框是刚刚重新创建的) def on_region_change(*args): if self.ds_region_var.get() == "Custom": self.ds_custom_entry.configure(state="normal") else: self.ds_custom_entry.configure(state="disabled") # 清理之前的 trace，防止多次刷新后绑定重复导致卡顿或报错 if hasattr(self, '_region_trace_id'): try: self.ds_region_var.trace_remove("write", self._region_trace_id) except Exception: pass self._region_trace_id = self.ds_region_var.trace_add("write", on_region_change) def log_dataset(self, msg): self._log_to_tab(msg, getattr(self, 'log_dataset_text', None)) def _browse_ds_local(self): # 🌟 初始目录设置为当前脚本所在目录，方便用户找 current_dir = os.path.dirname(os.path.abspath(__file__)) folder = filedialog.askdirectory(title="选择包含子文件夹的数据集母目录", initialdir=current_dir) if folder: self.ds_local_folder_var.set(folder) def _start_dataset_pipeline(self): local_dir = self.ds_local_folder_var.get().strip() # ==== 修改这里：获取前缀逻辑 ==== selected_region = self.ds_region_var.get() if selected_region == "Custom": region_prefix = self.ds_custom_region_var.get().strip().replace(" ", "_").replace("-", "_") if not region_prefix: messagebox.showwarning("错误", "您选择了自定义前缀，但输入框为空。") return else: region_prefix = selected_region # =============================== if not local_dir or not os.path.exists(local_dir): messagebox.showwarning("错误", "请选择有效的本地母文件夹。") return if not self._ensure_token(): return if not messagebox.askyesno("确认", f"即将对以下目录执行 5合1 流水线：\n{local_dir}\n前缀：{region_prefix}\n\n继续？"): return self.btn_ds_upload.configure(state="disabled") self.status_ds_var.set("处理中...") self.log_dataset(f"================ 开始 5合1 流水线 ================") self.log_dataset(f"目标目录: {local_dir} | 前缀: {region_prefix}") def worker(): api = self.api try: # 1. 清理 txt 换行符 self._clean_txt_newlines(local_dir) # 2. 清理 Mask 元数据 self._remove_mask_metadata(local_dir) # 3. 转换 PNG -> JPG self._convert_png_to_jpg(local_dir) # 4. 打包 Parquet 并更新 YAML temp_readme_path, valid_folders = self._generate_parquet_and_update_yaml(api, local_dir, region_prefix) # 5. 上传 self._upload_files_and_readme(api, local_dir, region_prefix, temp_readme_path, valid_folders) self.log_dataset("🎉 5合1 全流程处理完毕！") self.after(0, lambda: messagebox.showinfo("完成", "Dataset 流水线执行成功！")) self.after(0, lambda: self.status_ds_var.set("上传完成")) # 刷新 JSON 的拉取（可选，如果需要马上更新界面的话） self.after(0, self.load_remote) except Exception as e: self.log_exc("Dataset 流水线异常", is_model=True) self.after(0, lambda: self.status_ds_var.set("执行失败")) finally: self.after(0, lambda: self.btn_ds_upload.configure(state="normal")) threading.Thread(target=worker, daemon=True).start() # ---------------- 原有脚本的函数改造 ---------------- def _clean_txt_newlines(self, current_dir): import glob self.log_dataset("🧹 [任务 1/5] 正在清理所有 TXT 文件中的换行符...") txt_files = glob.glob(os.path.join(current_dir, "**", "*.txt"), recursive=True) if not txt_files: self.log_dataset(" ✅ 没有找到需要处理的 TXT 文件。") return cleaned_count = 0 for txt_path in txt_files: if os.path.dirname(txt_path) == current_dir: continue try: with open(txt_path, 'r', encoding='utf-8') as f: content = ''.join(line.strip() for line in f) with open(txt_path, 'w', encoding='utf-8') as f: f.write(content) cleaned_count += 1 except Exception as e: self.log_dataset(f" ⚠️ 处理 {os.path.basename(txt_path)} 失败: {e}") self.log_dataset(f" 🎉 TXT 清理完成！共处理了 {cleaned_count} 个文件。") def _remove_mask_metadata(self, current_dir): import glob self.log_dataset("🛡️ [任务 2/5] 正在剥离 Mask 蒙版图的元数据...") png_files = glob.glob(os.path.join(current_dir, "**", "*.png"), recursive=True) mask_files = [f for f in png_files if "masklabel" in os.path.basename(f).lower()] if not mask_files: self.log_dataset(" ✅ 没有找到 masklabel.png 文件，跳过此步骤。") return success_count = 0 for file_path in mask_files: try: with Image.open(file_path) as img: data = list(img.getdata()) clean_img = Image.new(img.mode, img.size) clean_img.putdata(data) clean_img.save(file_path, "PNG") success_count += 1 except Exception as e: self.log_dataset(f" ⚠️ 处理失败 [{os.path.basename(file_path)}]: {e}") self.log_dataset(f" 🎉 元数据清理完成！共处理了 {success_count} 张 Mask 图。") def _convert_png_to_jpg(self, current_dir): import glob self.log_dataset("🎬 [任务 3/5] 正在批量转换常规 PNG 为 JPG...") png_files = glob.glob(os.path.join(current_dir, "**", "*.png"), recursive=True) if not png_files: self.log_dataset(" ✅ 没有找到需要转换的 PNG 文件。") return converted_count = 0 for png_path in png_files: if "masklabel" in png_path.lower(): continue jpg_path = os.path.splitext(png_path)[0] + ".jpg" try: subprocess.run( ["ffmpeg", "-y", "-i", png_path, "-q:v", "2", jpg_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True ) os.remove(png_path) converted_count += 1 self.log_dataset(f" 🔄 转换: {os.path.basename(png_path)} -> {os.path.basename(jpg_path)}") except subprocess.CalledProcessError: self.log_dataset(f" ⚠️ 转换失败: {png_path}") except FileNotFoundError: self.log_dataset(" ❌ 错误: 找不到 ffmpeg！请确保已安装并配置了环境变量。") return self.log_dataset(f" 🎉 转换完成！共转换 {converted_count} 张图片。") def _generate_parquet_and_update_yaml(self, api, current_dir, region_prefix): import glob, tempfile, pandas as pd from datasets import Dataset, Features, Image as HFImage, Value from huggingface_hub import DatasetCard from PIL import Image as PILImage self.log_dataset("📦 [任务 4/5] 正在本地打包 Parquet 并更新 YAML 配置...") local_folder_name = os.path.basename(current_dir).replace(" ", "_").replace("-", "_") self.log_dataset(f" ☁️ 正在读取云端 {DATASET_REPO_ID} 的现有配置...") try: # 注意这里用 DATASET_REPO_ID 替换了原脚本的 REPO_ID remote_readme_path = api.hf_hub_download(repo_id=DATASET_REPO_ID, filename="README.md", repo_type="dataset") with open(remote_readme_path, "r", encoding="utf-8") as f: content = f.read() card = DatasetCard(content) self.log_dataset(" ✅ 成功读取云端 README。") except Exception as e: self.log_dataset(f" ⚠️ 找不到云端 README，将新建: {e}") card = DatasetCard("") yaml_data = card.data if card.data else {} if "configs" not in yaml_data: yaml_data["configs"] = [] yaml_data["dataset_info"] = { "features": [ {"name": "image", "dtype": "image"}, {"name": "file_name", "dtype": "string"}, {"name": "text", "dtype": "string"} ] } existing_config_names = [c.get("config_name") for c in yaml_data["configs"]] new_configs_added = False valid_folders = [] for sub_folder_name in os.listdir(current_dir): sub_folder_path = os.path.join(current_dir, sub_folder_name) if os.path.isdir(sub_folder_path) and not sub_folder_name.startswith('.'): jpg_files = glob.glob(os.path.join(sub_folder_path, "*.jpg")) + glob.glob(os.path.join(sub_folder_path, "*.jpeg")) if not jpg_files: self.log_dataset(f" ⚠️ 跳过子文件夹 [{sub_folder_name}]: 未找到任何 .jpg 图片。") continue self.log_dataset(f" 📦 正在打包: {sub_folder_name}") data_rows = [] for jpg_path in jpg_files: file_name = os.path.basename(jpg_path) base_name = os.path.splitext(file_name)[0] txt_path = os.path.join(sub_folder_path, f"{base_name}.txt") text_content = "" if os.path.exists(txt_path): with open(txt_path, "r", encoding="utf-8") as f: text_content = f.read().strip() try: with open(jpg_path, "rb") as f: image_bytes = {"bytes": f.read(), "path": None} except Exception as e: self.log_dataset(f" ⚠️ 跳过损坏图片 {file_name}: {e}") continue data_rows.append({ "image": image_bytes, "file_name": file_name, "text": text_content }) if data_rows: df = pd.DataFrame(data_rows) features = Features({ "image": HFImage(), "file_name": Value("string"), "text": Value("string") }) ds = Dataset.from_pandas(df, features=features) ds.to_parquet(os.path.join(sub_folder_path, "train.parquet")) valid_folders.append(sub_folder_name) self.log_dataset(f" ✅ 生成 train.parquet ({len(data_rows)} 条数据)") config_name = f"{local_folder_name}_{sub_folder_name}".lower() path_str = f"{region_prefix}/{local_folder_name}/{sub_folder_name}/train.parquet" if config_name not in existing_config_names: yaml_data["configs"].append({ "config_name": config_name, "data_files": [{"split": "train", "path": path_str}] }) new_configs_added = True self.log_dataset(f" ➕ 已添加配置: {config_name}") def get_priority(config): data_files = config.get("data_files", []) path = data_files[0].get("path", "") if data_files else "" priority_dict = {"CN": 1, "KR": 2, "JP": 3, "Fictional": 4, "WE": 5} path_prefix = path.split('/')[0] if path else "" return priority_dict.get(path_prefix, 99) yaml_data["configs"] = sorted(yaml_data["configs"], key=lambda c: (get_priority(c), c.get("config_name", ""))) self.log_dataset(" 🔄 已根据 [CN > KR > JP > Fictional > WE] 的顺序重排 YAML。") if new_configs_added: card.data = yaml_data with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', suffix='.md', delete=False) as temp_readme: temp_readme.write(str(card)) return temp_readme.name, valid_folders return None, valid_folders def _upload_files_and_readme(self, api, current_dir, region_prefix, temp_readme_path, valid_folders): self.log_dataset("🚀 [任务 5/5] 正在上传文件到 Hugging Face...") local_folder_name = os.path.basename(current_dir).replace(" ", "_").replace("-", "_") for sub_folder_name in valid_folders: sub_folder_path = os.path.join(current_dir, sub_folder_name) path_in_repo = f"{region_prefix}/{local_folder_name}/{sub_folder_name}" self.log_dataset(f" 📦 开始上传: {sub_folder_name} -> {path_in_repo}") try: api.upload_folder( folder_path=sub_folder_path, repo_id=DATASET_REPO_ID, repo_type="dataset", path_in_repo=path_in_repo, commit_message=f"Update files for {path_in_repo}", allow_patterns=["*.jpg", "*.jpeg", "*.txt", "*.png", "*.parquet"] ) self.log_dataset(f" 🎉 {sub_folder_name} 上传完成！") except Exception as e: self.log_dataset(f" ❌ {sub_folder_name} 上传失败: {e}") if temp_readme_path: self.log_dataset("\n 📝 正在推送最新的 README.md...") try: api.upload_file( path_or_fileobj=temp_readme_path, path_in_repo="README.md", repo_id=DATASET_REPO_ID, repo_type="dataset", commit_message="Auto-update README YAML configs" ) self.log_dataset(" 🎉 README.md 已成功更新到云端！") except Exception as e: self.log_dataset(f" ❌ README.md 上传失败: {e}") finally: os.remove(temp_readme_path) # ========================================================== # ==================== 区域 4: Showcase 管理 ================ # ========================================================== def _build_showcase_tab(self): parent = self.tab_showcase # 1. 输入区 frame = ttk.Frame(parent) frame.pack(fill="x", padx=10, pady=10) # 本地文件夹 ttk.Label(frame, text="本地文件夹:").grid(row=0, column=0, sticky="w", pady=4) self.sc_local_folder_var = tk.StringVar() ttk.Entry(frame, textvariable=self.sc_local_folder_var, width=70).grid(row=0, column=1, padx=6, sticky="we") ttk.Button(frame, text="📁 浏览...", width=18, command=self._browse_sc_local).grid(row=0, column=2) # 远端目录 (Treeview 改造，完全复用 Models 的样式) ttk.Label(frame, text="上传到 Showcase 目录:").grid(row=1, column=0, sticky="nw", pady=4) tree_frame = ttk.Frame(frame) tree_frame.grid(row=1, column=1, padx=6, sticky="nsew") # 滚动条 vsb = ttk.Scrollbar(tree_frame, orient="vertical") vsb.pack(side="right", fill="y") # 树形列表 self.sc_remote_tree = ttk.Treeview(tree_frame, height=6, show="tree",yscrollcommand=vsb.set, selectmode="browse") self.sc_remote_tree.pack(side="left", fill="both", expand=True) vsb.config(command=self.sc_remote_tree.yview) # 绑定树形点击事件 self.sc_remote_tree.bind("<>", self._on_sc_remote_tree_select) # 刷新按钮 ttk.Button(frame, text="🔄 拉取案例目录", width=18, command=self._fetch_showcase_dirs).grid(row=1, column=2, sticky="n") # 按钮区 btn_row = ttk.Frame(frame) btn_row.grid(row=2, column=0, columnspan=3, pady=15, sticky="w") self.btn_sc_upload = ttk.Button(btn_row, text="🢁 生成 JSON 并上传", command=self._start_showcase_pipeline) self.btn_sc_upload.pack(side="left") self.status_sc_var = tk.StringVar(value="就绪") ttk.Label(btn_row, textvariable=self.status_sc_var).pack(side="left", padx=10) # 🌟 新增：Showcase 链接展示区 result_row = ttk.Frame(frame) result_row.grid(row=3, column=0, columnspan=4, pady=(0, 10), sticky="we") ttk.Label(result_row, text="Showcase 链接:").pack(side="left") self.sc_result_var = tk.StringVar() self.sc_result_entry = ttk.Entry(result_row, textvariable=self.sc_result_var, state="readonly") self.sc_result_entry.pack(side="left", fill="x", expand=True, padx=6) # 复制按钮 ttk.Button(result_row, text="📋 复制", command=self._copy_sc_url).pack(side="left") frame.columnconfigure(1, weight=1) frame.rowconfigure(1, weight=1) frame.columnconfigure(1, weight=1) frame.rowconfigure(1, weight=1) # 2. 日志区 log_frame = ttk.LabelFrame(parent, text="Showcase 操作日志") log_frame.pack(fill="both", expand=True, padx=10, pady=6) self.log_sc_text = ScrolledText(log_frame, height=10, state="disabled", wrap="word") self.log_sc_text.pack(fill="both", expand=True) def _copy_sc_url(self): url = self.sc_result_var.get() if url: self.clipboard_clear() self.clipboard_append(url) messagebox.showinfo("成功", "链接已复制到剪贴板！") def log_showcase(self, msg): self._log_to_tab(msg, getattr(self, 'log_sc_text', None)) def _browse_sc_local(self): if getattr(sys, 'frozen', False): current_dir = os.path.dirname(sys.executable) else: current_dir = os.path.dirname(os.path.abspath(__file__)) folder = filedialog.askdirectory(title="选择包含 PNG 的 Showcase 文件夹", initialdir=current_dir) if folder: self.sc_local_folder_var.set(folder) def _fetch_showcase_dirs(self): if not self._ensure_token(): return self.status_sc_var.set("正在读取远端目录...") self.log_showcase("正在读取 Showcase 仓库目录列表...") def work(): try: api = self.api all_files = list(api.list_repo_files(repo_id=MODEL_REPO_ID, repo_type="model")) dirs = set() for f in all_files: # 核心修改：只提取 comfyui_showcase 下的文件夹 if f.startswith("comfyui_showcase/"): d = str(Path(f).parent) if d != "comfyui_showcase": dirs.add(d) sorted_dirs = sorted(list(dirs)) self.after(0, lambda: self._update_sc_remote_tree(sorted_dirs)) except Exception: self.log_exc("读取 Showcase 目录失败", is_model=True) finally: self.after(0, lambda: self.status_sc_var.set("就绪")) threading.Thread(target=work, daemon=True).start() def _update_sc_remote_tree(self, dirs): for item in self.sc_remote_tree.get_children(): self.sc_remote_tree.delete(item) tree_nodes = {} for d in dirs: safe_d = d.replace("\\", "/") parts = safe_d.split("/") parent_id = "" current_path = "" for i, part in enumerate(parts): current_path = f"{current_path}/{part}" if current_path else part if current_path not in tree_nodes: is_leaf = (i == len(parts) - 1) display_text = f"📁 {part}" if is_leaf else part node_id = self.sc_remote_tree.insert(parent_id, "end", text=display_text, values=(current_path,)) tree_nodes[current_path] = node_id parent_id = tree_nodes[current_path] for root_node in self.sc_remote_tree.get_children(""): self.sc_remote_tree.item(root_node, open=True) self.log_showcase(f"已加载 {len(dirs)} 个远端目录结构") def _on_sc_remote_tree_select(self, event): selection = self.sc_remote_tree.selection() if selection: item_id = selection[0] values = self.sc_remote_tree.item(item_id, "values") if values: self.sc_remote_dir = values[0] def _start_showcase_pipeline(self): local_dir = self.sc_local_folder_var.get().strip() remote_dir = getattr(self, 'sc_remote_dir', "").strip().strip("/") if not local_dir or not os.path.exists(local_dir): messagebox.showwarning("错误", "请选择有效的本地文件夹。") return if not remote_dir: messagebox.showwarning("错误", "请先在右侧选择要上传到的远端目录。") return if not self._ensure_token(): return # 提取本地文件夹名字，准备拼接到远端路径 local_folder_name = os.path.basename(os.path.normpath(local_dir)) final_target_path = f"{remote_dir}/{local_folder_name}" if not messagebox.askyesno("确认", f"即将增量上传并更新 JSON：\n{MODEL_REPO_ID}/{final_target_path}\n\n继续？"): return self.btn_sc_upload.configure(state="disabled") self.status_sc_var.set("处理中...") self.log_showcase("================ 开始 Showcase 增量流水线 ================") self.log_showcase(f"本地: {local_dir} -> 远端: {final_target_path}") def worker(): api = self.api try: # 1. 尝试拉取云端旧的 gallery.json remote_json_path = f"{final_target_path}/gallery.json" existing_data = [] self.log_showcase(f"☁️ 正在检查云端是否已存在 gallery.json...") try: local_tmp_json = api.hf_hub_download( repo_id=MODEL_REPO_ID, repo_type="model", filename=remote_json_path ) with open(local_tmp_json, "r", encoding="utf-8") as f: existing_data = json.load(f) self.log_showcase(f" ✅ 发现云端旧数据，包含 {len(existing_data)} 条记录。") except Exception: self.log_showcase(" ℹ️ 云端暂无旧 JSON，将生成全新 JSON。") existing_data = [] # 2. 提取本地新图片的数据 png_files = list(Path(local_dir).glob("*.png")) if not png_files: self.log_showcase("❌ 本地文件夹中没有找到 .png 图片。") self.after(0, lambda: self.status_sc_var.set("未找到图片")) return self.log_showcase(f"📦 发现 {len(png_files)} 张本地 PNG，开始提取数据...") new_gallery_data = [] for png in png_files: data = self._extract_single_png(png) if data: new_gallery_data.append(data) self.log_showcase(f" ✅ 解析: {png.name}") # 3. 增量合并 (由于名字不同，这里直接追加) # 如果你想防止重复上传完全一样的图，可以用 file_name 去重，这里我们用字典去重然后转回列表 merged_dict = {item["file_name"]: item for item in existing_data if "file_name" in item} for item in new_gallery_data: merged_dict[item["file_name"]] = item # 如果本地同名，直接覆盖老的；名字不同则追加 final_gallery_data = list(merged_dict.values()) json_path = os.path.join(local_dir, "gallery.json") with open(json_path, "w", encoding="utf-8") as f: json.dump(final_gallery_data, f, indent=2, ensure_ascii=False) self.log_showcase(f"📄 合并后的 gallery.json 已生成，共 {len(final_gallery_data)} 条数据。") # 4. 上传（只传新的图片和合并后的 JSON，云端原有的图保持不变） self.log_showcase(f"🚀 开始上传新的图片和 JSON...") api.upload_folder( folder_path=local_dir, repo_id=MODEL_REPO_ID, repo_type="model", path_in_repo=final_target_path, commit_message=f"Add new images and update gallery.json at {final_target_path}", allow_patterns=["*.png", "*.json"] # 没有 delete_patterns，所以云端旧图绝不会被删！ ) self.log_showcase("🎉 Showcase 增量上传完毕！") self.after(0, lambda: messagebox.showinfo("完成", "Showcase 增量更新成功！")) self.after(0, lambda: self.status_sc_var.set("上传完成")) showcase_url = f"https://huggingface.co/waveshaker/loras/tree/main/{final_target_path}" self.log_showcase("-" * 50) self.log_showcase(f"🔗 Showcase 页面: {showcase_url}") self.log_showcase("-" * 50) self.after(0, lambda: self.sc_result_var.set(showcase_url)) except Exception as e: self.log_exc("Showcase 流水线异常", is_model=True) self.after(0, lambda: self.status_sc_var.set("执行失败")) finally: self.after(0, lambda: self.btn_sc_upload.configure(state="normal")) threading.Thread(target=worker, daemon=True).start() # ========================================================== # ==================== Showcase 提取核心逻辑 ================ # ========================================================== def _clean_showcase_name(self, filepath): if not filepath or not isinstance(filepath, str): return "-" filename = filepath.replace("\\", "/").split("/")[-1] return filename.replace(".safetensors", "").replace(".ckpt", "") def _find_upstream_model(self, prompt_data, input_link, depth=0): if depth > 15: return "-" if not isinstance(input_link, list) or len(input_link) != 2: return "-" node = prompt_data.get(str(input_link[0])) if not node: return "-" # 🌟 核心新增：如果节点被静音/禁用，它的连线被短路，直接跳过它继续找上游 if node.get("mode") == 4: # 如果是 Bypass，输入口可能叫 model，继续往上找 bypassed_input = node.get("inputs", {}).get("model") if bypassed_input: return self._find_upstream_model(prompt_data, bypassed_input, depth+1) return "-" class_type = node.get("class_type", "") inputs = node.get("inputs", {}) if class_type in ["CheckpointLoaderSimple", "UNETLoader"]: return self._clean_showcase_name(inputs.get("ckpt_name") or inputs.get("unet_name")) elif class_type in ["LoraLoader", "LoraLoaderModelOnly"]: return self._find_upstream_model(prompt_data, inputs.get("model"), depth+1) elif inputs.get("model"): return self._find_upstream_model(prompt_data, inputs.get("model"), depth+1) return "-" def _find_specific_loader(self, prompt_data, target_class_types): for node in prompt_data.values(): if node.get("class_type") in target_class_types: inputs = node.get("inputs", {}) return self._clean_showcase_name(inputs.get("vae_name") or inputs.get("clip_name")) return "-" def _extract_single_png(self, png_path): try: with Image.open(png_path) as img: raw_prompt = img.info.get('prompt', '') dasiwa_text = img.info.get('parameters', '') or img.info.get('Description', '') report = {"size": "-", "loras": [], "model": "-", "vae": "-", "clip": "-","sampler": "-", "scheduler": "-", "seed": "-", "steps": "-", "cfg": "-", "denoise": "-", "pos": ""} if raw_prompt and raw_prompt.strip().startswith("{"): prompt_data = json.loads(raw_prompt) for node in prompt_data.values(): # 🌟 核心新增：跳过所有被禁用的节点 if node.get("mode") == 4: continue if node.get("class_type") in ["KSampler", "KSamplerAdvanced"]: inputs = node.get("inputs", {}) report["sampler"] = inputs.get("sampler_name", "-") report["scheduler"] = inputs.get("scheduler", "-") report["steps"] = inputs.get("steps", "-") report["seed"] = inputs.get("seed", "-") report["cfg"] = inputs.get("cfg", 1.0) report["denoise"] = inputs.get("denoise", 1.0) report["model"] = self._find_upstream_model(prompt_data, inputs.get("model")) break report["vae"] = self._find_specific_loader(prompt_data, ["VAELoader"]) report["clip"] = self._find_specific_loader(prompt_data, ["CLIPLoader", "DualCLIPLoader"]) neg_text = "" if dasiwa_text: pos_match = re.search(r"^(.*?)(?:Negative prompt:|$)", dasiwa_text, re.DOTALL | re.IGNORECASE) if pos_match: report["pos"] = pos_match.group(1).strip() neg_match = re.search(r"Negative prompt:\s*(.*?)(?:Steps:|$)", dasiwa_text, re.DOTALL | re.IGNORECASE) if neg_match: neg_text = neg_match.group(1).strip() size_match = re.search(r"Size:\s*(\d+x\d+)", dasiwa_text) if size_match: report["size"] = size_match.group(1) lora_matches = re.findall(r"]+)(?::[^>]+)?>", report["pos"]) report["loras"] = [self._clean_showcase_name(l) for l in lora_matches] if neg_text and "negative_prompt_not_found" not in neg_text: report["neg"] = neg_text report["file_name"] = os.path.basename(png_path) return report except Exception as e: self.log_showcase(f" ⚠️ 提取失败 {os.path.basename(png_path)}: {e}") return None if __name__ == "__main__": app = ModelsManager() app.mainloop()