GitHub账号被封？用Python一键迁移所有Star仓库

Be1k0

前言

最近因为某些原因有个Github账号被封了，导致里面积累的Star仓库需要手动迁移耗时费力且易遗漏，所以借助GitHub API和一段Python代码即可完成全部收藏仓库的迁移

代码

import requests
import time

# 你的新 GitHub 用户名
NEW_USERNAME = "***"
# 新账号的 GitHub Token（必须有 public_repo 权限）
GITHUB_TOKEN = "ghp_****"
# 被封禁的用户名
BANNED_USERNAME = "***"

# API 地址
STARRED_API = f"https://api.github.com/users/{BANNED_USERNAME}/starred"
HEADERS = {
    "Authorization": f"token {GITHUB_TOKEN}",
    "Accept": "application/vnd.github.v3+json"
}

def get_all_starred_repos():
    """获取被封禁账号的所有 Star 仓库列表"""
    repos = []
    page = 1
    print(f"正在获取 @{BANNED_USERNAME} 的所有 Star 仓库...")

    while True:
        url = f"{STARRED_API}?page={page}&per_page=100"
        resp = requests.get(url)

        if resp.status_code != 200:
            print(f"获取失败，状态码: {resp.status_code}")
            break

        data = resp.json()
        if not data:
            break

        for repo in data:
            repos.append({
                "full_name": repo["full_name"],
                "url": repo["html_url"]
            })

        print(f"已获取第 {page} 页，累计：{len(repos)} 个")
        page += 1
        time.sleep(1)  # 防限流

    print(f"\n总共获取到 {len(repos)} 个 Star 仓库")
    return repos

def star_repos(repos):
    """批量给新账号 Star 仓库"""
    success = 0
    failed = []
    total = len(repos)

    print(f"\n开始批量 Star 到 @{NEW_USERNAME}...")

    for i, repo in enumerate(repos, 1):
        full_name = repo["full_name"]
        url = f"https://api.github.com/user/starred/{full_name}"

        try:
            resp = requests.put(url, headers=HEADERS)

            if resp.status_code in (204, 404):
                # 204 = Star 成功
                # 404 = 仓库不存在/已删除，跳过
                success += 1
                print(f"[{i}/{total}] {full_name}")
            else:
                failed.append(full_name)
                print(f"[{i}/{total}] 失败 {full_name} | 状态码: {resp.status_code}")

            time.sleep(2)  # 防 GitHub API 限流

        except Exception as e:
            failed.append(full_name)
            print(f"[{i}/{total}] 异常 {full_name} | {str(e)}")

    # 最终统计
    print("\n" + "="*50)
    print(f"迁移完成！")
    print(f"成功 Star：{success} 个")
    print(f"失败/不存在：{len(failed)} 个")

    if failed:
        print("\n失败列表（可手动检查）：")
        for f in failed:
            print(f"  - {f}")

if __name__ == "__main__":
    starred_repos = get_all_starred_repos()
    star_repos(starred_repos)

使用方法

首先获取新账号的 GitHub Token

1. 打开 GitHub → 右上角头像 → Settings
2. 左侧最底下 → Developer settings → Personal access tokens → Tokens (classic)
   点击 Generate new token (classic)
3. 勾选权限：public_repo

4. 生成后复制 token

   然后补充脚本中

NEW_USERNAME = "你的新GitHub用户名"
GITHUB_TOKEN = "你刚复制的token"
BANNED_USERNAME = "你的旧Github用户名"

m_h

附件是更复杂的版本
syncfork.7z (14.55 KB, 下载次数: 11)

2026-4-10 11:25 上传

