shinkan-jinkendo/backend/progression_graph_planning_artifact.py

50 lines
1.7 KiB
Python
Raw Normal View History

"""Validierung und Normalisierung des Planungs-Artefakts am Progressionsgraph."""
from __future__ import annotations
import json
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field, field_validator
ARTIFACT_SCHEMA_VERSION = 1
_MAX_JSON_BYTES = 64_000
class GraphPlanningRoadmapArtifact(BaseModel):
schema_version: int = Field(default=ARTIFACT_SCHEMA_VERSION, ge=1, le=1)
goal_query: str = Field(default="", max_length=2000)
start_situation: Optional[str] = Field(default=None, max_length=2000)
target_state: Optional[str] = Field(default=None, max_length=2000)
roadmap_notes: Optional[str] = Field(default=None, max_length=2000)
max_steps: int = Field(default=5, ge=2, le=10)
progression_roadmap: Optional[Dict[str, Any]] = None
path_skill_expectations: Optional[Dict[str, Any]] = None
@field_validator("progression_roadmap", "path_skill_expectations", mode="before")
@classmethod
def _empty_dict_to_none(cls, v):
if v == {}:
return None
return v
def normalize_planning_roadmap_payload(raw: Any) -> Optional[Dict[str, Any]]:
"""None erlaubt (löschen); sonst validiertes Dict."""
if raw is None:
return None
if not isinstance(raw, dict):
raise ValueError("planning_roadmap muss ein JSON-Objekt sein")
artifact = GraphPlanningRoadmapArtifact.model_validate(raw)
out = artifact.model_dump(exclude_none=True)
blob = json.dumps(out, ensure_ascii=False)
if len(blob.encode("utf-8")) > _MAX_JSON_BYTES:
raise ValueError("planning_roadmap ist zu groß (max. 64 KB)")
return out
__all__ = [
"ARTIFACT_SCHEMA_VERSION",
"GraphPlanningRoadmapArtifact",
"normalize_planning_roadmap_payload",
]