feat: v1.3.0 — 多项目同时同步(并行度可调 1~8,默认 3)

- run_sync 改线程池并行:每轮最多 N 个项目同时克隆/推送,项目多时
  显著缩短单轮耗时(实测 6 个仓库 3 并发 3 倍加速)
- 按仓库防重入:同一仓库同时只允许一个同步,后到者提示
  「该仓库正在同步中,请稍候」,避免并发 git 操作损坏镜像
- 状态页单仓库「同步」按钮不再被整轮同步全局拦截(旧版报
  「已有同步任务正在进行」),与进行中的整轮真正并行
- 新增「同时同步项目数」设置(1~8),设置页可调,立即生效
- 统计计数与日志写入加锁,修正并行下的计数漂移与日志交错风险
- 首页/状态页显示当前并行数;manifest 1.2.0 → 1.3.0
This commit is contained in:
cbb
2026-10-13 20:14:57 +08:00
parent 870612ff93
commit 1acd60894f
3 changed files with 130 additions and 39 deletions
+127 -38
View File
@@ -20,6 +20,7 @@ import subprocess
import threading import threading
import time import time
import uuid import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path from pathlib import Path
@@ -50,6 +51,9 @@ LEGACY_GITEA_USER = "cbb" # 旧版硬编码的 Gitea 用户名,仅作
DEFAULT_INTERVAL = 10 # 同步间隔默认值(秒) DEFAULT_INTERVAL = 10 # 同步间隔默认值(秒)
MIN_INTERVAL = 5 # 最小同步间隔(秒) MIN_INTERVAL = 5 # 最小同步间隔(秒)
MAX_INTERVAL = 86400 # 最大同步间隔(秒,1 天) MAX_INTERVAL = 86400 # 最大同步间隔(秒,1 天)
DEFAULT_WORKERS = 3 # 同时同步的项目数默认值
MIN_WORKERS = 1
MAX_WORKERS = 8
PANEL_PORT = int(os.environ.get("PANEL_PORT", "8899")) PANEL_PORT = int(os.environ.get("PANEL_PORT", "8899"))
START_TS = time.time() START_TS = time.time()
NAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,100}$") NAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,100}$")
@@ -62,7 +66,8 @@ ENV["GIT_TERMINAL_PROMPT"] = "0"
# ---------------- 应用设置(settings.json) ---------------- # ---------------- 应用设置(settings.json) ----------------
cfg = {"gitea_user": LEGACY_GITEA_USER, "gitea_pass": "", "auto_create": True, 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): def norm_gitea_host(h):
@@ -120,6 +125,22 @@ def valid_interval(v):
return False 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(): def load_settings():
try: try:
d = json.loads(SETTINGS_FILE.read_text(encoding="utf-8")) 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)) cfg["auto_create"] = bool(d.get("auto_create", True))
if valid_interval(d.get("interval")): if valid_interval(d.get("interval")):
cfg["interval"] = int(d["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")) h = norm_gitea_host(d.get("gitea_host"))
if h: if h:
cfg["gitea_host"] = h cfg["gitea_host"] = h
@@ -241,17 +264,30 @@ def scrub(s, acc):
state = {"enabled": True, "syncing": False, "last_check": None, "checks": 0, state = {"enabled": True, "syncing": False, "last_check": None, "checks": 0,
"last_result": "尚未运行", "loop_last": None} "last_result": "尚未运行", "loop_last": None}
stats = {"total_checks": 0, "total_pushes": 0, "total_failures": 0} 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): def log(msg):
try: with log_lock:
DATA_DIR.mkdir(parents=True, exist_ok=True) try:
with open(LOG_FILE, "a", encoding="utf-8") as f: DATA_DIR.mkdir(parents=True, exist_ok=True)
f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n") with open(LOG_FILE, "a", encoding="utf-8") as f:
except Exception: f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n")
pass 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(): def write_gitcreds():
@@ -475,12 +511,24 @@ def fingerprint(repo):
def sync_repo(name): def sync_repo(name):
"""返回 'changed' / 'same' / 'error:...' """ """返回 'changed' / 'same' / 'error:...';同一仓库并发调用时后到者直接返回"""
if not valid_name(name): if not valid_name(name):
return "error: 非法仓库名" return "error: 非法仓库名"
acc = active_account() acc = active_account()
if not acc: if not acc:
return "error: 没有可用的 GitHub 账号,请到设置页添加" 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" repo = REPO_DIR / f"{name}.git"
try: try:
if not repo.exists(): if not repo.exists():
@@ -527,25 +575,41 @@ def sync_repo(name):
def run_sync(repos=None): def run_sync(repos=None):
"""跑一轮同步:多个项目按设置并行(默认 3 个同时)"""
with op_lock: with op_lock:
if state["syncing"]: if state["syncing"]:
return False, "已有同步任务正在进行" return False, "已有同步任务正在进行"
state["syncing"] = True state["syncing"] = True
try: try:
repos = repos if repos is not None else load_repos() 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 errs, changed = [], 0
for r in repos: for r in repos:
res = sync_repo(r) res = results.get(r, "error: 未执行")
if res.startswith("error"): if res.startswith("error"):
errs.append(f"{r} {res}") errs.append(f"{r} {res}")
stats["total_failures"] += 1
elif res == "changed": elif res == "changed":
changed += 1 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["last_check"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
state["checks"] += 1 state["checks"] += 1
stats["total_checks"] += 1 bump_stats(total_checks=1)
save_stats()
if errs: if errs:
state["last_result"] = "失败: " + "; ".join(errs)[:200] state["last_result"] = "失败: " + "; ".join(errs)[:200]
return False, state["last_result"] return False, state["last_result"]
@@ -651,6 +715,7 @@ def global_status(ui_url=None):
"syncing": state["syncing"], "syncing": state["syncing"],
"uptime": fmt_uptime(now - START_TS), "uptime": fmt_uptime(now - START_TS),
"interval": sync_interval(), "interval": sync_interval(),
"max_workers": sync_workers(),
"last_check": state["last_check"], "last_check": state["last_check"],
"next_in": next_in, "next_in": next_in,
"checks": state["checks"], "checks": state["checks"],
@@ -863,7 +928,7 @@ async function refresh(){
const s=await jget('api/state'); const s=await jget('api/state');
document.getElementById('sw').checked=s.enabled; document.getElementById('sw').checked=s.enabled;
document.getElementById('swlabel').textContent='自动同步:'+(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'); const p=document.getElementById('pill');
p.textContent=s.enabled?'运行中':'已停止';p.className='pill '+(s.enabled?'on':'off'); p.textContent=s.enabled?'运行中':'已停止';p.className='pill '+(s.enabled?'on':'off');
document.getElementById('sEnabled').textContent=s.enabled?('开启('+(s.interval||10)+'秒/次)'):'已停止'; 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('sPushes').textContent=s.stats.total_pushes;
document.getElementById('sFails').textContent=s.stats.total_failures; document.getElementById('sFails').textContent=s.stats.total_failures;
document.getElementById('sSize').textContent=s.total_size_mb+' MB'; 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=>{ document.getElementById('tb').innerHTML=s.repos.map(r=>{
const st=r.status==='已同步'?'<td class="ok">✓ 已同步</td>':'<td class="warn">⏳ 未同步</td>'; const st=r.status==='已同步'?'<td class="ok">✓ 已同步</td>':'<td class="warn">⏳ 未同步</td>';
const c=r.last_commit?`<div class="commit">${esc(r.last_commit.hash)} ${esc(r.last_commit.msg)} · ${esc(r.last_commit.author)} · ${esc(r.last_commit.time)}</div>`:'<div class="commit">—</div>'; const c=r.last_commit?`<div class="commit">${esc(r.last_commit.hash)} ${esc(r.last_commit.msg)} · ${esc(r.last_commit.author)} · ${esc(r.last_commit.time)}</div>`:'<div class="commit">—</div>';
@@ -1071,6 +1136,14 @@ PAGE_SETTINGS = """
<div class="swsub" id="intervalTip"></div> <div class="swsub" id="intervalTip"></div>
</div> </div>
</div> </div>
<div class="optrow">
<input id="syncWorkers" type="number" min="1" max="8" step="1" class="numin">
<button class="b-mini" id="btnWorkers" onclick="saveWorkers()">保存并发</button>
<div class="opttext">
<div class="swlabel">同时同步项目数</div>
<div class="swsub" id="workersTip"></div>
</div>
</div>
</div> </div>
<div class="msg" id="msgCfg"></div> <div class="msg" id="msgCfg"></div>
</div> </div>
@@ -1090,6 +1163,9 @@ async function refresh(){
const iv=document.getElementById('syncInterval'); const iv=document.getElementById('syncInterval');
iv.value=s.interval||10;iv.min=s.min_interval||5;iv.max=s.max_interval||86400; iv.value=s.interval||10;iv.min=s.min_interval||5;iv.max=s.max_interval||86400;
document.getElementById('intervalTip').textContent='每隔 '+fmtInterval(s.interval||10)+' 检查一次 Gitea 是否有新提交('+(s.min_interval||5)+'~'+(s.max_interval||86400)+' 秒),改完立即生效'; document.getElementById('intervalTip').textContent='每隔 '+fmtInterval(s.interval||10)+' 检查一次 Gitea 是否有新提交('+(s.min_interval||5)+'~'+(s.max_interval||86400)+' 秒),改完立即生效';
const wk=document.getElementById('syncWorkers');
wk.value=s.max_workers||3;wk.min=s.min_workers||1;wk.max=s.max_workers_limit||8;
document.getElementById('workersTip').textContent='每轮最多 '+(s.max_workers||3)+' 个项目同时克隆/推送(1~'+(s.max_workers_limit||8)+'),项目多时能显著加快;网络或 NAS 负载高就调小';
const cur=(s.accounts||[]).find(a=>a.id===s.active); const cur=(s.accounts||[]).find(a=>a.id===s.active);
// 当前账号无 Token 时,顶部提示为什么不能自动建库、怎么解锁 // 当前账号无 Token 时,顶部提示为什么不能自动建库、怎么解锁
document.getElementById('accBanner').innerHTML=(cur&&!cur.has_token) document.getElementById('accBanner').innerHTML=(cur&&!cur.has_token)
@@ -1188,6 +1264,15 @@ async function saveInterval(){
}catch(e){showMsg('msgCfg','请求失败: '+e,false)} }catch(e){showMsg('msgCfg','请求失败: '+e,false)}
finally{b.disabled=false;b.textContent='保存间隔';refresh();} finally{b.disabled=false;b.textContent='保存间隔';refresh();}
} }
async function saveWorkers(){
const v=parseInt(document.getElementById('syncWorkers').value,10);
const b=document.getElementById('btnWorkers');b.disabled=true;b.textContent='⏳ 保存中…';
try{
const r=await (await fetch('api/settings/save',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({max_workers:v})})).json();
showMsg('msgCfg',r.msg||JSON.stringify(r),!!r.ok);
}catch(e){showMsg('msgCfg','请求失败: '+e,false)}
finally{b.disabled=false;b.textContent='保存并发';refresh();}
}
refresh(); refresh();
</script> </script>
""" """
@@ -1242,6 +1327,7 @@ class Handler(BaseHTTPRequestHandler):
"last_check": state["last_check"], "checks": state["checks"], "last_check": state["last_check"], "checks": state["checks"],
"last_result": state["last_result"], "last_result": state["last_result"],
"interval": sync_interval(), "interval": sync_interval(),
"max_workers": sync_workers(),
"repo_count": len(repos), "repo_count": len(repos),
"synced_count": sum(1 for r in repos if state_path(acc, r).exists()), "synced_count": sum(1 for r in repos if state_path(acc, r).exists()),
"uptime": fmt_uptime(time.time() - START_TS), "uptime": fmt_uptime(time.time() - START_TS),
@@ -1257,6 +1343,9 @@ class Handler(BaseHTTPRequestHandler):
"interval": sync_interval(), "interval": sync_interval(),
"min_interval": MIN_INTERVAL, "min_interval": MIN_INTERVAL,
"max_interval": MAX_INTERVAL, "max_interval": MAX_INTERVAL,
"max_workers": sync_workers(),
"min_workers": MIN_WORKERS,
"max_workers_limit": MAX_WORKERS,
"gitea_host": gitea_host(), "gitea_host": gitea_host(),
"gitea_url": gitea_ui(self.headers.get("Host")), "gitea_url": gitea_ui(self.headers.get("Host")),
"active": active_id, "active": active_id,
@@ -1297,6 +1386,10 @@ class Handler(BaseHTTPRequestHandler):
if not valid_interval(data.get("interval")): if not valid_interval(data.get("interval")):
self._json({"ok": False, "msg": f"同步间隔须为 {MIN_INTERVAL} ~ {MAX_INTERVAL} 之间的整数(秒)"}) self._json({"ok": False, "msg": f"同步间隔须为 {MIN_INTERVAL} ~ {MAX_INTERVAL} 之间的整数(秒)"})
return return
if "max_workers" in data:
if not valid_workers(data.get("max_workers")):
self._json({"ok": False, "msg": f"同时同步项目数须为 {MIN_WORKERS} ~ {MAX_WORKERS} 之间的整数"})
return
# 预校验:新地址先探连通性;涉及账号的改动再用新凭据调一次 /user 验证 # 预校验:新地址先探连通性;涉及账号的改动再用新凭据调一次 /user 验证
trial = dict(cfg) trial = dict(cfg)
if new_user is not None: if new_user is not None:
@@ -1343,6 +1436,11 @@ class Handler(BaseHTTPRequestHandler):
cfg["interval"] = int(data.get("interval")) cfg["interval"] = int(data.get("interval"))
if cfg["interval"] != old_iv: if cfg["interval"] != old_iv:
log(f"同步间隔已调整: {old_iv}s -> {cfg['interval']}s") log(f"同步间隔已调整: {old_iv}s -> {cfg['interval']}s")
if "max_workers" in data:
old_w = sync_workers()
cfg["max_workers"] = int(data.get("max_workers"))
if cfg["max_workers"] != old_w:
log(f"同时同步项目数已调整: {old_w} -> {cfg['max_workers']}")
save_settings() save_settings()
notes = [] notes = []
if new_host is not None: if new_host is not None:
@@ -1443,26 +1541,18 @@ class Handler(BaseHTTPRequestHandler):
if not valid_name(name): if not valid_name(name):
self._json({"ok": False, "msg": "非法仓库名"}) self._json({"ok": False, "msg": "非法仓库名"})
return return
with op_lock: # 不再被整轮同步全局拦截:只禁止同一仓库重复同步(sync_repo 内部按仓库防重入),
if state["syncing"]: # 与正在进行的整轮同步互不阻塞,实现多项目同时同步
self._json({"ok": False, "msg": "已有同步任务正在进行"}) res = sync_repo(name)
return state["last_check"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
state["syncing"] = True if res.startswith("error"):
try: bump_stats(total_failures=1)
res = sync_repo(name) self._json({"ok": False, "msg": f"{name} {res}"})
state["last_check"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") elif res == "changed":
if res.startswith("error"): bump_stats(total_pushes=1)
stats["total_failures"] += 1 self._json({"ok": True, "msg": f"{name} 已同步到 GitHub"})
save_stats() else:
self._json({"ok": False, "msg": f"{name} {res}"}) self._json({"ok": True, "msg": f"{name} 已是最新,无需同步"})
elif res == "changed":
stats["total_pushes"] += 1
save_stats()
self._json({"ok": True, "msg": f"{name} 已同步到 GitHub"})
else:
self._json({"ok": True, "msg": f"{name} 已是最新,无需同步"})
finally:
state["syncing"] = False
elif path == "/api/repos/add": elif path == "/api/repos/add":
name = (data.get("name") or "").strip() name = (data.get("name") or "").strip()
if not valid_name(name): if not valid_name(name):
@@ -1493,8 +1583,7 @@ class Handler(BaseHTTPRequestHandler):
if res.startswith("error"): if res.startswith("error"):
self._json({"ok": False, "msg": f"已加入列表,但首次同步失败:{res}"}) self._json({"ok": False, "msg": f"已加入列表,但首次同步失败:{res}"})
elif res == "changed": elif res == "changed":
stats["total_pushes"] += 1 bump_stats(total_pushes=1)
save_stats()
self._json({"ok": True, "msg": f"{name} 添加成功,已完成首次同步"}) self._json({"ok": True, "msg": f"{name} 添加成功,已完成首次同步"})
else: else:
self._json({"ok": True, "msg": f"{name} 添加成功,内容已是最新"}) self._json({"ok": True, "msg": f"{name} 添加成功,内容已是最新"})
+1 -1
View File
@@ -1,5 +1,5 @@
appname = GiteaHubSync appname = GiteaHubSync
version = 1.2.0 version = 1.3.0
display_name = Gitea→GitHub 同步助手 display_name = Gitea→GitHub 同步助手
desc = 每次推送到飞牛 Gitea 的代码,10 秒内自动镜像到 GitHub 的 7 个私有仓库。内置控制面板:自动同步开关、立即同步、仓库状态、同步日志,一键打开 Gitea。数据全部保存在 NAS 本地。 desc = 每次推送到飞牛 Gitea 的代码,10 秒内自动镜像到 GitHub 的 7 个私有仓库。内置控制面板:自动同步开关、立即同步、仓库状态、同步日志,一键打开 Gitea。数据全部保存在 NAS 本地。
platform = all platform = all
+2
View File
@@ -14,6 +14,8 @@ NAS 重启自动恢复。控制面板带自动同步开关、立即同步、仓
- **多 GitHub 账号**:设置页可添加多个 GitHub 账号(用户名 + Personal Access Token), - **多 GitHub 账号**:设置页可添加多个 GitHub 账号(用户名 + Personal Access Token),
一键切换推送目标;每个账号独立记录同步进度,切回原账号不会重复推送 一键切换推送目标;每个账号独立记录同步进度,切回原账号不会重复推送
- **同步间隔可调**:设置页可改检查间隔(5 秒 ~ 1 天,默认 10 秒),改完立即生效 - **同步间隔可调**:设置页可改检查间隔(5 秒 ~ 1 天,默认 10 秒),改完立即生效
- **多项目并行同步**:每轮最多 N 个项目同时克隆/推送(默认 3,可调 1~8);
在状态页点单个仓库的「同步」也不再被整轮同步阻塞(同一仓库有防重入)
- **Gitea 地址可改**:设置页可改同步引擎访问的 Gitea 地址(默认 `http://127.0.0.1:3000`), - **Gitea 地址可改**:设置页可改同步引擎访问的 Gitea 地址(默认 `http://127.0.0.1:3000`),
保存前自动探测连通性;改地址后已有镜像的拉取地址自动更新,「打开 Gitea」按钮随访问地址自适应 保存前自动探测连通性;改地址后已有镜像的拉取地址自动更新,「打开 Gitea」按钮随访问地址自适应
- **自动创建仓库**:Gitea 上的新仓库加入同步时,若 GitHub 上不存在同名仓库, - **自动创建仓库**:Gitea 上的新仓库加入同步时,若 GitHub 上不存在同名仓库,