305 lines
12 KiB
Python
305 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
双向同步命令行工具
|
||
提供简单易用的命令行界面来执行双向同步操作
|
||
"""
|
||
|
||
import argparse
|
||
import os
|
||
import sys
|
||
from typing import Dict, Any
|
||
|
||
# 添加项目路径
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from issue_sync_module.clients.sync_service import IssueSyncService
|
||
|
||
|
||
def check_environment_variables(platforms: list) -> bool:
|
||
"""检查必要的环境变量是否已设置"""
|
||
missing_vars = []
|
||
|
||
for platform in platforms:
|
||
if platform.lower() == 'gitlink':
|
||
# GitLink支持Token或Cookie，任何一个都可以
|
||
token = os.getenv('GITLINK_TOKEN')
|
||
cookie = os.getenv('GITLINK_COOKIE')
|
||
if not token and not cookie:
|
||
missing_vars.append('GITLINK_TOKEN或GITLINK_COOKIE')
|
||
elif platform.lower() == 'github':
|
||
if not os.getenv('GITHUB_TOKEN'):
|
||
missing_vars.append('GITHUB_TOKEN')
|
||
elif platform.lower() == 'gitee':
|
||
if not os.getenv('GITEE_TOKEN'):
|
||
missing_vars.append('GITEE_TOKEN')
|
||
|
||
if missing_vars:
|
||
print(f"❌ 缺少环境变量: {', '.join(missing_vars)}")
|
||
print("请设置所需的认证信息:")
|
||
for var in missing_vars:
|
||
if 'GITLINK' in var:
|
||
print(f" GitLink (二选一):")
|
||
print(f" $env:GITLINK_TOKEN='your_token_here' # 或者")
|
||
print(f" $env:GITLINK_COOKIE='your_cookie_here'")
|
||
elif 'GITHUB' in var:
|
||
print(f" $env:GITHUB_TOKEN='your_github_token'")
|
||
elif 'GITEE' in var:
|
||
print(f" $env:GITEE_TOKEN='your_gitee_token'")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def print_sync_result(result: Dict[str, Any], platform_a: str, platform_b: str):
|
||
"""美化打印同步结果"""
|
||
print("=" * 60)
|
||
print("🎉 双向同步完成!")
|
||
print("=" * 60)
|
||
|
||
# 获取统计数据
|
||
key_a_to_b = f"{platform_a.lower()}_to_{platform_b.lower()}"
|
||
key_b_to_a = f"{platform_b.lower()}_to_{platform_a.lower()}"
|
||
|
||
a_to_b = result.get(key_a_to_b, {})
|
||
b_to_a = result.get(key_b_to_a, {})
|
||
|
||
# 显示详细统计
|
||
print(f"📤 {platform_a} → {platform_b}:")
|
||
print(f" ✅ 新建: {a_to_b.get('created', 0)}")
|
||
print(f" 🔄 更新: {a_to_b.get('updated', 0)}")
|
||
print(f" ⏭️ 跳过: {a_to_b.get('skipped', 0)}")
|
||
print(f" ❌ 失败: {a_to_b.get('failed', 0)}")
|
||
|
||
print(f"\n📥 {platform_b} → {platform_a}:")
|
||
print(f" ✅ 新建: {b_to_a.get('created', 0)}")
|
||
print(f" 🔄 更新: {b_to_a.get('updated', 0)}")
|
||
print(f" ⏭️ 跳过: {b_to_a.get('skipped', 0)}")
|
||
print(f" ❌ 失败: {b_to_a.get('failed', 0)}")
|
||
|
||
# 显示总体统计
|
||
total_created = a_to_b.get('created', 0) + b_to_a.get('created', 0)
|
||
total_updated = a_to_b.get('updated', 0) + b_to_a.get('updated', 0)
|
||
total_conflicts = result.get('conflicts_resolved', 0)
|
||
total_processed = result.get('total_processed', 0)
|
||
|
||
print(f"\n🔄 总体统计:")
|
||
print(f" 📊 处理Issue总数: {total_processed}")
|
||
print(f" ➕ 新建Issue: {total_created}")
|
||
print(f" 🔄 更新Issue: {total_updated}")
|
||
print(f" ⚖️ 解决冲突: {total_conflicts}")
|
||
|
||
# 显示里程碑同步结果（如果有）
|
||
if result.get('milestone_sync_result'):
|
||
print(f"\n🏁 里程碑同步:")
|
||
milestone_result = result['milestone_sync_result']
|
||
for direction, data in milestone_result.items():
|
||
if isinstance(data, dict):
|
||
created_key = [k for k in data.keys() if k.startswith('created_in_')]
|
||
if created_key:
|
||
created_count = data.get(created_key[0], 0)
|
||
direction_display = direction.replace('_to_', '→').replace('_', ' ').title()
|
||
print(f" {direction_display}: {created_count} 个")
|
||
|
||
|
||
def cmd_gitlink_github(args):
|
||
"""执行GitLink ↔ GitHub 双向同步"""
|
||
if not check_environment_variables(['gitlink', 'github']):
|
||
return False
|
||
|
||
print("🔄 开始GitLink ↔ GitHub 双向同步...")
|
||
print(f"GitLink: {args.gitlink_org}/{args.gitlink_repo}")
|
||
print(f"GitHub: {args.github_org}/{args.github_repo}")
|
||
print(f"冲突策略: {args.conflict_strategy}")
|
||
|
||
sync_service = IssueSyncService()
|
||
|
||
try:
|
||
result = sync_service.bidirectional_sync_gitlink_github(
|
||
gitlink_org=args.gitlink_org,
|
||
gitlink_repo=args.gitlink_repo,
|
||
github_org=args.github_org,
|
||
github_repo=args.github_repo,
|
||
conflict_strategy=args.conflict_strategy,
|
||
sync_milestones=args.sync_milestones,
|
||
enable_deletion=args.enable_deletion
|
||
)
|
||
|
||
print_sync_result(result, "GitLink", "GitHub")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 同步失败: {str(e)}")
|
||
return False
|
||
|
||
|
||
def cmd_gitlink_gitee(args):
|
||
"""执行GitLink ↔ Gitee 双向同步"""
|
||
if not check_environment_variables(['gitlink', 'gitee']):
|
||
return False
|
||
|
||
print("🔄 开始GitLink ↔ Gitee 双向同步...")
|
||
print(f"GitLink: {args.gitlink_org}/{args.gitlink_repo}")
|
||
print(f"Gitee: {args.gitee_org}/{args.gitee_repo}")
|
||
print(f"冲突策略: {args.conflict_strategy}")
|
||
|
||
sync_service = IssueSyncService()
|
||
|
||
try:
|
||
result = sync_service.bidirectional_sync_gitlink_gitee(
|
||
gitlink_org=args.gitlink_org,
|
||
gitlink_repo=args.gitlink_repo,
|
||
gitee_org=args.gitee_org,
|
||
gitee_repo=args.gitee_repo,
|
||
conflict_strategy=args.conflict_strategy,
|
||
sync_milestones=args.sync_milestones,
|
||
sync_comments=args.sync_comments,
|
||
enable_deletion=args.enable_deletion
|
||
)
|
||
|
||
print_sync_result(result, "GitLink", "Gitee")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 同步失败: {str(e)}")
|
||
return False
|
||
|
||
|
||
def cmd_github_gitee(args):
|
||
"""执行GitHub ↔ Gitee 双向同步"""
|
||
if not check_environment_variables(['github', 'gitee']):
|
||
return False
|
||
|
||
print("🔄 开始GitHub ↔ Gitee 双向同步...")
|
||
print(f"GitHub: {args.github_org}/{args.github_repo}")
|
||
print(f"Gitee: {args.gitee_org}/{args.gitee_repo}")
|
||
print(f"冲突策略: {args.conflict_strategy}")
|
||
|
||
sync_service = IssueSyncService()
|
||
|
||
try:
|
||
result = sync_service.bidirectional_sync_github_gitee(
|
||
github_org=args.github_org,
|
||
github_repo=args.github_repo,
|
||
gitee_org=args.gitee_org,
|
||
gitee_repo=args.gitee_repo,
|
||
conflict_strategy=args.conflict_strategy,
|
||
enable_deletion=args.enable_deletion
|
||
)
|
||
|
||
print_sync_result(result, "GitHub", "Gitee")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 同步失败: {str(e)}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
parser = argparse.ArgumentParser(
|
||
description="双向Issue同步工具 - 让两个代码托管平台的Issue实现并集同步",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
示例用法:
|
||
|
||
# GitLink ↔ GitHub 双向同步
|
||
python cli_bidirectional_sync.py gitlink-github \\
|
||
--gitlink-org myorg --gitlink-repo myproject \\
|
||
--github-org myorg --github-repo myproject
|
||
|
||
# GitLink ↔ Gitee 双向同步（包含评论）
|
||
python cli_bidirectional_sync.py gitlink-gitee \\
|
||
--gitlink-org myorg --gitlink-repo myproject \\
|
||
--gitee-org myorg --gitee-repo myproject \\
|
||
--sync-comments
|
||
|
||
# GitHub ↔ Gitee 双向同步（优先GitHub版本）
|
||
python cli_bidirectional_sync.py github-gitee \\
|
||
--github-org myorg --github-repo myproject \\
|
||
--gitee-org myorg --gitee-repo myproject \\
|
||
--conflict-strategy prefer_github
|
||
|
||
环境变量:
|
||
GITLINK_TOKEN GitLink访问令牌
|
||
GITHUB_TOKEN GitHub访问令牌
|
||
GITEE_TOKEN Gitee访问令牌
|
||
"""
|
||
)
|
||
|
||
# 创建子命令
|
||
subparsers = parser.add_subparsers(dest='command', help='选择同步平台组合')
|
||
|
||
# GitLink ↔ GitHub 子命令
|
||
parser_gl_gh = subparsers.add_parser('gitlink-github', help='GitLink ↔ GitHub 双向同步')
|
||
parser_gl_gh.add_argument('--gitlink-org', required=True, help='GitLink组织名')
|
||
parser_gl_gh.add_argument('--gitlink-repo', required=True, help='GitLink仓库名')
|
||
parser_gl_gh.add_argument('--github-org', required=True, help='GitHub组织名')
|
||
parser_gl_gh.add_argument('--github-repo', required=True, help='GitHub仓库名')
|
||
parser_gl_gh.add_argument('--conflict-strategy', default='prefer_newer',
|
||
choices=['prefer_newer', 'prefer_gitlink', 'prefer_github'],
|
||
help='冲突解决策略 (默认: prefer_newer)')
|
||
parser_gl_gh.add_argument('--sync-milestones', action='store_true', default=True, help='同步里程碑')
|
||
parser_gl_gh.add_argument('--no-sync-milestones', dest='sync_milestones', action='store_false', help='不同步里程碑')
|
||
parser_gl_gh.add_argument('--enable-deletion', action='store_true', help='启用删除同步')
|
||
parser_gl_gh.set_defaults(func=cmd_gitlink_github)
|
||
|
||
# GitLink ↔ Gitee 子命令
|
||
parser_gl_ge = subparsers.add_parser('gitlink-gitee', help='GitLink ↔ Gitee 双向同步')
|
||
parser_gl_ge.add_argument('--gitlink-org', required=True, help='GitLink组织名')
|
||
parser_gl_ge.add_argument('--gitlink-repo', required=True, help='GitLink仓库名')
|
||
parser_gl_ge.add_argument('--gitee-org', required=True, help='Gitee组织名')
|
||
parser_gl_ge.add_argument('--gitee-repo', required=True, help='Gitee仓库名')
|
||
parser_gl_ge.add_argument('--conflict-strategy', default='prefer_newer',
|
||
choices=['prefer_newer', 'prefer_gitlink', 'prefer_gitee'],
|
||
help='冲突解决策略 (默认: prefer_newer)')
|
||
parser_gl_ge.add_argument('--sync-milestones', action='store_true', default=True, help='同步里程碑')
|
||
parser_gl_ge.add_argument('--no-sync-milestones', dest='sync_milestones', action='store_false', help='不同步里程碑')
|
||
parser_gl_ge.add_argument('--sync-comments', action='store_true', help='同步评论')
|
||
parser_gl_ge.add_argument('--enable-deletion', action='store_true', help='启用删除同步')
|
||
parser_gl_ge.set_defaults(func=cmd_gitlink_gitee)
|
||
|
||
# GitHub ↔ Gitee 子命令
|
||
parser_gh_ge = subparsers.add_parser('github-gitee', help='GitHub ↔ Gitee 双向同步')
|
||
parser_gh_ge.add_argument('--github-org', required=True, help='GitHub组织名')
|
||
parser_gh_ge.add_argument('--github-repo', required=True, help='GitHub仓库名')
|
||
parser_gh_ge.add_argument('--gitee-org', required=True, help='Gitee组织名')
|
||
parser_gh_ge.add_argument('--gitee-repo', required=True, help='Gitee仓库名')
|
||
parser_gh_ge.add_argument('--conflict-strategy', default='prefer_newer',
|
||
choices=['prefer_newer', 'prefer_github', 'prefer_gitee'],
|
||
help='冲突解决策略 (默认: prefer_newer)')
|
||
parser_gh_ge.add_argument('--enable-deletion', action='store_true', help='启用删除同步')
|
||
parser_gh_ge.set_defaults(func=cmd_github_gitee)
|
||
|
||
# 解析参数
|
||
args = parser.parse_args()
|
||
|
||
if not args.command:
|
||
parser.print_help()
|
||
return
|
||
|
||
# 显示启动信息
|
||
print("🚀 双向Issue同步工具")
|
||
print("=" * 60)
|
||
|
||
# 执行对应的命令
|
||
success = args.func(args)
|
||
|
||
if success:
|
||
print("\n✅ 同步操作完成!")
|
||
sys.exit(0)
|
||
else:
|
||
print("\n❌ 同步操作失败!")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\n👋 用户中断，退出程序")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"\n💥 程序异常: {str(e)}")
|
||
sys.exit(1) |