""" TAPD API调用模块（任务二专用） 负责与TAPD Open API交互，查询需求（Story）信息 与任务一的区别： - 任务一：创建和管理Bug单 - 任务二：查询需求（Story）状态信息 """ import os import requests import time from typing import Dict, Optional, Any from requests.auth import HTTPBasicAuth # 导入任务二专用的日志模块 from src2.logger import get_task2_logger # ============================================================ # 自定义异常类 # ============================================================ class StoryNotFoundException(Exception): """当TAPD需求（Story）不存在时抛出的异常""" pass # TAPD状态值映射表 # 将API返回的状态代码转换为中文显示文本 STATUS_MAPPING = { "status_5": "进行中", "status_7": "未开始", "status_8": "已完成", "status_9": "待验收", "status_10": "联调", "status_12": "取消", "status_13": "待评审", } # 需求终态列表（这些状态不需要持续同步） TERMINAL_STATUSES = ['已完成', '取消'] def map_status(status_code: str) -> str: """ 将TAPD状态代码转换为中文显示文本 Args: status_code: TAPD API返回的状态代码（如 "status_5"） Returns: str: 中文显示文本（如 "进行中"），未知状态返回原始值 """ if not status_code: return "未知" return STATUS_MAPPING.get(status_code, status_code) class TAPDStoryApi: """TAPD需求API封装类（任务二专用）""" # TAPD API基础URL（与任务一相同） BASE_URL = "https://tapd-api.bilibili.co/tapd" # 发布计划字段名称 PLAN_FIELD_NAME = "release_id" def __init__(self, workspace_id: str, test_mode: bool = False): """ 初始化TAPD Story API Args: workspace_id: TAPD项目ID test_mode: 是否启用测试模式（显示API请求和响应） Raises: ValueError: 环境变量未设置时抛出 """ self.workspace_id = workspace_id self.test_mode = test_mode self.session = requests.Session() # 从环境变量读取认证信息（与任务一共用） self.api_user = os.environ.get('TAPD_API_USER') self.api_password = os.environ.get('TAPD_API_PASSWORD') if not self.api_user or not self.api_password: raise ValueError( "TAPD认证信息未设置。请设置环境变量:\n" " - TAPD_API_USER\n" " - TAPD_API_PASSWORD" ) # 设置Basic Auth self.auth = HTTPBasicAuth(self.api_user, self.api_password) # 初始化任务二专用的日志记录器 self.logger = get_task2_logger() # 计划字段映射缓存（ID -> 中文名称） self._plan_mapping = None print(f" ✓ TAPD Story API初始化完成 (workspace_id: {workspace_id})") if test_mode: print(f" ⚠ 测试模式已启用：将显示所有API调用的详细信息") def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict: """ 发起TAPD API GET请求的通用方法（支持429错误重试） Args: endpoint: API端点（如 "stories"） params: URL查询参数 Returns: Dict: API响应数据 Raises: RuntimeError: API调用失败时抛出 """ url = f"{self.BASE_URL}/{endpoint}" # 准备日志记录的请求数据 log_request_data = { "url": url, "method": "GET", "params": params, "auth_user": self.api_user } # 测试模式：显示请求信息 if self.test_mode: print("\n" + "=" * 60) print(f"【测试模式】TAPD API调用: {endpoint}") print("=" * 60) print(f"请求URL: {url}") if params: print(f"URL参数:") for key, value in params.items(): print(f" {key}: {value}") # 429错误重试逻辑：最多重试1次 max_retries = 1 retry_count = 0 while retry_count <= max_retries: try: response = self.session.get( url, params=params, auth=self.auth, timeout=30 ) # 测试模式：显示响应信息 if self.test_mode: print(f"\n响应状态码: {response.status_code}") try: import json result = response.json() print(f"响应数据:") print(json.dumps(result, ensure_ascii=False, indent=2)) except: print(f"响应内容: {response.text[:500]}") print("=" * 60) # 检查是否是429错误（在raise_for_status之前检查） if response.status_code == 429: if retry_count < max_retries: retry_count += 1 wait_seconds = 60 print(f"\n⚠️ 触发TAPD API限流 (429 Too Many Requests)") print(f" [开始等待] 等待 {wait_seconds} 秒后重试... (第 {retry_count}/{max_retries} 次重试)") print(f" [重要] 在等待期间，代码会阻塞在这里，不会继续处理其他记录") # 记录429错误日志 self.logger.log_api_call( api_type="tapd", operation=endpoint, request_data=log_request_data, response_data={"status_code": 429, "retry_count": retry_count}, success=False, error_message=f"429 Too Many Requests, 等待{wait_seconds}秒后重试" ) time.sleep(wait_seconds) print(f" [等待结束] 开始重试请求...") continue # 重试 else: # 已达到最大重试次数，抛出异常 error_msg = "TAPD API限流 (429 Too Many Requests)，重试后仍然失败" self.logger.log_api_call( api_type="tapd", operation=endpoint, request_data=log_request_data, response_data={"status_code": 429, "retry_count": retry_count}, success=False, error_message=error_msg ) raise RuntimeError(error_msg) # 对于非429错误，调用raise_for_status检查HTTP状态 response.raise_for_status() result = response.json() # 检查TAPD API返回的状态 if result.get('status') != 1: error_msg = result.get('info', '未知错误') self.logger.log_api_call( api_type="tapd", operation=endpoint, request_data=log_request_data, response_data=result, success=False, error_message=error_msg ) raise RuntimeError(f"TAPD API调用失败: {error_msg}") # 记录成功日志 self.logger.log_api_call( api_type="tapd", operation=endpoint, request_data=log_request_data, response_data=result, success=True ) return result except requests.exceptions.Timeout: error_msg = f"TAPD API请求超时: {endpoint}" self.logger.log_api_call( api_type="tapd", operation=endpoint, request_data=log_request_data, response_data={}, success=False, error_message=error_msg ) raise RuntimeError(error_msg) except requests.exceptions.RequestException as e: # 如果是HTTPError且状态码是429，由上面的status_code检查处理 # 这里不应该到达，因为429在response.status_code检查时已处理 error_msg = f"TAPD API请求失败: {e}" self.logger.log_api_call( api_type="tapd", operation=endpoint, request_data=log_request_data, response_data={}, success=False, error_message=error_msg ) raise RuntimeError(error_msg) # 理论上不会到达这里 raise RuntimeError("TAPD API请求失败：未知错误") def get_story(self, story_id: str) -> Dict: """ 获取需求详情 Args: story_id: 需求ID Returns: Dict: 需求详细信息 Raises: RuntimeError: 获取失败时抛出 """ params = { 'workspace_id': self.workspace_id, 'id': story_id } result = self._make_request("stories", params=params) # TAPD API返回格式: {"status": 1, "data": [{"Story": {...}}]} data = result.get('data', []) if not isinstance(data, list) or len(data) == 0: raise StoryNotFoundException(f"未找到需求: {story_id}") # 取第一个元素 first_item = data[0] # 提取Story对象 if isinstance(first_item, dict) and 'Story' in first_item: story_info = first_item['Story'] else: raise RuntimeError(f"API返回数据格式异常: {first_item}") if not story_info: raise StoryNotFoundException(f"未找到需求: {story_id}") # 转换状态为中文 raw_status = story_info.get('status', '') story_info['raw_status'] = raw_status story_info['status'] = map_status(raw_status) return story_info def get_story_url(self, story_id: str) -> str: """ 生成需求的访问URL Args: story_id: 需求ID Returns: str: 需求的访问URL """ return f"https://www.tapd.cn/{self.workspace_id}/prong/stories/view/{story_id}" def get_story_fields_info(self) -> Dict: """ 获取需求所有字段及候选值 Returns: Dict: 字段信息，包含各字段的名称、选项等 Raises: RuntimeError: 获取失败时抛出 """ params = { 'workspace_id': self.workspace_id } result = self._make_request("stories/get_fields_info", params=params) return result.get('data', {}) def get_plan_mapping(self) -> Dict[str, str]: """ 获取发布计划字段的ID到中文名称映射 Returns: Dict[str, str]: 发布计划ID到中文名称的映射 例如: {"1010104801000069739": "v2test", ...} """ # 获取字段信息 fields_info = self.get_story_fields_info() # 提取计划字段的options plan_field = fields_info.get(self.PLAN_FIELD_NAME, {}) options = plan_field.get('options', {}) # 缓存映射 self._plan_mapping = options if self.test_mode: print(f"\n【测试模式】计划字段映射:") for plan_id, plan_name in options.items(): print(f" {plan_id} -> {plan_name}") return options def map_plan_id_to_name(self, plan_id: str) -> str: """ 将发布计划ID转换为中文名称 Args: plan_id: 发布计划ID（如 "1010104801000069739"） Returns: str: 中文名称（如 "v2test"），未找到则返回空字符串 """ if not plan_id or plan_id == "0": return "" # 如果映射未初始化，先获取 if self._plan_mapping is None: self.get_plan_mapping() return self._plan_mapping.get(plan_id, "") if __name__ == "__main__": print("=== TAPD Story API 测试 ===\n") print("请使用 test_phase2.py 进行完整测试")