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

from __future__ import annotations

from typing import Any, Literal, TypedDict


RelationKind = Literal[
    "single-event",
    "event-sequence",
    "business-record",
    "aggregate",
    "calculated",
    "missing",
]
Completeness = Literal["complete", "partial", "missing"]


class CardDataSource(TypedDict, total=False):
    kind: Literal["database", "runtime-event", "artifact", "calculation", "log-anchor"]
    label: str
    ref: str
    fieldPath: str


class CardDataField(TypedDict):
    id: str
    label: str
    value: Any
    source: CardDataSource
    relation: Literal[
        "direct-input",
        "previous-result",
        "standing-constraint",
        "explicit-basis",
        "available-upstream",
        "persisted-output",
        "run-output",
    ]
    completeness: Completeness


class RuntimeUnit(TypedDict, total=False):
    id: str
    label: str
    status: str
    summary: str
    input: Any
    output: Any
    durationMs: int | None
    eventRef: str
    completeness: Completeness
    startedAt: str | None
    endedAt: str | None


def field(
    field_id: str,
    label: str,
    value: Any,
    *,
    source_kind: str,
    source_label: str,
    source_ref: str | None = None,
    field_path: str | None = None,
    relation: str,
    completeness: Completeness | None = None,
) -> CardDataField:
    missing = value is None or value == "" or value == [] or value == {}
    source: CardDataSource = {"kind": source_kind, "label": source_label}  # type: ignore[typeddict-item]
    if source_ref:
        source["ref"] = source_ref
    if field_path:
        source["fieldPath"] = field_path
    return {
        "id": field_id,
        "label": label,
        "value": value,
        "source": source,
        "relation": relation,  # type: ignore[typeddict-item]
        "completeness": completeness or ("missing" if missing else "complete"),
    }


def payload(
    card_kind: str,
    relation_kind: RelationKind,
    *,
    inputs: list[CardDataField] | None = None,
    units: list[RuntimeUnit] | None = None,
    outputs: list[CardDataField] | None = None,
    runtime_summary: str = "",
    calculation: str | None = None,
    notices: list[dict[str, str]] | None = None,
    business_modules: list[dict[str, Any]] | None = None,
    card_fields: list[dict[str, Any]] | None = None,
    completeness: Completeness | None = None,
) -> dict[str, Any]:
    inputs = inputs or []
    units = units or []
    outputs = outputs or []
    all_fields = [*inputs, *outputs]
    available = [item for item in all_fields if item["completeness"] != "missing"]
    if completeness is None:
        if not available and not units:
            completeness = "missing"
        elif any(item["completeness"] != "complete" for item in all_fields) or any(
            item.get("completeness") != "complete" for item in units
        ):
            completeness = "partial"
        else:
            completeness = "complete"
    if business_modules is None:
        business_modules = [
            {"id": "business-inputs", "title": "业务输入", "sourceIds": [item["id"] for item in inputs]},
            {"id": "runtime", "title": "运行过程 / I/O", "sourceIds": [item["id"] for item in all_fields]},
            {"id": "business-outputs", "title": "业务输出", "sourceIds": [item["id"] for item in outputs]},
        ]
    if card_fields is None:
        headline = next((item for item in outputs if item["completeness"] != "missing"), None)
        if headline is None:
            headline = next((item for item in inputs if item["completeness"] != "missing"), None)
        card_fields = [] if headline is None else [{
            "key": "headline",
            "label": headline["label"],
            "sourceIds": [headline["id"]],
            "transform": "使用结构化业务字段；卡片仅做视觉截行",
        }]
    used = {
        source_id
        for item in [*business_modules, *card_fields]
        for source_id in item.get("sourceIds", [])
    }
    runtime: dict[str, Any] = {"summary": runtime_summary, "units": units}
    if calculation:
        runtime["calculation"] = calculation
    return {
        "cardKind": card_kind,
        "relationKind": relation_kind,
        "completeness": completeness,
        "businessInputs": inputs,
        "runtime": runtime,
        "businessOutputs": outputs,
        "displayUse": {
            "businessModules": business_modules,
            "cardFields": card_fields,
            "unusedSourceIds": [item["id"] for item in all_fields if item["id"] not in used],
        },
        "notices": notices or [],
    }