"""Utility helpers for signal source repair service.""" from __future__ import annotations from .shared import Any, Optional, SignalSourceRegistry, os, re class SignalSourceRepairUtilsMixin: """Common utility helpers for repair workflow.""" @staticmethod def _safe_int(value: Any) -> Optional[int]: try: if value is None: return None return int(value) except (TypeError, ValueError): return None @staticmethod def _resolve_timeout_seconds() -> int: raw = ( os.getenv("SIGNAL_SOURCE_REPAIR_TIMEOUT_SECONDS") or os.getenv("LLM_REQUEST_TIMEOUT") or "60" ) try: timeout = int(float(raw)) except (TypeError, ValueError): timeout = 60 return max(10, min(timeout, 300)) @staticmethod def _build_repair_message(*, project_id: int, registry_id: int, source_name: str) -> str: return ( f"给项目ID: {project_id} 补录历史接入记录 registry#{registry_id}({source_name})," "请仅执行入库,不需要二次处理。" ) @staticmethod def _build_explicit_repair_message(*, project_id: int, row: SignalSourceRegistry) -> str: source_kind = str(row.source_kind or "").strip()[:40] source_name = str(row.source_name or "历史信号源").strip()[:80] source_identifier = str(row.source_identifier or "").strip()[:180] return ( f"项目 {int(project_id)} 新增信号源:{source_name}。" f"source_kind={source_kind}; source_identifier={source_identifier}。" "这是历史脏数据补录,请执行 signal_source.onboard 入库并生成消息台记录。" ) @staticmethod def _build_capture_message_id(*, project_id: int, registry_id: int, stage: str) -> str: phase = "explicit" if str(stage or "").strip().lower() == "explicit" else "base" return f"om_registry_repair_{int(project_id)}_{int(registry_id)}_{phase}"[:120] @staticmethod def _build_group_chat_id(row: SignalSourceRegistry) -> str: raw = (row.config or {}).get("chat_id") or row.source_identifier or f"repair_registry_{int(row.id)}" return str(raw)[:120] @staticmethod def _extract_route_status(capture_result: dict[str, Any]) -> str: rows = capture_result.get("results") if not isinstance(rows, list) or not rows: return "" first = rows[0] if isinstance(rows[0], dict) else {} return str(first.get("route_status") or "") @classmethod def _extract_signal_id(cls, capture_result: dict[str, Any]) -> Optional[int]: candidates: list[Any] = [capture_result.get("signal_id")] for key in ("signal_source_workflow", "signal_source_orchestration", "ai_execution"): payload = capture_result.get(key) if isinstance(payload, dict): candidates.append(payload.get("signal_id")) rows = capture_result.get("results") if isinstance(rows, list) and rows and isinstance(rows[0], dict): candidates.append(rows[0].get("signal_id")) for item in candidates: value = cls._safe_int(item) if value is not None: return value return None @staticmethod def _normalize_identifier_key(value: Any) -> str: key = re.sub(r"\s+", "", str(value or "").strip().lower()) if not key: return "" for prefix in ( "feishu:chat:", "wecom:chat:", "feishu_chat_", "wecom_chat_", "feishu:", "wecom:", ): if key.startswith(prefix): key = key[len(prefix) :] break return key[:180]