script-build-and-framework-tmp/visualization-old/backend/app/card_details/common.py

from __future__ import annotations

import re
from collections.abc import Callable, Iterable
from typing import Any

from .schema import RuntimeUnit, field


EventLoader = Callable[[int], dict[str, Any]]


def integer(value: Any) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def suffix_int(ref: str) -> int | None:
    return integer(str(ref).rsplit(":", 1)[-1])


def event_content(side: Any) -> Any:
    return side.get("content") if isinstance(side, dict) else None


def event_completeness(event: dict[str, Any]) -> str:
    has_input = event_content(event.get("input")) not in (None, "", [], {})
    has_output = event_content(event.get("output")) not in (None, "", [], {})
    if has_input and has_output:
        return "complete"
    if has_input or has_output or event:
        return "partial"
    return "missing"


def event_unit(event: dict[str, Any], *, label: str | None = None) -> RuntimeUnit:
    event_id = integer(event.get("id")) or 0
    status = str(event.get("status") or "unknown")
    event_label = label or str(event.get("businessLabel") or event.get("title") or event.get("event_name") or f"运行事件 {event_id}")
    return {
        "id": f"runtime:event:{event_id}",
        "label": event_label,
        "status": status,
        "summary": event_summary(event, event_label),
        "input": event.get("input"),
        "output": event.get("output"),
        "durationMs": integer(event.get("duration_ms")),
        "eventRef": f"event:{event_id}",
        "completeness": event_completeness(event),  # type: ignore[typeddict-item]
        "startedAt": event.get("started_at"),
        "endedAt": event.get("ended_at"),
    }


def event_summary(event: dict[str, Any], fallback: str) -> str:
    if event.get("error_message"):
        return f"技术失败：{event['error_message']}"
    output = event_content(event.get("output"))
    if isinstance(output, dict):
        for key in ("summary", "message", "conclusion"):
            if output.get(key):
                return str(output[key])
    if event.get("output_preview"):
        return str(event["output_preview"])
    return fallback


def load_events(loader: EventLoader, refs: Iterable[Any]) -> list[dict[str, Any]]:
    values: list[dict[str, Any]] = []
    seen: set[int] = set()
    for ref in refs:
        event_id = suffix_int(str(ref)) if isinstance(ref, str) else integer(ref)
        if event_id is None or event_id in seen:
            continue
        seen.add(event_id)
        try:
            values.append(loader(event_id))
        except LookupError:
            continue
    return values


def event_notices(events: Iterable[dict[str, Any]]) -> list[dict[str, str]]:
    notices: list[dict[str, str]] = []
    omitted = 0
    missing_body = 0
    for event in events:
        for side_name in ("input", "output"):
            side = event.get(side_name)
            if isinstance(side, dict) and side.get("truncated"):
                omitted += int(side.get("omittedCharacters") or 0)
        if event.get("input") is None and event.get("output") is None:
            missing_body += 1
    if omitted:
        notices.append({"level": "warning", "message": f"Event Body 超过读取上限，共省略 {omitted} 个字符。"})
    if missing_body:
        notices.append({"level": "warning", "message": f"{missing_body} 个运行事件没有保存可用的 Event Body。"})
    return notices


def find_by_id(items: Iterable[dict[str, Any]], record_id: int | None) -> dict[str, Any] | None:
    return next((item for item in items if integer(item.get("id")) == record_id), None)


def find_round(view: dict[str, Any], round_index: int) -> dict[str, Any] | None:
    return next((item for item in view.get("rounds") or [] if integer(item.get("roundIndex")) == round_index), None)


def find_raw_round(bundle: dict[str, Any], round_index: int) -> dict[str, Any] | None:
    return next((item for item in bundle.get("rounds") or [] if integer(item.get("round_index")) == round_index), None)


def find_branch(view: dict[str, Any], round_index: int, branch_id: int) -> dict[str, Any] | None:
    round_ = find_round(view, round_index)
    return next((item for item in (round_ or {}).get("branches") or [] if integer(item.get("branchId")) == branch_id), None)


def find_raw_branch(bundle: dict[str, Any], branch_id: int) -> dict[str, Any] | None:
    return next((item for item in bundle.get("branches") or [] if integer(item.get("branch_id")) == branch_id), None)


def find_detail(value: Any, detail_ref: str) -> dict[str, Any] | None:
    if isinstance(value, dict):
        if value.get("detailRef") == detail_ref:
            return value
        for child in value.values():
            found = find_detail(child, detail_ref)
            if found is not None:
                return found
    elif isinstance(value, list):
        for child in value:
            found = find_detail(child, detail_ref)
            if found is not None:
                return found
    return None


