git clone https://github.com/user/repo1.git
git clone https://github.com/user/repo2.git
git clone https://github.com/user/repo3.git
# ... 重復(fù)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:
        # 分頁(yè)請(qǐng)求：每頁(yè)100條，通過page參數(shù)控制
        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
            
        # 錯(cuò)誤碼處理矩陣
        if resp.status_code == 404:          # 用戶不存在
            break
        elif resp.status_code == 403 and "rate limit" in resp.text.lower():
            # 觸發(fā)速率限制
            break
        elif resp.status_code != 200:        # 其他錯(cuò)誤
            break
            
        data = resp.json()
        if not data:  # 無更多數(shù)據(jù)
            break
            
        # 服務(wù)端過濾：在客戶端應(yīng)用復(fù)合過濾條件
        for r in data:
            if language and r.get('language') != language:
                continue                    # 語言不匹配 → 跳過
            if r.get('stargazers_count', 0) < min_stars:
                continue                    # 星標(biāo)不足 → 跳過
            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                # 更新時(shí)間過早 → 跳過
                    
            repos.append(r)
            
            if max_repos > 0 and len(repos) >= max_repos:
                break                        # 達(dá)到上限 → 提前終止
        
        if max_repos > 0 and len(repos) >= max_repos:
            break                            # 提前終止分頁(yè)循環(huán)
            
        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:
        # 第一步：嘗試目標(biāo)分支/標(biāo)簽
        resp = try_download(target_ref)
        
        # 第二步：智能回退到默認(rèn)分支
        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))  # 實(shí)時(shí)更新進(jìn)度
                    
        # 第四步：智能解壓（清理分支后綴）
        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)  # 目標(biāo)目錄名（不含后綴）
        
        # 重命名：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  # 不可恢復(fù)錯(cuò)誤，直接拋出
    except Exception as e:
        raise RetryableError(str(e))  # 可恢復(fù)錯(cuò)誤，觸發(fā)重試

@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')):
        # ? 已存在 → 執(zhí)行Pull更新
        git_repo = Repo(repo_path)
        origin = git_repo.remotes.origin
        origin.pull()
        
        # 嘗試切換到目標(biāo)分支
        try:
            git_repo.git.checkout(target_ref)
        except GitCommandError:
            # 目標(biāo)分支不存在 → 回退到默認(rèn)分支
            if target_ref != fallback_ref:
                git_repo.git.checkout(fallback_ref)
    else:
        # ? 不存在 → 執(zhí)行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(...)  # 獲取倉(cāng)庫(kù)列表
    
    stats = {"success": 0, "failed": 0, "skipped": 0}
    stats_lock = threading.Lock()  # 線程安全計(jì)數(shù)器
    
    with Progress(...) as progress:
        overall_task = progress.add_task("[bold blue]Overall Progress", total=len(repos))
        
        # 創(chuàng)建線程池
        with ThreadPoolExecutor(max_workers=opts['max_workers']) as executor:
            # 提交所有任務(wù)
            futures = [
                executor.submit(process_repo, repo, opts, history, 
                              progress, overall_task, stats, stats_lock) 
                for repo in repos
            ]
            
            # 按完成順序收集結(jié)果（非阻塞）
            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):
    """單個(gè)倉(cāng)庫(kù)的處理邏輯（在線程池中運(yùn)行）"""
    repo_name = repo['name']
    task_id = progress.add_task(f"Processing {repo_name}...", total=100)
    
    # 增量同步檢查：比較時(shí)間戳
    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:
        # 執(zhí)行下載（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  # 保持舊時(shí)間戳以便下次重試

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

