# -*- coding: utf-8 -*- """Build sanitized run-flow snapshots from tasks and persisted diagnostics.""" from __future__ import annotations import json from collections import defaultdict from collections.abc import Mapping from datetime import datetime, timedelta, timezone from typing import Any, Dict, Iterable, List, Optional, Tuple from api.v1.schemas.run_flow import RunFlowSnapshot from src.analysis_context_pack_overview import extract_analysis_context_pack_overview from src.services.run_diagnostics import ( safe_diagnostic_key, sanitize_diagnostic_metadata, sanitize_diagnostic_text, ) from src.utils.data_processing import normalize_model_used, parse_json_field _LANES = [ {"id": "entry", "label": "入口", "order": 1}, {"id": "data_source", "label": "数据来源", "order": 2}, {"id": "analysis", "label": "分析引擎", "order": 3}, {"id": "artifact", "label": "产物", "order": 4}, ] _RUN_STATUS_MAP = { "pending": "pending", "processing": "running", "running": "running", "completed": "success", "success": "success", "failed": "failed", "cancel_requested": "cancel_requested", "cancelled": "cancelled", } _DATA_TYPE_LABELS = { "realtime_quote": "实时行情", "daily_data": "日线K线", "daily_bars": "日线K线", "technical": "技术指标", "news": "新闻舆情", "news_search": "新闻舆情", "fundamental": "基本面", "fundamentals": "基本面", "belong_boards": "所属板块", "chip": "筹码结构", } _DATA_TYPE_TO_BLOCK_KEY = { "realtime_quote": "quote", "daily_data": "daily_bars", "daily_bars": "daily_bars", "technical": "technical", "news": "news", "news_search": "news", "fundamental": "fundamentals", "fundamentals": "fundamentals", "belong_boards": "fundamentals", "chip": "chip", } _CONTEXT_STATUS_TO_FLOW = { "available": "success", "fallback": "fallback", "partial": "degraded", "stale": "degraded", "estimated": "degraded", "missing": "skipped", "not_supported": "skipped", "fetch_failed": "failed", } def build_task_run_flow_snapshot( task: Any, *, generated_at: Optional[datetime] = None, ) -> RunFlowSnapshot: """Build a skeleton run-flow snapshot from an in-memory task.""" status_value = _task_status_value(task) flow_status = _map_task_status(status_value) task_id = _safe_text(getattr(task, "task_id", None), max_length=96) or "unknown" trace_id = _safe_text(getattr(task, "trace_id", None), max_length=96) or task_id stock_code = _safe_text(getattr(task, "stock_code", None), max_length=32) or "unknown" stock_name = _safe_text(getattr(task, "stock_name", None), max_length=80) created_at = _datetime_to_iso(getattr(task, "created_at", None)) started_at = _datetime_to_iso(getattr(task, "started_at", None)) completed_at = _datetime_to_iso(getattr(task, "completed_at", None)) now_iso = _datetime_to_iso(generated_at or datetime.now()) or datetime.now().isoformat() nodes: Dict[str, Dict[str, Any]] = {} edges: List[Dict[str, Any]] = [] events: List[Dict[str, Any]] = [] _put_node( nodes, "request", lane="entry", kind="entry", label="用户请求", status="success" if created_at else "unknown", started_at=created_at, ended_at=created_at, message=_safe_text(getattr(task, "original_query", None), max_length=120) or "任务请求已创建", metadata={ "selection_source": getattr(task, "selection_source", None), "query_source": getattr(task, "query_source", None), "report_type": getattr(task, "report_type", None), "analysis_phase": getattr(task, "analysis_phase", None), }, ) _put_node( nodes, "task_queue", lane="entry", kind="queue", label="任务队列", status=flow_status, started_at=created_at, ended_at=completed_at, duration_ms=_elapsed_ms(getattr(task, "created_at", None), getattr(task, "completed_at", None)), message=getattr(task, "message", None) or _task_status_message(flow_status), metadata={ "progress": getattr(task, "progress", None), "error": getattr(task, "error", None), }, ) _append_edge(edges, "request", "task_queue", "control", flow_status, label="提交") if flow_status in {"pending", "running", "cancel_requested"}: _put_node( nodes, "analysis_pipeline", lane="analysis", kind="analysis", label="分析流程", status="running" if flow_status == "running" else flow_status, started_at=started_at, message=getattr(task, "message", None) or _task_status_message(flow_status), ) _append_edge(edges, "task_queue", "analysis_pipeline", "control", flow_status, label="调度") else: _put_skeleton_tail(nodes, edges, anchor_node_id="task_queue", status=flow_status) _append_task_events(events, task, flow_status) _append_active_flow_events( nodes, edges, events, _as_list(getattr(task, "flow_events", None)), flow_status=flow_status, ) _prune_active_skeleton_tail(nodes, edges) summary = _build_summary( nodes, edges, events, elapsed_ms=_elapsed_ms(getattr(task, "created_at", None), getattr(task, "completed_at", None)), ) return RunFlowSnapshot.model_validate( { "task_id": task_id, "trace_id": trace_id, "stock_code": stock_code, "stock_name": stock_name, "status": flow_status, "summary": summary, "lanes": _LANES, "nodes": list(nodes.values()), "edges": edges, "events": events, "generated_at": now_iso, } ) def build_history_run_flow_snapshot( record: Any, *, context_snapshot: Optional[Any] = None, raw_result: Optional[Any] = None, generated_at: Optional[datetime] = None, ) -> RunFlowSnapshot: """Build a run-flow snapshot from a persisted history record.""" snapshot = _as_mapping(context_snapshot if context_snapshot is not None else getattr(record, "context_snapshot", None)) raw = _as_mapping(raw_result if raw_result is not None else getattr(record, "raw_result", None)) diagnostics = _as_mapping(snapshot.get("diagnostics")) if snapshot else {} diagnostics = _normalize_history_diagnostics_for_record(record, snapshot, diagnostics) overview = extract_analysis_context_pack_overview(snapshot) if snapshot else None overview_metadata = overview.get("metadata") if isinstance((overview or {}).get("metadata"), Mapping) else {} query_id = _safe_text(getattr(record, "query_id", None), max_length=96) or diagnostics.get("query_id") or "unknown" task_id = _safe_text(diagnostics.get("task_id"), max_length=96) or query_id trace_id = ( _safe_text(diagnostics.get("trace_id"), max_length=96) or _safe_text(snapshot.get("trace_id") if snapshot else None, max_length=96) or _safe_text(raw.get("trace_id") if raw else None, max_length=96) or task_id ) stock_code = ( _safe_text(getattr(record, "code", None), max_length=32) or _safe_text(diagnostics.get("stock_code"), max_length=32) or _safe_text(raw.get("stock_code") or raw.get("code"), max_length=32) or "unknown" ) stock_name = _safe_text(getattr(record, "name", None), max_length=80) or _safe_text(raw.get("name"), max_length=80) created_at = _datetime_to_iso(getattr(record, "created_at", None)) now_iso = _datetime_to_iso(generated_at or datetime.now()) or datetime.now().isoformat() nodes: Dict[str, Dict[str, Any]] = {} edges: List[Dict[str, Any]] = [] events: List[Dict[str, Any]] = [] _put_node( nodes, "request", lane="entry", kind="entry", label="用户请求", status="success", started_at=created_at, ended_at=created_at, message="历史分析记录", metadata={ "query_id": query_id, "trigger_source": diagnostics.get("trigger_source") or overview_metadata.get("trigger_source"), "report_type": getattr(record, "report_type", None), }, ) _put_node( nodes, "task_queue", lane="entry", kind="queue", label="任务队列", status="success", started_at=created_at, ended_at=created_at, message="任务已完成并进入历史记录", ) _append_edge(edges, "request", "task_queue", "control", "success", label="提交") _append_event( events, "task_completed", node_id="task_queue", timestamp=created_at, severity="success", title="任务完成", message="历史记录已生成", ) provider_success_by_block = _append_provider_runs( nodes, edges, events, _as_list(diagnostics.get("provider_runs")), ) context_status = _context_pack_status(overview) _put_node( nodes, "context_pack", lane="analysis", kind="analysis", label="ContextPack", status=context_status, started_at=(overview or {}).get("created_at"), message=_context_pack_message(overview), metadata={ "pack_version": (overview or {}).get("pack_version"), "counts": (overview or {}).get("counts"), "warnings": (overview or {}).get("warnings"), "data_quality": (overview or {}).get("data_quality"), }, ) _append_context_blocks(nodes, edges, events, overview, provider_success_by_block) if not any(edge["to"] == "context_pack" for edge in edges): _append_edge(edges, "task_queue", "context_pack", "data", context_status, label="输入") last_analysis_node = _append_llm_runs( nodes, edges, events, _as_list(diagnostics.get("llm_runs")), raw, ) if last_analysis_node is None: _put_node( nodes, "llm", lane="analysis", kind="model", label="LLM 生成", status="unknown", provider=normalize_model_used(raw.get("model_used")) if raw else None, message="LLM 未记录诊断信息", ) _append_edge(edges, "context_pack", "llm", "data", "unknown", label="生成") last_analysis_node = "llm" last_artifact_node = _append_history_runs( nodes, edges, events, _as_list(diagnostics.get("history_runs")), anchor_node_id=last_analysis_node, fallback_created_at=created_at, ) if last_artifact_node is None: _put_node( nodes, "history_save", lane="artifact", kind="artifact", label="保存报告", status="success", ended_at=created_at, message="历史记录已存在", metadata={"analysis_history_id": getattr(record, "id", None)}, ) _append_edge(edges, last_analysis_node, "history_save", "data", "success", label="保存") last_artifact_node = "history_save" notification_count = _append_notification_runs( nodes, edges, events, _as_list(diagnostics.get("notification_runs")), anchor_node_id=last_artifact_node, ) if notification_count == 0: _put_node( nodes, "notification", lane="artifact", kind="notification", label="推送通知", status="unknown", message="通知结果未记录", ) _append_edge(edges, last_artifact_node, "notification", "control", "unknown", label="通知") summary = _build_summary(nodes, edges, events) status = _history_snapshot_status(nodes, diagnostics, overview) if summary.get("model") is None: summary["model"] = _safe_text(normalize_model_used(raw.get("model_used")) if raw else None, max_length=120) return RunFlowSnapshot.model_validate( { "task_id": task_id, "trace_id": trace_id, "stock_code": stock_code, "stock_name": stock_name, "status": status, "summary": summary, "lanes": _LANES, "nodes": list(nodes.values()), "edges": edges, "events": events, "generated_at": now_iso, } ) def _append_provider_runs( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], events: List[Dict[str, Any]], provider_runs: List[Any], ) -> Dict[str, str]: success_by_data_type = { data_type: any(_as_mapping(run).get("success") is True for run in runs) for data_type, runs in _group_provider_runs(provider_runs).items() } previous_node_by_type: Dict[str, Tuple[str, Dict[str, Any]]] = {} provider_success_by_block: Dict[str, str] = {} attempt_index_by_type: Dict[str, int] = defaultdict(int) for index, raw_run in enumerate(provider_runs, start=1): run = _as_mapping(raw_run) if not run: continue data_type = _safe_key(run.get("data_type") or "provider") attempt_index_by_type[data_type] += 1 attempt_index = attempt_index_by_type[data_type] provider = _safe_text(run.get("provider"), max_length=80) or "unknown" label = _DATA_TYPE_LABELS.get(data_type, data_type) node_id = f"provider_{_safe_key(data_type)}_{_safe_key(provider)}_{attempt_index}" success = run.get("success") is True had_previous_failure = data_type in previous_node_by_type and previous_node_by_type[data_type][1].get("success") is False status = _provider_run_status(run, had_previous_failure=had_previous_failure) duration_ms = _safe_int(run.get("latency_ms")) timestamp = _datetime_to_iso(run.get("created_at")) started_at = _started_at_from_end_and_duration(timestamp, duration_ms) message = _provider_run_message(label, provider, run, success=success) block_key = _DATA_TYPE_TO_BLOCK_KEY.get(data_type, data_type) _put_node( nodes, node_id, lane="data_source", kind="data_source", label=f"{label} · {provider}", status=status, provider=provider, started_at=started_at, ended_at=timestamp, duration_ms=duration_ms, attempts=1, record_count=_safe_int(run.get("record_count")), message=message, metadata={ "data_type": data_type, "operation": run.get("operation"), "attempt": attempt_index, "fallback_from": run.get("fallback_from"), "fallback_to": run.get("fallback_to"), "cache_hit": run.get("cache_hit"), "stale_seconds": run.get("stale_seconds"), "error_type": run.get("error_type"), "error_message": run.get("error_message_sanitized"), }, ) previous = previous_node_by_type.get(data_type) if previous: previous_node_id, previous_run = previous edge_kind = _provider_transition_kind(previous_run, run) _append_edge( edges, previous_node_id, node_id, edge_kind, status, label="降级" if edge_kind == "fallback" else "重试", message=_safe_text(run.get("fallback_from") or run.get("fallback_to"), max_length=120), ) else: _append_edge(edges, "task_queue", node_id, "control", status, label="调用") if success: provider_success_by_block[block_key] = node_id severity = "success" if success else ("warning" if success_by_data_type.get(data_type) else "danger") _append_event( events, "provider_run", node_id=node_id, timestamp=timestamp, severity=severity, title=f"{label}{'成功' if success else '失败'}", message=message, metadata={ "provider": provider, "data_type": data_type, "duration_ms": duration_ms, "record_count": run.get("record_count"), "fallback_from": run.get("fallback_from"), "fallback_to": run.get("fallback_to"), "error_type": run.get("error_type"), }, ) previous_node_by_type[data_type] = (node_id, run) return provider_success_by_block def _append_context_blocks( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], events: List[Dict[str, Any]], overview: Optional[Dict[str, Any]], provider_success_by_block: Dict[str, str], ) -> None: if not overview: return metadata = overview.get("metadata") if isinstance(overview.get("metadata"), Mapping) else {} overview_timestamp = overview.get("created_at") for block in _as_list(overview.get("blocks")): block_map = _as_mapping(block) key = _safe_key(block_map.get("key")) if not key: continue status = _CONTEXT_STATUS_TO_FLOW.get(str(block_map.get("status") or ""), "unknown") node_id = f"context_block_{key}" record_count = metadata.get("news_result_count") if key == "news" else None _put_node( nodes, node_id, lane="data_source", kind="data_source", label=_safe_text(block_map.get("label"), max_length=80) or key, status=status, provider=block_map.get("source"), started_at=overview_timestamp, ended_at=overview_timestamp, record_count=_safe_int(record_count), message=_context_block_message(block_map), metadata={ "block_key": key, "source": block_map.get("source"), "warnings": block_map.get("warnings"), "missing_reasons": block_map.get("missing_reasons"), }, ) provider_node_id = provider_success_by_block.get(key) if provider_node_id: _append_edge(edges, provider_node_id, node_id, "data", status, label="输入") else: _append_edge(edges, "task_queue", node_id, "data", status, label="输入") _append_edge(edges, node_id, "context_pack", "data", status, label="组装") if status != "success": _append_event( events, "context_block_status", node_id=node_id, timestamp=overview.get("created_at"), severity="danger" if status == "failed" else "warning", title=f"{block_map.get('label') or key}输入状态", message=_context_block_message(block_map), metadata={ "block_key": key, "status": block_map.get("status"), "warnings": block_map.get("warnings"), "missing_reasons": block_map.get("missing_reasons"), }, ) def _append_llm_runs( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], events: List[Dict[str, Any]], llm_runs: List[Any], raw_result: Dict[str, Any], ) -> Optional[str]: previous_node_id = "context_pack" last_node_id: Optional[str] = None for index, raw_run in enumerate(llm_runs, start=1): run = _as_mapping(raw_run) if not run: continue call_type = _safe_key(run.get("call_type") or "analysis") model = normalize_model_used(run.get("model")) or normalize_model_used(raw_result.get("model_used")) provider = _safe_text(run.get("provider"), max_length=80) node_id = f"llm_{call_type}_{index}" success = run.get("success") is True status = "success" if success else "failed" if success and (run.get("fallback_model") or index > 1): status = "fallback" timestamp = _datetime_to_iso(run.get("created_at")) duration_ms = _safe_int(run.get("duration_ms")) started_at = _started_at_from_end_and_duration(timestamp, duration_ms) message = _llm_run_message(model, run, success=success) edge_kind = "data" if index > 1: edge_kind = "fallback" if run.get("fallback_model") else "retry" _put_node( nodes, node_id, lane="analysis", kind="model", label="LLM 生成", status=status, provider=model or provider, started_at=started_at, ended_at=timestamp, duration_ms=duration_ms, attempts=1, message=message, metadata={ "provider": provider, "model": model, "call_type": call_type, "tokens": run.get("tokens"), "fallback_model": run.get("fallback_model"), "error_type": run.get("error_type"), "error_message": run.get("error_message_sanitized"), }, ) _append_edge(edges, previous_node_id, node_id, edge_kind, status, label="生成") _append_event( events, "llm_run", node_id=node_id, timestamp=timestamp, severity="success" if success else "danger", title=f"LLM {'成功' if success else '失败'}", message=message, metadata={ "provider": provider, "model": model, "call_type": call_type, "tokens": run.get("tokens"), "duration_ms": duration_ms, "fallback_model": run.get("fallback_model"), "error_type": run.get("error_type"), }, ) previous_node_id = node_id last_node_id = node_id return last_node_id def _append_history_runs( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], events: List[Dict[str, Any]], history_runs: List[Any], *, anchor_node_id: str, fallback_created_at: Optional[str], ) -> Optional[str]: last_node_id: Optional[str] = None previous_node_id = anchor_node_id for index, raw_run in enumerate(history_runs, start=1): run = _as_mapping(raw_run) if not run: continue success = run.get("report_saved") is True status = "success" if success else "failed" node_id = "history_save" if index == 1 else f"history_save_{index}" timestamp = _datetime_to_iso(run.get("created_at")) or fallback_created_at message = "报告历史已保存" if success else f"报告历史保存失败：{_safe_text(run.get('error_message_sanitized'), max_length=160) or '未知错误'}" _put_node( nodes, node_id, lane="artifact", kind="artifact", label="保存报告", status=status, ended_at=timestamp, message=message, metadata={ "metadata_saved": run.get("metadata_saved"), "analysis_history_id": run.get("analysis_history_id"), "error_message": run.get("error_message_sanitized"), }, ) _append_edge(edges, previous_node_id, node_id, "data", status, label="保存") _append_event( events, "history_run", node_id=node_id, timestamp=timestamp, severity="success" if success else "danger", title="历史保存成功" if success else "历史保存失败", message=message, metadata={ "metadata_saved": run.get("metadata_saved"), "analysis_history_id": run.get("analysis_history_id"), }, ) previous_node_id = node_id last_node_id = node_id return last_node_id def _append_notification_runs( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], events: List[Dict[str, Any]], notification_runs: List[Any], *, anchor_node_id: str, ) -> int: count = 0 for index, raw_run in enumerate(notification_runs, start=1): run = _as_mapping(raw_run) if not run: continue count += 1 channel = _safe_text(run.get("channel"), max_length=80) or "unknown" raw_status = _safe_text(run.get("status"), max_length=80) or "unknown" if raw_status in {"skipped", "not_configured"}: status = "skipped" elif run.get("success") is True: status = "success" elif run.get("success") is False: status = "failed" else: status = "unknown" node_id = f"notification_{_safe_key(channel)}_{index}" timestamp = _datetime_to_iso(run.get("created_at")) message = _notification_run_message(channel, run, status) _put_node( nodes, node_id, lane="artifact", kind="notification", label=f"推送通知 · {channel}", status=status, provider=channel, ended_at=timestamp, attempts=_safe_int(run.get("attempts")) if _safe_int(run.get("attempts")) is not None else 1, message=message, metadata={ "channel": channel, "status": raw_status, "attempts": run.get("attempts"), "error_message": run.get("error_message_sanitized"), }, ) _append_edge(edges, anchor_node_id, node_id, "control", status, label="通知") _append_event( events, "notification_run", node_id=node_id, timestamp=timestamp, severity="success" if status == "success" else ("warning" if status == "skipped" else "danger"), title="通知发送成功" if status == "success" else ("通知跳过" if status == "skipped" else "通知失败"), message=message, metadata={ "channel": channel, "status": raw_status, "attempts": run.get("attempts"), }, ) return count _STOCK_CONTEXT_PROVIDER_DATA_TYPES = { "realtime_quote", "daily_data", "daily_bars", "technical", "fundamental", "fundamentals", "belong_boards", "chip", } def _normalize_history_diagnostics_for_record( record: Any, snapshot: Dict[str, Any], diagnostics: Dict[str, Any], ) -> Dict[str, Any]: if not diagnostics: return diagnostics normalized = dict(diagnostics) report_type = _safe_key(getattr(record, "report_type", None)) code = _safe_text(getattr(record, "code", None), max_length=32) report_kind = _safe_key(snapshot.get("report_kind")) if snapshot else "" if report_type == "market_review" or report_kind == "market_review" or (code or "").upper() == "MARKET": normalized["stock_code"] = "MARKET" normalized.setdefault("scope", "market_review") normalized["provider_runs"] = [ run for run in _as_list(normalized.get("provider_runs")) if _safe_key(_as_mapping(run).get("data_type")) not in _STOCK_CONTEXT_PROVIDER_DATA_TYPES ] return normalized first_llm_at = _first_timestamp(_as_list(normalized.get("llm_runs"))) if first_llm_at is not None: normalized["history_runs"] = [ run for run in _as_list(normalized.get("history_runs")) if not _timestamp_before(_as_mapping(run).get("created_at"), first_llm_at) ] normalized["notification_runs"] = [ run for run in _as_list(normalized.get("notification_runs")) if not _timestamp_before(_as_mapping(run).get("created_at"), first_llm_at) ] first_stock_data_at = _first_timestamp( [ run for run in _as_list(normalized.get("provider_runs")) if _safe_key(_as_mapping(run).get("data_type")) in _STOCK_CONTEXT_PROVIDER_DATA_TYPES ] ) if first_stock_data_at is not None: normalized["provider_runs"] = [ run for run in _as_list(normalized.get("provider_runs")) if _safe_key(_as_mapping(run).get("data_type")) != "news_search" or not _timestamp_before(_as_mapping(run).get("created_at"), first_stock_data_at) ] return normalized def _first_timestamp(items: List[Any]) -> Optional[datetime]: timestamps = [ parsed for parsed in (_datetime_for_elapsed(_as_mapping(item).get("created_at")) for item in items) if parsed is not None ] return min(timestamps) if timestamps else None def _timestamp_before(value: Any, boundary: datetime) -> bool: parsed = _datetime_for_elapsed(value) if parsed is None: return False if parsed.tzinfo is None and boundary.tzinfo is not None: parsed = parsed.replace(tzinfo=boundary.tzinfo) elif parsed.tzinfo is not None and boundary.tzinfo is None: boundary = boundary.replace(tzinfo=parsed.tzinfo) return parsed < boundary def _put_skeleton_tail( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], *, anchor_node_id: str, status: str, ) -> None: downstream_status = "skipped" if status in {"failed", "cancelled"} else "unknown" _put_node( nodes, "context_pack", lane="analysis", kind="analysis", label="ContextPack", status=downstream_status, message="尚未记录输入上下文诊断", ) _put_node( nodes, "llm", lane="analysis", kind="model", label="LLM 生成", status=downstream_status, message="尚未记录 LLM 诊断", ) _put_node( nodes, "history_save", lane="artifact", kind="artifact", label="保存报告", status=downstream_status, message="尚未记录历史保存结果", ) _put_node( nodes, "notification", lane="artifact", kind="notification", label="推送通知", status=downstream_status, message="尚未记录通知结果", ) _append_edge(edges, anchor_node_id, "context_pack", "data", downstream_status, label="输入") _append_edge(edges, "context_pack", "llm", "data", downstream_status, label="生成") _append_edge(edges, "llm", "history_save", "data", downstream_status, label="保存") _append_edge(edges, "history_save", "notification", "control", downstream_status, label="通知") def _prune_active_skeleton_tail( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], ) -> None: remove_node_ids = set() if "llm" in nodes and any(node_id.startswith("llm_") for node_id in nodes): remove_node_ids.add("llm") if "notification" in nodes and any(node_id.startswith("notification_") for node_id in nodes): remove_node_ids.add("notification") if not remove_node_ids: return for node_id in remove_node_ids: nodes.pop(node_id, None) edges[:] = [ edge for edge in edges if edge.get("from") not in remove_node_ids and edge.get("to") not in remove_node_ids ] def _append_task_events(events: List[Dict[str, Any]], task: Any, flow_status: str) -> None: _append_event( events, "task_created", node_id="task_queue", timestamp=_datetime_to_iso(getattr(task, "created_at", None)), severity="info", title="任务已创建", message=getattr(task, "message", None) or "任务已加入队列", ) if getattr(task, "started_at", None): _append_event( events, "task_started", node_id="task_queue", timestamp=_datetime_to_iso(getattr(task, "started_at", None)), severity="info", title="任务开始执行", message=getattr(task, "message", None) or "任务执行中", ) if flow_status == "failed": _append_event( events, "task_failed", node_id="task_queue", timestamp=_datetime_to_iso(getattr(task, "completed_at", None)), severity="danger", title="任务失败", message=getattr(task, "error", None) or getattr(task, "message", None), ) elif flow_status in {"cancel_requested", "cancelled"}: _append_event( events, f"task_{flow_status}", node_id="task_queue", timestamp=_datetime_to_iso(getattr(task, "completed_at", None)), severity="warning", title="任务取消" if flow_status == "cancelled" else "任务请求取消", message=getattr(task, "message", None), ) elif flow_status == "success": _append_event( events, "task_completed", node_id="task_queue", timestamp=_datetime_to_iso(getattr(task, "completed_at", None)), severity="success", title="任务完成", message=getattr(task, "message", None) or "分析完成", ) def _append_active_flow_events( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], events: List[Dict[str, Any]], flow_events: List[Any], *, flow_status: str, ) -> None: if not flow_events: return known_node_ids = set(nodes) last_provider_node_by_type: Dict[str, Tuple[str, Dict[str, Any]]] = {} last_llm_node: Optional[str] = None last_history_node: Optional[str] = None for raw_event in flow_events: event = _as_mapping(raw_event) if not event: continue metadata = _sanitize_metadata(event.get("metadata") or {}) node_payload = metadata.get("node") if isinstance(metadata, Mapping) else None node_id = _safe_key(event.get("node_id")) if isinstance(node_payload, Mapping): raw_node_id = _safe_text(node_payload.get("id"), max_length=120) or node_id if raw_node_id: node_id = raw_node_id _put_node( nodes, node_id, lane=str(node_payload.get("lane") or "analysis"), kind=str(node_payload.get("kind") or "analysis"), label=str(node_payload.get("label") or node_id), status=str(node_payload.get("status") or flow_status), provider=node_payload.get("provider"), started_at=node_payload.get("started_at") or _started_at_from_end_and_duration( node_payload.get("ended_at") or event.get("timestamp"), node_payload.get("duration_ms"), ), ended_at=node_payload.get("ended_at") or event.get("timestamp"), duration_ms=node_payload.get("duration_ms"), attempts=node_payload.get("attempts"), record_count=node_payload.get("record_count"), message=node_payload.get("message") or event.get("message"), metadata={key: value for key, value in metadata.items() if key != "node"}, ) event_type = _safe_key(event.get("type")) or "event" provider_data_type = None provider_run = None if event_type in {"provider_run", "provider_run_started"} and node_id and node_id in nodes: provider_data_type = _safe_key(metadata.get("data_type") or "provider") provider_run = { "provider": metadata.get("provider") or nodes[node_id].get("provider"), "success": event.get("severity") == "success" or nodes[node_id].get("status") in {"success", "fallback"}, "fallback_from": metadata.get("fallback_from"), "fallback_to": metadata.get("fallback_to"), } if node_id and node_id in nodes and node_id in known_node_ids: _refresh_incoming_edge_status(edges, node_id, nodes[node_id].get("status")) if provider_data_type and provider_run: last_provider_node_by_type[provider_data_type] = (node_id, provider_run) elif event_type in {"llm_run", "llm_run_started"}: last_llm_node = node_id elif event_type == "history_run": last_history_node = node_id if node_id and node_id in nodes and node_id not in known_node_ids: if provider_data_type and provider_run: previous_provider = last_provider_node_by_type.get(provider_data_type) if previous_provider: previous_provider_node, previous_provider_run = previous_provider edge_kind = _provider_transition_kind(previous_provider_run, provider_run) _append_edge( edges, previous_provider_node, node_id, edge_kind, nodes[node_id].get("status", "unknown"), label="降级" if edge_kind == "fallback" else ("重试" if edge_kind == "retry" else "调用"), ) else: _append_edge(edges, "task_queue", node_id, "control", nodes[node_id].get("status", "unknown"), label="调用") last_provider_node_by_type[provider_data_type] = (node_id, provider_run) elif event_type in {"llm_run", "llm_run_started"}: anchor = "analysis_pipeline" if "analysis_pipeline" in nodes else "task_queue" _append_edge(edges, anchor, node_id, "data", nodes[node_id].get("status", "unknown"), label="生成") last_llm_node = node_id elif event_type == "history_run": anchor = last_llm_node or ("analysis_pipeline" if "analysis_pipeline" in nodes else "task_queue") _append_edge(edges, anchor, node_id, "data", nodes[node_id].get("status", "unknown"), label="保存") last_history_node = node_id elif event_type == "notification_run": anchor = last_history_node or last_llm_node or ("analysis_pipeline" if "analysis_pipeline" in nodes else "task_queue") _append_edge(edges, anchor, node_id, "control", nodes[node_id].get("status", "unknown"), label="通知") known_node_ids.add(node_id) _append_external_event(events, event) def _append_external_event(events: List[Dict[str, Any]], event: Dict[str, Any]) -> None: event_id = _safe_text(event.get("id"), max_length=96) or f"flow_{len(events) + 1:04d}" if any(existing.get("id") == event_id for existing in events): return metadata = _sanitize_metadata(event.get("metadata") or {}) if isinstance(metadata, Mapping) and "node" in metadata: metadata = {key: value for key, value in metadata.items() if key != "node"} events.append( { "id": event_id, "timestamp": _datetime_to_iso(event.get("timestamp")), "severity": event.get("severity") if event.get("severity") in {"info", "success", "warning", "danger"} else "info", "type": _safe_key(event.get("type")) or "event", "node_id": _safe_text(event.get("node_id"), max_length=120), "title": _safe_text(event.get("title"), max_length=100) or "运行事件", "message": _safe_text(event.get("message"), max_length=220), "metadata": metadata, } ) def _group_provider_runs(provider_runs: List[Any]) -> Dict[str, List[Dict[str, Any]]]: grouped: Dict[str, List[Dict[str, Any]]] = defaultdict(list) for run in provider_runs: run_map = _as_mapping(run) if run_map: grouped[_safe_key(run_map.get("data_type") or "provider")].append(run_map) return grouped def _provider_run_status(run: Dict[str, Any], *, had_previous_failure: bool) -> str: if run.get("success") is True: if run.get("fallback_from") or had_previous_failure: return "fallback" return "success" error_type = str(run.get("error_type") or "").lower() if "timeout" in error_type: return "timeout" return "failed" def _provider_transition_kind(previous_run: Dict[str, Any], current_run: Dict[str, Any]) -> str: previous_provider = _safe_text(previous_run.get("provider"), max_length=80) current_provider = _safe_text(current_run.get("provider"), max_length=80) if previous_run.get("fallback_to") or current_run.get("fallback_from"): return "fallback" if previous_provider and previous_provider == current_provider: return "retry" if previous_run.get("success") is False: return "fallback" return "data" def _history_snapshot_status( nodes: Dict[str, Dict[str, Any]], diagnostics: Dict[str, Any], overview: Optional[Dict[str, Any]], ) -> str: statuses = [node.get("status") for node in nodes.values()] has_diagnostics = bool(diagnostics) has_overview = bool(overview) if not has_diagnostics and not has_overview: return "unknown" if any( node.get("kind") in {"model", "artifact"} and node.get("status") in {"failed", "timeout"} for node in nodes.values() ): return "failed" if any(status in {"failed", "timeout", "degraded", "fallback"} for status in statuses): return "degraded" return "success" def _context_pack_status(overview: Optional[Dict[str, Any]]) -> str: if not overview: return "unknown" block_statuses = [ _CONTEXT_STATUS_TO_FLOW.get(str(_as_mapping(block).get("status") or ""), "unknown") for block in _as_list(overview.get("blocks")) ] if not block_statuses: return "unknown" if any(status in {"failed", "fallback", "degraded", "skipped"} for status in block_statuses): return "degraded" if all(status == "success" for status in block_statuses): return "success" return "unknown" def _context_pack_message(overview: Optional[Dict[str, Any]]) -> str: if not overview: return "未记录 AnalysisContextPack overview" counts = overview.get("counts") if isinstance(counts, Mapping): available = counts.get("available", 0) return f"输入上下文已组装，可用块 {available}" return "输入上下文已组装" def _context_block_message(block: Dict[str, Any]) -> str: status = str(block.get("status") or "") if status == "available": return "已进入本次分析输入" if status == "fallback": return "本次分析输入使用降级数据" if status == "partial": return "本次分析输入仅部分可用" if status == "stale": return "本次分析输入使用过期数据" if status == "estimated": return "本次分析输入使用估算数据" if status == "fetch_failed": return "输入块抓取失败" if status == "missing": reasons = _as_list(block.get("missing_reasons")) reason = _safe_text(reasons[0], max_length=120) if reasons else None return f"未进入本次分析输入：{reason}" if reason else "未进入本次分析输入" if status == "not_supported": return "当前市场或链路不支持该输入块" return f"输入块状态为 {status or 'unknown'}" def _provider_run_message(label: str, provider: str, run: Dict[str, Any], *, success: bool) -> str: if success: record_count = _safe_int(run.get("record_count")) suffix = f"，返回 {record_count} 条" if record_count is not None else "" return f"{label} {provider} 成功{suffix}" error = _safe_text(run.get("error_message_sanitized") or run.get("error_type"), max_length=160) return f"{label} {provider} 失败：{error or '未知错误'}" def _llm_run_message(model: Optional[str], run: Dict[str, Any], *, success: bool) -> str: display_model = _safe_text(model or run.get("provider") or "unknown", max_length=120) if success: if run.get("fallback_model"): return f"LLM {display_model} 成功，期间发生模型切换" return f"LLM {display_model} 成功" error = _safe_text(run.get("error_message_sanitized") or run.get("error_type"), max_length=160) return f"LLM {display_model} 失败：{error or '未知错误'}" def _notification_run_message(channel: str, run: Dict[str, Any], status: str) -> str: if status == "success": return f"{channel} 通知发送成功" if status == "skipped": return f"{channel} 通知跳过" if status == "failed": error = _safe_text(run.get("error_message_sanitized") or run.get("status"), max_length=160) return f"{channel} 通知失败：{error or '未知错误'}" return f"{channel} 通知结果未知" def _build_summary( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], events: List[Dict[str, Any]], *, elapsed_ms: Optional[int] = None, ) -> Dict[str, Any]: bottleneck_node_id = None max_duration = -1 for node_id, node in nodes.items(): duration_ms = _safe_int(node.get("duration_ms")) if duration_ms is not None and duration_ms > max_duration: max_duration = duration_ms bottleneck_node_id = node_id if elapsed_ms is None: elapsed_ms = _events_elapsed_ms(events) if elapsed_ms is None and max_duration >= 0: elapsed_ms = sum( _safe_int(node.get("duration_ms")) or 0 for node in nodes.values() ) or None failed_attempts = sum( 1 for node in nodes.values() if node.get("status") in {"failed", "timeout"} and node.get("kind") in {"data_source", "model", "artifact", "notification"} ) fallback_count = sum( 1 for edge in edges if edge.get("kind") in {"fallback", "retry"} ) data_source_count = sum(1 for node in nodes.values() if node.get("kind") == "data_source") model = next( ( _safe_text(node.get("provider"), max_length=120) for node in nodes.values() if node.get("kind") == "model" and node.get("provider") ), None, ) return { "elapsed_ms": elapsed_ms, "bottleneck_node_id": bottleneck_node_id, "failed_attempts": failed_attempts, "fallback_count": fallback_count, "model": model, "data_source_count": data_source_count, "event_count": len(events), } def _put_node( nodes: Dict[str, Dict[str, Any]], node_id: str, *, lane: str, kind: str, label: str, status: str, provider: Optional[Any] = None, started_at: Optional[Any] = None, ended_at: Optional[Any] = None, duration_ms: Optional[Any] = None, attempts: Optional[Any] = None, record_count: Optional[Any] = None, message: Optional[Any] = None, metadata: Optional[Any] = None, ) -> None: payload = { "id": node_id, "lane": lane, "kind": kind, "label": _safe_text(label, max_length=80) or node_id, "status": _valid_status(status), "provider": _safe_text(provider, max_length=120), "started_at": _datetime_to_iso(started_at), "ended_at": _datetime_to_iso(ended_at), "duration_ms": _safe_int(duration_ms), "attempts": _safe_int(attempts), "record_count": _safe_int(record_count), "message": _safe_text(message, max_length=220), "metadata": _sanitize_metadata(metadata or {}), } nodes[node_id] = {key: value for key, value in payload.items() if value not in (None, {}, [])} def _append_edge( edges: List[Dict[str, Any]], from_node: str, to_node: str, kind: str, status: str, *, label: Optional[Any] = None, message: Optional[Any] = None, metadata: Optional[Any] = None, ) -> None: edge_id = f"{from_node}_to_{to_node}_{kind}" for edge in edges: if edge["id"] != edge_id: continue edge["status"] = _valid_status(status) safe_label = _safe_text(label, max_length=40) if safe_label: edge["label"] = safe_label safe_message = _safe_text(message, max_length=180) if safe_message: edge["message"] = safe_message safe_metadata = _sanitize_metadata(metadata or {}) if safe_metadata: edge["metadata"] = safe_metadata return edges.append( { "id": edge_id, "from": from_node, "to": to_node, "kind": kind if kind in {"data", "control", "fallback", "retry"} else "data", "status": _valid_status(status), "label": _safe_text(label, max_length=40), "message": _safe_text(message, max_length=180), "metadata": _sanitize_metadata(metadata or {}), } ) def _refresh_incoming_edge_status( edges: List[Dict[str, Any]], node_id: Optional[str], status: Optional[Any], ) -> None: if not node_id or status is None: return valid_status = _valid_status(status) for edge in edges: if edge.get("to") == node_id: edge["status"] = valid_status def _append_event( events: List[Dict[str, Any]], event_type: str, *, node_id: Optional[str], timestamp: Optional[Any], severity: str, title: str, message: Optional[Any] = None, metadata: Optional[Any] = None, ) -> None: event_id = f"evt_{len(events) + 1:04d}" events.append( { "id": event_id, "timestamp": _datetime_to_iso(timestamp), "severity": severity if severity in {"info", "success", "warning", "danger"} else "info", "type": _safe_key(event_type) or "event", "node_id": node_id, "title": _safe_text(title, max_length=100) or event_type, "message": _safe_text(message, max_length=220), "metadata": _sanitize_metadata(metadata or {}), } ) def _task_status_value(task: Any) -> str: status = getattr(task, "status", None) value = getattr(status, "value", status) return str(value or "unknown").strip().lower() def _map_task_status(status_value: str) -> str: return _RUN_STATUS_MAP.get(status_value, "unknown") def _task_status_message(status: str) -> str: return { "pending": "任务已加入队列", "running": "任务执行中", "success": "任务已完成", "failed": "任务失败", "cancel_requested": "任务请求取消", "cancelled": "任务已取消", }.get(status, "任务状态未知") def _valid_status(value: Any) -> str: text = str(value or "unknown").strip().lower() if text in { "pending", "running", "success", "failed", "degraded", "fallback", "timeout", "cancel_requested", "cancelled", "skipped", "unknown", }: return text return "unknown" def _safe_text(value: Any, *, max_length: int = 300) -> Optional[str]: return sanitize_diagnostic_text(value, max_length=max_length) def _safe_key(value: Any) -> str: return safe_diagnostic_key(value) def _safe_int(value: Any) -> Optional[int]: if isinstance(value, bool) or value is None: return None try: parsed = int(value) except (TypeError, ValueError): return None return parsed if parsed >= 0 else None def _sanitize_metadata(value: Any, *, depth: int = 0) -> Any: return sanitize_diagnostic_metadata(value, depth=depth) def _as_mapping(value: Any) -> Dict[str, Any]: parsed = parse_json_field(value) if isinstance(parsed, Mapping): return dict(parsed) if isinstance(parsed, str) and parsed.strip(): try: loaded = json.loads(parsed) except (TypeError, ValueError, json.JSONDecodeError): return {} return dict(loaded) if isinstance(loaded, Mapping) else {} return {} def _as_list(value: Any) -> List[Any]: return value if isinstance(value, list) else [] def _datetime_to_iso(value: Any) -> Optional[str]: if isinstance(value, datetime): return value.isoformat() if isinstance(value, str) and value.strip(): return _safe_text(value, max_length=80) return None def _elapsed_ms(start: Any, end: Any) -> Optional[int]: start_dt = _datetime_for_elapsed(start) end_dt = _datetime_for_elapsed(end) if start_dt is None or end_dt is None: return None seconds = (end_dt - start_dt).total_seconds() if seconds < 0: return None return int(seconds * 1000) def _started_at_from_end_and_duration(end: Any, duration_ms: Any) -> Optional[str]: duration = _safe_int(duration_ms) if duration is None: return None if isinstance(end, datetime): parsed = end elif isinstance(end, str) and "T" in end: normalized = end[:-1] + "+00:00" if end.endswith("Z") else end try: parsed = datetime.fromisoformat(normalized) except ValueError: return None else: return None return (parsed - timedelta(milliseconds=duration)).isoformat() def _local_timezone(): return datetime.now().astimezone().tzinfo or timezone.utc def _datetime_for_elapsed(value: Any) -> Optional[datetime]: if isinstance(value, datetime): parsed = value elif isinstance(value, str) and "T" in value: normalized = value[:-1] + "+00:00" if value.endswith("Z") else value try: parsed = datetime.fromisoformat(normalized) except ValueError: return None else: return None if parsed.tzinfo is None or parsed.utcoffset() is None: parsed = parsed.replace(tzinfo=_local_timezone()) return parsed.astimezone(timezone.utc).replace(tzinfo=None) def _events_elapsed_ms(events: Iterable[Dict[str, Any]]) -> Optional[int]: timestamps: List[datetime] = [] for event in events: parsed = _datetime_for_elapsed(event.get("timestamp")) if parsed is not None: timestamps.append(parsed) if len(timestamps) < 2: return None elapsed = (max(timestamps) - min(timestamps)).total_seconds() return int(elapsed * 1000) if elapsed >= 0 else None