点击文件名下载附件
V2

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
GitHub Fork 批量同步脚本 (API动态获取版)
特性：自动拉取Fork列表 / 代理支持 / 智能限速防封 / 断点续传 / 详细日志
"""

import os
import sys
import time
import json
import random
import requests
from datetime import datetime, timedelta

# ================= 配置区 =================
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN", "")
PROXY_URL = os.getenv("GITHUB_PROXY", "socks5://192.168.2.x:xxxx")
DELAY = 2.0                 # 基础同步间隔（秒），代理越慢建议设越大
MAX_RETRIES = 3             # 失败重试次数
SYNC_STATE_FILE = "sync_state.json"
FORKS_CACHE_FILE = "forks_cache.json"  # API 拉取结果缓存（防重复请求）
# ==========================================

PROXIES = {"http": PROXY_URL, "https": PROXY_URL}
HEADERS = {
    "Authorization": f"token {GITHUB_TOKEN}",
    "Accept": "application/vnd.github.v3+json",
    "User-Agent": "ForkSyncer-API/1.0"
}

def check_rate_limit(resp):
    """通用速率限制检查，返回是否需要等待"""
    remaining = int(resp.headers.get("X-RateLimit-Remaining", 5000))
    if remaining <= 10:
        reset_ts = int(resp.headers.get("X-RateLimit-Reset", time.time() + 300))
        wait = max(reset_ts - time.time(), 0) + 5
        print(f"  ⚠️ API 额度即将耗尽 (剩余 {remaining} 次)。安全等待 {wait:.0f} 秒...")
        time.sleep(wait)
        return True
    return False

def fetch_forked_repos():
    """通过 API 分页获取当前用户的所有 Fork 仓库"""
    if os.path.exists(FORKS_CACHE_FILE):
        with open(FORKS_CACHE_FILE, "r", encoding="utf-8") as f:
            cached = json.load(f)
        # 缓存有效期 24 小时
        if datetime.now().isoformat() < cached.get("expires", "1970-01-01"):
            print(f"📦 使用缓存的 Fork 列表 ({len(cached['repos'])} 个)")
            return cached["repos"]

    repos = []
    page = 1
    per_page = 100
    print("📡 正在通过 GitHub API 获取 Fork 列表...")

    while True:
        # sort=full_name 保证每次顺序一致，断点续传更可靠
        url = f"https://api.github.com/user/repos?type=forks&per_page={per_page}&page={page}&sort=full_name"
        try:
            resp = requests.get(url, headers=HEADERS, proxies=PROXIES, timeout=30)
            if check_rate_limit(resp):
                continue
            resp.raise_for_status()
            data = resp.json()
            if not data:
                break

            for repo in data:
                repos.append(f"{repo['owner']['login']}/{repo['name']}")

            print(f"  📄 第 {page} 页完成 | 当前累计: {len(repos)} 个")
            page += 1
            time.sleep(0.3)  # 拉取列表时轻微延迟，避免突发请求
        except requests.exceptions.RequestException as e:
            print(f"  ⚠️ 获取第 {page} 页失败: {e}。2秒后重试...")
            time.sleep(2)
            continue

    # 保存缓存
    cache_data = {
        "repos": repos,
        "fetched_at": datetime.now().isoformat(),
        "expires": (datetime.now() + timedelta(hours=24)).isoformat()
    }
    with open(FORKS_CACHE_FILE, "w", encoding="utf-8") as f:
        json.dump(cache_data, f, indent=2)
    print(f"✅ Fork 列表获取完毕，共 {len(repos)} 个仓库。已缓存。")
    return repos

def load_sync_state():
    if os.path.exists(SYNC_STATE_FILE):
        with open(SYNC_STATE_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

def save_sync_state(state):
    with open(SYNC_STATE_FILE, "w", encoding="utf-8") as f:
        json.dump(state, f, indent=2, ensure_ascii=False)

def sync_single_repo(owner, repo):
    """同步单个 Fork（仅同步默认分支）"""
    url = f"https://api.github.com/repos/{owner}/{repo}/merge-upstream"
    payload = {"branch": "main", "sync_method": "merge"}

    for attempt in range(MAX_RETRIES):
        try:
            resp = requests.post(url, headers=HEADERS, json=payload, proxies=PROXIES, timeout=30)
            #if check_rate_limit(resp):
            #    continue

            if resp.status_code == 200:
                return True, "同步成功"
            elif resp.status_code == 409:
                return True, "已是最新或存在合并冲突"
            elif resp.status_code == 404:
                return False, "仓库不存在或无权限 (404)"
            elif resp.status_code in (403, 429):
                retry_after = int(resp.headers.get("Retry-After", 60))
                print(f"  ⚠️ 触发限流，等待 {retry_after}s...")
                time.sleep(retry_after)
                continue
            else:
                resp.raise_for_status()
        except requests.exceptions.RequestException as e:
            if attempt < MAX_RETRIES - 1:
                wait = (attempt + 1) * 3
                print(f"  ⚠️ 网络异常: {e}。{wait}s 后重试...")
                time.sleep(wait)
            else:
                return False, f"网络异常: {str(e)}"

    # 兼容 master 分支的老仓库
    payload["branch"] = "master"
    try:
        resp = requests.post(url, headers=HEADERS, json=payload, proxies=PROXIES, timeout=30)
        if resp.status_code in (200, 409):
            return True, "同步成功 (master)"
    except:
        pass

    return False, "同步失败 (main & master 均无效)"

def main():
    if not GITHUB_TOKEN or GITHUB_TOKEN.startswith("ghp_你的"):
        print("❌ 请设置 GITHUB_TOKEN 环境变量或修改脚本配置。")
        sys.exit(1)

    print(f"🌐 代理: {PROXY_URL}")
    print(f"⏱️ 同步间隔: {DELAY}s + 随机抖动")
    print("-" * 50)

    # 1. 获取仓库列表
    repos = fetch_forked_repos()
    if not repos:
        print("❌ 未找到任何 Fork 仓库。")
        sys.exit(0)

    # 2. 加载同步状态
    state = load_sync_state()
    total = len(repos)
    processed = sum(1 for r in repos if r in state)
    print(f"🚀 开始同步 | 总计: {total} | 已处理: {processed} | 待同步: {total - processed}")
    print("-" * 50)

    # 3. 执行同步
    for i, repo in enumerate(repos, 1):
        if repo in state:
            print(f"[{i}/{total}] ⏭️ 跳过 {repo} (状态: {state[repo]['status']})")
            continue

        owner, name = repo.split("/", 1)
        print(f"[{i}/{total}] 🔄 同步 {repo} ...")

        success, msg = sync_single_repo(owner, name)
        state[repo] = {
            "status": "success" if success else "failed",
            "message": msg,
            "time": datetime.now().isoformat()
        }
        save_sync_state(state)  # 实时保存，防中断丢失
        print(f"  {'✅' if success else '❌'} {msg}")

        #if i < total:
            #jitter = random.uniform(0.5, 1.5)
            #sleep_time = DELAY + jitter
            #print(f"  💤 等待 {sleep_time:.2f}s ...")
            #time.sleep(sleep_time)

    print("\n🎉 全部处理完成！")
    failed_count = sum(1 for v in state.values() if v["status"] == "failed")
    print(f"📊 成功: {total - failed_count} | 失败: {failed_count}")
    print(f"📄 详细报告已保存至: sync_state.json")

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n⛔ 用户中断，状态已保存，下次运行将自动续传。")
    except Exception as e:
        print(f"\n💥 发生未捕获异常: {e}")
        sys.exit(1)

同步更新上游fork

jake77

感谢大佬分享！

Tsuki0402

感谢，收藏下来以备不时之需

allxxall

只能访问对应公开的仓库，那私有的还是需要手动操作？

Be1k0

allxxall 发表于 2026-4-6 18:19
只能访问对应公开的仓库，那私有的还是需要手动操作？

新账号也不会有原私有仓库的权限吧，所以这只支持公开的

Artstation

m_h 发表于 2026-4-6 20:47
[md]````python

#!/usr/bin/env python3

话说为什么GitHub账号被封？

传闻中的喜哥哥

头一次听书github被封 虽然被微软收购  但政策没变

m_h

Artstation 发表于 2026-4-8 13:59
话说为什么GitHub账号被封？

我的脚本 用来更新自己账号fork 太多手动点击效率太低用的。

pptx

感谢大佬分享。学习了以备不时之需