src/planner/validation.py

from typing import Any, Dict

from src.planner.schema import SemanticPlan

_LIST_FIELDS = {
    "primary_entities",
    "secondary_entities",
    "visual_attributes",
    "style",
    "mood_emotion",
    "narrative_tone",
    "audio_intent",
    "audio_elements",
    "must_include",
    "must_avoid",
}

_LIST_PATHS = [
    ("core_semantics", "main_subjects"),
    ("core_semantics", "actions"),
    ("style_controls", "visual_style"),
    ("style_controls", "color_palette"),
    ("style_controls", "lighting"),
    ("style_controls", "camera"),
    ("style_controls", "mood_emotion"),
    ("style_controls", "narrative_tone"),
    ("image_constraints", "must_include"),
    ("image_constraints", "must_avoid"),
    ("image_constraints", "objects"),
    ("image_constraints", "environment_details"),
    ("image_constraints", "composition"),
    ("audio_constraints", "audio_intent"),
    ("audio_constraints", "sound_sources"),
    ("audio_constraints", "ambience"),
    ("audio_constraints", "must_include"),
    ("audio_constraints", "must_avoid"),
    ("text_constraints", "must_include"),
    ("text_constraints", "must_avoid"),
    ("text_constraints", "keywords"),
]

_STRING_PATHS = [
    ("scene_summary",),
    ("domain",),
    ("core_semantics", "setting"),
    ("core_semantics", "time_of_day"),
    ("core_semantics", "weather"),
    ("audio_constraints", "tempo"),
    ("text_constraints", "length"),
]


def _as_list(value: Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(v) for v in value if str(v).strip()]
    if isinstance(value, str):
        return [value] if value.strip() else []
    return [str(value)]


def _as_str(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, list):
        for item in value:
            text = str(item).strip()
            if text:
                return text
        return ""
    return str(value)


def _get_parent(data: Dict[str, Any], path: tuple[str, ...]) -> Dict[str, Any] | None:
    cur: Any = data
    for key in path[:-1]:
        if not isinstance(cur, dict) or key not in cur:
            return None
        cur = cur[key]
    if not isinstance(cur, dict):
        return None
    return cur


def _normalize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    for key in _LIST_FIELDS:
        if key in data:
            data[key] = _as_list(data[key])

    for path in _LIST_PATHS:
        parent = _get_parent(data, path)
        if parent is not None and path[-1] in parent:
            parent[path[-1]] = _as_list(parent[path[-1]])

    for path in _STRING_PATHS:
        parent = _get_parent(data, path)
        if parent is not None and path[-1] in parent:
            parent[path[-1]] = _as_str(parent[path[-1]])

    return data


def validate_semantic_plan_dict(data: Dict[str, Any]) -> None:
    data = _normalize_fields(data)
    SemanticPlan(**data)