将脚本从Gitee同步至Github改为Github同步至Gitee

Signed-off-by: gfdgd_xi <3025613752@qq.com>

sync-github-to-gitee/sync.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+将 GitHub 上的 GXDE-OS 组织仓库镜像同步到 Gitee（GXDE-OS 组织）。
+
+用法:
+    python3 sync.py <github用户名> <githubToken> <gitee用户名> <gitee密码或Token>
+
+流程:
+    1. 通过 GitHub REST API 分页获取 GXDE-OS 组织下的全部仓库列表（带 token 失败时回退匿名）
+    2. 多线程 clone --bare 拉取 GitHub 仓库
+    3. 逐个 push --mirror 推送到 Gitee，失败自动重试并在结束时汇总报告
+
+注意:
+    - Gitee 侧的同名仓库需要预先存在，push --mirror 会覆盖其分支与标签
+    - GitHub 只接受 Personal Access Token 作为密码（需具备组织仓库的读取权限）
+"""
+import json
+import os
+import subprocess
+import sys
+import time
+import urllib.parse
+import urllib.request
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+programPath = os.path.split(os.path.realpath(__file__))[0]
+cloneDir = os.path.join(programPath, "git-clone")
+orgName = "GXDE-OS"
+perPage = 100
+threadMax = 8
+
+
+def usage():
+    print("用法: python3 sync.py <github用户名> <githubToken> <gitee用户名> <gitee密码或Token>")
+    sys.exit(1)
+
+
+def apiGet(path, params, token):
+    """调用 GitHub REST API 的单个请求"""
+    params["per_page"] = str(perPage)
+    url = "https://api.github.com" + path + "?" + urllib.parse.urlencode(params)
+    headers = {
+        "User-Agent": "GXDE-Sync/1.0",
+        "Accept": "application/vnd.github+json",
+        "X-GitHub-Api-Version": "2022-11-28",
+    }
+    if token:
+        headers["Authorization"] = "Bearer " + token
+    req = urllib.request.Request(url, headers=headers)
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        return json.loads(resp.read().decode("utf-8"))
+
+
+def getGithubRepos(token):
+    """分页获取 GitHub 组织下全部仓库的 name（即 URL 路径中的仓库名）
+
+    带 token 时 type=all 可包含私有仓库；匿名请求只返回公开仓库。
+    GitHub 的 name 可直接拼接 clone/push URL，无需像 Gitee 那样取 path 字段。
+    """
+    repoList = []
+    page = 1
+    while True:
+        params = {"page": str(page), "type": "all"}
+        data = apiGet(f"/orgs/{orgName}/repos", params, token)
+        if not isinstance(data, list) or len(data) == 0:
+            break
+        repoList.extend([i["name"] for i in data])
+        if len(data) < perPage:
+            break
+        page += 1
+    return repoList
+
+
+def runGit(args, cwd=None):
+    return subprocess.run(args, cwd=cwd, stdout=subprocess.PIPE,
+                          stderr=subprocess.STDOUT, text=True)
+
+
+def syncRepo(name, pullUrl, pushUrl):
+    repoDir = os.path.join(cloneDir, name + ".git")
+    try:
+        # 1. 从 GitHub clone --bare（重试 3 次）
+        ok = False
+        lastOut = ""
+        for attempt in range(3):
+            result = runGit(["git", "clone", "--bare", pullUrl + name, repoDir])
+            if result.returncode == 0:
+                ok = True
+                break
+            lastOut = result.stdout
+            time.sleep(5)
+        if not ok:
+            print(f"[FAIL] clone 失败: {name}\n{lastOut}")
+            return False
+
+        # 2. push --mirror 到 Gitee（重试 3 次）
+        ok = False
+        lastOut = ""
+        for attempt in range(3):
+            result = runGit(["git", "-C", repoDir, "push", "--mirror", pushUrl + name])
+            if result.returncode == 0:
+                ok = True
+                break
+            lastOut = result.stdout
+            time.sleep(5)
+        if not ok:
+            print(f"[FAIL] push 失败: {name}\n{lastOut}")
+            return False
+
+        print(f"[ OK ] 同步完成: {name}")
+        return True
+    finally:
+        # 清理临时目录
+        runGit(["rm", "-rf", repoDir])
+
+
+def main():
+    if len(sys.argv) < 5:
+        usage()
+    githubUser, githubToken = sys.argv[1], sys.argv[2]
+    giteeUser, giteeToken = sys.argv[3], sys.argv[4]
+
+    githubAuth = urllib.parse.quote(githubUser, safe="") + ":" + urllib.parse.quote(githubToken, safe="")
+    giteeAuth = urllib.parse.quote(giteeUser, safe="") + ":" + urllib.parse.quote(giteeToken, safe="")
+    pullUrl = f"https://{githubAuth}@github.com/{orgName}/"
+    pushUrl = f"https://{giteeAuth}@gitee.com/{orgName}/"
+
+    os.makedirs(cloneDir, exist_ok=True)
+
+    # 获取仓库列表：优先带 token（可含私有仓库），失败则回退到匿名（仅公开仓库）
+    try:
+        repoList = getGithubRepos(githubToken)
+    except Exception:
+        repoList = getGithubRepos("")
+    print(f"共获取到 {len(repoList)} 个仓库")
+    if not repoList:
+        print("未能获取仓库列表，退出")
+        sys.exit(1)
+
+    failed = []
+    with ThreadPoolExecutor(max_workers=threadMax) as pool:
+        futures = {pool.submit(syncRepo, n, pullUrl, pushUrl): n for n in repoList}
+        for future in as_completed(futures):
+            if not future.result():
+                failed.append(futures[future])
+
+    print("=" * 50)
+    print(f"同步完成: 成功 {len(repoList) - len(failed)} / {len(repoList)}")
+    if failed:
+        print("以下仓库同步失败:")
+        for name in failed:
+            print("  -", name)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()