def decision_context(node: dict[str, Any] | None) -> dict[str, Any]:
    if not isinstance(node, dict):
        return {}
    decision = node.get("decision")
    if isinstance(decision, dict):
        body = decision.get("body")
        if isinstance(body, dict):
            return {**decision, **body}
        return decision
    return node


def context_input_fields(
    context: dict[str, Any],
    prefix: str,
    events: list[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
    relation_map = {
        "direct-read": "direct-input",
        "standing-constraint": "standing-constraint",
        "persisted-record": "previous-result",
        "explicit-basis": "explicit-basis",
    }
    fields: list[dict[str, Any]] = []
    events_by_ref = {
        f"event:{integer(event.get('id'))}": event
        for event in events or []
        if integer(event.get("id"))
    }
    for index, item in enumerate(context.get("inputs") or [], 1):
        if not isinstance(item, dict):
            continue
        raw_relation = str(item.get("relation") or "")
        relation = relation_map.get(raw_relation, "available-upstream")
        detail_ref = str(item.get("detailRef") or "")
        label = str(item.get("label") or f"上游输入 {index}")
        if not detail_ref:
            detail_ref = _implicit_context_ref(prefix, label)
        event = events_by_ref.get(detail_ref) or {}
        event_name = str(event.get("event_name") or "")
        source_kind, source_label, field_path = _context_source(
            detail_ref, raw_relation, event_name
        )
        fields.append(field(
            f"{prefix}:input:{index}",
            label,
            item.get("summary"),
            source_kind=source_kind,
            source_label=source_label,
            source_ref=detail_ref or None,
            field_path=field_path,
            relation=relation,
        ))
    return fields


def _context_source(
    detail_ref: str,
    raw_relation: str,
    event_name: str,
) -> tuple[str, str, str]:
    """Classify context sources from their real reference contract."""
    event_paths = {
        "save_script_direction": "input.content.script_direction",
        "begin_round": "input.content.goal",
        "record_multipath_plan": "input.content",
    }
    if event_name or detail_ref.startswith("event:"):
        return (
            "runtime-event",
            "明确读取的运行事件" if raw_relation == "direct-read" else "上游运行事件",
            event_paths.get(event_name, ""),
        )
    database_refs = (
        (r"^run:objective$", "script_build_record", "script_direction"),
        (r"^round:\d+:goal$", "script_build_round", "goal"),
        (r"^multipath-decision:\d+$", "script_build_multipath_decision", "decision"),
        (r"^data-decision:\d+$", "script_build_data_decision", "decision"),
        (r"^domain-info:\d+$", "script_build_domain_info", "content"),
    )
    for pattern, table, field_path in database_refs:
        if re.match(pattern, detail_ref):
            return "database", table, field_path
    if detail_ref.startswith("artifact:"):
        return "artifact", "脚本或候选产物", ""
    return "database", "上游业务结果", ""


def _implicit_context_ref(prefix: str, label: str) -> str:
    """Recover only deterministic upstream refs omitted by historical views."""
    if prefix == "objective" and label in {"选题", "账号人设"}:
        return ""
    if re.match(r"^round:\d+:goal$", prefix) and label == "当前保存的创作目标":
        return "run:objective"
    match = re.match(r"^(round:\d+):plan$", prefix)
    if match and label == "本轮目标":
        return f"{match.group(1)}:goal"
    return ""


def decision_event_refs(context: dict[str, Any]) -> list[str]:
    refs: list[str] = []
    for key in ("revisionRefs", "technicalRefs", "eventRefs"):
        for ref in context.get(key) or []:
            if isinstance(ref, str) and ref.startswith("event:") and ref not in refs:
                refs.append(ref)
    for key in ("currentRevisionRef", "eventRef"):
        ref = context.get(key)
        if isinstance(ref, str) and ref.startswith("event:") and ref not in refs:
            refs.append(ref)
    for item in context.get("inputs") or []:
        ref = item.get("detailRef") if isinstance(item, dict) else None
        if isinstance(ref, str) and ref.startswith("event:") and ref not in refs:
            refs.append(ref)
    for item in context.get("revisions") or []:
        ref = item.get("detailRef") if isinstance(item, dict) else None
        if isinstance(ref, str) and ref.startswith("event:") and ref not in refs:
            refs.append(ref)
    return refs


def output_field_from_event(event: dict[str, Any], prefix: str, label: str, relation: str = "persisted-output") -> dict[str, Any]:
    event_id = integer(event.get("id")) or 0
    return field(
        f"{prefix}:event:{event_id}:output",
        label,
        event_content(event.get("output")),
        source_kind="runtime-event",
        source_label="Event Body 输出",
        source_ref=f"event:{event_id}",
        field_path="output.content",
        relation=relation,
        completeness=event_completeness(event),
    )