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 [], }