feat(GitMirror): 新增 Gitea 到 GitHub 实时镜像服务(webhook+定时兜底+LFS)
This commit is contained in:
1 parent
0ba3aeee2d
commit
eb4ff5197d
6 files changed
+811
No files matched your search
@@ -0,0 +1,4 @@
|
||||
config.ini
|
||||
data/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
@@ -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/<repo> 手动同步单仓(X-Admin-Token)
|
||||
- GET /status 镜像状态(X-Admin-Token)
|
||||
- GET /healthz 健康检查
|
||||
|
||||
后台:工作线程池执行镜像;定时线程按 SyncIntervalMinutes 做全量兜底。
|
||||
|
||||
用法:
|
||||
python app.py serve
|
||||
python app.py backfill
|
||||
python app.py sync <repo>
|
||||
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/<RepoName>')
|
||||
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())
|
||||
@@ -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 =
|
||||
@@ -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'
|
||||
@@ -0,0 +1,2 @@
|
||||
flask>=3.0
|
||||
requests>=2.31
|
||||
@@ -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
|
||||
Reference in new issue
Block a user