# ============================================================================= # 企微IT智能服务台 — 待办数据源 Service 层(抽象基类 + 企微审批实现) # ============================================================================= # 说明:定义待办数据源的统一抽象接口,并提供企微审批数据源的具体实现。 # - TodoSourceService: 抽象基类,定义 get_todo_list / get_todo_detail 接口 # - ApprovalTodoService: 企微审批实现,聚合 getapprovaldata + getapprovaldetail # # 设计原则: # 1. 策略模式 — 不同数据源实现同一接口,TodoAggregatorService 可透明替换 # 2. 容错隔离 — 单个数据源失败不影响其他数据源 # 3. 并发优化 — 企微审批详情查询使用 Semaphore 限制并发,防止 API 限流 # ============================================================================= import asyncio import logging import time from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional from urllib.parse import parse_qs, urlparse import httpx import redis.asyncio as aioredis from app.api.approval import ( APPROVAL_TEMPLATES, _extract_current_approver, get_approval_detail, ) from app.utils.token_manager import TokenManager logger = logging.getLogger(__name__) # 企微 getapprovalinfo API 地址(旧接口 getapprovaldata 已废弃,改用 getapprovalinfo) WECOM_GETAPPROVALINFO_URL = "https://qyapi.weixin.qq.com/cgi-bin/oa/getapprovalinfo" # 企微审批详情并发查询上限(Semaphore),防止触发企微 API 限流 APPROVAL_DETAIL_CONCURRENCY = 10 # 查询审批数据的时间范围(最近 N 天) APPROVAL_QUERY_DAYS = 30 # 企微 getapprovaldata 单页查询上限 APPROVAL_PAGE_SIZE = 100 # =========================================================================== # 模块级工具函数 # =========================================================================== def _extract_template_ids_from_templates() -> List[str]: """从 APPROVAL_TEMPLATES 中提取企微审批模板 ID 列表。 APPROVAL_TEMPLATES 中每个模板的 url 字段可能包含 template_id 查询参数 (仅 location=="企微审批" 的模板才有)。此函数解析所有 URL,提取有效的 template_id,用于 getapprovaldata 的 filters 过滤。 Returns: List[str]: 企微审批模板 ID 列表(去重) """ template_ids: List[str] = [] seen: set = set() for template in APPROVAL_TEMPLATES.values(): url = template.get("url", "") if not url: continue # 解析 URL 中的 query 参数 parsed = urlparse(url) # 企微审批 URL 的 query 在 fragment 中(#/?template_id=xxx) # urlparse 会把 # 后面的内容放入 fragment fragment = parsed.fragment or "" query_string = "" if "?" in fragment: query_string = fragment.split("?", 1)[1] elif parsed.query: query_string = parsed.query if query_string: params = parse_qs(query_string) tid_list = params.get("template_id", []) for tid in tid_list: if tid and tid not in seen: seen.add(tid) template_ids.append(tid) return template_ids def _build_template_id_name_map() -> Dict[str, str]: """构建 企微template_id → 模板名称 的映射表。 用于在映射 TodoItemData 时,通过 template_id 查找对应的审批模板名称。 Returns: Dict[str, str]: {企微template_id: 模板名称} """ mapping: Dict[str, str] = {} for template in APPROVAL_TEMPLATES.values(): url = template.get("url", "") name = template.get("name", "") if not url: continue parsed = urlparse(url) fragment = parsed.fragment or "" query_string = "" if "?" in fragment: query_string = fragment.split("?", 1)[1] elif parsed.query: query_string = parsed.query if query_string: params = parse_qs(query_string) tid_list = params.get("template_id", []) for tid in tid_list: if tid: mapping[tid] = name return mapping def _apply_time_to_iso(apply_time: Any) -> str: """将企微 apply_time(秒级时间戳)转换为 ISO 8601 格式字符串。 Args: apply_time: 企微审批的 apply_time 字段(int 秒级时间戳,或已格式化字符串) Returns: str: ISO 8601 格式时间字符串 """ if not apply_time: return "" try: if isinstance(apply_time, (int, float)): # 企微 apply_time 为秒级时间戳 return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(int(apply_time))) return str(apply_time) except Exception: return str(apply_time) # 模块级缓存:企微审批模板 ID 列表(启动时计算一次) _APPROVAL_TEMPLATE_IDS: List[str] = _extract_template_ids_from_templates() # 模块级缓存:template_id → 模板名称映射 _TEMPLATE_ID_NAME_MAP: Dict[str, str] = _build_template_id_name_map() # =========================================================================== # 抽象基类 # =========================================================================== class TodoSourceService(ABC): """待办数据源抽象基类。 所有待办数据源(企微审批、ITSM 工单等)需实现此接口, 以便 TodoAggregatorService 统一聚合。 """ @abstractmethod async def get_todo_list(self) -> List[Dict[str, Any]]: """获取待办列表。 Returns: List[Dict[str, Any]]: TodoItemData 格式的待办列表 """ ... @abstractmethod async def get_todo_detail(self, item_id: str) -> Optional[Dict[str, Any]]: """获取单条待办详情。 Args: item_id: 待办原始 ID(不含类型前缀) Returns: Optional[Dict[str, Any]]: TodoItemData 格式的待办详情,不存在返回 None """ ... # =========================================================================== # 企微审批数据源实现 # =========================================================================== class ApprovalTodoService(TodoSourceService): """企微审批待办数据源实现。 通过企微 OA API 获取当前坐席待处理的审批单: 1. getapprovaldata — 按 sp_status=1 + 模板 ID 过滤,获取审批单号列表 2. getapprovaldetail — 并发获取每个审批单的详情 3. _extract_current_approver — 过滤当前审批人是当前坐席的审批单 4. _map_to_todo_item — 映射为统一 TodoItemData 格式 Attributes: agent_userid: 当前坐席的企微 userid redis: Redis 异步客户端(用于获取 access_token) """ def __init__(self, agent_userid: str, redis: aioredis.Redis): """初始化企微审批数据源服务。 Args: agent_userid: 当前坐席的企微 userid redis: Redis 异步客户端实例 """ self.agent_userid = agent_userid self.redis = redis # ------------------------------------------------------------------ # 公开接口 # ------------------------------------------------------------------ async def get_todo_list(self) -> List[Dict[str, Any]]: """获取当前坐席待处理的企微审批列表。 流程: 1. 获取审批 access_token 2. 调用 getapprovaldata 获取审批单号列表(sp_status=1,最近7天) 3. 并发调用 getapprovaldetail 获取每个审批单详情(Semaphore 限流) 4. 用 _extract_current_approver 过滤出当前审批人是当前坐席的审批单 5. 映射为统一 TodoItemData 格式 异常处理:任何步骤失败都返回空列表并记日志,不抛出异常。 Returns: List[Dict[str, Any]]: TodoItemData 格式的待办列表 """ try: # 1. 获取 access_token access_token = await self._get_access_token() if not access_token: logger.error("获取企微审批 access_token 失败,返回空列表") return [] # 2. 获取审批单号列表 sp_no_list = await self._fetch_approval_sp_no_list(access_token) if not sp_no_list: logger.info("企微审批待处理列表为空") return [] logger.info(f"企微审批待处理审批单号列表: {len(sp_no_list)} 条") # 3. 并发获取审批详情 details = await self._fetch_approval_details(access_token, sp_no_list) if not details: logger.info("企微审批详情获取失败或为空") return [] # 3.5 按模板 ID 过滤(企微 API 每个 key 只能出现一次, # 无法在 API 层按多个 template_id 过滤,需在代码层过滤) if _APPROVAL_TEMPLATE_IDS: before_count = len(details) details = [ d for d in details if d.get("info", {}).get("template_id", "") in _APPROVAL_TEMPLATE_IDS ] logger.info( f"企微审批按模板ID过滤后: {len(details)}/{before_count} 条" ) # 4. 过滤当前审批人是当前坐席的审批单 filtered = self._filter_by_current_approver(details) logger.info( f"企微审批过滤后(当前审批人={self.agent_userid}): " f"{len(filtered)}/{len(details)} 条" ) # 5. 映射为 TodoItemData todo_items = [self._map_to_todo_item(d) for d in filtered] return todo_items except Exception as e: logger.error(f"获取企微审批待办列表失败: {e}", exc_info=True) return [] async def get_todo_detail(self, item_id: str) -> Optional[Dict[str, Any]]: """获取单条企微审批详情。 Args: item_id: 审批单号 sp_no(不含 "approval:" 前缀) Returns: Optional[Dict[str, Any]]: TodoItemData 格式的审批详情 """ try: access_token = await self._get_access_token() if not access_token: logger.error("获取企微审批 access_token 失败") return None detail = await self._fetch_approval_detail(access_token, item_id) if not detail: return None return self._map_to_todo_item(detail) except Exception as e: logger.error(f"获取企微审批详情失败: sp_no={item_id}, error={e}", exc_info=True) return None # ------------------------------------------------------------------ # 私有方法 # ------------------------------------------------------------------ async def _get_access_token(self) -> str: """获取企微 access_token(使用IT支持应用Secret,IP已在白名单中)。""" manager = TokenManager(self.redis) try: return await manager.get_token() finally: await manager.close() async def _fetch_approval_sp_no_list(self, access_token: str) -> List[str]: """调用企微 getapprovalinfo API 获取审批单号列表。 使用 new_cursor 分页循环,查询最近 7 天内 sp_status=1(审批中)的审批单, 并按预置的模板 ID 列表过滤。 注意:旧接口 getapprovaldata 已废弃(返回404),改用 getapprovalinfo。 Args: access_token: 企微审批 access_token Returns: List[str]: 审批单号列表 """ # 时间范围:最近 7 天 endtime = int(time.time()) starttime = endtime - APPROVAL_QUERY_DAYS * 24 * 3600 # 构建 filters:仅 sp_status=1(审批中) # 注意:企微 getapprovalinfo API 每个 key 只能出现一次, # 不能在 API 层按多个 template_id 过滤,需在代码层面过滤。 filters: List[Dict[str, Any]] = [ {"key": "sp_status", "value": 1}, ] sp_no_list: List[str] = [] new_cursor = "" async with httpx.AsyncClient( timeout=httpx.Timeout(timeout=30.0, connect=10.0, read=30.0) ) as client: while True: payload = { "starttime": str(starttime), "endtime": str(endtime), "new_cursor": new_cursor, "size": APPROVAL_PAGE_SIZE, "filters": filters, } params = {"access_token": access_token} response = await client.post( WECOM_GETAPPROVALINFO_URL, params=params, json=payload ) result = response.json() if result.get("errcode") != 0: logger.error( f"getapprovalinfo 调用失败: errcode={result.get('errcode')}, " f"errmsg={result.get('errmsg')}" ) break # getapprovalinfo 返回 sp_no_list(字符串数组),非旧接口的 data page_sp_no_list = result.get("sp_no_list", []) sp_no_list.extend(page_sp_no_list) # 检查是否还有下一页(new_next_cursor 为空表示无更多数据) next_cursor = result.get("new_next_cursor", "") if not next_cursor or next_cursor == new_cursor: break new_cursor = next_cursor return sp_no_list async def _fetch_approval_detail(self, access_token: str, sp_no: str) -> Optional[dict]: """调用企微 getapprovaldetail API 获取单条审批详情。 复用 approval.py 中已有的 get_approval_detail 函数。 Args: access_token: 企微审批 access_token sp_no: 审批单号 Returns: Optional[dict]: 企微 API 返回的完整审批详情,失败返回 None """ try: return await get_approval_detail(access_token, sp_no) except Exception as e: logger.warning(f"获取审批详情失败: sp_no={sp_no}, error={e}") return None async def _fetch_approval_details( self, access_token: str, sp_no_list: List[str] ) -> List[dict]: """并发获取多个审批单的详情。 使用 asyncio.Semaphore 限制并发数(默认 10),防止企微 API 限流。 单个审批单查询失败不影响其他审批单。 Args: access_token: 企微审批 access_token sp_no_list: 审批单号列表 Returns: List[dict]: 成功获取的审批详情列表 """ semaphore = asyncio.Semaphore(APPROVAL_DETAIL_CONCURRENCY) async def _fetch_one(sp_no: str) -> Optional[dict]: async with semaphore: return await self._fetch_approval_detail(access_token, sp_no) tasks = [_fetch_one(sp_no) for sp_no in sp_no_list] results = await asyncio.gather(*tasks, return_exceptions=True) details: List[dict] = [] for result in results: if isinstance(result, Exception): logger.warning(f"审批详情查询异常: {result}") continue if result is not None: details.append(result) return details def _filter_by_current_approver(self, details: List[dict]) -> List[dict]: """过滤当前审批人是当前坐席的审批单。 使用 approval.py 中的 _extract_current_approver 提取当前审批人 userid, 仅保留当前审批人等于 agent_userid 的审批单。 Args: details: 企微 getapprovaldetail 返回的审批详情列表 Returns: List[dict]: 过滤后的审批详情列表 """ filtered: List[dict] = [] for detail in details: current_approver = _extract_current_approver(detail) if current_approver and current_approver == self.agent_userid: filtered.append(detail) return filtered def _map_to_todo_item(self, detail: dict) -> Dict[str, Any]: """将企微审批详情映射为统一 TodoItemData 格式。 映射规则参考系统设计文档 §8.1: - id: "approval:{sp_no}" - type: "approval" - title: sp_name - priority: "high"(企微无优先级概念,默认 high) - status: "pending"(sp_status=1 审批中统一映射为 pending) - description: 包含 sp_no、template_name、applicant 等字段 Args: detail: 企微 getapprovaldetail 返回的完整审批详情 Returns: Dict[str, Any]: TodoItemData 格式的待办事项 """ info = detail.get("info", {}) sp_no = info.get("sp_no", "") sp_name = info.get("sp_name", "") sp_status = info.get("sp_status", 1) template_id = info.get("template_id", "") apply_time = info.get("apply_time", 0) applyer_userid = info.get("applyer", {}).get("userid", "") current_approver = _extract_current_approver(detail) template_name = _TEMPLATE_ID_NAME_MAP.get(template_id, sp_name) apply_time_iso = _apply_time_to_iso(apply_time) return { "id": f"approval:{sp_no}", "type": "approval", "title": sp_name or template_name or "企微审批", "priority": "high", "description": { "sp_no": sp_no, "template_name": template_name, "template_id": template_id, "applicant": applyer_userid, "apply_time": apply_time, "sp_status": sp_status, "current_approver": current_approver or "", }, "status": "pending", "assigned_agent_id": current_approver, "corp_id": "", "created_at": apply_time_iso, "updated_at": apply_time_iso, }