Python构建高性能GitHub仓库批量下载工具的技术实践指南

git clone https://github.com/user/repo1.git
git clone https://github.com/user/repo2.git
git clone https://github.com/user/repo3.git
# ... 重复100次

from selenium import webdriver
driver = webdriver.Chrome()
# 模拟浏览器操作...

import requests
repos = requests.get('https://api.github.com/users/xxx/repos').json()
for repo in repos:
    os.system(f'git clone {repo["clone_url"]}')

def get_repos(username, token, language, min_stars, updated_after, max_repos=0):
    repos = []
    page = 1
    headers = {"Accept": "application/vnd.github.v3+json"}
    
    if token:
        headers["Authorization"] = f"token {token}"
        
    while True:
        # 分页请求：每页100条，通过page参数控制
        url = f"https://api.github.com/users/{username}/repos?per_page=100&page={page}"
        
        try:
            resp = requests.get(url, headers=headers, timeout=15)
        except requests.exceptions.RequestException as e:
            logger.error("Network error while accessing GitHub API: %s", e)
            break
            
        # 错误码处理矩阵
        if resp.status_code == 404:          # 用户不存在
            break
        elif resp.status_code == 403 and "rate limit" in resp.text.lower():
            # 触发速率限制
            break
        elif resp.status_code != 200:        # 其他错误
            break
            
        data = resp.json()
        if not data:  # 无更多数据
            break
            
        # 服务端过滤：在客户端应用复合过滤条件
        for r in data:
            if language and r.get('language') != language:
                continue                    # 语言不匹配 → 跳过
            if r.get('stargazers_count', 0) < min_stars:
                continue                    # 星标不足 → 跳过
            if updated_after:
                r_date = datetime.strptime(r['updated_at'], "%Y-%m-%dT%H:%M:%SZ")
                f_date = datetime.strptime(updated_after, "%Y-%m-%d")
                if r_date < f_date:
                    continue                # 更新时间过早 → 跳过
                    
            repos.append(r)
            
            if max_repos > 0 and len(repos) >= max_repos:
                break                        # 达到上限 → 提前终止
        
        if max_repos > 0 and len(repos) >= max_repos:
            break                            # 提前终止分页循环
            
        page += 1
        
    return repos

@retry(retry=retry_if_exception_type(RetryableError), 
       stop=stop_after_attempt(3), 
       wait=wait_fixed(2))
def download_zip(repo, opts, progress, task_id):
    repo_name = repo['name']
    target_ref = opts.get('target_ref')
    fallback_ref = repo.get('default_branch', 'main')
    
    save_path = opts['save_path']
    
    def try_download(ref):
        # 尝试Branch URL
        urls_to_try = [
            f"https://github.com/{repo['owner']['login']}/{repo_name}/archive/refs/heads/{ref}.zip",
            f"https://github.com/{repo['owner']['login']}/{repo_name}/archive/refs/tags/{ref}.zip"
        ]
        
        for url in urls_to_try:
            resp = requests.get(url, headers=headers, stream=True)
            if resp.status_code == 200:
                return resp
            if resp.status_code in [500, 502, 503, 504]:
                raise RetryableError(f"Server error {resp.status_code}")
                
        return None

    try:
        # 第一步：尝试目标分支/标签
        resp = try_download(target_ref)
        
        # 第二步：智能回退到默认分支
        if not resp and target_ref != fallback_ref:
            progress.update(task_id, description=f"Fallback to {fallback_ref}")
            resp = try_download(fallback_ref)
            
        if not resp:
            raise NonRetryableError(f"Branch '{target_ref}' not found")
            
        # 第三步：流式下载（支持大文件）
        total_size = int(resp.headers.get('content-length', 0))
        zip_path = os.path.join(save_path, f"{repo_name}.zip")
        
        with open(zip_path, 'wb') as f:
            for chunk in resp.iter_content(chunk_size=8192):  # 8KB chunks
                if chunk:
                    f.write(chunk)
                    progress.advance(task_id, len(chunk))  # 实时更新进度
                    
        # 第四步：智能解压（清理分支后缀）
        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            top_level = next(iter(zip_ref.namelist())).split('/')[0]  # e.g., "repo-main"
            zip_ref.extractall(save_path)
            
        unzipped_folder = os.path.join(save_path, top_level)
        extract_dir = os.path.join(save_path, repo_name)  # 目标目录名（不含后缀）
        
        # 重命名：repo-main → repo
        if os.path.exists(extract_dir):
            shutil.rmtree(extract_dir)
        os.rename(unzipped_folder, extract_dir)
        else:
            os.rename(unzipped_folder, extract_dir)
        
        # 第五步：可选保留ZIP文件
        if not opts['keep_zip']:
            os.remove(zip_path)
            
        return "Success"
        
    except NonRetryableError as e:
        raise e  # 不可恢复错误，直接抛出
    except Exception as e:
        raise RetryableError(str(e))  # 可恢复错误，触发重试

