reposync/issue_sync_module/clients/gitee_client.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Gitee Issue API 客户端
"""

import requests
import json
import os
from typing import List, Dict, Optional

# 兼容导入：优先使用环境变量，后备使用src模块
try:
    from src.base import config
    from src.utils.logger import logger
except ImportError:
    # 独立运行时使用环境变量和内置日志
    import logging
    logger = logging.getLogger(__name__)
    logging.basicConfig(level=logging.INFO)
    config = None


class GiteeIssueClient:
    """Gitee Issue操作客户端"""
    
    def __init__(self, organization: str, repo_name: str):
        self.organization = organization
        self.repo_name = repo_name
        
        # 硬编码Gitee Token和API地址（便于迁移）
        self.token = 'e93fe21b329bb12c3c9c11f590fdd653'
        self.base_url = 'https://gitee.com/api/v5/repos'
        
    def get_issues(self, state: str = 'all') -> List[Dict]:
        """获取仓库的Issue列表"""
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues"
        params = {
            'access_token': self.token,
            'state': state,
            'per_page': 100
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            
            issues = response.json()
            logger.info(f"获取到Gitee仓库 {self.organization}/{self.repo_name} 的 {len(issues)} 个Issue")
            return issues
            
        except Exception as e:
            logger.error(f"获取Gitee Issues失败: {str(e)}")
            return []
    
    def get_issue(self, issue_number: int) -> Optional[Dict]:
        """获取单个Issue详情"""
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues/{issue_number}"
        params = {
            'access_token': self.token
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            
            issue = response.json()
            logger.info(f"获取Gitee Issue #{issue_number} 成功")
            return issue
            
        except Exception as e:
            logger.error(f"获取Gitee Issue #{issue_number} 失败: {str(e)}")
            return None
    
    def create_issue(self, title: str, body: str = "", priority: str = None, 
                    labels: str = None, assignee: str = None) -> Optional[Dict]:
        """创建新Issue"""
        # 根据Gitee官方API文档，正确的URL格式
        url = f"{self.base_url}/{self.organization}/issues"
        
        # 使用JSON格式发送数据，包含repo参数
        data = {
            'access_token': self.token,
            'repo': self.repo_name,  # 仓库名作为参数
            'title': title,
            'body': body or ""
        }
        
        if priority:
            data['priority'] = self._get_priority_value(priority)
        if labels:
            data['labels'] = labels
        if assignee:
            data['assignee'] = assignee
        
        # 设置JSON请求头
        headers = {
            'Content-Type': 'application/json;charset=UTF-8'
        }
        
        try:
            response = requests.post(url, json=data, headers=headers)
            response.raise_for_status()
            
            issue = response.json()
            logger.info(f"创建Gitee Issue成功: #{issue['number']} - {title}")
            return issue
            
        except Exception as e:
            logger.error(f"创建Gitee Issue失败: {str(e)}")
            return None
    
    def update_issue(self, issue_number: int, title: str = None, body: str = None, 
                    state: str = None, priority: str = None, labels: str = None, 
                    assignee: str = None) -> Optional[Dict]:
        """更新Issue"""
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues/{issue_number}"
        
        data = {
            'access_token': self.token
        }
        
        if title is not None:
            data['title'] = title
        if body is not None:
            data['body'] = body
        if state is not None:
            data['state'] = state
        if priority is not None:
            data['priority'] = self._get_priority_value(priority)
        if labels is not None:
            data['labels'] = labels
        if assignee is not None:
            data['assignee'] = assignee
        
        try:
            response = requests.patch(url, data=data)
            response.raise_for_status()
            
            issue = response.json()
            logger.info(f"更新Gitee Issue #{issue_number} 成功")
            return issue
            
        except Exception as e:
            logger.error(f"更新Gitee Issue #{issue_number} 失败: {str(e)}")
            return None
    
    def close_issue(self, issue_number: int) -> bool:
        """关闭Issue"""
        return self.update_issue(issue_number, state="closed") is not None
    
    def delete_issue(self, issue_number: int) -> bool:
        """删除Issue"""
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues/{issue_number}"
        params = {
            'access_token': self.token
        }
        
        try:
            response = requests.delete(url, params=params)
            response.raise_for_status()
            
            logger.info(f"删除Gitee Issue #{issue_number} 成功")
            return True
            
        except Exception as e:
            logger.error(f"删除Gitee Issue #{issue_number} 失败: {str(e)}")
            return False
    
    def _get_priority_value(self, priority: str) -> int:
        """获取优先级数值"""
        priority_map = {
            "不重要": 0,
            "次要": 1,
            "主要": 2,
            "严重": 3,
            "致命": 4
        }
        return priority_map.get(priority, 1)
    
    def find_issue_by_title(self, title: str) -> Optional[Dict]:
        """根据标题查找Issue（避免重复创建）"""
        issues = self.get_issues()
        for issue in issues:
            if issue.get('title', '') == title:
                return issue
        return None
    
    # =============== 评论相关API ===============
    
    def get_issue_comments(self, issue_number: int, per_page: int = 100) -> List[Dict]:
        """获取Issue的评论列表
        
        Args:
            issue_number: Issue编号
            per_page: 每页数量
            
        Returns:
            评论列表
        """
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues/{issue_number}/comments"
        params = {
            'access_token': self.token,
            'per_page': per_page
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            
            comments = response.json()
            logger.info(f"获取到Gitee Issue #{issue_number} 的 {len(comments)} 条评论")
            return comments
            
        except Exception as e:
            logger.error(f"获取Gitee Issue #{issue_number} 评论失败: {str(e)}")
            return []
    
    def get_issue_comment(self, comment_id: int) -> Optional[Dict]:
        """获取单个评论详情
        
        Args:
            comment_id: 评论ID
            
        Returns:
            评论对象
        """
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues/comments/{comment_id}"
        params = {
            'access_token': self.token
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            
            comment = response.json()
            logger.info(f"获取Gitee评论 #{comment_id} 成功")
            return comment
            
        except Exception as e:
            logger.error(f"获取Gitee评论 #{comment_id} 失败: {str(e)}")
            return None
    
    def create_issue_comment(self, issue_number: int, body: str) -> Optional[Dict]:
        """创建Issue评论
        
        Args:
            issue_number: Issue编号
            body: 评论内容
            
        Returns:
            创建的评论对象
        """
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues/{issue_number}/comments"
        
        data = {
            'access_token': self.token,
            'body': body
        }
        
        try:
            response = requests.post(url, data=data)
            response.raise_for_status()
            
            comment = response.json()
            comment_content = comment.get('body', '')[:50] + ('...' if len(comment.get('body', '')) > 50 else '')
            logger.info(f"✅ 成功创建Gitee Issue #{issue_number} 评论: #{comment['id']} - '{comment_content}'")
            return comment
            
        except Exception as e:
            logger.error(f"创建Gitee Issue #{issue_number} 评论失败: {str(e)}")
            return None
    
    def update_issue_comment(self, comment_id: int, body: str) -> Optional[Dict]:
        """更新评论内容
        
        Args:
            comment_id: 评论ID
            body: 新的评论内容
            
        Returns:
            更新后的评论对象
        """
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues/comments/{comment_id}"
        
        data = {
            'access_token': self.token,
            'body': body
        }
        
        try:
            response = requests.patch(url, data=data)
            response.raise_for_status()
            
            comment = response.json()
            logger.info(f"更新Gitee评论 #{comment_id} 成功")
            return comment
            
        except Exception as e:
            logger.error(f"更新Gitee评论 #{comment_id} 失败: {str(e)}")
            return None
    
    def delete_issue_comment(self, comment_id: int) -> bool:
        """删除评论
        
        Args:
            comment_id: 评论ID
            
        Returns:
            是否删除成功
        """
        url = f"{self.base_url}/{self.organization}/{self.repo_name}/issues/comments/{comment_id}"
        params = {
            'access_token': self.token
        }
        
        try:
            response = requests.delete(url, params=params)
            response.raise_for_status()
            
            logger.info(f"删除Gitee评论 #{comment_id} 成功")
            return True
            
        except Exception as e:
            logger.error(f"删除Gitee评论 #{comment_id} 失败: {str(e)}")
            return False