feat(approval): 新增审批流程核心模块
实现了完整的审批功能链路，包含审批请求投递、模板渲染、回复解析、引擎管理以及原生交互式卡片支持
This commit is contained in:
parent
90b16ce9b1
commit
79ee930957
30
backend/package/yuxi/channel/approval/__init__.py
Normal file
30
backend/package/yuxi/channel/approval/__init__.py
Normal file
@ -0,0 +1,30 @@
|
||||
from yuxi.channel.approval.delivery import DeliveryResult, deliver_approval_request
|
||||
from yuxi.channel.approval.engine import ApprovalEngine
|
||||
from yuxi.channel.approval.native_runtime import (
|
||||
NativeApprovalCard,
|
||||
NativeApprovalRenderer,
|
||||
get_native_renderer,
|
||||
)
|
||||
from yuxi.channel.approval.parser import ParseResult, parse_approval_reply, parse_channel_approval_reply
|
||||
from yuxi.channel.approval.templates import (
|
||||
ApprovalTemplateConfig,
|
||||
ApprovalTemplateRenderer,
|
||||
TemplateStyle,
|
||||
get_template_renderer,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"ApprovalEngine",
|
||||
"ApprovalTemplateConfig",
|
||||
"ApprovalTemplateRenderer",
|
||||
"DeliveryResult",
|
||||
"NativeApprovalCard",
|
||||
"NativeApprovalRenderer",
|
||||
"ParseResult",
|
||||
"TemplateStyle",
|
||||
"deliver_approval_request",
|
||||
"get_native_renderer",
|
||||
"get_template_renderer",
|
||||
"parse_approval_reply",
|
||||
"parse_channel_approval_reply",
|
||||
]
|
||||
139
backend/package/yuxi/channel/approval/delivery.py
Normal file
139
backend/package/yuxi/channel/approval/delivery.py
Normal file
@ -0,0 +1,139 @@
|
||||
"""审批投递适配器 — 将审批请求通过渠道消息投递给审批人"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from yuxi.channel.protocols import (
|
||||
ApprovalDecision,
|
||||
ApprovalRequest,
|
||||
OutboundProtocol,
|
||||
)
|
||||
from yuxi.channel.plugins.registry import ChannelPluginRegistry
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_TIMEOUT = 300
|
||||
POLL_INTERVAL = 3
|
||||
|
||||
|
||||
@dataclass
|
||||
class DeliveryResult:
|
||||
success: bool
|
||||
request_id: str
|
||||
decision: ApprovalDecision | None = None
|
||||
decided_by: str | None = None
|
||||
reason: str | None = None
|
||||
error: str | None = None
|
||||
elapsed_seconds: float = 0.0
|
||||
delivery_targets: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
async def deliver_approval_request(
|
||||
request: ApprovalRequest,
|
||||
approver_peer_ids: list[str],
|
||||
*,
|
||||
timeout: float = DEFAULT_TIMEOUT,
|
||||
) -> DeliveryResult:
|
||||
"""将审批请求投递到审批人，等待审批结果。
|
||||
|
||||
投递时格式化为结构化审批消息，包含「批准/拒绝」操作指引。
|
||||
并发发送通知给所有审批人，轮询解析审批回复。
|
||||
|
||||
返回 DeliveryResult 包含投递过程详情。
|
||||
"""
|
||||
started_at = time.monotonic()
|
||||
|
||||
plugin = ChannelPluginRegistry.get(request.channel_type)
|
||||
if plugin is None:
|
||||
return _delivery_expired(request.id, "渠道插件未注册", started_at)
|
||||
|
||||
if not isinstance(plugin, OutboundProtocol):
|
||||
return _delivery_expired(request.id, "渠道不支持消息发送", started_at)
|
||||
|
||||
notify_text = _format_approval_message(request)
|
||||
|
||||
send_tasks = [
|
||||
_send_notification(plugin, peer_id, notify_text, request)
|
||||
for peer_id in approver_peer_ids
|
||||
]
|
||||
send_results = await asyncio.gather(*send_tasks)
|
||||
delivery_targets = [peer_id for peer_id, ok in send_results if ok]
|
||||
|
||||
if not delivery_targets:
|
||||
return _delivery_expired(request.id, "所有审批通知发送失败", started_at)
|
||||
|
||||
deadline = time.monotonic() + timeout
|
||||
poll_count = 0
|
||||
while time.monotonic() < deadline:
|
||||
poll_count += 1
|
||||
await asyncio.sleep(POLL_INTERVAL)
|
||||
|
||||
try:
|
||||
if callable(getattr(plugin, "check_approval_status", None)):
|
||||
result = await plugin.check_approval_status({}, request.id)
|
||||
if result is not None:
|
||||
logger.info(
|
||||
"审批结果: req=%s, decision=%s, poll=%d, elapsed=%.1fs",
|
||||
request.id,
|
||||
result.decision,
|
||||
poll_count,
|
||||
time.monotonic() - started_at,
|
||||
)
|
||||
return DeliveryResult(
|
||||
success=True,
|
||||
request_id=request.id,
|
||||
decision=result.decision,
|
||||
decided_by=result.decided_by,
|
||||
reason=result.reason,
|
||||
elapsed_seconds=time.monotonic() - started_at,
|
||||
delivery_targets=delivery_targets,
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("轮询审批状态异常: req=%s", request.id)
|
||||
|
||||
return _delivery_expired(request.id, "审批等待超时", started_at, delivery_targets)
|
||||
|
||||
|
||||
async def _send_notification(
|
||||
plugin,
|
||||
peer_id: str,
|
||||
notify_text: str,
|
||||
request: ApprovalRequest,
|
||||
) -> tuple[str, bool]:
|
||||
try:
|
||||
await plugin.send_text(peer_id, notify_text, account_id=request.account_id)
|
||||
return peer_id, True
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"发送审批通知失败: channel=%s, peer=%s", request.channel_type, peer_id
|
||||
)
|
||||
return peer_id, False
|
||||
|
||||
|
||||
def _format_approval_message(request: ApprovalRequest) -> str:
|
||||
from yuxi.channel.approval.templates import get_template_renderer
|
||||
|
||||
renderer = get_template_renderer("default")
|
||||
return renderer.render_request(request)
|
||||
|
||||
|
||||
def _delivery_expired(
|
||||
request_id: str,
|
||||
reason: str,
|
||||
started_at: float,
|
||||
delivery_targets: list[str] | None = None,
|
||||
) -> DeliveryResult:
|
||||
return DeliveryResult(
|
||||
success=False,
|
||||
request_id=request_id,
|
||||
decision=ApprovalDecision.EXPIRED,
|
||||
decided_by="system",
|
||||
reason=reason,
|
||||
elapsed_seconds=time.monotonic() - started_at,
|
||||
delivery_targets=delivery_targets or [],
|
||||
error=reason,
|
||||
)
|
||||
272
backend/package/yuxi/channel/approval/engine.py
Normal file
272
backend/package/yuxi/channel/approval/engine.py
Normal file
@ -0,0 +1,272 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
|
||||
from yuxi.channel.protocols import (
|
||||
ApprovalAction,
|
||||
ApprovalDecision,
|
||||
ApprovalProtocol,
|
||||
ApprovalRequest,
|
||||
ApprovalResult,
|
||||
)
|
||||
from yuxi.channel.plugins.registry import ChannelPluginRegistry
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_TERMINAL_DECISIONS = frozenset({
|
||||
ApprovalDecision.APPROVED,
|
||||
ApprovalDecision.DENIED,
|
||||
ApprovalDecision.CANCELLED,
|
||||
ApprovalDecision.EXPIRED,
|
||||
})
|
||||
|
||||
|
||||
class ApprovalEngine:
|
||||
def __init__(self):
|
||||
self._pending: dict[str, ApprovalRequest] = {}
|
||||
self._results: dict[str, ApprovalResult] = {}
|
||||
self._events: dict[str, asyncio.Event] = {}
|
||||
self._cleanup_task: asyncio.Task | None = None
|
||||
|
||||
@property
|
||||
def pending_count(self) -> int:
|
||||
return len(self._pending)
|
||||
|
||||
@property
|
||||
def results_count(self) -> int:
|
||||
return len(self._results)
|
||||
|
||||
def get_pending_request(self, request_id: str) -> ApprovalRequest | None:
|
||||
return self._pending.get(request_id)
|
||||
|
||||
def get_result(self, request_id: str) -> ApprovalResult | None:
|
||||
return self._results.get(request_id)
|
||||
|
||||
def _is_terminal(self, request_id: str) -> bool:
|
||||
existing = self._results.get(request_id)
|
||||
return existing is not None and existing.decision in _TERMINAL_DECISIONS
|
||||
|
||||
def cancel_request(self, request_id: str, reason: str = "已取消") -> bool:
|
||||
existing = self._results.get(request_id)
|
||||
if existing is not None:
|
||||
logger.warning(
|
||||
"审批请求已达终态，无法取消: req=%s, current=%s",
|
||||
request_id,
|
||||
existing.decision.value,
|
||||
)
|
||||
return False
|
||||
if request_id not in self._pending:
|
||||
return False
|
||||
del self._pending[request_id]
|
||||
result = ApprovalResult(
|
||||
request_id=request_id,
|
||||
decision=ApprovalDecision.CANCELLED,
|
||||
reason=reason,
|
||||
)
|
||||
self._results[request_id] = result
|
||||
self._signal(request_id)
|
||||
logger.info("审批请求已取消: req=%s, reason=%s", request_id, reason)
|
||||
return True
|
||||
|
||||
def register_decision(
|
||||
self,
|
||||
request_id: str,
|
||||
decision: ApprovalDecision,
|
||||
decided_by: str = "external",
|
||||
reason: str | None = None,
|
||||
) -> bool:
|
||||
existing = self._results.get(request_id)
|
||||
if existing is not None:
|
||||
logger.warning(
|
||||
"审批请求已达终态，无法重新决策: req=%s, current=%s, attempted=%s",
|
||||
request_id,
|
||||
existing.decision.value,
|
||||
decision.value,
|
||||
)
|
||||
return False
|
||||
request = self._pending.get(request_id)
|
||||
if request is None:
|
||||
return False
|
||||
if not request.approver_ids:
|
||||
logger.warning(
|
||||
"审批身份校验失败: req=%s, 审批人列表为空",
|
||||
request_id,
|
||||
)
|
||||
return False
|
||||
if decided_by not in request.approver_ids:
|
||||
logger.warning(
|
||||
"审批身份校验失败: req=%s, decided_by=%s 不在审批人列表 %s 中",
|
||||
request_id,
|
||||
decided_by,
|
||||
request.approver_ids,
|
||||
)
|
||||
return False
|
||||
del self._pending[request_id]
|
||||
result = ApprovalResult(
|
||||
request_id=request_id,
|
||||
decision=decision,
|
||||
decided_by=decided_by,
|
||||
reason=reason,
|
||||
)
|
||||
self._results[request_id] = result
|
||||
self._signal(request_id)
|
||||
logger.info(
|
||||
"审批决策已注册: req=%s, decision=%s, decided_by=%s, reason=%s",
|
||||
request_id,
|
||||
decision.value,
|
||||
decided_by,
|
||||
reason,
|
||||
)
|
||||
return True
|
||||
|
||||
def cleanup_expired(self, max_age_seconds: float = 300) -> int:
|
||||
removed = 0
|
||||
now = time.monotonic()
|
||||
stale = [
|
||||
rid for rid, req in self._pending.items()
|
||||
if now - req.created_at > max_age_seconds and rid not in self._results
|
||||
]
|
||||
for rid in stale:
|
||||
del self._pending[rid]
|
||||
result = ApprovalResult(
|
||||
request_id=rid,
|
||||
decision=ApprovalDecision.EXPIRED,
|
||||
reason="清理过期审批",
|
||||
)
|
||||
self._results[rid] = result
|
||||
self._signal(rid)
|
||||
removed += 1
|
||||
if removed > 0:
|
||||
logger.info("清理过期审批请求: count=%d", removed)
|
||||
return removed
|
||||
|
||||
def cleanup_results(self, max_age_seconds: float = 600) -> int:
|
||||
removed = 0
|
||||
now = time.monotonic()
|
||||
stale = [rid for rid, res in self._results.items() if now - res.decided_at > max_age_seconds]
|
||||
for rid in stale:
|
||||
del self._results[rid]
|
||||
removed += 1
|
||||
if removed > 0:
|
||||
logger.info("清理过期审批结果: count=%d", removed)
|
||||
return removed
|
||||
|
||||
def start_cleanup_task(self, interval: float = 60) -> None:
|
||||
if self._cleanup_task is not None and not self._cleanup_task.done():
|
||||
return
|
||||
self._cleanup_task = asyncio.create_task(self._run_cleanup_loop(interval))
|
||||
|
||||
def stop_cleanup_task(self) -> None:
|
||||
if self._cleanup_task is not None and not self._cleanup_task.done():
|
||||
self._cleanup_task.cancel()
|
||||
|
||||
async def _run_cleanup_loop(self, interval: float) -> None:
|
||||
try:
|
||||
while True:
|
||||
await asyncio.sleep(interval)
|
||||
self.cleanup_expired()
|
||||
self.cleanup_results()
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
def _signal(self, request_id: str) -> None:
|
||||
event = self._events.pop(request_id, None)
|
||||
if event is not None:
|
||||
event.set()
|
||||
|
||||
async def request_approval(
|
||||
self,
|
||||
channel_type: str,
|
||||
account_id: str,
|
||||
config: dict,
|
||||
action: ApprovalAction,
|
||||
initiator_peer_id: str,
|
||||
description: str,
|
||||
context: dict,
|
||||
) -> ApprovalRequest | None:
|
||||
plugin = ChannelPluginRegistry.get(channel_type)
|
||||
if plugin is None or not isinstance(plugin, ApprovalProtocol):
|
||||
return None
|
||||
|
||||
need = await plugin.check_approval_required(config, action, initiator_peer_id)
|
||||
if not need:
|
||||
return None
|
||||
|
||||
request = await plugin.create_approval_request(
|
||||
config,
|
||||
action,
|
||||
initiator_peer_id,
|
||||
description,
|
||||
context,
|
||||
)
|
||||
approvers = plugin.get_approver_ids(config)
|
||||
if not approvers:
|
||||
logger.warning(
|
||||
"审批请求创建失败: channel=%s, 审批人列表为空",
|
||||
channel_type,
|
||||
)
|
||||
return None
|
||||
request.approver_ids = list(approvers)
|
||||
sent = await plugin.send_approval_notification(config, request, approvers)
|
||||
if not sent:
|
||||
logger.warning(
|
||||
"审批通知发送失败: req=%s, channel=%s, approvers=%s",
|
||||
request.id,
|
||||
channel_type,
|
||||
approvers,
|
||||
)
|
||||
|
||||
self._pending[request.id] = request
|
||||
return request
|
||||
|
||||
async def wait_for_decision(
|
||||
self, channel_type: str, config: dict, request_id: str, timeout: float = 300
|
||||
) -> ApprovalResult:
|
||||
existing = self._results.get(request_id)
|
||||
if existing is not None:
|
||||
return existing
|
||||
|
||||
event = self._events.get(request_id)
|
||||
if event is None:
|
||||
event = asyncio.Event()
|
||||
self._events[request_id] = event
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + timeout
|
||||
while True:
|
||||
remaining = deadline - loop.time()
|
||||
if remaining <= 0:
|
||||
break
|
||||
|
||||
plugin_result = await self._check_and_update(channel_type, config, request_id)
|
||||
if plugin_result is not None:
|
||||
self._pending.pop(request_id, None)
|
||||
self._results[request_id] = plugin_result
|
||||
self._events.pop(request_id, None)
|
||||
return plugin_result
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(event.wait(), timeout=min(remaining, 5))
|
||||
except asyncio.TimeoutError:
|
||||
continue
|
||||
|
||||
stored = self._results.get(request_id)
|
||||
if stored is not None:
|
||||
self._events.pop(request_id, None)
|
||||
return stored
|
||||
|
||||
self._events.pop(request_id, None)
|
||||
expired = ApprovalResult(
|
||||
request_id=request_id,
|
||||
decision=ApprovalDecision.EXPIRED,
|
||||
reason="审批超时",
|
||||
)
|
||||
self._pending.pop(request_id, None)
|
||||
self._results[request_id] = expired
|
||||
return expired
|
||||
|
||||
async def _check_and_update(self, channel_type: str, config: dict, request_id: str) -> ApprovalResult | None:
|
||||
plugin = ChannelPluginRegistry.get(channel_type)
|
||||
if plugin is None or not isinstance(plugin, ApprovalProtocol):
|
||||
return None
|
||||
return await plugin.check_approval_status(config, request_id)
|
||||
373
backend/package/yuxi/channel/approval/native_runtime.py
Normal file
373
backend/package/yuxi/channel/approval/native_runtime.py
Normal file
@ -0,0 +1,373 @@
|
||||
"""审批原生运行时 — 通过渠道原生交互式卡片渲染审批 UI
|
||||
|
||||
相比 delivery.py 的纯文本投递，native_runtime 使用渠道的交互式卡片消息
|
||||
渲染审批按钮，提供更好的用户体验。
|
||||
|
||||
支持的渠道卡片格式：
|
||||
- 飞书: interactive card (lark_md + button actions)
|
||||
- 钉钉: action_card (Markdown + btns)
|
||||
- 其他渠道: 降级为纯文本投递
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from yuxi.channel.protocols import (
|
||||
ApprovalDecision,
|
||||
ApprovalRequest,
|
||||
ApprovalResult,
|
||||
)
|
||||
from yuxi.channel.plugins.registry import ChannelPluginRegistry
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
FEISHU = "feishu"
|
||||
DINGTALK = "dingtalk"
|
||||
DEFAULT_TIMEOUT_MINUTES = 5
|
||||
DEFAULT_TIMEOUT = 300
|
||||
POLL_INTERVAL = 3
|
||||
|
||||
|
||||
@dataclass
|
||||
class NativeApprovalCard:
|
||||
channel_type: str
|
||||
card_content: dict
|
||||
request_id: str
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"channel_type": self.channel_type,
|
||||
"card_content": self.card_content,
|
||||
"request_id": self.request_id,
|
||||
}
|
||||
|
||||
|
||||
class NativeApprovalRenderer:
|
||||
def render_card(self, request: ApprovalRequest, channel_type: str) -> NativeApprovalCard:
|
||||
if channel_type == FEISHU:
|
||||
card = _build_feishu_approval_card(request)
|
||||
elif channel_type == DINGTALK:
|
||||
card = _build_dingtalk_approval_card(request)
|
||||
else:
|
||||
card = _build_fallback_card(request)
|
||||
return NativeApprovalCard(
|
||||
channel_type=channel_type,
|
||||
card_content=card,
|
||||
request_id=request.id,
|
||||
)
|
||||
|
||||
def render_resolved_card(self, result: ApprovalResult, request: ApprovalRequest | None, channel_type: str) -> dict:
|
||||
if channel_type == FEISHU:
|
||||
return _build_feishu_resolved_card(result, request)
|
||||
elif channel_type == DINGTALK:
|
||||
return _build_dingtalk_resolved_card(result, request)
|
||||
return _build_fallback_resolved_card(result, request)
|
||||
|
||||
async def deliver_native_approval(
|
||||
self,
|
||||
request: ApprovalRequest,
|
||||
approver_peer_ids: list[str],
|
||||
*,
|
||||
timeout: float = DEFAULT_TIMEOUT,
|
||||
) -> ApprovalResult:
|
||||
"""通过原生交互式卡片投递审批请求。
|
||||
|
||||
1. 构建渠道原生卡片
|
||||
2. 通过 send_card 发送给审批人
|
||||
3. 轮询等待审批结果
|
||||
4. 获得结果后更新卡片为已处理状态
|
||||
|
||||
若渠道不支持 send_card，降级为纯文本投递。
|
||||
"""
|
||||
from yuxi.channel.approval.delivery import deliver_approval_request
|
||||
|
||||
plugin = ChannelPluginRegistry.get(request.channel_type)
|
||||
if plugin is None:
|
||||
return _native_expired(request.id, "渠道插件未注册")
|
||||
|
||||
if not _supports_card(plugin):
|
||||
logger.info("渠道 %s 不支持卡片消息，降级为纯文本投递", request.channel_type)
|
||||
return await deliver_approval_request(request, approver_peer_ids, timeout=timeout)
|
||||
|
||||
card = self.render_card(request, request.channel_type)
|
||||
sent_message_ids: dict[str, str] = {}
|
||||
|
||||
for peer_id in approver_peer_ids:
|
||||
try:
|
||||
raw = await plugin.send_card(peer_id, card.card_content)
|
||||
msg_id = _extract_msg_id(raw)
|
||||
if msg_id:
|
||||
sent_message_ids[peer_id] = msg_id
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"发送审批卡片失败: channel=%s, peer=%s",
|
||||
request.channel_type,
|
||||
peer_id,
|
||||
)
|
||||
|
||||
if not sent_message_ids:
|
||||
return _native_expired(request.id, "所有审批通知发送失败")
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
deadline = loop.time() + timeout
|
||||
try:
|
||||
while loop.time() < deadline:
|
||||
await asyncio.sleep(POLL_INTERVAL)
|
||||
|
||||
if callable(getattr(plugin, "check_approval_status", None)):
|
||||
result = await plugin.check_approval_status({}, request.id)
|
||||
if result is not None:
|
||||
resolved_card = self.render_resolved_card(result, request, request.channel_type)
|
||||
await _edit_cards(plugin, sent_message_ids, resolved_card)
|
||||
return result
|
||||
|
||||
expired = _native_expired(request.id, "审批等待超时")
|
||||
resolved_card = self.render_resolved_card(expired, request, request.channel_type)
|
||||
await _edit_cards(plugin, sent_message_ids, resolved_card)
|
||||
return expired
|
||||
except asyncio.CancelledError:
|
||||
cancelled = ApprovalResult(
|
||||
request_id=request.id,
|
||||
decision=ApprovalDecision.CANCELLED,
|
||||
decided_by="system",
|
||||
reason="审批被取消",
|
||||
)
|
||||
resolved_card = self.render_resolved_card(cancelled, request, request.channel_type)
|
||||
await _edit_cards(plugin, sent_message_ids, resolved_card)
|
||||
raise
|
||||
|
||||
|
||||
_native_renderer = NativeApprovalRenderer()
|
||||
|
||||
|
||||
def get_native_renderer() -> NativeApprovalRenderer:
|
||||
return _native_renderer
|
||||
|
||||
|
||||
def _supports_card(plugin: Any) -> bool:
|
||||
has_send = callable(getattr(plugin, "send_card", None))
|
||||
has_edit = (
|
||||
callable(getattr(plugin, "edit_card", None))
|
||||
or callable(getattr(plugin, "update_card", None))
|
||||
)
|
||||
return has_send and has_edit
|
||||
|
||||
|
||||
def _extract_msg_id(raw: Any) -> str | None:
|
||||
if raw is None:
|
||||
return None
|
||||
if isinstance(raw, str):
|
||||
return raw
|
||||
if isinstance(raw, dict):
|
||||
return raw.get("msg_id") or raw.get("message_id")
|
||||
return None
|
||||
|
||||
|
||||
async def _edit_cards(plugin: Any, sent_message_ids: dict[str, str], resolved_card: dict) -> None:
|
||||
edit_fn = getattr(plugin, "edit_card", None) or getattr(plugin, "update_card", None)
|
||||
if edit_fn is None:
|
||||
return
|
||||
for peer_id, msg_id in sent_message_ids.items():
|
||||
try:
|
||||
await edit_fn(peer_id, msg_id, resolved_card)
|
||||
except Exception:
|
||||
logger.debug("更新审批卡片失败: %s/%s", peer_id, msg_id)
|
||||
|
||||
|
||||
def _native_expired(request_id: str, reason: str) -> ApprovalResult:
|
||||
return ApprovalResult(
|
||||
request_id=request_id,
|
||||
decision=ApprovalDecision.EXPIRED,
|
||||
decided_by="system",
|
||||
reason=reason,
|
||||
)
|
||||
|
||||
|
||||
# ── 飞书交互式卡片 ──────────────────────────────────────
|
||||
|
||||
|
||||
def _build_feishu_approval_card(request: ApprovalRequest) -> dict:
|
||||
timeout_minutes = DEFAULT_TIMEOUT_MINUTES
|
||||
return {
|
||||
"config": {"wide_screen_mode": True},
|
||||
"header": {
|
||||
"title": {"tag": "plain_text", "content": "🔐 审批请求"},
|
||||
"template": "blue",
|
||||
},
|
||||
"elements": [
|
||||
{
|
||||
"tag": "div",
|
||||
"fields": [
|
||||
{"is_short": True, "text": {"tag": "lark_md", "content": f"**请求 ID**\n`{request.id[:12]}...`"}},
|
||||
{"is_short": True, "text": {"tag": "lark_md", "content": f"**操作类型**\n{request.action.value}"}},
|
||||
],
|
||||
},
|
||||
{
|
||||
"tag": "div",
|
||||
"text": {"tag": "lark_md", "content": f"**描述**\n{request.description}"},
|
||||
},
|
||||
{
|
||||
"tag": "div",
|
||||
"fields": [
|
||||
{
|
||||
"is_short": True,
|
||||
"text": {"tag": "lark_md", "content": f"**发起人**\n{request.initiator_peer_id}"},
|
||||
},
|
||||
{"is_short": True, "text": {"tag": "lark_md", "content": f"**有效期**\n{timeout_minutes} 分钟"}},
|
||||
],
|
||||
},
|
||||
{"tag": "hr"},
|
||||
{
|
||||
"tag": "action",
|
||||
"actions": [
|
||||
{
|
||||
"tag": "button",
|
||||
"text": {"tag": "lark_md", "content": "✅ 批准"},
|
||||
"type": "primary",
|
||||
"value": f'{{"action":"approve","req":"{request.id}"}}',
|
||||
},
|
||||
{
|
||||
"tag": "button",
|
||||
"text": {"tag": "lark_md", "content": "❌ 拒绝"},
|
||||
"type": "danger",
|
||||
"value": f'{{"action":"deny","req":"{request.id}"}}',
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _build_feishu_resolved_card(result: ApprovalResult, request: ApprovalRequest | None) -> dict:
|
||||
if result.decision == ApprovalDecision.APPROVED:
|
||||
template = "green"
|
||||
title = "✅ 审批已通过"
|
||||
reason_text = f"**通过原因**: {result.reason}" if result.reason else ""
|
||||
elif result.decision == ApprovalDecision.DENIED:
|
||||
template = "red"
|
||||
title = "❌ 审批已拒绝"
|
||||
reason_text = f"**拒绝原因**: {result.reason}" if result.reason else ""
|
||||
elif result.decision == ApprovalDecision.CANCELLED:
|
||||
template = "grey"
|
||||
title = "🚫 审批已取消"
|
||||
reason_text = f"**取消原因**: {result.reason}" if result.reason else ""
|
||||
else:
|
||||
template = "grey"
|
||||
title = "⏰ 审批已过期"
|
||||
reason_text = f"**原因**: {result.reason}" if result.reason else ""
|
||||
|
||||
elements: list[dict] = [
|
||||
{
|
||||
"tag": "div",
|
||||
"text": {"tag": "lark_md", "content": f"请求 ID: `{result.request_id[:12]}...`"},
|
||||
},
|
||||
]
|
||||
|
||||
if result.decided_by:
|
||||
elements.append(
|
||||
{
|
||||
"tag": "div",
|
||||
"text": {"tag": "lark_md", "content": f"决定人: {result.decided_by}"},
|
||||
}
|
||||
)
|
||||
|
||||
if reason_text:
|
||||
elements.append(
|
||||
{
|
||||
"tag": "div",
|
||||
"text": {"tag": "lark_md", "content": reason_text},
|
||||
}
|
||||
)
|
||||
|
||||
if request:
|
||||
elements.insert(
|
||||
0,
|
||||
{
|
||||
"tag": "div",
|
||||
"text": {"tag": "lark_md", "content": f"原请求: {request.description}"},
|
||||
},
|
||||
)
|
||||
|
||||
return {
|
||||
"config": {"wide_screen_mode": True},
|
||||
"header": {
|
||||
"title": {"tag": "plain_text", "content": title},
|
||||
"template": template,
|
||||
},
|
||||
"elements": elements,
|
||||
}
|
||||
|
||||
|
||||
# ── 钉钉 Action Card ────────────────────────────────────
|
||||
|
||||
|
||||
def _build_dingtalk_approval_card(request: ApprovalRequest) -> dict:
|
||||
timeout_minutes = DEFAULT_TIMEOUT_MINUTES
|
||||
markdown = (
|
||||
f"## 🔐 审批请求\n\n"
|
||||
f"**请求 ID**: `{request.id[:12]}...`\n\n"
|
||||
f"**操作类型**: {request.action.value}\n\n"
|
||||
f"**描述**: {request.description}\n\n"
|
||||
f"**发起人**: {request.initiator_peer_id}\n\n"
|
||||
f"**有效期**: {timeout_minutes} 分钟\n\n"
|
||||
f"> 请点击下方按钮做出决定"
|
||||
)
|
||||
return {
|
||||
"title": "🔐 审批请求",
|
||||
"text": markdown,
|
||||
"btnOrientation": "1",
|
||||
"btns": [
|
||||
{
|
||||
"title": "✅ 批准",
|
||||
"actionURL": f"dingtalk://yuxi/approval/approve?req={request.id}",
|
||||
},
|
||||
{
|
||||
"title": "❌ 拒绝",
|
||||
"actionURL": f"dingtalk://yuxi/approval/deny?req={request.id}",
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _build_dingtalk_resolved_card(result: ApprovalResult, request: ApprovalRequest | None) -> dict:
|
||||
if result.decision == ApprovalDecision.APPROVED:
|
||||
title = "✅ 审批已通过"
|
||||
elif result.decision == ApprovalDecision.DENIED:
|
||||
title = "❌ 审批已拒绝"
|
||||
elif result.decision == ApprovalDecision.CANCELLED:
|
||||
title = "🚫 审批已取消"
|
||||
else:
|
||||
title = "⏰ 审批已过期"
|
||||
|
||||
lines = [f"## {title}", "", f"**请求 ID**: `{result.request_id[:12]}...`"]
|
||||
|
||||
if result.decided_by:
|
||||
lines.append(f"**决定人**: {result.decided_by}")
|
||||
if result.reason:
|
||||
lines.append(f"**原因**: {result.reason}")
|
||||
|
||||
markdown = "\n\n".join(lines)
|
||||
return {"title": title, "text": markdown}
|
||||
|
||||
|
||||
# ── 降级卡片（纯文本格式的卡片消息）────────────────────
|
||||
|
||||
|
||||
def _build_fallback_card(request: ApprovalRequest) -> dict:
|
||||
from yuxi.channel.approval.templates import get_template_renderer
|
||||
|
||||
renderer = get_template_renderer("default")
|
||||
text = renderer.render_request(request)
|
||||
return {"text": text, "type": "plain_text"}
|
||||
|
||||
|
||||
def _build_fallback_resolved_card(result: ApprovalResult, request: ApprovalRequest | None) -> dict:
|
||||
from yuxi.channel.approval.templates import get_template_renderer
|
||||
|
||||
renderer = get_template_renderer("default")
|
||||
text = renderer.render_result(result, request)
|
||||
return {"text": text, "type": "plain_text"}
|
||||
229
backend/package/yuxi/channel/approval/parser.py
Normal file
229
backend/package/yuxi/channel/approval/parser.py
Normal file
@ -0,0 +1,229 @@
|
||||
"""审批响应解析器 — 解析审批人回复文本中的批准/拒绝信号"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from yuxi.channel.protocols import ApprovalDecision
|
||||
|
||||
APPROVE_PATTERNS = [
|
||||
re.compile(r"^批准|^同意|^通过|^approve|^yes|^ok|^可以|^确认"),
|
||||
re.compile(r"批准|同意|通过|approve|yes|ok", re.IGNORECASE),
|
||||
]
|
||||
|
||||
DENY_PATTERNS = [
|
||||
re.compile(r"^拒绝|^驳回|^deny|^no(?!t)\b|^不行"),
|
||||
re.compile(r"拒绝|驳回|deny|\bno(?!t)\b", re.IGNORECASE),
|
||||
]
|
||||
|
||||
# 当 "no" 后面紧跟这些词时，视为否定表达而非拒绝信号
|
||||
_NO_EXCLUDE_SUFFIX = re.compile(r"\s+(?:approve|agree|ok|yes|pass|可以|批准|同意|通过)", re.IGNORECASE)
|
||||
|
||||
CANCEL_PATTERNS = [
|
||||
re.compile(r"^取消|^撤销|^cancel|^撤回"),
|
||||
re.compile(r"取消|cancel", re.IGNORECASE),
|
||||
]
|
||||
|
||||
# 否定词模式 — 要求紧邻匹配关键词前（允许可选空白）
|
||||
# 包括独立的否定词 + 中文复合否定词（不能、不会、没法、无法等）
|
||||
_NEGATION_PATTERNS = [
|
||||
re.compile(r"不\s*$"),
|
||||
re.compile(r"没\s*$"),
|
||||
re.compile(r"未\s*$"),
|
||||
re.compile(r"别\s*$"),
|
||||
re.compile(r"勿\s*$"),
|
||||
re.compile(r"(?:不能|不会|不要|不可|不准|不想|不愿|不敢|不必|不用)\s*$"),
|
||||
re.compile(r"(?:没法|没有|无法)\s*$"),
|
||||
re.compile(r"(?:不可以|不应该|不可能|不允许|不准许|不值得)\s*$"),
|
||||
re.compile(r"non[-\s]?$", re.IGNORECASE),
|
||||
re.compile(r"not\s*$", re.IGNORECASE),
|
||||
re.compile(r"no\s*$", re.IGNORECASE),
|
||||
re.compile(r"never\s*$", re.IGNORECASE),
|
||||
re.compile(r"don'?t\s*$", re.IGNORECASE),
|
||||
re.compile(r"doesn'?t\s*$", re.IGNORECASE),
|
||||
re.compile(r"didn'?t\s*$", re.IGNORECASE),
|
||||
re.compile(r"won'?t\s*$", re.IGNORECASE),
|
||||
re.compile(r"wouldn'?t\s*$", re.IGNORECASE),
|
||||
re.compile(r"can'?t\s*$", re.IGNORECASE),
|
||||
re.compile(r"couldn'?t\s*$", re.IGNORECASE),
|
||||
re.compile(r"shouldn'?t\s*$", re.IGNORECASE),
|
||||
]
|
||||
|
||||
# 转折/让步词 — 当这些词出现在批准关键词之后时，降低置信度或视为拒绝
|
||||
_CONCESSION_PATTERNS = [
|
||||
re.compile(r"[,，。；\n]\s*但是"),
|
||||
re.compile(r"[,，。；\n]\s*但"),
|
||||
re.compile(r"[,，。；\n]\s*不过"),
|
||||
re.compile(r"[,，。；\n]\s*然而"),
|
||||
re.compile(r"[,，。；\n]\s*只是"),
|
||||
re.compile(r"[,，。；\n]\s*except"),
|
||||
re.compile(r"[,，。；\n]\s*but"),
|
||||
re.compile(r"[,，。；\n]\s*however"),
|
||||
]
|
||||
|
||||
|
||||
def _is_negated(text: str, match_start: int) -> bool:
|
||||
"""检查匹配位置前是否紧邻否定词。
|
||||
|
||||
只检查匹配位置之前的有限文本片段（最多20个字符），
|
||||
要求否定词紧邻关键词前（中间仅允许空白字符），
|
||||
避免"不是不行"中"不行"被误判，也避免远距离否定词干扰。
|
||||
同时避免否定词属于前一句子的情况。
|
||||
"""
|
||||
if match_start <= 0:
|
||||
return False
|
||||
window_start = max(0, match_start - 20)
|
||||
prefix = text[window_start:match_start]
|
||||
|
||||
sentence_breaks = [". ", "。", "? ", "？", "! ", "！", "\n"]
|
||||
for breaker in sentence_breaks:
|
||||
idx = prefix.rfind(breaker)
|
||||
if idx >= 0:
|
||||
prefix = prefix[idx + len(breaker):]
|
||||
|
||||
for neg_pattern in _NEGATION_PATTERNS:
|
||||
if neg_pattern.search(prefix):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _has_concession_after(text: str, match_end: int) -> bool:
|
||||
"""检查匹配位置后是否出现转折/让步词。"""
|
||||
if match_end >= len(text):
|
||||
return False
|
||||
suffix = text[match_end:]
|
||||
for con_pattern in _CONCESSION_PATTERNS:
|
||||
if con_pattern.search(suffix):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParseResult:
|
||||
decision: ApprovalDecision | None
|
||||
reason: str | None = None
|
||||
confidence: float = 0.0
|
||||
matched_pattern: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class ApprovalContext:
|
||||
channel_type: str
|
||||
account_id: str
|
||||
request_id: str
|
||||
approver_id: str
|
||||
reply_text: str
|
||||
reply_at: float = 0.0
|
||||
metadata: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
def parse_approval_reply(
|
||||
reply_text: str,
|
||||
*,
|
||||
strict: bool = False,
|
||||
) -> ParseResult:
|
||||
"""解析审批回复文本，识别审批决定。
|
||||
|
||||
优先级: 批准 > 拒绝 > 取消
|
||||
严格模式下仅匹配行首关键字（如「批准」「拒绝」）。
|
||||
非严格模式下会检测否定词前缀，避免"不批准"被误判为批准。
|
||||
"""
|
||||
original = reply_text.strip()
|
||||
cleaned = original.lower()
|
||||
|
||||
if strict:
|
||||
for pattern in APPROVE_PATTERNS[:1]:
|
||||
match = pattern.match(cleaned)
|
||||
if match and not _is_negated(cleaned, match.start()):
|
||||
if _has_concession_after(cleaned, match.end()):
|
||||
return ParseResult(
|
||||
decision=ApprovalDecision.DENIED,
|
||||
confidence=0.75,
|
||||
matched_pattern=pattern.pattern,
|
||||
reason=original,
|
||||
)
|
||||
return ParseResult(
|
||||
decision=ApprovalDecision.APPROVED,
|
||||
confidence=0.95,
|
||||
matched_pattern=pattern.pattern,
|
||||
)
|
||||
for pattern in DENY_PATTERNS[:1]:
|
||||
match = pattern.match(cleaned)
|
||||
if match and not _is_negated(cleaned, match.start()):
|
||||
return ParseResult(
|
||||
decision=ApprovalDecision.DENIED,
|
||||
confidence=0.95,
|
||||
matched_pattern=pattern.pattern,
|
||||
reason=original,
|
||||
)
|
||||
for pattern in CANCEL_PATTERNS[:1]:
|
||||
match = pattern.match(cleaned)
|
||||
if match and not _is_negated(cleaned, match.start()):
|
||||
return ParseResult(
|
||||
decision=ApprovalDecision.CANCELLED,
|
||||
confidence=0.95,
|
||||
matched_pattern=pattern.pattern,
|
||||
reason=original,
|
||||
)
|
||||
return ParseResult(decision=None, confidence=0.0)
|
||||
|
||||
for pattern in APPROVE_PATTERNS:
|
||||
match = pattern.search(cleaned)
|
||||
if match and not _is_negated(cleaned, match.start()):
|
||||
if _has_concession_after(cleaned, match.end()):
|
||||
return ParseResult(
|
||||
decision=ApprovalDecision.DENIED,
|
||||
confidence=0.70,
|
||||
matched_pattern=pattern.pattern,
|
||||
reason=original,
|
||||
)
|
||||
return ParseResult(
|
||||
decision=ApprovalDecision.APPROVED,
|
||||
confidence=0.85,
|
||||
matched_pattern=pattern.pattern,
|
||||
)
|
||||
|
||||
for pattern in DENY_PATTERNS:
|
||||
match = pattern.search(cleaned)
|
||||
if match and not _is_negated(cleaned, match.start()):
|
||||
matched_text = match.group().lower()
|
||||
if matched_text == "no" and _NO_EXCLUDE_SUFFIX.match(cleaned[match.end() :]):
|
||||
continue
|
||||
return ParseResult(
|
||||
decision=ApprovalDecision.DENIED,
|
||||
confidence=0.85,
|
||||
matched_pattern=pattern.pattern,
|
||||
reason=original,
|
||||
)
|
||||
|
||||
for pattern in CANCEL_PATTERNS:
|
||||
match = pattern.search(cleaned)
|
||||
if match and not _is_negated(cleaned, match.start()):
|
||||
return ParseResult(
|
||||
decision=ApprovalDecision.CANCELLED,
|
||||
confidence=0.70,
|
||||
matched_pattern=pattern.pattern,
|
||||
reason="审批人取消",
|
||||
)
|
||||
|
||||
return ParseResult(decision=None, confidence=0.0)
|
||||
|
||||
|
||||
async def parse_channel_approval_reply(
|
||||
channel_type: str,
|
||||
reply_text: str,
|
||||
*,
|
||||
strict: bool = False,
|
||||
) -> ParseResult:
|
||||
"""渠道感知的审批回复解析。
|
||||
|
||||
各渠道可自定义解析逻辑。当前统一使用文本模式匹配，
|
||||
预留渠道扩展点。
|
||||
"""
|
||||
result = parse_approval_reply(reply_text, strict=strict)
|
||||
if result.decision is not None:
|
||||
return result
|
||||
|
||||
return result
|
||||
164
backend/package/yuxi/channel/approval/templates.py
Normal file
164
backend/package/yuxi/channel/approval/templates.py
Normal file
@ -0,0 +1,164 @@
|
||||
"""审批展示模板 — 可自定义的审批信息展示格式"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html as _html
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from enum import StrEnum
|
||||
|
||||
from yuxi.channel.protocols import ApprovalAction, ApprovalDecision, ApprovalRequest, ApprovalResult
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_MD_ESCAPE_RE = re.compile(r"([\\`*_{}\[\]()#+\-.!])")
|
||||
|
||||
|
||||
class TemplateStyle(StrEnum):
|
||||
PLAIN = "plain"
|
||||
MARKDOWN = "markdown"
|
||||
HTML = "html"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ApprovalTemplateConfig:
|
||||
title: str = "审批请求"
|
||||
show_request_id: bool = True
|
||||
show_action: bool = True
|
||||
show_description: bool = True
|
||||
show_initiator: bool = True
|
||||
show_expiry: bool = True
|
||||
style: TemplateStyle = TemplateStyle.PLAIN
|
||||
|
||||
approve_label: str = "批准"
|
||||
deny_label: str = "拒绝"
|
||||
timeout_minutes: int = 5
|
||||
|
||||
|
||||
_DEFAULT_CONFIG = ApprovalTemplateConfig()
|
||||
|
||||
|
||||
class ApprovalTemplateRenderer:
|
||||
def __init__(self, config: ApprovalTemplateConfig | None = None):
|
||||
self._config = config or _DEFAULT_CONFIG
|
||||
|
||||
@property
|
||||
def config(self) -> ApprovalTemplateConfig:
|
||||
return self._config
|
||||
|
||||
def render_request(self, request: ApprovalRequest) -> str:
|
||||
cfg = self._config
|
||||
lines: list[str] = [f"🔐 {self._bold(cfg.title)}"]
|
||||
|
||||
if cfg.show_request_id:
|
||||
lines.append(f"请求 ID: {self._code(request.id)}")
|
||||
if cfg.show_action:
|
||||
lines.append(f"操作类型: {_action_label(request.action)}")
|
||||
if cfg.show_description:
|
||||
lines.append(f"描述: {self._escape(request.description)}")
|
||||
if cfg.show_initiator:
|
||||
lines.append(f"发起人: {self._escape(request.initiator_peer_id)}")
|
||||
if cfg.show_expiry and cfg.timeout_minutes > 0:
|
||||
lines.append(f"请在 {cfg.timeout_minutes} 分钟内回复")
|
||||
|
||||
lines.append("")
|
||||
lines.append(f"回复「{self._bold(cfg.approve_label)}」或「{self._bold(cfg.deny_label)}」做出决定")
|
||||
return "\n".join(lines)
|
||||
|
||||
def render_result(self, result: ApprovalResult, request: ApprovalRequest | None = None) -> str:
|
||||
if result.decision == ApprovalDecision.APPROVED:
|
||||
icon = "✅"
|
||||
text = "审批已通过"
|
||||
elif result.decision == ApprovalDecision.DENIED:
|
||||
icon = "❌"
|
||||
text = "审批已拒绝"
|
||||
elif result.decision == ApprovalDecision.CANCELLED:
|
||||
icon = "🚫"
|
||||
text = "审批已取消"
|
||||
elif result.decision == ApprovalDecision.EXPIRED:
|
||||
icon = "⏰"
|
||||
text = "审批已过期"
|
||||
else:
|
||||
icon = "❓"
|
||||
text = "审批状态未知"
|
||||
|
||||
lines = [f"{icon} {self._bold(text)}"]
|
||||
lines.append(f"请求 ID: {self._code(result.request_id)}")
|
||||
if result.reason:
|
||||
lines.append(f"原因: {self._escape(result.reason)}")
|
||||
if result.decided_by:
|
||||
lines.append(f"决定人: {self._escape(result.decided_by)}")
|
||||
return "\n".join(lines)
|
||||
|
||||
def render_expiry_warning(self, request: ApprovalRequest, remaining_seconds: int) -> str:
|
||||
request_id = self._code(request.id)
|
||||
if remaining_seconds <= 0:
|
||||
return f"⏳ 审批请求 {request_id} 已过期。"
|
||||
if remaining_seconds < 60:
|
||||
return f"⏳ 审批请求 {request_id} 将在 {remaining_seconds} 秒后过期，请尽快处理。"
|
||||
minutes = remaining_seconds // 60
|
||||
return f"⏳ 审批请求 {request_id} 将在 {minutes} 分钟后过期，请尽快处理。"
|
||||
|
||||
def _bold(self, text: str) -> str:
|
||||
style = self._config.style
|
||||
if style == TemplateStyle.MARKDOWN:
|
||||
return f"**{text}**"
|
||||
elif style == TemplateStyle.HTML:
|
||||
return f"<b>{_html.escape(text)}</b>"
|
||||
return text
|
||||
|
||||
def _code(self, text: str) -> str:
|
||||
style = self._config.style
|
||||
if style == TemplateStyle.MARKDOWN:
|
||||
return f"`{text}`"
|
||||
elif style == TemplateStyle.HTML:
|
||||
return f"<code>{_html.escape(text)}</code>"
|
||||
return f"'{text}'"
|
||||
|
||||
def _escape(self, text: str) -> str:
|
||||
style = self._config.style
|
||||
if style == TemplateStyle.HTML:
|
||||
return _html.escape(text)
|
||||
elif style == TemplateStyle.MARKDOWN:
|
||||
return _MD_ESCAPE_RE.sub(r"\\\1", text)
|
||||
return text
|
||||
|
||||
|
||||
def _action_label(action: ApprovalAction) -> str:
|
||||
labels: dict[ApprovalAction, str] = {
|
||||
ApprovalAction.APPROVE_EXEC: "执行操作",
|
||||
ApprovalAction.APPROVE_READ: "读取操作",
|
||||
}
|
||||
return labels.get(action, action.value)
|
||||
|
||||
|
||||
_renderer_cache: dict[str, ApprovalTemplateRenderer] = {}
|
||||
|
||||
|
||||
def get_template_renderer(style: str = "default") -> ApprovalTemplateRenderer:
|
||||
if style not in _renderer_cache:
|
||||
if style == "default":
|
||||
_renderer_cache[style] = ApprovalTemplateRenderer()
|
||||
elif style == "compact":
|
||||
_renderer_cache[style] = ApprovalTemplateRenderer(
|
||||
ApprovalTemplateConfig(
|
||||
title="审批",
|
||||
show_request_id=False,
|
||||
show_action=False,
|
||||
show_initiator=False,
|
||||
show_expiry=False,
|
||||
)
|
||||
)
|
||||
elif style == "detailed":
|
||||
_renderer_cache[style] = ApprovalTemplateRenderer(
|
||||
ApprovalTemplateConfig(
|
||||
title="操作审批请求",
|
||||
style=TemplateStyle.MARKDOWN,
|
||||
timeout_minutes=10,
|
||||
)
|
||||
)
|
||||
else:
|
||||
logger.warning("未知模板风格 '%s'，使用默认配置", style)
|
||||
_renderer_cache[style] = ApprovalTemplateRenderer()
|
||||
return _renderer_cache[style]
|
||||
Loading…
Reference in New Issue
Block a user