385 lines
15 KiB
Python
385 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
Webhook处理器 - 自动同步Gitee和GitLink之间的Issues
|
||
支持防循环机制，避免无限同步
|
||
"""
|
||
|
||
import time
|
||
import json
|
||
import hashlib
|
||
import requests
|
||
from typing import Dict, Any, Optional
|
||
from datetime import datetime, timedelta
|
||
|
||
# 兼容导入：优先使用环境变量，后备使用src模块
|
||
try:
|
||
from src.utils.logger import logger
|
||
except ImportError:
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
logging.basicConfig(level=logging.INFO)
|
||
|
||
|
||
class LoopDetector:
|
||
"""循环检测器 - 防止无限同步循环"""
|
||
|
||
def __init__(self, window_minutes: int = 5):
|
||
self.window_minutes = window_minutes
|
||
self.recent_syncs = {} # {issue_key: timestamp}
|
||
|
||
def _get_issue_key(self, platform: str, org: str, repo: str, issue_title: str) -> str:
|
||
"""生成Issue的唯一标识"""
|
||
return f"{platform}:{org}/{repo}:{issue_title}"
|
||
|
||
def should_sync(self, platform: str, org: str, repo: str, issue_title: str) -> bool:
|
||
"""检查是否应该同步（避免循环）"""
|
||
issue_key = self._get_issue_key(platform, org, repo, issue_title)
|
||
current_time = datetime.now()
|
||
|
||
# 清理过期记录
|
||
self._cleanup_old_records(current_time)
|
||
|
||
# 检查是否在时间窗口内已经同步过
|
||
if issue_key in self.recent_syncs:
|
||
last_sync = self.recent_syncs[issue_key]
|
||
if current_time - last_sync < timedelta(minutes=self.window_minutes):
|
||
logger.info(f"🔄 跳过同步 - Issue '{issue_title}' 在 {self.window_minutes} 分钟内已同步过")
|
||
return False
|
||
|
||
# 记录本次同步
|
||
self.recent_syncs[issue_key] = current_time
|
||
logger.info(f"✅ 允许同步 - Issue '{issue_title}' 可以进行同步")
|
||
return True
|
||
|
||
def _cleanup_old_records(self, current_time: datetime):
|
||
"""清理过期的同步记录"""
|
||
cutoff_time = current_time - timedelta(minutes=self.window_minutes * 2)
|
||
expired_keys = [
|
||
key for key, timestamp in self.recent_syncs.items()
|
||
if timestamp < cutoff_time
|
||
]
|
||
for key in expired_keys:
|
||
del self.recent_syncs[key]
|
||
|
||
|
||
class WebhookHandler:
|
||
"""Webhook处理器"""
|
||
|
||
def __init__(self):
|
||
self.loop_detector = LoopDetector(window_minutes=5)
|
||
self.sync_api_url = "http://localhost:8002/sync/immediate"
|
||
|
||
# Webhook密码验证
|
||
self.webhook_password = "hzk200407140238"
|
||
|
||
# 仓库配置
|
||
self.repo_config = {
|
||
"gitee": {
|
||
"org": "ttk00",
|
||
"repo": "testdemo"
|
||
},
|
||
"gitlink": {
|
||
"org": "qinxinqi",
|
||
"repo": "testdemo"
|
||
}
|
||
}
|
||
|
||
def detect_platform(self, webhook_data: Dict[str, Any], headers: Dict[str, str] = None) -> Optional[str]:
|
||
"""检测webhook来源平台"""
|
||
|
||
# 检测Gitee webhook - 通过headers
|
||
if headers:
|
||
if 'x-gitee-event' in headers or 'x-git-oschina-event' in headers:
|
||
logger.info(f"🔍 检测到Gitee webhook (通过headers)")
|
||
return "gitee"
|
||
|
||
# 检测GitLink webhook - 通过headers (GitLink使用Gitea/Gogs格式)
|
||
if 'x-gitea-event' in headers or 'x-gogs-event' in headers or 'x-github-event' in headers:
|
||
logger.info(f"🔍 检测到GitLink webhook (通过headers)")
|
||
return "gitlink"
|
||
|
||
# 检测Gitee webhook - 通过数据结构
|
||
if 'repository' in webhook_data and 'html_url' in webhook_data.get('repository', {}):
|
||
repo_url = webhook_data['repository']['html_url']
|
||
if 'gitee.com' in repo_url:
|
||
logger.info(f"🔍 检测到Gitee webhook: {repo_url}")
|
||
return "gitee"
|
||
|
||
# 检测Gitee webhook - 通过project字段
|
||
if 'project' in webhook_data and 'html_url' in webhook_data.get('project', {}):
|
||
project_url = webhook_data['project']['html_url']
|
||
if 'gitee.com' in project_url:
|
||
logger.info(f"🔍 检测到Gitee webhook: {project_url}")
|
||
return "gitee"
|
||
|
||
# 检测GitLink webhook - 通过project字段和mirror_url
|
||
if 'project' in webhook_data and 'identifier' in webhook_data.get('project', {}):
|
||
project = webhook_data['project']
|
||
mirror_url = project.get('mirror_url', '')
|
||
if 'gitee.com' in mirror_url:
|
||
logger.info(f"🔍 检测到GitLink webhook (镜像自Gitee): {project.get('identifier', 'unknown')}")
|
||
return "gitlink"
|
||
else:
|
||
logger.info(f"🔍 检测到GitLink webhook: {project.get('identifier', 'unknown')}")
|
||
return "gitlink"
|
||
|
||
# 通过User-Agent检测
|
||
if headers and 'user-agent' in headers:
|
||
user_agent = headers['user-agent'].lower()
|
||
if 'git-oschina-hook' in user_agent:
|
||
return "gitee"
|
||
elif 'gitlink' in user_agent:
|
||
return "gitlink"
|
||
|
||
logger.warning(f"⚠️ 无法识别webhook来源平台，数据结构: {list(webhook_data.keys())}")
|
||
if headers:
|
||
logger.warning(f"⚠️ Headers: {list(headers.keys())}")
|
||
return None
|
||
|
||
def extract_issue_info(self, webhook_data: Dict[str, Any], platform: str) -> Optional[Dict[str, str]]:
|
||
"""提取Issue信息"""
|
||
try:
|
||
if platform == "gitee":
|
||
# Gitee webhook结构 - 支持新格式
|
||
issue = webhook_data.get('issue', {})
|
||
return {
|
||
"title": issue.get('title', webhook_data.get('title', '')),
|
||
"action": webhook_data.get('action', ''),
|
||
"number": str(issue.get('number', webhook_data.get('iid', ''))),
|
||
"state": issue.get('state', webhook_data.get('state', '')),
|
||
"body": issue.get('body', issue.get('description', '')),
|
||
"user": issue.get('user', {}).get('login', '')
|
||
}
|
||
|
||
elif platform == "gitlink":
|
||
# GitLink webhook结构 - 支持新格式
|
||
issue = webhook_data.get('issue', {})
|
||
action = webhook_data.get('action', '')
|
||
|
||
# 处理GitLink的状态信息
|
||
status = issue.get('status', {})
|
||
status_name = status.get('name', '') if isinstance(status, dict) else issue.get('status_name', '')
|
||
|
||
return {
|
||
"title": issue.get('subject', ''),
|
||
"action": action,
|
||
"number": str(issue.get('id', issue.get('project_issues_index', ''))),
|
||
"state": status_name,
|
||
"body": issue.get('description', ''),
|
||
"user": issue.get('author', {}).get('login', ''),
|
||
"event_type": webhook_data.get('journal', {}).get('notes', '') if 'journal' in webhook_data else ''
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 提取Issue信息失败: {str(e)}")
|
||
return None
|
||
|
||
return None
|
||
|
||
def verify_webhook_password(self, webhook_data: Dict[str, Any], headers: Dict[str, str] = None) -> bool:
|
||
"""验证webhook密码"""
|
||
|
||
# 检查数据中的密码字段
|
||
if 'password' in webhook_data:
|
||
provided_password = webhook_data['password']
|
||
if provided_password == self.webhook_password:
|
||
logger.info("✅ Webhook密码验证通过")
|
||
return True
|
||
else:
|
||
logger.warning(f"❌ Webhook密码验证失败: 提供的密码不匹配")
|
||
return False
|
||
|
||
# 检查headers中的token
|
||
if headers:
|
||
gitee_token = headers.get('x-gitee-token', '')
|
||
if gitee_token == self.webhook_password:
|
||
logger.info("✅ Gitee Token验证通过")
|
||
return True
|
||
|
||
logger.warning("⚠️ 未找到有效的密码或token，跳过验证")
|
||
return True # 如果没有密码字段，暂时允许通过
|
||
|
||
def should_trigger_sync(self, webhook_data: Dict[str, Any], platform: str, headers: Dict[str, str] = None) -> bool:
|
||
"""判断是否应该触发同步"""
|
||
|
||
# 检查是否是Issue相关事件
|
||
if 'issue' not in webhook_data:
|
||
logger.info("📋 非Issue事件，跳过同步")
|
||
return False
|
||
|
||
# 检查事件类型 - GitLink可能是评论事件
|
||
if headers:
|
||
event_type = headers.get('x-gitea-event', headers.get('x-gogs-event', headers.get('x-github-event', '')))
|
||
if event_type == 'issue_comment':
|
||
logger.info("📋 Issue评论事件，暂不触发同步")
|
||
return False
|
||
|
||
# 检查动作类型
|
||
action = webhook_data.get('action', '')
|
||
valid_actions = ['opened', 'open', 'created', 'edited', 'updated', 'closed', 'reopened']
|
||
|
||
if action not in valid_actions:
|
||
logger.info(f"📋 动作 '{action}' 不需要同步")
|
||
return False
|
||
|
||
# 提取Issue信息
|
||
issue_info = self.extract_issue_info(webhook_data, platform)
|
||
if not issue_info or not issue_info.get('title'):
|
||
logger.warning("⚠️ 无法提取Issue标题，跳过同步")
|
||
return False
|
||
|
||
# 检查是否是同步机器人创建的Issue（防循环）
|
||
if self._is_sync_created_issue(webhook_data, platform):
|
||
logger.info(f"🤖 检测到同步机器人创建的Issue，跳过同步")
|
||
return False
|
||
|
||
# 使用循环检测器
|
||
source_config = self.repo_config[platform]
|
||
return self.loop_detector.should_sync(
|
||
platform,
|
||
source_config["org"],
|
||
source_config["repo"],
|
||
issue_info["title"]
|
||
)
|
||
|
||
def _is_sync_created_issue(self, webhook_data: Dict[str, Any], platform: str) -> bool:
|
||
"""检查是否是同步机器人创建的Issue"""
|
||
|
||
if platform == "gitee":
|
||
# 检查Gitee的用户信息
|
||
issue = webhook_data.get('issue', {})
|
||
user = issue.get('user', {})
|
||
username = user.get('login', '').lower()
|
||
|
||
# 如果是同步相关的用户名，认为是机器人操作
|
||
sync_usernames = ['sync-bot', 'reposync', 'auto-sync']
|
||
if any(sync_name in username for sync_name in sync_usernames):
|
||
return True
|
||
|
||
elif platform == "gitlink":
|
||
# 检查GitLink的用户信息
|
||
issue = webhook_data.get('issue', {})
|
||
author = issue.get('author', {})
|
||
username = author.get('login', '').lower()
|
||
|
||
# 如果是同步相关的用户名，认为是机器人操作
|
||
sync_usernames = ['sync-bot', 'reposync', 'auto-sync']
|
||
if any(sync_name in username for sync_name in sync_usernames):
|
||
return True
|
||
|
||
return False
|
||
|
||
def call_sync_api(self, source_platform: str) -> Dict[str, Any]:
|
||
"""调用同步API"""
|
||
|
||
# 根据源平台确定目标平台
|
||
if source_platform == "gitee":
|
||
target_platform = "gitlink"
|
||
source_config = self.repo_config["gitee"]
|
||
target_config = self.repo_config["gitlink"]
|
||
elif source_platform == "gitlink":
|
||
target_platform = "gitee"
|
||
source_config = self.repo_config["gitlink"]
|
||
target_config = self.repo_config["gitee"]
|
||
else:
|
||
return {"success": False, "message": f"不支持的源平台: {source_platform}"}
|
||
|
||
# 构建同步请求
|
||
sync_request = {
|
||
"source_platform": source_platform,
|
||
"source_org": source_config["org"],
|
||
"source_repo": source_config["repo"],
|
||
"target_platform": target_platform,
|
||
"target_org": target_config["org"],
|
||
"target_repo": target_config["repo"],
|
||
"sync_comments": False,
|
||
"sync_milestones": True,
|
||
"update_existing": True,
|
||
"enable_deletion": False
|
||
}
|
||
|
||
try:
|
||
logger.info(f"🚀 调用同步API: {source_platform} → {target_platform}")
|
||
logger.info(f"📋 同步参数: {json.dumps(sync_request, indent=2, ensure_ascii=False)}")
|
||
|
||
response = requests.post(
|
||
self.sync_api_url,
|
||
json=sync_request,
|
||
headers={"Content-Type": "application/json"},
|
||
timeout=150 # 增加到150秒，适应2分钟左右的同步时间
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
logger.info(f"✅ 同步API调用成功: {result}")
|
||
return result
|
||
else:
|
||
error_msg = f"同步API调用失败: HTTP {response.status_code}"
|
||
logger.error(f"❌ {error_msg}")
|
||
return {"success": False, "message": error_msg}
|
||
|
||
except requests.exceptions.Timeout:
|
||
error_msg = "同步API调用超时"
|
||
logger.error(f"❌ {error_msg}")
|
||
return {"success": False, "message": error_msg}
|
||
except Exception as e:
|
||
error_msg = f"同步API调用异常: {str(e)}"
|
||
logger.error(f"❌ {error_msg}")
|
||
return {"success": False, "message": error_msg}
|
||
|
||
def process_webhook(self, webhook_data: Dict[str, Any], headers: Dict[str, str] = None) -> Dict[str, Any]:
|
||
"""处理webhook请求"""
|
||
|
||
logger.info("=" * 60)
|
||
logger.info("🎣 收到新的Webhook请求")
|
||
logger.info(f"📅 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
# 验证密码
|
||
if not self.verify_webhook_password(webhook_data, headers):
|
||
return {
|
||
"success": False,
|
||
"message": "Webhook密码验证失败",
|
||
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
|
||
}
|
||
|
||
# 检测平台
|
||
platform = self.detect_platform(webhook_data, headers)
|
||
if not platform:
|
||
return {
|
||
"success": False,
|
||
"message": "无法识别webhook来源平台",
|
||
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
|
||
}
|
||
|
||
logger.info(f"🔍 识别平台: {platform}")
|
||
|
||
# 判断是否需要同步
|
||
if not self.should_trigger_sync(webhook_data, platform, headers):
|
||
return {
|
||
"success": True,
|
||
"message": f"Webhook已接收，但不需要触发同步",
|
||
"platform": platform,
|
||
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
|
||
}
|
||
|
||
# 提取Issue信息用于日志
|
||
issue_info = self.extract_issue_info(webhook_data, platform)
|
||
if issue_info:
|
||
logger.info(f"📋 Issue信息: {issue_info['title']} (动作: {issue_info['action']}, 用户: {issue_info['user']})")
|
||
|
||
# 调用同步API
|
||
sync_result = self.call_sync_api(platform)
|
||
|
||
logger.info("=" * 60)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": f"Webhook处理完成: {platform} → {'gitlink' if platform == 'gitee' else 'gitee'}",
|
||
"platform": platform,
|
||
"issue_info": issue_info,
|
||
"sync_result": sync_result,
|
||
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
|
||
}
|