@retry(retry=retry_if_exception_type(RetryableError),
       stop=stop_after_attempt(3),
       wait=wait_fixed(2))
def clone_git(repo, opts, progress, task_id):
    repo_path = os.path.join(opts['save_path'], repo['name'])
    
    # 判断：本地是否已存在？
    if os.path.exists(os.path.join(repo_path, '.git')):
        # ✅ 已存在 → 执行Pull更新
        git_repo = Repo(repo_path)
        origin = git_repo.remotes.origin
        origin.pull()
        
        # 尝试切换到目标分支
        try:
            git_repo.git.checkout(target_ref)
        except GitCommandError:
            # 目标分支不存在 → 回退到默认分支
            if target_ref != fallback_ref:
                git_repo.git.checkout(fallback_ref)
    else:
        # ❌ 不存在 → 执行Clone
        try:
            git_repo = Repo.clone_from(clone_url, repo_path, branch=target_ref)
        except GitCommandError as e:
            # Clone失败且原因是分支不存在 → 回退
            if target_ref != fallback_ref:
                git_repo = Repo.clone_from(clone_url, repo_path, branch=fallback_ref)
            else:
                raise e
                
    return "Success"

def run():
    opts = parse_and_merge_args()
    repos = get_repos(...)  # 获取仓库列表
    
    stats = {"success": 0, "failed": 0, "skipped": 0}
    stats_lock = threading.Lock()  # 线程安全计数器
    
    with Progress(...) as progress:
        overall_task = progress.add_task("[bold blue]Overall Progress", total=len(repos))
        
        # 创建线程池
        with ThreadPoolExecutor(max_workers=opts['max_workers']) as executor:
            # 提交所有任务
            futures = [
                executor.submit(process_repo, repo, opts, history, 
                              progress, overall_task, stats, stats_lock) 
                for repo in repos
            ]
            
            # 按完成顺序收集结果（非阻塞）
            for future in as_completed(futures):
                repo, status, new_updated_at = future.result()
                statuses[repo['name']] = status
                
                # 更新历史记录（仅成功/跳过的）
                if status in ('success', 'skipped'):
                    history[repo['name']] = {"updated_at": new_updated_at}

def process_repo(repo, opts, history, progress, overall_task, stats, stats_lock):
    """单个仓库的处理逻辑（在线程池中运行）"""
    repo_name = repo['name']
    task_id = progress.add_task(f"Processing {repo_name}...", total=100)
    
    # 增量同步检查：比较时间戳
    last_updated = history.get(repo_name, {}).get("updated_at")
    current_updated = repo['updated_at']
    
    if last_updated == current_updated:
        # 未变更 → 跳过
        progress.update(task_id, description=f"[yellow]Skipped {repo_name}[/yellow]", completed=100)
        with stats_lock:
            stats["skipped"] += 1
        return repo, "skipped", current_updated
        
    try:
        # 执行下载（Git或ZIP）
        if opts['mode'] == 'zip':
            download_zip(repo, opts, progress, task_id)
        else:
            clone_git(repo, opts, progress, task_id)
            
        # 成功
        with stats_lock:
            stats["success"] += 1
        return repo, "success", current_updated
        
    except Exception as e:
        # 失败
        with stats_lock:
            stats["failed"] += 1
        return repo, "failed", last_updated  # 保持旧时间戳以便下次重试

concurrency:
  max_workers: 5  # 推荐值：3-10

def load_sync_history(save_path):
    """加载上次同步的时间戳"""
    history_file = os.path.join(save_path, "last_sync.json")
    if os.path.exists(history_file):
        with open(history_file, 'r', encoding='utf-8') as f:
            return json.load(f) or {}
    return {}

def save_sync_history(save_path, history):
    """保存本次同步的时间戳"""
    os.makedirs(save_path, exist_ok=True)
    history_file = os.path.join(save_path, "last_sync.json")
    with open(history_file, 'w', encoding='utf-8') as f:
        json.dump(history, indent=4)

def generate_report(repos, statuses, opts):
    """生成Markdown或CSV格式的报表"""
    format = opts.get('report_format', 'markdown').lower()
    out_dir = opts.get('report_dir', '.')
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")  # 时间戳命名
    
    if format == 'csv':
        filepath = os.path.join(out_dir, f"repo_report_{ts}.csv")
        # ... CSV写入逻辑
    else:
        filepath = os.path.join(out_dir, f"repo_report_{ts}.md")
        # ... Markdown表格生成

def download_with_fallback(target_ref, fallback_ref):
    """
    三级回退策略：
    Level 1: 尝试目标分支/标签
    Level 2: 如果失败且不是默认分支 → 回退到默认分支
    Level 3: 如果还是失败 → 抛出不可恢复异常
    """
    
    # Level 1: 目标引用
    resp = try_download(target_ref)
    
    # Level 2: 智能回退
    if not resp and target_ref != fallback_ref:
        log(f"Target '{target_ref}' not found, falling back to '{fallback_ref}'")
        resp = try_download(fallback_ref)
    
    # Level 3: 彻底失败
    if not resp:
        raise NonRetryableError(f"Neither '{target_ref}' nor '{fallback_ref}' exists")
    
    return process_response(resp)

