From eb4ff5197d7ba44918e8ffc4fc0425e14c113f76 Mon Sep 17 00:00:00 2001 From: TermiNexus Date: Fri, 23 Oct 2026 13:45:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(GitMirror):=20=E6=96=B0=E5=A2=9E=20Gitea?= =?UTF-8?q?=20=E5=88=B0=20GitHub=20=E5=AE=9E=E6=97=B6=E9=95=9C=E5=83=8F?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=EF=BC=88webhook+=E5=AE=9A=E6=97=B6=E5=85=9C?= =?UTF-8?q?=E5=BA=95+LFS=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- GitMirror/.gitignore | 4 + GitMirror/app.py | 327 ++++++++++++++++++++++++++ GitMirror/config.example.ini | 28 +++ GitMirror/mirror.py | 434 +++++++++++++++++++++++++++++++++++ GitMirror/requirements.txt | 2 + GitMirror/run.ps1 | 16 ++ 6 files changed, 811 insertions(+) create mode 100644 GitMirror/.gitignore create mode 100644 GitMirror/app.py create mode 100644 GitMirror/config.example.ini create mode 100644 GitMirror/mirror.py create mode 100644 GitMirror/requirements.txt create mode 100644 GitMirror/run.ps1 diff --git a/GitMirror/.gitignore b/GitMirror/.gitignore new file mode 100644 index 00000000..97adf6ed --- /dev/null +++ b/GitMirror/.gitignore @@ -0,0 +1,4 @@ +config.ini +data/ +__pycache__/ +*.pyc diff --git a/GitMirror/app.py b/GitMirror/app.py new file mode 100644 index 00000000..f83a4f2d --- /dev/null +++ b/GitMirror/app.py @@ -0,0 +1,327 @@ +"""GitMirror 服务入口 + +把 git.gvsds.com/GVSDS/* 实时镜像到 github.com/GVSDS/*： + +- POST /hooks/gitea Gitea 组织级 webhook（HMAC-SHA256 验签）→ 入队同步 +- POST /backfill 手动全量回填（X-Admin-Token） +- POST /sync/ 手动同步单仓（X-Admin-Token） +- GET /status 镜像状态（X-Admin-Token） +- GET /healthz 健康检查 + +后台：工作线程池执行镜像；定时线程按 SyncIntervalMinutes 做全量兜底。 + +用法： + python app.py serve + python app.py backfill + python app.py sync + python app.py register-webhook --url https://mirror.example.com/hooks/gitea + python app.py status +""" + +from __future__ import annotations + +import argparse +import hashlib +import hmac +import json +import logging +import os +import queue +import sys +import threading +import time + +from flask import Flask, jsonify, request + +from mirror import ( + GiteaClient, + GithubClient, + MirrorConfig, + MirrorError, + MirrorManager, + MirrorStore, +) + +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +CONFIG_PATH = os.environ.get('GITMIRROR_CONFIG') or os.path.join(BASE_DIR, 'config.ini') + +logger = logging.getLogger('gitmirror') + +# --------------------------------------------------------------------------- +# 全局组件（CLI 与 HTTP 共用） +# --------------------------------------------------------------------------- + +Cfg = MirrorConfig(CONFIG_PATH) +Gitea = GiteaClient(Cfg.GiteaBaseUrl, Cfg.GiteaOrg, Cfg.GiteaToken) +Github = GithubClient(Cfg.GithubOrg, Cfg.GithubToken) +Store = MirrorStore(Cfg.DbFile) +Manager = MirrorManager(Cfg, Gitea, Github, Store) + + +class SyncWorker: + """去重的同步队列 + 工作线程（每仓串行，失败退避重试）""" + + def __init__(self, Workers: int): + self.Queue: queue.Queue[tuple[str, dict | None]] = queue.Queue() + self._Pending: set[str] = set() + self._Guard = threading.Lock() + self._Workers = Workers + self._Threads: list[threading.Thread] = [] + + def Start(self) -> None: + for Index in range(self._Workers): + Thread = threading.Thread(target=self._Run, name=f'GitMirrorWorker-{Index}', daemon=True) + Thread.start() + self._Threads.append(Thread) + + def Put(self, RepoName: str, Repo: dict | None = None) -> bool: + """入队；已在队列中的同名仓库不重复入队""" + with self._Guard: + if RepoName in self._Pending: + return False + self._Pending.add(RepoName) + self.Queue.put((RepoName, Repo)) + return True + + def PendingCount(self) -> int: + with self._Guard: + return len(self._Pending) + + def _Run(self) -> None: + while True: + RepoName, Repo = self.Queue.get() + try: + self._SyncWithRetry(RepoName, Repo) + except Exception: # 兜底，避免工作线程退出 + logger.exception('同步任务异常：%s', RepoName) + finally: + with self._Guard: + self._Pending.discard(RepoName) + self.Queue.task_done() + + def _SyncWithRetry(self, RepoName: str, Repo: dict | None) -> None: + Delays = [5, 15, 45] + LastError: Exception | None = None + for Attempt, Delay in enumerate([0] + Delays): + if Delay: + time.sleep(Delay) + try: + Info = Repo or Gitea.GetRepo(RepoName) + if not Info: + raise MirrorError(f'Gitea 中不存在仓库 {RepoName}') + Manager.Sync(Info) + return + except MirrorError as Error: + LastError = Error + logger.warning('同步失败（第 %s 次）%s：%s', Attempt + 1, RepoName, Error) + except Exception as Error: # 网络异常等 + LastError = Error + logger.warning('同步异常（第 %s 次）%s：%s', Attempt + 1, RepoName, Error) + Store.SetStatus(RepoName, Status='failed', LastError=str(LastError)[:1000]) + logger.error('同步最终失败 %s：%s', RepoName, LastError) + + +Worker = SyncWorker(Cfg.Concurrency) + + +# --------------------------------------------------------------------------- +# 任务：全量回填 / 删除处理 +# --------------------------------------------------------------------------- + +def Backfill() -> dict: + """枚举组织全部仓库并入队（幂等）""" + Repos = Gitea.ListRepos() + Queued = 0 + for Repo in Repos: + Name = Repo.get('name') or '' + if not Name: + continue + Store.SetStatus(Name, SourcePrivate=int(bool(Repo.get('private', True)))) + if Worker.Put(Name, Repo): + Queued += 1 + logger.info('全量回填：共 %s 个仓库，新入队 %s 个', len(Repos), Queued) + return {'Total': len(Repos), 'Queued': Queued} + + +def HandleRepoDeleted(RepoName: str) -> None: + try: + Manager.HandleRepoDeleted(RepoName) + except Exception: + logger.exception('处理源仓库删除失败：%s', RepoName) + + +class Scheduler: + """定时兜底：每 SyncIntervalMinutes 一次全量回填（防 webhook 丢失）""" + + def __init__(self, IntervalMinutes: int, RunOnStart: bool): + self.Interval = max(1, IntervalMinutes) * 60 + self.RunOnStart = RunOnStart + self._Stop = threading.Event() + + def Start(self) -> None: + threading.Thread(target=self._Run, name='GitMirrorScheduler', daemon=True).start() + + def _Run(self) -> None: + if self.RunOnStart: + try: + Backfill() + except Exception: + logger.exception('启动回填失败') + while not self._Stop.wait(self.Interval): + try: + Backfill() + except Exception: + logger.exception('定时回填失败') + + +# --------------------------------------------------------------------------- +# Flask +# --------------------------------------------------------------------------- + +App = Flask(__name__) + + +def _VerifySignature(RawBody: bytes, Signature: str | None) -> bool: + if not Signature: + return False + Expected = hmac.new(Cfg.WebhookSecret.encode('utf-8'), RawBody, hashlib.sha256).hexdigest() + return hmac.compare_digest(Expected, Signature) + + +def _IsAdmin() -> bool: + return bool(Cfg.AdminToken) and request.headers.get('X-Admin-Token') == Cfg.AdminToken + + +@App.get('/healthz') +def Healthz(): + return jsonify({'ok': True, 'pending': Worker.PendingCount(), 'gitea': Cfg.GiteaOrg, 'github': Cfg.GithubOrg}) + + +@App.post('/hooks/gitea') +def GiteaHook(): + Raw = request.get_data() or b'' + if not _VerifySignature(Raw, request.headers.get('X-Gitea-Signature')): + return jsonify({'code': 1, 'msg': '签名校验失败'}), 401 + Event = request.headers.get('X-Gitea-Event', '') + Payload = request.get_json(silent=True) or {} + Repo = Payload.get('repository') or {} + RepoName = (Repo.get('name') or '').strip() + if not RepoName: + return jsonify({'code': 0, 'msg': 'ignored'}), 202 + + Action = (Payload.get('action') or '').strip() + if Event == 'repository' and Action == 'deleted': + threading.Thread(target=HandleRepoDeleted, args=(RepoName,), daemon=True).start() + return jsonify({'code': 0, 'msg': 'repo deleted handled'}), 202 + + Worker.Put(RepoName, Repo) + logger.info('webhook 入队 event=%s action=%s repo=%s', Event, Action, RepoName) + return jsonify({'code': 0, 'msg': 'queued'}), 202 + + +@App.post('/backfill') +def BackfillRoute(): + if not _IsAdmin(): + return jsonify({'code': 1, 'msg': 'unauthorized'}), 401 + return jsonify({'code': 0, 'data': Backfill()}) + + +@App.post('/sync/') +def SyncRoute(RepoName: str): + if not _IsAdmin(): + return jsonify({'code': 1, 'msg': 'unauthorized'}), 401 + Queued = Worker.Put(RepoName) + return jsonify({'code': 0, 'data': {'RepoName': RepoName, 'Queued': Queued}}) + + +@App.get('/status') +def StatusRoute(): + if not _IsAdmin(): + return jsonify({'code': 1, 'msg': 'unauthorized'}), 401 + return jsonify({'code': 0, 'data': {'Pending': Worker.PendingCount(), 'Repos': Store.List()}}) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def _SetupLogging() -> None: + logging.basicConfig( + level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(name)s: %(message)s', + ) + + +def _CmdServe() -> None: + Worker.Start() + Scheduler(Cfg.SyncIntervalMinutes, Cfg.BackfillOnStart).Start() + logger.info('GitMirror 启动：%s -> github.com/%s，每 %s 分钟兜底，webhook secret 已就绪', + f'{Cfg.GiteaBaseUrl}/{Cfg.GiteaOrg}', Cfg.GithubOrg, Cfg.SyncIntervalMinutes) + logger.info('监听 %s:%s（webhook 路径 /hooks/gitea）', Cfg.ServerHost, Cfg.ServerPort) + App.run(host=Cfg.ServerHost, port=Cfg.ServerPort, threaded=True) + + +def _CmdBackfill() -> None: + Result = Backfill() + Worker.Start() + logger.info('已入队 %s 个仓库，等待完成（Ctrl+C 可提前退出）', Result['Queued']) + while Worker.PendingCount() > 0: + time.sleep(1) + logger.info('回填结束') + + +def _CmdSync(RepoName: str) -> None: + Info = Gitea.GetRepo(RepoName) + if not Info: + raise MirrorError(f'Gitea 中不存在仓库 {RepoName}') + Manager.Sync(Info) + logger.info('同步完成：%s', RepoName) + + +def _CmdRegisterWebhook(Url: str) -> None: + Action = Gitea.RegisterWebhook(Url, Cfg.WebhookSecret) + logger.info('组织 webhook 已 %s：%s（events: push/create/delete/repository）', Action, Url) + logger.info('Webhook Secret：%s', Cfg.WebhookSecret) + + +def _CmdStatus() -> None: + print(json.dumps({'Pending': Worker.PendingCount(), 'Repos': Store.List()}, ensure_ascii=False, indent=2)) + + +def main() -> int: + _SetupLogging() + Parser = argparse.ArgumentParser(description='Gitea → GitHub 仓库镜像服务') + Sub = Parser.add_subparsers(dest='command') + + Sub.add_parser('serve', help='启动 HTTP 服务（webhook + 定时兜底）') + Sub.add_parser('backfill', help='全量回填一次') + SyncParser = Sub.add_parser('sync', help='同步单个仓库') + SyncParser.add_argument('repo') + HookParser = Sub.add_parser('register-webhook', help='注册/更新 Gitea 组织级 webhook') + HookParser.add_argument('--url', required=True, help='本服务对外地址，如 https://mirror.example.com/hooks/gitea') + Sub.add_parser('status', help='打印镜像状态') + + Args = Parser.parse_args() + Command = Args.command or 'serve' + try: + if Command == 'serve': + _CmdServe() + elif Command == 'backfill': + _CmdBackfill() + elif Command == 'sync': + _CmdSync(Args.repo) + elif Command == 'register-webhook': + _CmdRegisterWebhook(Args.url) + elif Command == 'status': + _CmdStatus() + else: + Parser.print_help() + return 2 + except MirrorError as Error: + logger.error('%s', Error) + return 1 + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/GitMirror/config.example.ini b/GitMirror/config.example.ini new file mode 100644 index 00000000..0dd07996 --- /dev/null +++ b/GitMirror/config.example.ini @@ -0,0 +1,28 @@ +[gitea] +BaseUrl = https://git.gvsds.com +OrgName = GVSDS +# 留空则从 SecretsIni 的 [gitea] API_TOKEN 读取 +ApiToken = +SecretsIni = D:\SMANAGE\secrets.ini + +[github] +OrgName = GVSDS +# Fine-grained PAT：需 Contents 读写 + Administration 读写（自动建仓/归档） +Token = +# 源仓库删除时：archive=归档 | delete=真删 | ignore=忽略 +OnRepoDeleted = archive + +[mirror] +WorkDir = D:\SMANAGE\GitMirror\data +SyncIntervalMinutes = 15 +BackfillOnStart = true +LfsEnabled = true +Concurrency = 1 +# 留空则自动生成并写入 WorkDir/runtime.json +WebhookSecret = + +[server] +Host = 0.0.0.0 +Port = 8600 +# 留空则自动生成并写入 WorkDir/runtime.json（请求头 X-Admin-Token） +AdminToken = diff --git a/GitMirror/mirror.py b/GitMirror/mirror.py new file mode 100644 index 00000000..df2b889b --- /dev/null +++ b/GitMirror/mirror.py @@ -0,0 +1,434 @@ +"""模块：Gitea → GitHub 仓库镜像核心 + +职责： +- MirrorConfig 读取 config.ini，自动生成并持久化 webhook secret / admin token +- GiteaClient 枚举组织仓库、注册组织级 webhook +- GithubClient 确保目标仓库存在（不存在则创建）、归档/删除 +- MirrorManager bare 镜像的 clone/fetch/push（含 LFS）并记录状态 +- MirrorStore SQLite 状态表（断点、重试、可观测） + +安全： +- 令牌只写入工作目录下的 git-credentials（等价 git credential store）， + 命令仅引用该文件路径，令牌不出现在 argv 中； +- 所有日志输出都会把令牌替换为 ***。 +""" + +from __future__ import annotations + +import configparser +import datetime +import json +import logging +import os +import secrets as py_secrets +import sqlite3 +import subprocess +import threading + +import requests + +logger = logging.getLogger(__name__) + +GITHUB_API = 'https://api.github.com' +GITHUB_AUTH_USER = 'x-access-token' + + +class MirrorError(Exception): + """镜像流程中的可预期错误（Status 为上游 HTTP 状态码，便于区分 404）""" + + def __init__(self, Message: str, Status: int | None = None): + super().__init__(Message) + self.Status = Status + + +def _Now() -> str: + """本地时间字符串（秒级）""" + return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + +# --------------------------------------------------------------------------- +# 配置 +# --------------------------------------------------------------------------- + +class MirrorConfig: + """config.ini + 自动生成的运行期密钥""" + + def __init__(self, Path: str): + Parser = configparser.ConfigParser(interpolation=None) + if not Parser.read(Path, encoding='utf-8'): + raise MirrorError(f'配置文件不存在：{Path}') + self.Path = Path + self.GiteaBaseUrl = (Parser.get('gitea', 'BaseUrl', fallback='https://git.gvsds.com') or '').rstrip('/') + self.GiteaOrg = Parser.get('gitea', 'OrgName', fallback='GVSDS') + self.GiteaToken = (Parser.get('gitea', 'ApiToken', fallback='') or '').strip() + self.SecretsIni = Parser.get('gitea', 'SecretsIni', fallback='') + + self.GithubOrg = Parser.get('github', 'OrgName', fallback='GVSDS') + self.GithubToken = (Parser.get('github', 'Token', fallback='') or '').strip() + self.OnRepoDeleted = (Parser.get('github', 'OnRepoDeleted', fallback='archive') or 'archive').strip().lower() + + self.WorkDir = Parser.get('mirror', 'WorkDir', fallback=os.path.join(os.path.dirname(os.path.abspath(Path)), 'data')) + self.SyncIntervalMinutes = Parser.getint('mirror', 'SyncIntervalMinutes', fallback=15) + self.BackfillOnStart = Parser.getboolean('mirror', 'BackfillOnStart', fallback=True) + self.LfsEnabled = Parser.getboolean('mirror', 'LfsEnabled', fallback=True) + self.Concurrency = max(1, Parser.getint('mirror', 'Concurrency', fallback=1)) + self.WebhookSecret = (Parser.get('mirror', 'WebhookSecret', fallback='') or '').strip() + + self.ServerHost = Parser.get('server', 'Host', fallback='0.0.0.0') + self.ServerPort = Parser.getint('server', 'Port', fallback=8600) + self.AdminToken = (Parser.get('server', 'AdminToken', fallback='') or '').strip() + + os.makedirs(self.WorkDir, exist_ok=True) + self.ReposDir = os.path.join(self.WorkDir, 'repos') + os.makedirs(self.ReposDir, exist_ok=True) + self.CredFile = os.path.join(self.WorkDir, 'git-credentials') + self.DbFile = os.path.join(self.WorkDir, 'mirror.db') + + # Gitea 令牌缺省从 secrets.ini 的 [gitea] API_TOKEN 读取（与主站共用） + if not self.GiteaToken and self.SecretsIni: + SecretParser = configparser.ConfigParser(interpolation=None) + SecretParser.read(self.SecretsIni, encoding='utf-8') + self.GiteaToken = (SecretParser.get('gitea', 'API_TOKEN', fallback='') or '').strip() + + self._LoadRuntimeSecrets() + + if not self.GithubToken: + raise MirrorError('未配置 github.Token（Fine-grained PAT）') + if not self.GiteaToken: + raise MirrorError('未配置 gitea.ApiToken，且 secrets.ini 中也没有 [gitea] API_TOKEN') + + def _LoadRuntimeSecrets(self) -> None: + """webhook secret / admin token：配置为空时自动生成并落盘（不入库）""" + RuntimePath = os.path.join(self.WorkDir, 'runtime.json') + Runtime = {} + if os.path.exists(RuntimePath): + try: + with open(RuntimePath, 'r', encoding='utf-8') as Fh: + Runtime = json.load(Fh) or {} + except (OSError, ValueError): + Runtime = {} + Changed = False + if not self.WebhookSecret: + self.WebhookSecret = Runtime.get('WebhookSecret') or py_secrets.token_hex(32) + Runtime['WebhookSecret'] = self.WebhookSecret + Changed = True + if not self.AdminToken: + self.AdminToken = Runtime.get('AdminToken') or py_secrets.token_hex(32) + Runtime['AdminToken'] = self.AdminToken + Changed = True + if Changed: + with open(RuntimePath, 'w', encoding='utf-8') as Fh: + json.dump(Runtime, Fh, ensure_ascii=False, indent=2) + + +# --------------------------------------------------------------------------- +# 状态存储 +# --------------------------------------------------------------------------- + +class MirrorStore: + """SQLite 状态表：每个仓库一行""" + + def __init__(self, Path: str): + self.Path = Path + self._Lock = threading.Lock() + self._Init() + + def _Connect(self) -> sqlite3.Connection: + Conn = sqlite3.connect(self.Path, timeout=15) + Conn.row_factory = sqlite3.Row + return Conn + + def _Init(self) -> None: + with self._Connect() as Conn: + Conn.execute(''' + CREATE TABLE IF NOT EXISTS RepoMirror ( + RepoName TEXT PRIMARY KEY, + GithubName TEXT, + Status TEXT, + LastSyncTime TEXT, + LastError TEXT, + SourcePrivate INTEGER DEFAULT 1, + Archived INTEGER DEFAULT 0 + ) + ''') + Conn.commit() + + def SetStatus(self, RepoName: str, **Fields) -> None: + Allowed = {'GithubName', 'Status', 'LastSyncTime', 'LastError', 'SourcePrivate', 'Archived'} + Data = {K: V for K, V in Fields.items() if K in Allowed} + if not Data: + return + with self._Lock, self._Connect() as Conn: + Conn.execute( + 'INSERT INTO RepoMirror (RepoName) VALUES (?) ON CONFLICT(RepoName) DO NOTHING', + (RepoName,), + ) + Assign = ', '.join(f'{K} = ?' for K in Data) + Conn.execute( + f'UPDATE RepoMirror SET {Assign} WHERE RepoName = ?', + (*Data.values(), RepoName), + ) + Conn.commit() + + def List(self) -> list[dict]: + with self._Connect() as Conn: + Rows = Conn.execute('SELECT * FROM RepoMirror ORDER BY RepoName').fetchall() + return [dict(Row) for Row in Rows] + + def Get(self, RepoName: str) -> dict | None: + with self._Connect() as Conn: + Row = Conn.execute('SELECT * FROM RepoMirror WHERE RepoName = ?', (RepoName,)).fetchone() + return dict(Row) if Row else None + + +# --------------------------------------------------------------------------- +# Gitea +# --------------------------------------------------------------------------- + +class GiteaClient: + """Gitea REST API（组织维度）""" + + def __init__(self, BaseUrl: str, Org: str, Token: str): + self.BaseUrl = BaseUrl.rstrip('/') + self.Org = Org + self.Token = Token + + def _Request(self, Method: str, Path: str, **Kwargs): + Url = f'{self.BaseUrl}/api/v1{Path}' + Headers = {'Authorization': f'token {self.Token}', 'Accept': 'application/json'} + Resp = requests.request(Method, Url, headers=Headers, timeout=30, **Kwargs) + if Resp.status_code >= 400: + raise MirrorError( + f'Gitea {Method} {Path} -> {Resp.status_code} {Resp.text[:200]}', + Status=Resp.status_code, + ) + if not Resp.content: + return None + return Resp.json() + + def GetRepo(self, Name: str) -> dict | None: + """查询单个仓库，不存在返回 None""" + try: + return self._Request('GET', f'/repos/{self.Org}/{Name}') + except MirrorError as Error: + if Error.Status == 404: + return None + raise + + def ListRepos(self) -> list[dict]: + """枚举组织下全部仓库（自动翻页）""" + Result: list[dict] = [] + Page = 1 + while True: + Batch = self._Request('GET', f'/orgs/{self.Org}/repos', params={'page': Page, 'limit': 50}) or [] + Result.extend(Batch) + if len(Batch) < 50: + break + Page += 1 + return Result + + def RegisterWebhook(self, Url: str, Secret: str) -> str: + """创建或更新组织级 webhook，返回 created/updated""" + Events = {'push': True, 'create': True, 'delete': True, 'repository': True} + Config = {'url': Url, 'content_type': 'json', 'secret': Secret} + Hooks = self._Request('GET', f'/orgs/{self.Org}/hooks') or [] + for Hook in Hooks: + if (Hook.get('config') or {}).get('url') == Url: + self._Request('PATCH', f'/orgs/{self.Org}/hooks/{Hook["id"]}', json={ + 'active': True, 'events': Events, 'config': Config, + }) + return 'updated' + self._Request('POST', f'/orgs/{self.Org}/hooks', json={ + 'type': 'gitea', 'active': True, 'events': Events, 'config': Config, + }) + return 'created' + + +# --------------------------------------------------------------------------- +# GitHub +# --------------------------------------------------------------------------- + +class GithubClient: + """GitHub REST API（组织维度，Fine-grained PAT）""" + + def __init__(self, Org: str, Token: str): + self.Org = Org + self.Token = Token + + def _Request(self, Method: str, Path: str, Json=None): + Headers = { + 'Authorization': f'Bearer {self.Token}', + 'Accept': 'application/vnd.github+json', + 'X-GitHub-Api-Version': '2022-11-28', + } + Resp = requests.request(Method, f'{GITHUB_API}{Path}', headers=Headers, json=Json, timeout=30) + return Resp + + def EnsureRepo(self, Name: str, Private: bool) -> str: + """确保仓库存在且可写；返回 full_name。已归档的会自动取消归档。""" + Resp = self._Request('GET', f'/repos/{self.Org}/{Name}') + if Resp.status_code == 200: + Info = Resp.json() + if Info.get('archived'): + self._Request('PATCH', f'/repos/{self.Org}/{Name}', Json={'archived': False}) + return Info.get('full_name') or f'{self.Org}/{Name}' + if Resp.status_code != 404: + raise MirrorError(f'GitHub 查询仓库失败 {Name} -> {Resp.status_code} {Resp.text[:200]}') + Created = self._Request('POST', f'/orgs/{self.Org}/repos', Json={ + 'name': Name, + 'private': bool(Private), + 'auto_init': False, + 'has_issues': False, + 'has_wiki': False, + }) + if Created.status_code >= 400: + raise MirrorError(f'GitHub 创建仓库失败 {Name} -> {Created.status_code} {Created.text[:200]}') + return (Created.json() or {}).get('full_name') or f'{self.Org}/{Name}' + + def ArchiveRepo(self, Name: str) -> None: + Resp = self._Request('PATCH', f'/repos/{self.Org}/{Name}', Json={'archived': True}) + if Resp.status_code >= 400 and Resp.status_code != 404: + raise MirrorError(f'GitHub 归档仓库失败 {Name} -> {Resp.status_code} {Resp.text[:200]}') + + def DeleteRepo(self, Name: str) -> None: + Resp = self._Request('DELETE', f'/repos/{self.Org}/{Name}') + if Resp.status_code >= 400 and Resp.status_code != 404: + raise MirrorError(f'GitHub 删除仓库失败 {Name} -> {Resp.status_code} {Resp.text[:200]}') + + +# --------------------------------------------------------------------------- +# 镜像执行 +# --------------------------------------------------------------------------- + +class MirrorManager: + """单个仓库的镜像同步（bare clone + fetch + push）""" + + def __init__(self, Cfg: MirrorConfig, Gitea: GiteaClient, Github: GithubClient, Store: MirrorStore): + self.Cfg = Cfg + self.Gitea = Gitea + self.Github = Github + self.Store = Store + self._Locks: dict[str, threading.Lock] = {} + self._LocksGuard = threading.Lock() + self._LfsAvailable = self._DetectLfs() + self._WriteCredentials() + + # -- 基础设施 -- + + def _DetectLfs(self) -> bool: + if not self.Cfg.LfsEnabled: + return False + try: + Result = subprocess.run(['git', 'lfs', 'version'], capture_output=True, text=True, + encoding='utf-8', errors='replace', timeout=20) + return Result.returncode == 0 + except (OSError, subprocess.SubprocessError): + logger.warning('未检测到 git lfs，将跳过 LFS 对象同步') + return False + + def _WriteCredentials(self) -> None: + """写入 git credential store 文件（600 权限）""" + Lines = [ + f'https://oauth2:{self.Cfg.GiteaToken}@{self._GiteaHost()}', + f'https://{GITHUB_AUTH_USER}:{self.Cfg.GithubToken}@github.com', + ] + with open(self.Cfg.CredFile, 'w', encoding='utf-8') as Fh: + Fh.write('\n'.join(Lines) + '\n') + try: + os.chmod(self.Cfg.CredFile, 0o600) + except OSError: + pass + + def _GiteaHost(self) -> str: + return self.Cfg.GiteaBaseUrl.split('://', 1)[-1].rstrip('/') + + def LockFor(self, RepoName: str) -> threading.Lock: + with self._LocksGuard: + return self._Locks.setdefault(RepoName, threading.Lock()) + + def _Sanitize(self, Text: str) -> str: + if not Text: + return '' + for Secret in (self.Cfg.GithubToken, self.Cfg.GiteaToken): + if Secret: + Text = Text.replace(Secret, '***') + return Text + + def _Git(self, Args: list[str], Cwd: str | None = None) -> str: + Cmd = [ + 'git', + '-c', 'core.longpaths=true', + '-c', f'credential.helper=store --file={self.Cfg.CredFile}', + ] + Args + Env = os.environ.copy() + Env['GIT_TERMINAL_PROMPT'] = '0' + Env['LC_ALL'] = 'C' + try: + Result = subprocess.run( + Cmd, cwd=Cwd, env=Env, capture_output=True, text=True, + encoding='utf-8', errors='replace', timeout=3600, + ) + except subprocess.SubprocessError as Error: + raise MirrorError(f'git 执行失败：{Error}') from Error + Output = self._Sanitize((Result.stdout or '') + (Result.stderr or '')) + if Result.returncode != 0: + Tail = '\n'.join(Output.strip().splitlines()[-8:]) + raise MirrorError(f'git {" ".join(Args[:3])} 失败：{Tail}') + return Output + + # -- 主流程 -- + + def Sync(self, Repo: dict) -> dict: + """同步单个仓库（幂等）：确保 GitHub 仓库 → 本地 bare → fetch → push（含 LFS）""" + RepoName = (Repo.get('name') or '').strip() + if not RepoName: + raise MirrorError('仓库缺少 name 字段') + GithubName = self._GithubName(RepoName) + Private = bool(Repo.get('private', True)) + Local = os.path.join(self.Cfg.ReposDir, f'{RepoName}.git') + SourceUrl = f'{self.Cfg.GiteaBaseUrl}/{self.Cfg.GiteaOrg}/{RepoName}.git' + TargetUrl = f'https://github.com/{self.Cfg.GithubOrg}/{GithubName}.git' + + with self.LockFor(RepoName): + logger.info('开始镜像 %s -> %s/%s', RepoName, self.Cfg.GithubOrg, GithubName) + self.Store.SetStatus(RepoName, Status='syncing', SourcePrivate=int(Private), + GithubName=GithubName, LastError='') + try: + self.Github.EnsureRepo(GithubName, Private) + if not os.path.isdir(Local): + self._Git(['clone', '--mirror', SourceUrl, Local]) + else: + self._Git(['-C', Local, 'remote', 'set-url', 'origin', SourceUrl]) + self._Git(['-C', Local, 'fetch', '--prune', '--prune-tags', '--force', 'origin']) + if self._LfsAvailable: + self._Git(['-C', Local, 'lfs', 'fetch', '--all']) + self._Git(['-C', Local, 'push', '--mirror', TargetUrl]) + if self._LfsAvailable: + self._Git(['-C', Local, 'lfs', 'push', '--all', TargetUrl]) + except MirrorError as Error: + self.Store.SetStatus(RepoName, Status='failed', LastError=str(Error)[:1000]) + raise + self.Store.SetStatus(RepoName, Status='ok', LastSyncTime=_Now(), LastError='') + logger.info('镜像完成 %s', RepoName) + return {'RepoName': RepoName, 'GithubName': GithubName, 'Status': 'ok'} + + def HandleRepoDeleted(self, RepoName: str) -> None: + """源仓库被删除：按配置归档（默认）/ 真删 / 忽略""" + GithubName = self._GithubName(RepoName) + Mode = self.Cfg.OnRepoDeleted + if Mode == 'ignore': + logger.info('源仓库 %s 已删除，配置为忽略', RepoName) + return + if Mode == 'delete': + self.Github.DeleteRepo(GithubName) + logger.info('源仓库 %s 已删除，GitHub 仓库已删除', RepoName) + else: + self.Github.ArchiveRepo(GithubName) + logger.info('源仓库 %s 已删除，GitHub 仓库已归档', RepoName) + self.Store.SetStatus(RepoName, Status='deleted', Archived=int(Mode != 'delete'), LastSyncTime=_Now()) + + @staticmethod + def _GithubName(RepoName: str) -> str: + """GitHub 仓库名只允许字母数字 . _ -（其余替换为 -）""" + Clean = ''.join(C if (C.isalnum() or C in '._-') else '-' for C in RepoName) + return Clean.strip('-') or 'repo' diff --git a/GitMirror/requirements.txt b/GitMirror/requirements.txt new file mode 100644 index 00000000..a0d407c7 --- /dev/null +++ b/GitMirror/requirements.txt @@ -0,0 +1,2 @@ +flask>=3.0 +requests>=2.31 diff --git a/GitMirror/run.ps1 b/GitMirror/run.ps1 new file mode 100644 index 00000000..0047d6dc --- /dev/null +++ b/GitMirror/run.ps1 @@ -0,0 +1,16 @@ +$ErrorActionPreference = 'Stop' +Set-Location $PSScriptRoot + +if (-not (Test-Path 'config.ini')) { + Copy-Item 'config.example.ini' 'config.ini' + Write-Host '已生成 config.ini，请填写 [github] Token 后重新运行。' + exit 1 +} + +python -c "import flask, requests" 2>$null +if ($LASTEXITCODE -ne 0) { + Write-Host '安装依赖...' + python -m pip install -r requirements.txt +} + +python app.py serve