github_parser.py

"""
GitHub事件解析器
用于解析GitHub Webhook事件并提取关键信息
"""

import logging
from datetime import datetime
from typing import Dict, Optional, Any

logger = logging.getLogger(__name__)


class GitHubEventParser:
    """GitHub事件解析器类"""
    
    def __init__(self):
        """初始化解析器"""
        self.supported_events = {
            'issues': self._parse_issue_event,
            'pull_request': self._parse_pull_request_event,
            'push': self._parse_push_event
        }
    
    def parse_event(self, event_type: str, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """
        解析GitHub事件
        
        Args:
            event_type: 事件类型
            payload: 事件载荷数据
        
        Returns:
            解析后的消息数据，如果不支持该事件类型则返回None
        """
        if event_type not in self.supported_events:
            logger.info(f"不支持的事件类型: {event_type}")
            return None
        
        try:
            return self.supported_events[event_type](payload)
        except Exception as e:
            logger.error(f"解析{event_type}事件时发生错误: {e}")
            return None
    
    def _parse_issue_event(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """
        解析Issue事件
        
        Args:
            payload: Issue事件载荷
        
        Returns:
            格式化的消息数据
        """
        action = payload.get('action', '')
        issue = payload.get('issue', {})
        repository = payload.get('repository', {})
        sender = payload.get('sender', {})
        
        # 只处理特定的Issue动作
        if action not in ['opened', 'closed', 'reopened']:
            return None
        
        # 动作映射
        action_map = {
            'opened': '创建了',
            'closed': '关闭了',
            'reopened': '重新打开了'
        }
        
        return {
            'type': 'issue',
            'action': action,
            'action_text': action_map.get(action, action),
            'title': issue.get('title', ''),
            'number': issue.get('number', ''),
            'url': issue.get('html_url', ''),
            'body': issue.get('body', '')[:200] + '...' if len(issue.get('body', '')) > 200 else issue.get('body', ''),
            'state': issue.get('state', ''),
            'labels': [label.get('name', '') for label in issue.get('labels', [])],
            'assignees': [assignee.get('login', '') for assignee in issue.get('assignees', [])],
            'repository': {
                'name': repository.get('name', ''),
                'full_name': repository.get('full_name', ''),
                'url': repository.get('html_url', '')
            },
            'sender': {
                'login': sender.get('login', ''),
                'avatar_url': sender.get('avatar_url', ''),
                'url': sender.get('html_url', '')
            },
            'created_at': issue.get('created_at', ''),
            'updated_at': issue.get('updated_at', '')
        }
    
    def _parse_pull_request_event(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """
        解析Pull Request事件
        
        Args:
            payload: Pull Request事件载荷
        
        Returns:
            格式化的消息数据
        """
        action = payload.get('action', '')
        pull_request = payload.get('pull_request', {})
        repository = payload.get('repository', {})
        sender = payload.get('sender', {})
        
        # 只处理特定的PR动作
        if action not in ['opened', 'closed', 'reopened', 'merged']:
            return None
        
        # 检查是否为合并
        if action == 'closed' and pull_request.get('merged', False):
            action = 'merged'
        
        # 动作映射
        action_map = {
            'opened': '创建了',
            'closed': '关闭了',
            'reopened': '重新打开了',
            'merged': '合并了'
        }
        
        return {
            'type': 'pull_request',
            'action': action,
            'action_text': action_map.get(action, action),
            'title': pull_request.get('title', ''),
            'number': pull_request.get('number', ''),
            'url': pull_request.get('html_url', ''),
            'body': pull_request.get('body', '')[:200] + '...' if len(pull_request.get('body', '')) > 200 else pull_request.get('body', ''),
            'state': pull_request.get('state', ''),
            'merged': pull_request.get('merged', False),
            'draft': pull_request.get('draft', False),
            'base_branch': pull_request.get('base', {}).get('ref', ''),
            'head_branch': pull_request.get('head', {}).get('ref', ''),
            'commits': pull_request.get('commits', 0),
            'additions': pull_request.get('additions', 0),
            'deletions': pull_request.get('deletions', 0),
            'changed_files': pull_request.get('changed_files', 0),
            'labels': [label.get('name', '') for label in pull_request.get('labels', [])],
            'assignees': [assignee.get('login', '') for assignee in pull_request.get('assignees', [])],
            'reviewers': [reviewer.get('login', '') for reviewer in pull_request.get('requested_reviewers', [])],
            'repository': {
                'name': repository.get('name', ''),
                'full_name': repository.get('full_name', ''),
                'url': repository.get('html_url', '')
            },
            'sender': {
                'login': sender.get('login', ''),
                'avatar_url': sender.get('avatar_url', ''),
                'url': sender.get('html_url', '')
            },
            'created_at': pull_request.get('created_at', ''),
            'updated_at': pull_request.get('updated_at', '')
        }
    
    def _parse_push_event(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """
        解析Push事件
        
        Args:
            payload: Push事件载荷
        
        Returns:
            格式化的消息数据
        """
        repository = payload.get('repository', {})
        sender = payload.get('sender', {})
        commits = payload.get('commits', [])
        ref = payload.get('ref', '')
        
        # 提取分支名
        branch = ref.replace('refs/heads/', '') if ref.startswith('refs/heads/') else ref
        
        # 过滤掉删除分支的推送
        if payload.get('deleted', False):
            return None
        
        # 处理提交信息
        commit_messages = []
        for commit in commits[:5]:  # 最多显示5个提交
            message = commit.get('message', '').split('\n')[0]  # 只取第一行
            if len(message) > 50:
                message = message[:47] + '...'
            commit_messages.append({
                'id': commit.get('id', '')[:7],  # 短SHA
                'message': message,
                'url': commit.get('url', ''),
                'author': commit.get('author', {}).get('name', ''),
                'timestamp': commit.get('timestamp', '')
            })
        
        return {
            'type': 'push',
            'action': 'pushed',
            'action_text': '推送了',
            'branch': branch,
            'ref': ref,
            'commits_count': len(commits),
            'commits': commit_messages,
            'compare_url': payload.get('compare', ''),
            'forced': payload.get('forced', False),
            'repository': {
                'name': repository.get('name', ''),
                'full_name': repository.get('full_name', ''),
                'url': repository.get('html_url', '')
            },
            'sender': {
                'login': sender.get('login', ''),
                'avatar_url': sender.get('avatar_url', ''),
                'url': sender.get('html_url', '')
            },
            'head_commit': payload.get('head_commit', {}),
            'before': payload.get('before', ''),
            'after': payload.get('after', '')
        }
    
    @staticmethod
    def format_datetime(datetime_str: str) -> str:
        """
        格式化日期时间字符串
        
        Args:
            datetime_str: ISO格式的日期时间字符串
        
        Returns:
            格式化后的日期时间字符串
        """
        try:
            dt = datetime.fromisoformat(datetime_str.replace('Z', '+00:00'))
            return dt.strftime('%Y-%m-%d %H:%M:%S')
        except Exception:
            return datetime_str