simon/wecom_it_smart_desk
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(agent/backend): 坐席端审批线降级跳转 + 回调最终一致回写

Browse Source 
实现 Phase 0 审批线 T01+T02（依据 PRD-REQ-坐席-011 + U-1 技术验证结论）。

前端（T01）：
- TaskDetailView.handleAction 移除 mock toast，审批类仅 console.info
- ApprovalDetail 通过/拒绝/转交 + 打开按钮接线为企微审批深链真实跳转（<a target="_blank">）
- useWebSocket 新增 todo_status_changed 实时刷新分支

后端（T02）：
- approval.py 新增 writeback_approval_todo_status 主入口 + /approval/callback 接线
- approval_webhook.py 新增 _writeback_agent_todo 复用回调→WS 推送通道
- 以 approval:{sp_no} 为关联键，缓存就地改写 + 7天快照 + WS 推送，最终一致

测试：src/backend/tests/test_approval_todo_writeback.py（51 例全绿）
文档：PRD-REQ-坐席-011 v0.1、技术验证-U-1 v1.0
This commit is contained in:
Simon
2026-08-09 00:46:31 +08:00
parent 2fd2e7df02
commit 9292f41763
8 changed files with 1782 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/app/api/approval.py
+355 -19
View File
@@ -8,9 +8,11 @@
# =============================================================================
import asyncio
import json
import logging
import os
from typing import Optional
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import httpx
from fastapi import APIRouter, Depends, Query
@@ -29,6 +31,63 @@ router = APIRouter()
# IT资产升级申请模板ID（回调时用于识别审批类型，触发年限核查推送）
ASSET_UPGRADE_TEMPLATE_ID = "Bs7ucTGsPuFhxfk8pn8EydxrWxkVetB4JR8Pb6PHS"
# ---------------------------------------------------------------------------
# 坐席待办回写相关常量（企微审批回调 → 服务台待办最终一致）
# ---------------------------------------------------------------------------
# 坐席待办列表缓存 Key 前缀（与 TodoAggregatorService._cache_key 保持一致）
# 实际格式：todo:cache:{agent_userid}:{todo_type_or_all}
TODO_CACHE_KEY_PREFIX = "todo:cache:"
# 待办列表缓存兜底 TTL（秒）—— 与 TodoAggregatorService.CACHE_TTL 对齐
TODO_CACHE_FALLBACK_TTL = 45
# 审批状态回写快照 Key（供排障/审计，以及缓存过期后的状态追溯）
TODO_APPROVAL_STATUS_KEY = "todo:approval:status:{sp_no}"
# 状态快照保留时长（秒）：7 天
TODO_APPROVAL_STATUS_TTL = 7 * 24 * 3600
# status_change_event → 事件语义（企微 sys_approval_change）
APPROVAL_EVENT_MAP: Dict[int, str] = {
1: "submitted", # 提单
2: "approved", # 同意
3: "rejected", # 驳回
4: "transferred", # 转审
5: "reminded", # 催办
6: "revoked", # 撤销
8: "revoked_after_approved", # 通过后撤销
10: "commented", # 添加备注
}
# status_change_event → 本地待办状态（pending/processing/resolved）
# 说明：
# - 2 同意 / 3 驳回 / 6 撤销 / 8 通过后撤销 → 审批单已终结，移出坐席待办
# - 4 转审 → 对「当前这位审批人」而言同样已终结（单子转给了别人）
# - 1 提单 / 5 催办 / 10 备注 → 审批单仍在流转，待办保持 pending
APPROVAL_EVENT_TODO_STATUS: Dict[int, str] = {
1: "pending",
2: "resolved",
3: "resolved",
4: "resolved",
5: "pending",
6: "resolved",
8: "resolved",
10: "pending",
}
# sp_status → 本地待办状态（回调未带 status_change_event 时的兜底映射）
# sp_status: 1-审批中 2-已通过 3-已驳回 4-已撤销 6-通过后撤销 7-已删除 10-已支付
APPROVAL_SP_STATUS_TODO_STATUS: Dict[int, str] = {
1: "pending",
2: "resolved",
3: "resolved",
4: "resolved",
6: "resolved",
7: "resolved",
10: "resolved",
}
# Redis客户端（依赖注入）
async def get_redis() -> aioredis.Redis:
"""获取Redis客户端依赖"""
@@ -775,6 +834,288 @@ async def _do_asset_urge(sp_no: str, redis: aioredis.Redis) -> dict:
return {"success": False, "message": f"推送失败: {e}", "sp_no": sp_no}
# =============================================================================
# 审批回调 → 坐席待办状态回写（降级跳转方案的「最终一致」闭环）
# =============================================================================
# 背景：
# 企微官方不提供「服务端代审批人执行同意/拒绝/转交」的接口，坐席端的审批动作
# 只能降级为跳转企微原系统由本人操作。因此服务台侧的待办状态无法在动作发生的
# 那一刻同步更新，只能依赖企微 sys_approval_change 回调回写，达成最终一致。
#
# 关联键（approval_code / sp_no → 本地待办）：
# 坐席待办的 id 由 ApprovalTodoService._map_to_todo_item 生成为 "approval:{sp_no}"，
# description.sp_no 亦为同一值。因此企微回调携带的 sp_no 就是本地待办的反查键，
# 无需额外建立映射表。
#
# 存储现状（重要）：
# 坐席待办当前**不落库**——todo_items 表（TodoItem 模型）虽已定义但全链路未接线，
# 列表由 TodoAggregatorService 实时聚合企微审批 + ITSM，并缓存在 Redis
# （todo:cache:{agent_userid}:{type}，TTL 45s）。
# 因此本回写作用于两处：
# 1) 就地改写命中的待办列表缓存条目（覆盖缓存未过期的 45s 窗口）；
# 2) 写一份状态快照 todo:approval:status:{sp_no}（TTL 7 天）供排障/审计。
# 缓存过期后由聚合层重新拉取企微权威数据，天然一致。
# =============================================================================
def _map_approval_todo_status(status_change_event: int, sp_status: int) -> str:
"""将企微审批回调映射为本地待办状态。
优先使用 status_change_event（语义更精确，可区分「转审」），
未命中时回退到 sp_status 映射，再兜底为 pending。
Args:
status_change_event: 企微状态变化类型（1提单/2同意/3驳回/4转审/
5催办/6撤销/8通过后撤销/10备注）
sp_status: 企微审批单状态（1审批中/2已通过/3已驳回/4已撤销/
6通过后撤销/7已删除/10已支付）
Returns:
str: 本地待办状态（pending/processing/resolved）
"""
todo_status = APPROVAL_EVENT_TODO_STATUS.get(status_change_event)
if todo_status:
return todo_status
return APPROVAL_SP_STATUS_TODO_STATUS.get(sp_status, "pending")
def _parse_agent_userid_from_cache_key(cache_key: str) -> str:
"""从待办缓存 Key 中解析坐席 userid。
Key 格式：todo:cache:{agent_userid}:{todo_type_or_all}
userid 理论上不含冒号，但仍按「去掉前缀与末段」的方式解析以增强容错。
Args:
cache_key: Redis 缓存 Key（已 decode 为 str）
Returns:
str: 坐席 userid，解析失败返回空字符串
"""
if not cache_key.startswith(TODO_CACHE_KEY_PREFIX):
return ""
remainder = cache_key[len(TODO_CACHE_KEY_PREFIX):]
if ":" not in remainder:
return ""
# 末段是 todo_type（all/approval/ticket），其余部分是 userid
return remainder.rsplit(":", 1)[0]
async def _patch_todo_cache(
redis: aioredis.Redis,
sp_no: str,
todo_status: str,
sp_status: int,
) -> List[str]:
"""就地改写待办列表缓存中命中的审批条目，并返回受影响的坐席 userid 列表。
做什么：扫描 todo:cache:*，找到 items 中 id == "approval:{sp_no}" 的条目，
更新其 status 与 description.sp_status，然后按剩余 TTL 写回。
为什么：待办不落库，缓存就是坐席端当前看到的「本地待办」；不改写的话，
坐席在缓存过期前仍会看到已终结的审批单。
单个 Key 处理失败不影响其他 Key。
Args:
redis: Redis 异步客户端
sp_no: 企微审批单号（本地待办反查键）
todo_status: 回写后的本地待办状态
sp_status: 企微审批单状态（同步写入 description.sp_status）
Returns:
List[str]: 命中该审批单的坐席 userid 列表（去重，顺序稳定）
"""
item_id = f"approval:{sp_no}"
affected_agents: List[str] = []
seen_agents: set = set()
try:
keys = await redis.keys(f"{TODO_CACHE_KEY_PREFIX}*")
except Exception as e:
logger.warning(f"扫描待办缓存失败: sp_no={sp_no}, error={e}")
return affected_agents
for raw_key in keys or []:
key = raw_key.decode("utf-8") if isinstance(raw_key, bytes) else str(raw_key)
try:
raw_value = await redis.get(key)
if not raw_value:
continue
if isinstance(raw_value, bytes):
raw_value = raw_value.decode("utf-8")
payload: Dict[str, Any] = json.loads(raw_value)
items = payload.get("items")
if not isinstance(items, list):
continue
matched = False
for item in items:
if not isinstance(item, dict) or item.get("id") != item_id:
continue
item["status"] = todo_status
description = item.get("description")
if isinstance(description, dict):
description["sp_status"] = sp_status
matched = True
if not matched:
continue
# 保留剩余 TTL 写回（拿不到有效 TTL 时用兜底值，避免写成永不过期）
ttl = await redis.ttl(key)
if not isinstance(ttl, int) or ttl <= 0:
ttl = TODO_CACHE_FALLBACK_TTL
await redis.setex(key, ttl, json.dumps(payload, ensure_ascii=False))
agent_userid = _parse_agent_userid_from_cache_key(key)
if agent_userid and agent_userid not in seen_agents:
seen_agents.add(agent_userid)
affected_agents.append(agent_userid)
except Exception as e:
logger.warning(f"回写待办缓存失败: key={key}, sp_no={sp_no}, error={e}")
continue
return affected_agents
async def _save_approval_status_snapshot(
redis: aioredis.Redis,
sp_no: str,
snapshot: Dict[str, Any],
) -> None:
"""持久化一份审批状态回写快照（TTL 7 天）。
用途：待办列表缓存只有 45s，快照可用于排障、审计以及回调乱序时的追溯。
Args:
redis: Redis 异步客户端
sp_no: 审批单号
snapshot: 快照内容
"""
try:
await redis.setex(
TODO_APPROVAL_STATUS_KEY.format(sp_no=sp_no),
TODO_APPROVAL_STATUS_TTL,
json.dumps(snapshot, ensure_ascii=False),
)
except Exception as e:
logger.warning(f"写入审批状态快照失败: sp_no={sp_no}, error={e}")
async def _push_todo_status_to_agents(
agent_userids: List[str], payload: Dict[str, Any]
) -> None:
"""向相关坐席推送待办状态变更事件（复用现有 WS 推送通道）。
推送失败不抛异常（坐席可能不在线），由前端下次拉取兜底。
Args:
agent_userids: 目标坐席 userid 列表
payload: 事件数据（对应前端 msg.data）
"""
if not agent_userids:
return
# 延迟导入，避免 api 层与 services 层在模块加载期形成循环依赖
from app.services.ws_manager import manager as ws_manager
message = {"type": "todo_status_changed", "data": payload}
for agent_userid in agent_userids:
try:
await ws_manager.send_to_agent(agent_userid, message)
except Exception as e:
logger.warning(
f"推送待办状态变更失败: agent={agent_userid}, "
f"sp_no={payload.get('sp_no')}, error={e}"
)
async def writeback_approval_todo_status(
sp_no: str,
sp_status: int,
status_change_event: int,
redis: aioredis.Redis,
template_id: str = "",
) -> Dict[str, Any]:
"""企微审批回调 → 坐席待办状态回写 + WS 推送（最终一致）。
流程：
1. 映射 status_change_event/sp_status → 本地待办状态
2. 就地改写命中的待办列表缓存条目，得到受影响坐席
3. 写入状态快照（TTL 7 天）
4. 向受影响坐席推送 todo_status_changed 事件
全流程 try/except 保护：回写失败只记日志，绝不影响回调 ACK
（企微回调失败会重试，且服务台侧有 45s 缓存过期兜底）。
Args:
sp_no: 审批单号（本地待办反查键，本地 id = "approval:{sp_no}"）
sp_status: 企微审批单状态
status_change_event: 企微状态变化类型
redis: Redis 异步客户端
template_id: 审批模板 ID（可选，仅用于日志与快照）
Returns:
Dict[str, Any]: 回写结果 {success, sp_no, status, agents}
"""
if not sp_no:
return {"success": False, "sp_no": sp_no, "message": "sp_no 为空，跳过回写"}
event_type = APPROVAL_EVENT_MAP.get(
status_change_event, f"unknown_{status_change_event}"
)
todo_status = _map_approval_todo_status(status_change_event, sp_status)
updated_at = datetime.now(timezone.utc).isoformat()
try:
# 1. 回写待办列表缓存，拿到受影响坐席
affected_agents = await _patch_todo_cache(redis, sp_no, todo_status, sp_status)
# 2. 持久化状态快照
snapshot: Dict[str, Any] = {
"sp_no": sp_no,
"template_id": template_id,
"sp_status": sp_status,
"status_change_event": status_change_event,
"event_type": event_type,
"todo_status": todo_status,
"affected_agents": affected_agents,
"updated_at": updated_at,
}
await _save_approval_status_snapshot(redis, sp_no, snapshot)
# 3. 推送待办状态变更（仅在有命中坐席时推送）
await _push_todo_status_to_agents(
affected_agents,
{
"item_id": f"approval:{sp_no}",
"todo_type": "approval",
"sp_no": sp_no,
"sp_status": sp_status,
"status": todo_status,
"event_type": event_type,
"updated_at": updated_at,
},
)
logger.info(
f"审批待办状态回写完成: sp_no={sp_no}, event={event_type}, "
f"todo_status={todo_status}, agents={affected_agents}"
)
return {
"success": True,
"sp_no": sp_no,
"status": todo_status,
"event_type": event_type,
"agents": affected_agents,
}
except Exception as e:
logger.error(f"审批待办状态回写失败: sp_no={sp_no}, error={e}", exc_info=True)
return {"success": False, "sp_no": sp_no, "message": f"回写失败: {e}"}
# =============================================================================
# API 端点
# =============================================================================
@@ -935,26 +1276,21 @@ async def approval_callback(
"""
logger.info(f"审批回调: sp_no={sp_no}, status={sp_status}, event={status_change_event}")
# TODO: 根据业务需求处理审批状态变化
# 例如：
# - 审批通过后，更新IT服务台待办状态
# - 审批驳回后，通知申请人
# - 审批撤销后，关闭相关工单
event_map = {
1: "submitted",
2: "approved",
3: "rejected",
4: "transferred",
5: "reminded",
6: "revoked",
8: "revoked_after_approved",
10: "commented"
}
event_type = event_map.get(status_change_event, f"unknown_{status_change_event}")
event_type = APPROVAL_EVENT_MAP.get(status_change_event, f"unknown_{status_change_event}")
logger.info(f"审批事件类型: {event_type}")
# 回写坐席端待办状态 + WS 推送（异步执行，不阻塞回调响应）
# 说明：审批动作只能由审批人在企微原系统完成，服务台通过本回调达成最终一致。
asyncio.create_task(
writeback_approval_todo_status(
sp_no=sp_no,
sp_status=sp_status,
status_change_event=status_change_event,
redis=redis,
template_id=template_id,
)
)
# IT资产升级申请提单时，自动触发年限核查+推送（异步执行，不阻塞回调响应）
if status_change_event == 1 and template_id == ASSET_UPGRADE_TEMPLATE_ID:
logger.info(f"检测到IT资产升级申请提单: sp_no={sp_no}")