diff --git a/GiteaHubSync/app/backend/panel.py b/GiteaHubSync/app/backend/panel.py index 888b0d8..3629fbe 100644 --- a/GiteaHubSync/app/backend/panel.py +++ b/GiteaHubSync/app/backend/panel.py @@ -20,6 +20,7 @@ import subprocess import threading import time import uuid +from concurrent.futures import ThreadPoolExecutor from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -50,6 +51,9 @@ LEGACY_GITEA_USER = "cbb" # 旧版硬编码的 Gitea 用户名,仅作 DEFAULT_INTERVAL = 10 # 同步间隔默认值(秒) MIN_INTERVAL = 5 # 最小同步间隔(秒) MAX_INTERVAL = 86400 # 最大同步间隔(秒,1 天) +DEFAULT_WORKERS = 3 # 同时同步的项目数默认值 +MIN_WORKERS = 1 +MAX_WORKERS = 8 PANEL_PORT = int(os.environ.get("PANEL_PORT", "8899")) START_TS = time.time() NAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,100}$") @@ -62,7 +66,8 @@ ENV["GIT_TERMINAL_PROMPT"] = "0" # ---------------- 应用设置(settings.json) ---------------- cfg = {"gitea_user": LEGACY_GITEA_USER, "gitea_pass": "", "auto_create": True, - "interval": DEFAULT_INTERVAL, "gitea_host": DEFAULT_GITEA_HOST} + "interval": DEFAULT_INTERVAL, "gitea_host": DEFAULT_GITEA_HOST, + "max_workers": DEFAULT_WORKERS} def norm_gitea_host(h): @@ -120,6 +125,22 @@ def valid_interval(v): return False +def sync_workers(): + """同时同步的项目数,始终返回合法值""" + try: + n = int(cfg.get("max_workers", DEFAULT_WORKERS)) + except Exception: + n = DEFAULT_WORKERS + return min(MAX_WORKERS, max(MIN_WORKERS, n)) + + +def valid_workers(v): + try: + return MIN_WORKERS <= int(v) <= MAX_WORKERS + except Exception: + return False + + def load_settings(): try: d = json.loads(SETTINGS_FILE.read_text(encoding="utf-8")) @@ -129,6 +150,8 @@ def load_settings(): cfg["auto_create"] = bool(d.get("auto_create", True)) if valid_interval(d.get("interval")): cfg["interval"] = int(d["interval"]) + if valid_workers(d.get("max_workers")): + cfg["max_workers"] = int(d["max_workers"]) h = norm_gitea_host(d.get("gitea_host")) if h: cfg["gitea_host"] = h @@ -241,17 +264,30 @@ def scrub(s, acc): state = {"enabled": True, "syncing": False, "last_check": None, "checks": 0, "last_result": "尚未运行", "loop_last": None} stats = {"total_checks": 0, "total_pushes": 0, "total_failures": 0} -op_lock = threading.Lock() +op_lock = threading.Lock() # 整轮同步互斥(同一轮只跑一次) +stats_lock = threading.Lock() # 保护 stats 计数 +log_lock = threading.Lock() # 保护日志文件写入 +repo_busy_lock = threading.Lock() # 保护 repo_in_progress +repo_in_progress = set() # 正在同步的仓库名(按仓库防重入) # ---------------- 基础工具 ---------------- def log(msg): - try: - DATA_DIR.mkdir(parents=True, exist_ok=True) - with open(LOG_FILE, "a", encoding="utf-8") as f: - f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n") - except Exception: - pass + with log_lock: + try: + DATA_DIR.mkdir(parents=True, exist_ok=True) + with open(LOG_FILE, "a", encoding="utf-8") as f: + f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n") + except Exception: + pass + + +def bump_stats(**kw): + """线程安全地累加统计计数""" + with stats_lock: + for k, v in kw.items(): + stats[k] = stats.get(k, 0) + v + save_stats() def write_gitcreds(): @@ -475,12 +511,24 @@ def fingerprint(repo): def sync_repo(name): - """返回 'changed' / 'same' / 'error:...' """ + """返回 'changed' / 'same' / 'error:...';同一仓库并发调用时后到者直接返回""" if not valid_name(name): return "error: 非法仓库名" acc = active_account() if not acc: return "error: 没有可用的 GitHub 账号,请到设置页添加" + with repo_busy_lock: + if name in repo_in_progress: + return "error: 该仓库正在同步中,请稍候" + repo_in_progress.add(name) + try: + return _sync_repo_locked(name, acc) + finally: + with repo_busy_lock: + repo_in_progress.discard(name) + + +def _sync_repo_locked(name, acc): repo = REPO_DIR / f"{name}.git" try: if not repo.exists(): @@ -527,25 +575,41 @@ def sync_repo(name): def run_sync(repos=None): + """跑一轮同步:多个项目按设置并行(默认 3 个同时)""" with op_lock: if state["syncing"]: return False, "已有同步任务正在进行" state["syncing"] = True try: repos = repos if repos is not None else load_repos() + results = {} + workers = sync_workers() + + def worker(name): + try: + results[name] = sync_repo(name) + except Exception as e: + results[name] = f"error: {e}" + + if repos: + with ThreadPoolExecutor(max_workers=workers, + thread_name_prefix="ghs-sync") as ex: + # map 会阻塞到全部完成;异常已在 worker 内捕获 + list(ex.map(worker, repos)) errs, changed = [], 0 for r in repos: - res = sync_repo(r) + res = results.get(r, "error: 未执行") if res.startswith("error"): errs.append(f"{r} {res}") - stats["total_failures"] += 1 elif res == "changed": changed += 1 - stats["total_pushes"] += 1 + if errs: + bump_stats(total_failures=len(errs)) + if changed: + bump_stats(total_pushes=changed) state["last_check"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") state["checks"] += 1 - stats["total_checks"] += 1 - save_stats() + bump_stats(total_checks=1) if errs: state["last_result"] = "失败: " + "; ".join(errs)[:200] return False, state["last_result"] @@ -651,6 +715,7 @@ def global_status(ui_url=None): "syncing": state["syncing"], "uptime": fmt_uptime(now - START_TS), "interval": sync_interval(), + "max_workers": sync_workers(), "last_check": state["last_check"], "next_in": next_in, "checks": state["checks"], @@ -863,7 +928,7 @@ async function refresh(){ const s=await jget('api/state'); document.getElementById('sw').checked=s.enabled; document.getElementById('swlabel').textContent='自动同步:'+(s.enabled?'开':'关'); - document.getElementById('swsub').textContent='开启后每 '+(s.interval||10)+' 秒检查一次 Gitea,有新提交自动推送到 GitHub(间隔可在设置页修改)'; + document.getElementById('swsub').textContent='开启后每 '+(s.interval||10)+' 秒检查一次 Gitea,'+(s.max_workers||1)+' 个项目并行同步,有新提交自动推送到 GitHub(间隔/并发可在设置页调整)'; const p=document.getElementById('pill'); p.textContent=s.enabled?'运行中':'已停止';p.className='pill '+(s.enabled?'on':'off'); document.getElementById('sEnabled').textContent=s.enabled?('开启('+(s.interval||10)+'秒/次)'):'已停止'; @@ -928,7 +993,7 @@ async function refresh(){ document.getElementById('sPushes').textContent=s.stats.total_pushes; document.getElementById('sFails').textContent=s.stats.total_failures; document.getElementById('sSize').textContent=s.total_size_mb+' MB'; - document.getElementById('sTarget').textContent='同步目标: '+(s.account?s.account.gh_user+'('+s.account.method+')':'未配置')+' · Gitea '+(s.gitea_online?'在线 ✓':'离线 ✗'); + document.getElementById('sTarget').textContent='同步目标: '+(s.account?s.account.gh_user+'('+s.account.method+')':'未配置')+' · '+(s.max_workers||1)+' 个并行 · Gitea '+(s.gitea_online?'在线 ✓':'离线 ✗'); document.getElementById('tb').innerHTML=s.repos.map(r=>{ const st=r.status==='已同步'?'