shinkan-jinkendo/backend/planning_catalog_context.py

"""
Katalog-Kontext für Progressionsgraph-Planung — Fokusbereich, Stil, Trainingsstil, Zielgruppe.

Explizite Trainer-Auswahl ergänzt Freitext/LLM; ersetzt kein Roadmap-Didaktik-Modell.
"""
from __future__ import annotations

from typing import Any, Dict, List, Mapping, Optional, Sequence

from pydantic import BaseModel, Field

from planning_exercise_profiles import PlanningTargetProfile, _normalize_weight_map
from planning_exercise_target_pipeline import (
    SCENARIO_FREE_SEARCH,
    merge_query_overlay_into_target,
)
from planning_exercise_text_signals import resolve_planning_text_to_catalog_weights


class PlanningCatalogContextItem(BaseModel):
    id: int = Field(..., ge=1)
    is_primary: bool = False
    weight: float = Field(default=1.0, ge=0.1, le=1.0)


class ProgressionPlanningCatalogContext(BaseModel):
    focus_areas: List[PlanningCatalogContextItem] = Field(default_factory=list)
    style_directions: List[PlanningCatalogContextItem] = Field(default_factory=list)
    training_types: List[PlanningCatalogContextItem] = Field(default_factory=list)
    target_groups: List[PlanningCatalogContextItem] = Field(default_factory=list)


def catalog_context_has_items(catalog: Optional[ProgressionPlanningCatalogContext]) -> bool:
    if catalog is None:
        return False
    return bool(
        catalog.focus_areas
        or catalog.style_directions
        or catalog.training_types
        or catalog.target_groups
    )


def catalog_items_to_weight_map(
    items: Sequence[PlanningCatalogContextItem],
    *,
    primary_weight: float = 0.95,
    secondary_weight: float = 0.78,
) -> Dict[int, float]:
    out: Dict[int, float] = {}
    for item in items or []:
        base = primary_weight if item.is_primary else secondary_weight
        w = base * float(item.weight)
        iid = int(item.id)
        out[iid] = max(out.get(iid, 0.0), w)
    return _normalize_weight_map(out) if out else out


def merge_catalog_context_into_target(
    target: PlanningTargetProfile,
    catalog: Optional[ProgressionPlanningCatalogContext],
    *,
    emphasis: str = "replace",
) -> PlanningTargetProfile:
    """Trainer-Katalog-Kontext ins Erwartungsprofil — beeinflusst Retrieval-Scoring."""
    if not catalog_context_has_items(catalog):
        return target

    focus = catalog_items_to_weight_map(catalog.focus_areas)
    style = catalog_items_to_weight_map(catalog.style_directions, primary_weight=0.9, secondary_weight=0.72)
    tt = catalog_items_to_weight_map(catalog.training_types, primary_weight=0.9, secondary_weight=0.72)
    tg = catalog_items_to_weight_map(catalog.target_groups, primary_weight=0.88, secondary_weight=0.7)

    merged = merge_query_overlay_into_target(
        target,
        focus=focus,
        style=style,
        tt=tt,
        tg=tg,
        skills={},
        emphasis=emphasis,
        scenario=SCENARIO_FREE_SEARCH,
    )
    sources = list(merged.sources or [])
    if "catalog_context" not in sources:
        sources.append("catalog_context")
    merged.sources = sources
    return merged


def enrich_target_from_planning_text_blobs(
    cur,
    target: PlanningTargetProfile,
    *text_blobs: Optional[str],
) -> PlanningTargetProfile:
    """Additive Katalog-Signale aus Freitext (Anfrage, Start/Ziel, Notizen)."""
    combined = " ".join(str(t or "").strip() for t in text_blobs if (t or "").strip())
    if len(combined) < 4:
        return target
    focus, style, tt, tg, skills = resolve_planning_text_to_catalog_weights(cur, combined)
    if not (focus or style or tt or tg or skills):
        return target
    merged = merge_query_overlay_into_target(
        target,
        focus=focus,
        style=style,
        tt=tt,
        tg=tg,
        skills=skills,
        emphasis="additive",
        scenario=SCENARIO_FREE_SEARCH,
    )
    sources = list(merged.sources or [])
    if "text_catalog_signals" not in sources:
        sources.append("text_catalog_signals")
    merged.sources = sources
    return merged


def catalog_context_from_mapping(raw: Any) -> Optional[ProgressionPlanningCatalogContext]:
    if not raw or not isinstance(raw, Mapping):
        return None
    try:
        ctx = ProgressionPlanningCatalogContext.model_validate(dict(raw))
    except Exception:
        return None
    return ctx if catalog_context_has_items(ctx) else None


def load_catalog_context_from_graph_row(
    planning_roadmap: Any,
) -> Optional[ProgressionPlanningCatalogContext]:
    if not isinstance(planning_roadmap, dict):
        return None
    return catalog_context_from_mapping(planning_roadmap.get("planning_catalog_context"))


__all__ = [
    "PlanningCatalogContextItem",
    "ProgressionPlanningCatalogContext",
    "catalog_context_from_mapping",
    "catalog_context_has_items",
    "catalog_items_to_weight_map",
    "enrich_target_from_planning_text_blobs",
    "load_catalog_context_from_graph_row",
    "merge_catalog_context_into_target",
]