#!/usr/bin/env python3 """ 多仓库同步工具 - 支持多方向同步 支持动态环境变量前缀，适用于 A->B, B->C 等多场景 """ import yaml import os import sys import subprocess from pathlib import Path import tempfile import shutil import time import argparse def expand_env_vars(text, env_prefix): """使用指定的前缀扩展环境变量""" if not text: return text # 构建完整的变量名 var_mappings = { 'USERNAME': f'{env_prefix}_USERNAME', 'TOKEN': f'{env_prefix}_TOKEN', } # 替换 ${VAR} 和 $VAR 格式 for placeholder, env_var in var_mappings.items(): value = os.environ.get(env_var, '') if not value: raise ValueError(f"环境变量 {env_var} 未设置") # 替换 ${VAR} text = text.replace(f'${{{placeholder}}}', value) # 替换 $VAR text = text.replace(f'${placeholder}', value) # 如果还有未替换的变量，报错 if '${' in text: raise ValueError(f"URL 中包含未解析的变量: {text}") return text def sync_repository(repo_config, env_prefix): """同步单个仓库""" name = repo_config.get('name', 'unnamed-repo') source_url_template = repo_config.get('source_url', '') target_url_template = repo_config.get('target_url', '') branch = repo_config.get('branch', 'master') if not source_url_template or not target_url_template: print(f"❌ [{name}] 错误: 缺少 source_url 或 target_url") return False try: # 替换环境变量 source_url = expand_env_vars(source_url_template, env_prefix) target_url = expand_env_vars(target_url_template, env_prefix) except ValueError as e: print(f"❌ [{name}] 错误: {e}") return False # 打印信息（隐藏敏感信息） def hide_credentials(url): if '@' in url: parts = url.split('@') return f"***:***@{parts[1]}" return url print(f"\n{'='*60}") print(f"开始同步仓库: {name}") print(f"源地址: {hide_credentials(source_url)}") print(f"目标地址: {hide_credentials(target_url)}") print(f"分支: {branch}") print(f"{'='*60}") # 创建临时工作目录 work_dir = tempfile.mkdtemp(prefix=f"sync_{name}_") try: # 克隆目标仓库 print(f"\n[1/6] 克隆目标仓库...") result = subprocess.run( ['git', 'clone', target_url, work_dir], capture_output=True, text=True, timeout=300 ) if result.returncode != 0: print(f"❌ 克隆目标仓库失败:") print(result.stderr) return False os.chdir(work_dir) # 添加上游远程 print(f"[2/6] 添加上游远程...") result = subprocess.run( ['git', 'remote', 'add', 'upstream', source_url], capture_output=True, text=True ) if result.returncode != 0: print(f"❌ 添加上游远程失败:") print(result.stderr) return False # 获取上游更改 print(f"[3/6] 获取上游更改...") max_retries = 3 for attempt in range(1, max_retries + 1): try: result = subprocess.run( ['git', 'fetch', 'upstream'], capture_output=True, text=True, timeout=200 ) if result.returncode == 0: break print(f"❌ 获取上游更改失败 (尝试 {attempt}/{max_retries}):") print(result.stderr) except subprocess.TimeoutExpired: print(f"❌ 获取上游更改超时 (尝试 {attempt}/{max_retries})，200秒限制") if attempt < max_retries: print(f"⏳ 等待5秒后重试...") time.sleep(5) else: print(f"\n网络诊断信息:") print(f"- 错误类型: DNS解析或网络连接失败") print(f"- 可能原因:") print(f" 1. DNS配置问题") print(f" 2. 无法访问外部网络") print(f" 3. 上游仓库服务器限制") return False # 检查目标分支是否存在 print(f"[4/6] 检查分支...") result = subprocess.run( ['git', 'branch', '-a'], capture_output=True, text=True ) remote_branch = f'upstream/{branch}' if remote_branch not in result.stdout: print(f"❌ 上游分支 {branch} 不存在!") print(f"可用的分支:") print(result.stdout) return False # 同步操作：重置到上游 print(f"[5/6] 同步到上游分支 (使用 reset --hard)...") subprocess.run(['git', 'fetch', 'origin'], capture_output=True) result = subprocess.run( ['git', 'reset', '--hard', f'upstream/{branch}'], capture_output=True, text=True ) if result.returncode != 0: print(f"❌ 同步失败:") print(result.stderr) return False # 推送到目标 print(f"[6/6] 推送到目标仓库...") result = subprocess.run( ['git', 'push', '--force', '--tags', 'origin', f'refs/heads/{branch}'], capture_output=True, text=True, timeout=600 ) if result.returncode != 0: print(f"❌ 推送到目标仓库失败:") print(result.stderr) return False print(f"✅ [{name}] 同步成功!") return True except subprocess.TimeoutExpired: print(f"❌ [{name}] 操作超时!") return False except Exception as e: print(f"❌ [{name}] 发生错误: {e}") return False finally: # 返回原始目录并清理 os.chdir('/') if Path(work_dir).exists(): shutil.rmtree(work_dir, ignore_errors=True) def load_config(config_file): """加载配置文件""" if not os.path.exists(config_file): print(f"错误: 配置文件 {config_file} 不存在!") sys.exit(1) try: with open(config_file, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) if not isinstance(config, dict) or 'repositories' not in config: print(f"错误: 配置文件格式不正确，需要包含 'repositories' 键") sys.exit(1) repos = config['repositories'] if not isinstance(repos, list) or len(repos) == 0: print(f"警告: 配置文件中未找到仓库配置") return [], None # 获取同步任务名称（可选） sync_name = config.get('name', 'unnamed-sync') # 过滤掉注释掉的或空配置 valid_repos = [ repo for repo in repos if isinstance(repo, dict) and repo.get('name') and not repo.get('name', '').strip().startswith('#') and (repo.get('source_url') or '').strip() ] print(f"✓ 配置任务: {sync_name}") print(f"✓ 找到 {len(valid_repos)} 个仓库配置") return valid_repos, sync_name except yaml.YAMLError as e: print(f"错误: 解析YAML配置文件失败: {e}") sys.exit(1) except Exception as e: print(f"错误: 读取配置文件失败: {e}") sys.exit(1) def main(): parser = argparse.ArgumentParser(description='多仓库同步工具') parser.add_argument('config_file', help='配置文件路径') parser.add_argument('--env-prefix', required=True, help='环境变量前缀 (例如: SYNC_A_B)') args = parser.parse_args() print("="*80) print("多仓库同步工具") print("="*80) print(f"\n开始时间: {time.strftime('%Y-%m-%d %H:%M:%S')}") print(f"配置文件: {args.config_file}") print(f"环境前缀: {args.env_prefix}") # 加载配置 repos, sync_name = load_config(args.config_file) if not repos: print("\n⚠️ 没有需要同步的仓库，退出") sys.exit(0) # 检查必要的环境变量 required_vars = [ f'{args.env_prefix}_USERNAME', f'{args.env_prefix}_TOKEN' ] missing_vars = [var for var in required_vars if not os.environ.get(var)] if missing_vars: print(f"\n❌ 错误: 缺少必要的环境变量: {', '.join(missing_vars)}") print(f"请设置以下环境变量:") print(f" - {args.env_prefix}_USERNAME") print(f" - {args.env_prefix}_TOKEN") sys.exit(1) print(f"✓ 所有必要环境变量已设置") # 同步所有仓库 results = [] for i, repo in enumerate(repos, 1): print(f"\n[{i}/{len(repos)}] 正在同步...") success = sync_repository(repo, args.env_prefix) results.append((repo['name'], success)) # 在仓库之间等待一下，避免过于频繁的操作 if i < len(repos): time.sleep(5) # 总结报告 print("\n" + "="*80) print("同步完成报告") print("="*80) for name, success in results: status = "✅ 成功" if success else "❌ 失败" print(f"{status} {name}") successful = sum(1 for _, s in results if s) failed = len(results) - successful print(f"\n总计: {len(results)} 个仓库") print(f"成功: {successful} 个") print(f"失败: {failed} 个") if failed > 0: print("\n❌ 有仓库同步失败，请查看详细日志") sys.exit(1) else: print("\n✅ 所有仓库同步成功!") print(f"\n结束时间: {time.strftime('%Y-%m-%d %H:%M:%S')}") if __name__ == '__main__': main()