from __future__ import annotations import json from hashlib import sha1 from dataclasses import dataclass from threading import RLock from time import monotonic from typing import Any, Iterable from ..execution_builder import ExecutionViewBuilder from ..inspector_projection import ( project_activity_detail, project_event_detail, project_round_detail, ) from ..providers import LocalDatabaseProvider from ..sanitizer import sanitize from .log_context import locate_log_module from .repository import AuditRepository class ModuleNotFound(LookupError): pass @dataclass(frozen=True) class _RunSnapshot: bundle: dict[str, Any] view: dict[str, Any] modules: list[dict[str, Any]] expires_at: float class ModuleAuditService: def __init__( self, provider: LocalDatabaseProvider | None = None, builder: ExecutionViewBuilder | None = None, repository: AuditRepository | None = None, ): self.provider = provider or LocalDatabaseProvider() self.builder = builder or ExecutionViewBuilder() self.repository = repository or AuditRepository() self._snapshot_cache: dict[int, _RunSnapshot] = {} self._detail_cache: dict[tuple[int, str], tuple[float, dict[str, Any]]] = {} self._event_cache: dict[tuple[int, int], tuple[float, dict[str, Any] | None]] = {} self._log_cache: dict[int, tuple[float, str | None]] = {} self._snapshot_lock = RLock() def _snapshot(self, script_build_id: int) -> _RunSnapshot: now = monotonic() with self._snapshot_lock: cached = self._snapshot_cache.get(script_build_id) if cached is not None and cached.expires_at > now: return cached bundle = self.provider.load_bundle(script_build_id) view = self.builder.from_bundle(bundle) modules = _collect_modules(view) status = str((view.get("header") or {}).get("status") or "").lower() ttl = 300.0 if status in {"completed", "success", "failed", "interrupted", "cancelled"} else 3.0 snapshot = _RunSnapshot( bundle=bundle, view=view, modules=modules, expires_at=monotonic() + ttl, ) with self._snapshot_lock: self._snapshot_cache[script_build_id] = snapshot if len(self._snapshot_cache) > 12: stale_ids = sorted( self._snapshot_cache, key=lambda item: self._snapshot_cache[item].expires_at, )[:-12] for stale_id in stale_ids: self._snapshot_cache.pop(stale_id, None) return snapshot def index(self, script_build_id: int) -> dict[str, Any]: snapshot = self._snapshot(script_build_id) view = snapshot.view modules = snapshot.modules return sanitize( { "schemaVersion": "1", "run": { "id": script_build_id, "status": (view.get("header") or {}).get("status"), "capturedAt": view.get("capturedAt"), "dataShape": view.get("dataShape"), }, "columns": ["模块实际使用", "模块实际读取", "原始记录与上下文"], "modules": [ { "id": item["id"], "title": item["title"], "group": item["group"], "path": item["path"], "kind": item["kind"], "role": item.get("role"), "roundIndex": item.get("roundIndex"), "branchId": item.get("branchId"), "detailRef": item.get("detailRef"), "displayVariant": item.get("displayVariant", "expanded"), "cardCharacters": _characters(_card_use(item)), "completeness": "available", } for item in modules ], "summary": { "moduleCount": len(modules), "groupCount": len({item["group"] for item in modules}), "cardCharacters": sum(_characters(_card_use(item)) for item in modules), }, }, max_text=None, ) def detail(self, script_build_id: int, module_id: str) -> dict[str, Any]: snapshot = self._snapshot(script_build_id) cache_key = (script_build_id, module_id) now = monotonic() with self._snapshot_lock: cached_detail = self._detail_cache.get(cache_key) if cached_detail is not None and cached_detail[0] > now: return cached_detail[1] bundle = snapshot.bundle view = snapshot.view module = next((item for item in snapshot.modules if item["id"] == module_id), None) if module is None: raise ModuleNotFound(module_id) inspector, inspector_notice = self._inspector(script_build_id, module, view, bundle) card = _card_use(module) reads = self._reads(script_build_id, module, inspector, bundle) reads = _annotate_usage_areas(reads, inspector) use_by_consumer = { "card": _strings(card), "inspector": _strings(inspector), } use_snippets = sorted( set(use_by_consumer["card"] + use_by_consumer["inspector"]), key=len, reverse=True, ) contexts = self._contexts( script_build_id, reads, use_snippets, expires_at=snapshot.expires_at, ) reads, contexts = _annotate_lineage(module["id"], reads, contexts) completeness = _completeness(inspector_notice, reads, contexts) payload = sanitize( { "schemaVersion": "1", "id": module["id"], "title": module["title"], "group": module["group"], "path": module["path"], "actualUse": { "card": card, "inspector": inspector, "inspectorNotice": inspector_notice, "characters": { "card": _characters(card), "inspectorBusiness": _characters((inspector or {}).get("businessSections")), "inspectorDecision": _characters((inspector or {}).get("blocks")), "inspectorChanges": _characters((inspector or {}).get("changes")), "inspectorTechnical": _characters((inspector or {}).get("technical")), }, }, "actualReads": [ _public_read(item, use_by_consumer) for item in reads ], "contexts": contexts, "completeness": completeness, }, max_text=None, ) with self._snapshot_lock: self._detail_cache[cache_key] = (snapshot.expires_at, payload) if len(self._detail_cache) > 512: stale_keys = sorted( self._detail_cache, key=lambda item: self._detail_cache[item][0], )[:-512] for stale_key in stale_keys: self._detail_cache.pop(stale_key, None) return payload def _inspector( self, script_build_id: int, module: dict[str, Any], view: dict[str, Any], bundle: dict[str, Any], ) -> tuple[dict[str, Any] | None, str | None]: detail_ref = str(module.get("detailRef") or "") if not detail_ref: return None, "该模块没有 Inspector 入口。" if module["kind"] == "final-result" or detail_ref.startswith("artifact:"): return None, "该按钮进入脚本表查看器,不属于 Inspector。" try: if module["kind"] == "round-summary": return project_round_detail( script_build_id, module["payload"], bundle, ), None if detail_ref.startswith("event:"): event_id = int(detail_ref.rsplit(":", 1)[1]) event_detail = self.provider.load_event_detail(script_build_id, event_id) try: return project_activity_detail( script_build_id, detail_ref, view, bundle, event_detail=event_detail, ), None except KeyError: return project_event_detail( script_build_id, event_detail, run_status=(view.get("header") or {}).get("status"), ), None return project_activity_detail( script_build_id, detail_ref, view, bundle, ), None except (KeyError, ValueError, LookupError) as exc: return None, f"Inspector 记录不可用:{type(exc).__name__}" def _reads( self, script_build_id: int, module: dict[str, Any], inspector: dict[str, Any] | None, bundle: dict[str, Any], ) -> list[dict[str, Any]]: reads = _explicit_reads(module, bundle) technical = (inspector or {}).get("technical") if isinstance(technical, dict): for key, value in technical.items(): if key in {"scriptBuildId", "table"} or value in (None, "", [], {}): continue source = _source_from_technical(key, value) reads.append( { "consumer": "inspector", "label": _technical_label(key), "disposition": "selected", "relation": "exact-record" if isinstance(value, (dict, list)) else "projected", "source": source, "fieldPath": f"technical.{key}", "value": value, "characters": _characters(value), "context": value, } ) event_ids = _event_ids(module, reads) known = { int(item["source"]["eventId"]) for item in reads if isinstance(item.get("source"), dict) and item["source"].get("eventId") is not None } for event_id in event_ids: if event_id in known: continue event = self.repository.load_event(script_build_id, event_id) if event: reads.append(_event_read(event, "module", "关联运行事件", "calculation-input")) return _dedupe_reads(reads) def _contexts( self, script_build_id: int, reads: list[dict[str, Any]], use_snippets: list[str], *, expires_at: float, ) -> list[dict[str, Any]]: contexts: list[dict[str, Any]] = [] event_ids: set[int] = set() for read in reads: source = read.get("source") or {} event_id = _integer(source.get("eventId")) if event_id is not None: event_ids.add(event_id) if source.get("sourceKind") == "database": continue content = read.get("context") if content in (None, "", [], {}): continue function_result = source.get("sourceKind") == "function" contexts.append( { "id": _context_id(source, read.get("fieldPath")), "kind": "function-result" if function_result else "database-record", "title": ( f"函数处理结果 · {read.get('label') or '数据'}" if function_result else read.get("label") or "原始数据库记录" ), "source": source, "sourceKey": _source_key(source), "relatedSourceKey": ( f"database:event:{source.get('eventId')}" if function_result and source.get("eventId") is not None else None ), "content": content, "characters": _characters(content), "readPaths": ["$"] if function_result else [read.get("fieldPath")], "usedFragments": _matching_snippets(content, use_snippets), "match": read.get("relation"), "completeness": "complete", } ) log_content = self._load_log(script_build_id, expires_at) if event_ids else None for event_id in sorted(event_ids): event = self._load_event(script_build_id, event_id, expires_at) if not event: contexts.append( { "id": f"event:{event_id}:missing", "kind": "event", "title": f"Event {event_id}", "source": { "sourceKind": "database", "table": "script_build_event + script_build_event_body", "eventId": event_id, }, "sourceKey": f"database:event:{event_id}", "content": "事件记录不存在。", "characters": 8, "readPaths": [], "usedFragments": [], "match": "missing", "completeness": "missing", } ) continue contexts.append( { "id": f"event:{event_id}", "kind": "event-record", "title": f"script_build_event #{event_id}", "source": { "sourceKind": "database", "table": "script_build_event + script_build_event_body", "eventId": event_id, }, "sourceKey": f"database:event:{event_id}", "content": event, "characters": _characters(event), "readPaths": ["eventBody.input_content", "eventBody.output_content", "inputData", "outputPreview"], "usedFragments": _matching_snippets(event, use_snippets), "match": "exact-record", "completeness": "complete" if event.get("eventBody") else "body-missing", } ) log_module = locate_log_module(log_content, event) if log_module: contexts.append( { "id": f"event:{event_id}:log", "kind": "log-module", "title": f"锚定日志模块 · Event {event_id}", "source": { "sourceKind": "log", "table": "script_build_log", "eventId": event_id, "anchor": log_module["anchor"], }, "sourceKey": f"log:event:{event_id}", "relatedSourceKey": f"database:event:{event_id}", "content": log_module["content"], "characters": log_module["characters"], "readPaths": [], "usedFragments": _matching_snippets(log_module["content"], use_snippets), "match": "anchored", "completeness": "complete", } ) return _dedupe_contexts(contexts) def _load_event( self, script_build_id: int, event_id: int, expires_at: float, ) -> dict[str, Any] | None: key = (script_build_id, event_id) now = monotonic() with self._snapshot_lock: cached = self._event_cache.get(key) if cached is not None and cached[0] > now: return cached[1] event = self.repository.load_event(script_build_id, event_id) self._event_cache[key] = (expires_at, event) if len(self._event_cache) > 512: stale_keys = sorted( self._event_cache, key=lambda item: self._event_cache[item][0], )[:-512] for stale_key in stale_keys: self._event_cache.pop(stale_key, None) return event def _load_log(self, script_build_id: int, expires_at: float) -> str | None: now = monotonic() with self._snapshot_lock: cached = self._log_cache.get(script_build_id) if cached is not None and cached[0] > now: return cached[1] log_content = self.repository.load_log(script_build_id) self._log_cache[script_build_id] = (expires_at, log_content) if len(self._log_cache) > 12: stale_ids = sorted( self._log_cache, key=lambda item: self._log_cache[item][0], )[:-12] for stale_id in stale_ids: self._log_cache.pop(stale_id, None) return log_content def _collect_modules(view: dict[str, Any]) -> list[dict[str, Any]]: modules: list[dict[str, Any]] = [] def add( payload: Any, *, group: str, path: list[str], kind: str, round_index: int | None = None, branch_id: int | None = None, display_variant: str = "expanded", forced_id: str | None = None, fallback_id: str | None = None, fallback_title: str | None = None, fallback_detail_ref: str | None = None, ) -> None: if not isinstance(payload, dict): return module_id = str(forced_id or payload.get("id") or fallback_id or "").strip() if not module_id: return modules.append( { "id": module_id, "title": str(payload.get("title") or fallback_title or module_id), "group": group, "path": path, "kind": kind, "role": payload.get("role"), "roundIndex": round_index, "branchId": branch_id, "detailRef": payload.get("detailRef") or fallback_detail_ref, "displayVariant": display_variant, "payload": payload, } ) add(view.get("planning"), group="构建前", path=["构建前"], kind="story") add(view.get("objective"), group="构建前", path=["构建前"], kind="story") for round_ in view.get("rounds") or []: round_index = _integer(round_.get("roundIndex")) or 0 group = f"第 {round_index} 轮" add(round_.get("goal"), group=group, path=[group, "主 Agent"], kind="story", round_index=round_index) add((round_.get("plan") or {}).get("node"), group=group, path=[group, "主 Agent"], kind="story", round_index=round_index) for branch in round_.get("branches") or []: branch_id = _integer(branch.get("branchId")) or 0 branch_label = f"方案 {branch_id}" add(branch.get("task"), group=group, path=[group, branch_label], kind="story", round_index=round_index, branch_id=branch_id) stage = branch.get("retrievalStage") or {} add(stage, group=group, path=[group, branch_label, "取数"], kind="retrieval-stage", round_index=round_index, branch_id=branch_id, fallback_title="取数阶段") for direct in stage.get("directToolGroups") or []: add(direct, group=group, path=[group, branch_label, "取数", "工具取数"], kind="retrieval-direct", round_index=round_index, branch_id=branch_id) for call in direct.get("calls") or []: add(call, group=group, path=[group, branch_label, "取数", "工具调用"], kind="retrieval-call", round_index=round_index, branch_id=branch_id, fallback_title=call.get("businessLabel")) for agent in stage.get("agentRuns") or []: add(agent, group=group, path=[group, branch_label, "取数", "Agent 取数"], kind="retrieval-agent", round_index=round_index, branch_id=branch_id, fallback_title=agent.get("businessLabel")) for attempt in agent.get("attempts") or []: add(attempt, group=group, path=[group, branch_label, "取数", "单次查询"], kind="query-attempt", round_index=round_index, branch_id=branch_id, fallback_title=attempt.get("queryLabel")) for decision in branch.get("dataDecisions") or []: add(decision.get("node") or decision, group=group, path=[group, branch_label], kind="story", round_index=round_index, branch_id=branch_id) add((branch.get("output") or {}).get("node"), group=group, path=[group, branch_label], kind="story", round_index=round_index, branch_id=branch_id) for batch in round_.get("convergenceBatches") or []: add(batch.get("review") or batch.get("missingReview"), group=group, path=[group, "收敛"], kind="review", round_index=round_index) add((batch.get("decision") or {}).get("node") or batch.get("decision") or batch.get("missingDecision"), group=group, path=[group, "收敛"], kind="decision", round_index=round_index) add(round_.get("overallEvaluation"), group=group, path=[group, "整体评审"], kind="story", round_index=round_index) add(round_.get("result"), group=group, path=[group, "本轮产出"], kind="story", round_index=round_index) add(view.get("finalResult"), group="最终结果", path=["最终结果"], kind="final-result") return _unique_modules(modules) def _card_use(module: dict[str, Any]) -> dict[str, Any]: payload = module["payload"] card = payload.get("card") if not isinstance(card, dict): card = ((payload.get("decisionProjection") or {}).get("card") if isinstance(payload.get("decisionProjection"), dict) else None) if isinstance(card, dict): fields = [] primary = card.get("primary") if isinstance(primary, dict): fields.append({"label": primary.get("label"), "value": primary.get("value"), "kind": "primary"}) for item in card.get("secondary") or []: if isinstance(item, dict): fields.append({"label": item.get("label"), "value": item.get("value"), "kind": "secondary"}) return {"title": module["title"], "fields": fields} kind = module["kind"] if kind == "round-summary": summary = payload.get("summary") or {} return { "title": f"第 {module.get('roundIndex')} 轮摘要", "fields": [ {"label": "本轮目标", "value": (summary.get("goal") or {}).get("headline")}, {"label": "尝试方案", "value": summary.get("approach")}, {"label": "主 Agent 决策", "value": summary.get("decision")}, {"label": "本轮产出", "value": summary.get("output")}, ], } if kind == "branch-summary": summary = payload.get("summary") or {} return {"title": module["title"], "fields": [{"label": "任务", "value": summary.get("task")}, {"label": "产出", "value": summary.get("output")}, {"label": "状态", "value": payload.get("status")}]} if kind == "retrieval-stage": return {"title": "取数阶段", "fields": [{"label": "模式", "value": payload.get("observedMode")}, {"label": "状态", "value": payload.get("status")}, {"label": "取数模块", "value": f"{len(payload.get('directToolGroups') or [])} 个工具组、{len(payload.get('agentRuns') or [])} 个 Agent"}]} if kind == "retrieval-direct": sources = "、".join(str(item.get("businessLabel") or "") for item in payload.get("sources") or [] if isinstance(item, dict)) result = f"{payload.get('failureCount')} 次失败" if payload.get("failureCount") else f"{payload.get('interruptedCount')} 次中断" if payload.get("interruptedCount") else "进行中" if payload.get("runningCount") else "已完成" return {"title": sources or "工具取数", "fields": [{"label": "读取", "value": f"{payload.get('callCount') or 0} 次"}, {"label": "结果", "value": result}]} if kind == "retrieval-agent": query = payload.get("querySummary") or {} return {"title": payload.get("businessLabel") or "Agent 取数", "fields": [{"label": "取数目标", "value": payload.get("taskPreview")}, {"label": "查询过程", "value": query}, {"label": "初筛整理", "value": (payload.get("screening") or {}).get("preview")}]} if kind in {"query-attempt", "retrieval-call"}: return {"title": module["title"], "fields": [{"label": "状态", "value": payload.get("status")}, {"label": "结果数量", "value": payload.get("resultCount")}, {"label": "结果摘要", "value": payload.get("resultSummary")}]} if kind == "final-result": return {"title": module["title"], "fields": [{"label": "执行结果", "value": payload.get("summary")}, {"label": "状态", "value": payload.get("status")}, {"label": "轮次", "value": payload.get("roundCount")}, {"label": "方案", "value": payload.get("branchCounts")}]} return {"title": module["title"], "fields": [{"label": "模块数据", "value": payload}]} def _explicit_reads(module: dict[str, Any], bundle: dict[str, Any]) -> list[dict[str, Any]]: reads: list[dict[str, Any]] = [] role = str(module.get("role") or "") round_index = module.get("roundIndex") branch_id = module.get("branchId") raw_round = _find(bundle.get("rounds"), "round_index", round_index) raw_branch = _find(bundle.get("branches"), "branch_id", branch_id) if role == "planning-analysis": _append_related_events(reads, module, bundle, "card", "规划事件") elif role == "objective": record = bundle.get("record") or {} reads.append(_record_read("card", "创作目标", "script_build_record", record, "script_direction")) elif role == "goal" and raw_round: reads.append(_record_read("card", "本轮目标", "script_build_round", raw_round, "goal")) elif role == "plan" and raw_round: for field, label in (("multipath_plan", "实现路线"), ("race_or_divide", "组织方式"), ("plan_note", "规划理由")): if raw_round.get(field) not in (None, "", [], {}): reads.append(_record_read("card", label, "script_build_round", raw_round, field)) elif role == "branch-task" and raw_branch: reads.append(_record_read("card", "实现任务", "script_build_branch", raw_branch, "impl_task")) task_event = next( ( event for event in reversed(bundle.get("events") or []) if event.get("event_name") == "script_implementer" and _integer(event.get("round_index")) == round_index and _integer(event.get("branch_id")) == branch_id ), None, ) if task_event: event_value = (task_event.get("inputData") or {}).get("task") if isinstance(task_event.get("inputData"), dict) else None reads.append(_event_summary_read(task_event, "inspector", "实现 Agent 派发任务", "inputData.task", event_value, "selected" if event_value else "missing")) reads.append({**_record_read("inspector", "历史回退任务", "script_build_branch", raw_branch, "impl_task"), "disposition": "fallback" if event_value else "selected"}) else: reads.append({**_record_read("inspector", "实现任务", "script_build_branch", raw_branch, "impl_task"), "disposition": "selected"}) elif role == "data-decision": row = _find(bundle.get("dataDecisions"), "id", _suffix_int(module["id"])) if row: reads.append(_whole_record_read("card", "取数取舍记录", "script_build_data_decision", row)) elif role in {"domain-output", "candidate-output"} and raw_branch: reads.append(_whole_record_read("card", "实现方案记录", "script_build_branch", raw_branch)) if role == "domain-output": for row in bundle.get("domainInfo") or []: if _integer(row.get("branch_id")) == branch_id: reads.append(_whole_record_read("card", "领域事实", "script_build_domain_info", row)) elif role == "multipath-decision": row = _find(bundle.get("multipathDecisions"), "id", _suffix_int(module["id"])) if row: reads.append(_whole_record_read("card", "多方案决定", "script_build_multipath_decision", row)) elif role in {"round-evaluation", "round-result"} and raw_round: reads.append(_whole_record_read("card", "轮次记录", "script_build_round", raw_round)) _append_related_events(reads, module, bundle, "card", "轮次运行事件") kind = module["kind"] payload = module["payload"] if kind == "round-summary" and raw_round: reads.append(_whole_record_read("card", "折叠态轮次记录", "script_build_round", raw_round)) elif kind == "branch-summary" and raw_branch: reads.append(_whole_record_read("card", "折叠态方案记录", "script_build_branch", raw_branch)) elif kind.startswith("retrieval") or kind == "query-attempt": event_ids = _event_ids_from_payload(payload) for event_id in event_ids: event = next((item for item in bundle.get("events") or [] if _integer(item.get("id")) == event_id), None) if event: reads.append(_event_summary_read(event, "card", "取数运行事件", "event", event, "calculation-input")) elif kind == "review": _append_related_events(reads, module, bundle, "card", "评审事件") elif kind == "final-result": reads.append(_whole_record_read("card", "构建记录", "script_build_record", bundle.get("record") or {})) reads.append(_whole_record_read("card", "当前主脚本快照", "script_build_artifact", bundle.get("currentArtifact") or {})) if not reads: reads.append( { "consumer": "module", "label": "来源未确认", "disposition": "unconfirmed", "relation": "unconfirmed", "source": {"type": "unknown", "sourceKind": "unknown"}, "fieldPath": None, "value": "当前审计规则无法可靠确认这个模块的字段来源。", "characters": 24, "context": None, } ) return reads def _record_read(consumer: str, label: str, table: str, row: dict[str, Any], field: str) -> dict[str, Any]: value = row.get(field) return { "consumer": consumer, "label": label, "disposition": "selected", "relation": ( "projected" if consumer == "card" else "exact" if isinstance(value, str) else "structured" ), "source": { "type": "business-record", "sourceKind": "database", "table": table, "recordId": row.get("id"), }, "fieldPath": field, "value": value, "characters": _characters(value), "context": row, } def _whole_record_read(consumer: str, label: str, table: str, row: dict[str, Any]) -> dict[str, Any]: return { "consumer": consumer, "label": label, "disposition": "selected", "relation": ( "combined" if table == "script_build_artifact" else "projected" if consumer == "card" else "exact-record" ), "source": { "type": "business-record", "sourceKind": "database", "table": table, "recordId": row.get("id"), }, "fieldPath": "$", "value": row, "characters": _characters(row), "context": row, } def _event_summary_read(event: dict[str, Any], consumer: str, label: str, field_path: str, value: Any, disposition: str) -> dict[str, Any]: return { "consumer": consumer, "label": label, "disposition": disposition, "relation": ( "calculation-input" if disposition == "calculation-input" else "exact" if isinstance(value, str) else "structured" ), "source": { "type": "runtime-event", "sourceKind": "database", "table": "script_build_event + script_build_event_body", "eventId": event.get("id"), "eventName": event.get("event_name"), }, "fieldPath": field_path, "value": value, "characters": _characters(value), "context": None, } def _event_read(event: dict[str, Any], consumer: str, label: str, disposition: str) -> dict[str, Any]: return { "consumer": consumer, "label": label, "disposition": disposition, "relation": "exact-record", "source": { "type": "runtime-event", "sourceKind": "database", "table": "script_build_event + script_build_event_body", "eventId": event.get("id"), "eventName": event.get("event_name"), }, "fieldPath": "$", "value": event, "characters": _characters(event), "context": None, } def _append_related_events(reads: list[dict[str, Any]], module: dict[str, Any], bundle: dict[str, Any], consumer: str, label: str) -> None: for event_id in _event_ids_from_payload(module.get("payload") or {}): event = next((item for item in bundle.get("events") or [] if _integer(item.get("id")) == event_id), None) if event: reads.append(_event_summary_read(event, consumer, label, "event", event, "calculation-input")) def _source_from_technical(key: str, value: Any) -> dict[str, Any]: event_id = None table = None if isinstance(value, dict): event_id = _integer(value.get("eventId") or value.get("id")) if "event" in key.lower() else None table = value.get("table") return { "type": "inspector-projection", "sourceKind": "function", "functionName": "Inspector 详情组装", "table": table, "eventId": event_id, "technicalGroup": key, } def _event_ids(module: dict[str, Any], reads: list[dict[str, Any]]) -> list[int]: values = _event_ids_from_payload(module.get("payload") or {}) values.extend( _integer((item.get("source") or {}).get("eventId")) for item in reads ) return sorted({value for value in values if value is not None}) def _event_ids_from_payload(value: Any) -> list[int]: result: list[int] = [] if isinstance(value, dict): for key, nested in value.items(): if key in {"eventId", "implementerEventId", "eventRef"}: parsed = _integer(nested) if parsed is None: parsed = _suffix_int(nested) if parsed is not None: result.append(parsed) elif key == "detailRef" and str(nested or "").startswith("event:"): parsed = _suffix_int(nested) if parsed is not None: result.append(parsed) elif key in {"technicalRefs", "runtimeRevisionRefs"} and isinstance(nested, list): result.extend(_suffix_int(item) for item in nested if _suffix_int(item) is not None) elif key not in {"card", "summary", "body", "content_snapshot", "candidate_snapshot"}: result.extend(_event_ids_from_payload(nested)) elif isinstance(value, list): for item in value: result.extend(_event_ids_from_payload(item)) return sorted({item for item in result if item is not None}) def _public_read( value: dict[str, Any], use_by_consumer: dict[str, list[str]] ) -> dict[str, Any]: public = {key: item for key, item in value.items() if key != "context"} source = value.get("source") or {} public["sourceKey"] = _source_key(source) disposition = str(value.get("disposition") or "") relation = str(value.get("relation") or "") public["sourcePriority"] = { "selected": "已采用来源", "fallback": "备用来源", "calculation-input": "计算输入", "unconfirmed": "来源未确认", "missing": "来源缺失", }.get(disposition, "已读取来源") public["transformation"] = { "exact": "直接摘取", "exact-record": "完整记录读取", "structured": "结构化字段读取", "projected": "展示处理", "combined": "多来源重组", "calculation-input": "参与计算", "unconfirmed": "无法确认", }.get(relation, relation or "未标记") ratio = _usage_ratio( value.get("value"), use_by_consumer.get(str(value.get("consumer") or ""), []), relation, disposition, ) public["usageRatio"] = ( { "calculable": True, "value": ratio, "label": f"{ratio * 100:.1f}%", } if ratio is not None else { "calculable": False, "value": None, "label": "不可直接计算", } ) return public def _annotate_lineage( module_id: str, reads: list[dict[str, Any]], contexts: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Attach field-level pointers without inventing lineage for derived values.""" contexts_by_source: dict[str, list[dict[str, Any]]] = {} for context in contexts: contexts_by_source.setdefault(str(context.get("sourceKey") or ""), []).append(context) for index, read in enumerate(reads): source_key = _source_key(read.get("source") or {}) identity = "|".join( ( module_id, str(read.get("consumer") or "module"), source_key, str(read.get("fieldPath") or "$"), str(read.get("label") or "data"), str(index), ) ) read_id = f"read:{sha1(identity.encode('utf-8')).hexdigest()[:14]}" read["readId"] = read_id refs: list[dict[str, Any]] = [] for context in contexts_by_source.get(source_key, []): context_path, relation = _lineage_path(read, context) refs.append( { "contextId": context.get("id"), "sourceKey": source_key, "contextFieldPath": context_path, "relation": relation, "exact": relation in {"direct", "normalized-alias", "whole-record"}, } ) read["contextRefs"] = refs read_ids_by_context: dict[str, list[str]] = {} paths_by_context: dict[str, list[str]] = {} for read in reads: for ref in read.get("contextRefs") or []: context_id = str(ref.get("contextId") or "") if not context_id: continue read_ids_by_context.setdefault(context_id, []).append(str(read["readId"])) field_path = str(ref.get("contextFieldPath") or "$") paths_by_context.setdefault(context_id, []).append(field_path) for context in contexts: context_id = str(context.get("id") or "") context["readIds"] = list(dict.fromkeys(read_ids_by_context.get(context_id, []))) mapped_paths = list(dict.fromkeys(paths_by_context.get(context_id, []))) if mapped_paths: context["readPaths"] = mapped_paths return reads, contexts def _lineage_path( read: dict[str, Any], context: dict[str, Any] ) -> tuple[str, str]: field_path = str(read.get("fieldPath") or "$") content = context.get("content") source_kind = str((read.get("source") or {}).get("sourceKind") or "") if source_kind == "function": return "$", "transformed" if field_path in {"$", "event"}: return "$", "whole-record" candidates = [field_path] if field_path.startswith("inputData."): suffix = field_path.removeprefix("inputData.") candidates.append(f"eventBody.input_content.{suffix}") elif field_path == "inputData": candidates.append("eventBody.input_content") if field_path.startswith("agentOutputData."): suffix = field_path.removeprefix("agentOutputData.") candidates.append(f"eventBody.output_content.{suffix}") elif field_path in {"agentOutputData", "outputPreview", "output_preview"}: candidates.append("eventBody.output_content") for candidate in candidates: if _value_at_path(content, candidate) is not _MISSING: relation = "direct" if candidate == field_path else "normalized-alias" return candidate, relation return "$", "record-only" _MISSING = object() def _value_at_path(value: Any, field_path: str) -> Any: if not field_path or field_path == "$": return value current = value for part in field_path.removeprefix("$.").split("."): if isinstance(current, str): text = current.strip() if text[:1] in {"{", "["}: try: current = json.loads(text) except (TypeError, ValueError, json.JSONDecodeError): return _MISSING if not isinstance(current, dict) or part not in current: return _MISSING current = current[part] return current def _annotate_usage_areas( reads: list[dict[str, Any]], inspector: dict[str, Any] | None ) -> list[dict[str, Any]]: has_business = bool( (inspector or {}).get("businessSections") or (inspector or {}).get("blocks") ) has_changes = bool((inspector or {}).get("changes")) has_technical = isinstance((inspector or {}).get("technical"), dict) for read in reads: consumer = read.get("consumer") field_path = str(read.get("fieldPath") or "") disposition = read.get("disposition") areas: list[str] = [] if disposition == "fallback": read["usageAreas"] = areas continue if consumer == "card": areas.append("卡片") if has_business: areas.append("Inspector 业务详情") if has_changes: areas.append("Inspector 产出与改动") elif consumer == "inspector": areas.append( "Inspector 技术详情" if field_path.startswith("technical.") else "Inspector 业务详情" ) elif consumer == "module": areas.append("模块处理") read["usageAreas"] = areas if has_technical: technical_refs = { _underlying_record_key(read.get("source") or {}) for read in reads if str(read.get("fieldPath") or "").startswith("technical.") } technical_refs.discard(None) for read in reads: source = read.get("source") or {} if ( source.get("sourceKind") == "database" and _underlying_record_key(source) in technical_refs and read.get("disposition") != "fallback" and "Inspector 技术详情" not in read["usageAreas"] ): read["usageAreas"].append("Inspector 技术详情") return reads def _underlying_record_key(source: dict[str, Any]) -> str | None: event_id = _integer(source.get("eventId")) if event_id is not None: return f"event:{event_id}" table = source.get("table") record_id = source.get("recordId") if table and record_id is not None: return f"record:{table}:{record_id}" return None def _source_key(source: dict[str, Any]) -> str: kind = str(source.get("sourceKind") or "unknown") underlying = _underlying_record_key(source) if kind == "database" and underlying: return f"database:{underlying}" if kind == "log": return f"log:event:{source.get('eventId') or 'unknown'}" if kind == "function": group = source.get("technicalGroup") or source.get("functionName") or "detail" event = source.get("eventId") return f"function:{group}:{event or 'module'}" return f"unknown:{source.get('type') or 'source'}" def _usage_ratio( value: Any, snippets: list[str], relation: str, disposition: str ) -> float | None: if ( disposition != "selected" or relation != "exact" or not isinstance(value, str) or not value ): return None normalized_source = value.strip() if not normalized_source: return None candidates = [] for snippet in snippets: normalized = snippet.strip().removesuffix("…").rstrip() if len(normalized) >= 8 and normalized in normalized_source: candidates.append(normalized) if not candidates: return None return min(1.0, len(max(candidates, key=len)) / len(normalized_source)) def _dedupe_reads(reads: list[dict[str, Any]]) -> list[dict[str, Any]]: seen: set[tuple[str, str, str, str]] = set() result = [] for item in reads: source = item.get("source") or {} key = (str(item.get("consumer")), str(source.get("table")), str(source.get("eventId") or source.get("recordId")), str(item.get("fieldPath"))) if key in seen: continue seen.add(key) result.append(item) return result def _dedupe_contexts(contexts: list[dict[str, Any]]) -> list[dict[str, Any]]: result = [] seen = set() for item in contexts: key = item.get("id") if key in seen: continue seen.add(key) result.append(item) return result def _unique_modules(modules: list[dict[str, Any]]) -> list[dict[str, Any]]: seen = set() result = [] for item in modules: if item["id"] in seen: continue seen.add(item["id"]) result.append(item) return result def _find(values: Any, key: str, expected: Any) -> dict[str, Any] | None: if expected is None: return None return next((item for item in values or [] if _integer(item.get(key)) == _integer(expected)), None) def _characters(value: Any) -> int: if value is None: return 0 if isinstance(value, str): return len(value) try: return len(json.dumps(value, ensure_ascii=False, sort_keys=True, default=str, separators=(",", ":"))) except (TypeError, ValueError): return len(str(value)) def _strings(value: Any) -> list[str]: result: list[str] = [] if isinstance(value, str): if len(value.strip()) >= 8: result.append(value.strip()) elif isinstance(value, dict): for nested in value.values(): result.extend(_strings(nested)) elif isinstance(value, list): for nested in value: result.extend(_strings(nested)) return sorted(set(result), key=len, reverse=True) def _matching_snippets(value: Any, snippets: list[str]) -> list[str]: text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str) return [snippet for snippet in snippets if snippet in text][:20] def _technical_label(key: str) -> str: return { "source": "数据来源与定位", "association": "关联方式与完整性", "input": "原始输入", "output": "原始输出", "cost": "耗时与成本", "record": "原始业务记录", "event": "原始运行事件", "taskEvent": "实现任务事件", "taskSource": "任务来源判定", "rawEvent": "运行事件摘要", "branch": "实现方案记录", "round": "轮次记录", }.get(key, key) def _context_id(source: dict[str, Any], field_path: Any) -> str: return ":".join(str(item) for item in (source.get("table") or source.get("type"), source.get("recordId") or source.get("eventId") or "record", field_path or "$") ) def _completeness(notice: str | None, reads: list[dict[str, Any]], contexts: list[dict[str, Any]]) -> dict[str, Any]: missing = [item for item in contexts if item.get("completeness") in {"missing", "body-missing"}] unconfirmed = [item for item in reads if item.get("disposition") == "unconfirmed"] return { "state": "partial" if notice or missing or unconfirmed else "complete", "inspectorAvailable": notice is None, "sourceCount": len(reads), "contextCount": len(contexts), "missingContextCount": len(missing), "unconfirmedSourceCount": len(unconfirmed), "notice": notice, "redacted": True, } def _suffix_int(value: Any) -> int | None: text = str(value or "") try: return int(text.rsplit(":", 1)[-1]) except (TypeError, ValueError): return None def _integer(value: Any) -> int | None: try: return int(value) if value is not None else None except (TypeError, ValueError): return None