def load_sync_history(save_path):
    """加載上次同步的時(shí)間戳"""
    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):
    """保存本次同步的時(shí)間戳"""
    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格式的報(bào)表"""
    format = opts.get('report_format', 'markdown').lower()
    out_dir = opts.get('report_dir', '.')
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")  # 時(shí)間戳命名
    
    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):
    """
    三級(jí)回退策略：
    Level 1: 嘗試目標(biāo)分支/標(biāo)簽
    Level 2: 如果失敗且不是默認(rèn)分支 → 回退到默認(rèn)分支
    Level 3: 如果還是失敗 → 拋出不可恢復(fù)異常
    """
    
    # Level 1: 目標(biāo)引用
    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):
    """可重試錯(cuò)誤：臨時(shí)性故障，重試可能成功"""
    pass

class NonRetryableError(Exception):
    """不可重試錯(cuò)誤：永久性故障，重試也無用"""
    pass

@retry(
    retry=retry_if_exception_type(RetryableError),  # 僅對(duì)可重試錯(cuò)誤生效
    stop=stop_after_attempt(3),                      # 最多重試3次
    wait=wait_fixed(2)                               # 固定間隔2秒
)
def download_zip(...):
    ...

# Layer 1: 默認(rèn)值 (Hardcoded)
# Layer 2: YAML配置文件 (config.yaml)
# Layer 3: CLI命令行參數(shù) (最高優(yōu)先級(jí))

def parse_and_merge_args():
    args = parser.parse_args()           # 解析CLI參數(shù)
    config = load_config(args.config)    # 加載YAML
    
    # 合并策略：CLI參數(shù)覆蓋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),
        # ...
    }

# 場(chǎng)景1：使用YAML默認(rèn)配置
uv run main.py
# 場(chǎng)景2：臨時(shí)覆蓋部分參數(shù)
uv run main.py --username octocat --language Python --max-repos 10
# 場(chǎng)景3：完全自定義（忽略YAML）
uv run main.py --username octocat --token ghp_xxx --mode zip --max-workers 8

# ? 一次性讀取整個(gè)文件到內(nèi)存（OOM風(fēng)險(xiǎn)）
content = resp.content
with open(file, 'wb') as f:
    f.write(content)

# ? 流式分塊寫入（內(nèi)存友好）
for chunk in resp.iter_content(chunk_size=8192):
    f.write(chunk)
    progress.advance(task_id, len(chunk))  # 即時(shí)更新UI

# 優(yōu)化建議（未來改進(jìn)）
session = requests.Session()
session.headers.update({"Authorization": f"token {token}"})

# 所有請(qǐng)求共用TCP連接
for repo in repos:
    session.get(url)  # 復(fù)用連接，減少握手開銷

# config.yaml
github:
  username: "tiangolo"
  token: "ghp_your_token"
filter:
  language: "Python"
  min_stars: 1000  # 只要高質(zhì)量項(xiàng)目
download:
  mode: "git"      # 保留完整Git歷史
  save_path: "./learning/tiangolo"
# 執(zhí)行
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點(diǎn)執(zhí)行
# 0 2 * * 0 /opt/github-backup/backup.sh >> backup.log 2>&1

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

# 瀏覽器自動(dòng)化方案
from selenium import webdriver
driver = webdriver.Chrome()
driver.get("https://github.com/user?tab=repositories")
# ... 模擬點(diǎn)擊、滾動(dòng)、提取鏈接

# ? 錯(cuò)誤做法：提交到版本控制
github:
  token: "ghp_abc123..."
# ? 正確做法：使用環(huán)境變量或Secrets管理
github:
  token: ""  # 通過 --token 參數(shù)或環(huán)境變量傳入

# 根據(jù)網(wǎng)絡(luò)帶寬調(diào)整
# 家庭寬帶 (10Mbps): max_workers=3
# 企業(yè)網(wǎng)絡(luò) (100Mbps): max_workers=5-8
# 數(shù)據(jù)中心 (1Gbps+): max_workers=10-20
# 監(jiān)控指標(biāo)：
# - 如果頻繁出現(xiàn)429 (Rate Limit) → 降低workers
# - 如果CPU利用率 < 50% → 可以適當(dāng)提高

# 預(yù)估磁盤空間
# 平均倉(cāng)庫(kù)大小: 50-200 MB (含.git目錄)
# 100個(gè)倉(cāng)庫(kù) ≈ 5-20 GB
# 1000個(gè)倉(cāng)庫(kù) ≈ 50-200 GB
# 建議：
# 1. 使用SSD存儲(chǔ)（IOPS更高）
# 2. 定期清理舊版本（Git GC）
# 3. 啟用壓縮文件系統(tǒng)（ZFS/Btrfs）

# 定期檢查日志
tail -f app.log | grep ERROR
# 關(guān)鍵告警規(guī)則：
# - 單次運(yùn)行失敗率 > 10%
# - 連續(xù)3次觸發(fā)Rate Limit
# - 單個(gè)倉(cāng)庫(kù)下載時(shí)間 > 5分鐘