class RetryableError(Exception):
    """可重试错误：临时性故障，重试可能成功"""
    pass

class NonRetryableError(Exception):
    """不可重试错误：永久性故障，重试也无用"""
    pass

@retry(
    retry=retry_if_exception_type(RetryableError),  # 仅对可重试错误生效
    stop=stop_after_attempt(3),                      # 最多重试3次
    wait=wait_fixed(2)                               # 固定间隔2秒
)
def download_zip(...):
    ...

# Layer 1: 默认值 (Hardcoded)
# Layer 2: YAML配置文件 (config.yaml)
# Layer 3: CLI命令行参数 (最高优先级)

def parse_and_merge_args():
    args = parser.parse_args()           # 解析CLI参数
    config = load_config(args.config)    # 加载YAML
    
    # 合并策略：CLI参数覆盖YAML配置
    return {
        "username": args.username or c_github.get('username'),
        "mode": args.mode or c_dw.get('mode', 'git'),
        "max_workers": args.max_workers or c_conc.get('max_workers', 5),
        # ...
    }

# 场景1：使用YAML默认配置
uv run main.py
# 场景2：临时覆盖部分参数
uv run main.py --username octocat --language Python --max-repos 10
# 场景3：完全自定义（忽略YAML）
uv run main.py --username octocat --token ghp_xxx --mode zip --max-workers 8

# ❌ 一次性读取整个文件到内存（OOM风险）
content = resp.content
with open(file, 'wb') as f:
    f.write(content)

# ✅ 流式分块写入（内存友好）
for chunk in resp.iter_content(chunk_size=8192):
    f.write(chunk)
    progress.advance(task_id, len(chunk))  # 即时更新UI

# 优化建议（未来改进）
session = requests.Session()
session.headers.update({"Authorization": f"token {token}"})

# 所有请求共用TCP连接
for repo in repos:
    session.get(url)  # 复用连接，减少握手开销

# config.yaml
github:
  username: "tiangolo"
  token: "ghp_your_token"
filter:
  language: "Python"
  min_stars: 1000  # 只要高质量项目
download:
  mode: "git"      # 保留完整Git历史
  save_path: "./learning/tiangolo"
# 执行
uv run main.py

# backup.sh (Cron Job)
#!/bin/bash
cd /opt/github-backup
uv run main.py \
  --username mycompany \
  --token $GITHUB_TOKEN \
  --mode git \
  --save_path ./archives/$(date +%Y%m%d) \
  --max-workers 3 \
  --report-format csv
# Crontab: 每周日凌晨2点执行
# 0 2 * * 0 /opt/github-backup/backup.sh >> backup.log 2>&1

# .github/workflows/pull-deps.yml
name: Pull Dependencies
on: [push, workflow_dispatch]
jobs:
  pull-repos:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: astral-sh/setup-uv@v3
      - name: Download dependencies
        run: |
          uv run main.py \
            --username upstream-org \
            --token ${{ secrets.GITHUB_TOKEN }} \
            --language TypeScript \
            --mode zip \          # ZIP更快，不需要Git历史
            --save_path ./deps \
            --max-workers 10
      - name: Build services
        run: ./build.sh

uv run main.py \
  --username gin-gonic \
  --language Go \
  --updated-after 2024-01-01 \
  --min-stars 5000 \
  --max-repos 20 \
  --mode zip \
  --save_path ./research/go-frameworks

# 浏览器自动化方案
from selenium import webdriver
driver = webdriver.Chrome()
driver.get("https://github.com/user?tab=repositories")
# ... 模拟点击、滚动、提取链接

# ❌ 错误做法：提交到版本控制
github:
  token: "ghp_abc123..."
# ✅ 正确做法：使用环境变量或Secrets管理
github:
  token: ""  # 通过 --token 参数或环境变量传入

# 根据网络带宽调整
# 家庭宽带 (10Mbps): max_workers=3
# 企业网络 (100Mbps): max_workers=5-8
# 数据中心 (1Gbps+): max_workers=10-20
# 监控指标：
# - 如果频繁出现429 (Rate Limit) → 降低workers
# - 如果CPU利用率 < 50% → 可以适当提高

# 预估磁盘空间
# 平均仓库大小: 50-200 MB (含.git目录)
# 100个仓库 ≈ 5-20 GB
# 1000个仓库 ≈ 50-200 GB
# 建议：
# 1. 使用SSD存储（IOPS更高）
# 2. 定期清理旧版本（Git GC）
# 3. 启用压缩文件系统（ZFS/Btrfs）

# 定期检查日志
tail -f app.log | grep ERROR
# 关键告警规则：
# - 单次运行失败率 > 10%
# - 连续3次触发Rate Limit
# - 单个仓库下载时间 > 5分钟