Lars
7265cd5a01
All checks were successful
Deploy Development / deploy (push) Successful in 43s
Test Suite / pytest-backend (push) Successful in 45s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 14s
Test Suite / k6 /health Baseline (push) Successful in 34s
Test Suite / playwright-tests (push) Successful in 1m34s
- Introduced `findings_stale` field in `GraphPlanningRoadmapArtifact` to track the freshness of findings. - Updated `ProgressionGraphEditor` to manage `findingsStale` state across various functions, ensuring accurate representation of evaluation status. - Modified related utility functions and tests to accommodate the new state, enhancing overall functionality and user feedback in the progression graph management process.
79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
"""Validierung und Normalisierung des Planungs-Artefakts am Progressionsgraph."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
ARTIFACT_SCHEMA_VERSION = 1
|
|
_MAX_JSON_BYTES = 64_000
|
|
|
|
|
|
class SlotExerciseContent(BaseModel):
|
|
kind: str = Field(default="empty", pattern=r"^(empty|library|proposal)$")
|
|
exercise_id: Optional[int] = Field(default=None, ge=1)
|
|
variant_id: Optional[int] = Field(default=None, ge=1)
|
|
title: Optional[str] = Field(default=None, max_length=500)
|
|
variant_name: Optional[str] = Field(default=None, max_length=200)
|
|
proposal_key: Optional[str] = Field(default=None, max_length=120)
|
|
ai_suggestion: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
class SlotContentEntry(BaseModel):
|
|
major_step_index: int = Field(ge=0, le=20)
|
|
primary: SlotExerciseContent = Field(default_factory=SlotExerciseContent)
|
|
siblings: List[SlotExerciseContent] = Field(default_factory=list)
|
|
|
|
|
|
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
|
|
slot_contents: Optional[List[SlotContentEntry]] = None
|
|
last_findings: Optional[Dict[str, Any]] = None
|
|
findings_stale: bool = Field(default=False)
|
|
planning_catalog_context: Optional[Dict[str, Any]] = None
|
|
|
|
@field_validator("progression_roadmap", "path_skill_expectations", "last_findings", "planning_catalog_context", mode="before")
|
|
@classmethod
|
|
def _empty_dict_to_none(cls, v):
|
|
if v == {}:
|
|
return None
|
|
return v
|
|
|
|
@field_validator("slot_contents", mode="before")
|
|
@classmethod
|
|
def _empty_slot_list_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",
|
|
"SlotContentEntry",
|
|
"SlotExerciseContent",
|
|
"normalize_planning_roadmap_payload",
|
|
]
|