Lars
78cf89c0fa
All checks were successful
Deploy Trainer_LLM to llm-node / deploy (push) Successful in 1s
121 lines
3.4 KiB
Python
121 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
plan_router.py – v0.12.2 (WP-15)
|
||
|
||
Minimal-CRUD für Plan-Templates & Pläne (POST/GET) + Idempotenz via Fingerprint.
|
||
NEU/Änderungen ggü. v0.12.0:
|
||
- GET /plan_templates (Liste/Filter)
|
||
- GET /plans (Liste/Filter)
|
||
- Fix: Filter `section` bei `/plans` nutzt materialisierte `plan_section_names`
|
||
- POST /plan materialisiert `plan_section_names` automatisch
|
||
- Ausführlichere Swagger-Doku
|
||
"""
|
||
from fastapi import APIRouter, HTTPException, Query
|
||
from pydantic import BaseModel, Field
|
||
from typing import List, Optional, Dict, Any
|
||
from uuid import uuid4
|
||
from datetime import datetime, timezone
|
||
import hashlib
|
||
import json
|
||
import os
|
||
|
||
from clients import model, qdrant
|
||
from qdrant_client.models import PointStruct, Filter, FieldCondition, MatchValue, VectorParams, Distance
|
||
|
||
router = APIRouter(tags=["plans"])
|
||
|
||
# -----------------
|
||
# Konfiguration
|
||
# -----------------
|
||
PLAN_COLLECTION = os.getenv("PLAN_COLLECTION") or os.getenv("QDRANT_COLLECTION_PLANS", "plans")
|
||
PLAN_TEMPLATE_COLLECTION = os.getenv("PLAN_TEMPLATE_COLLECTION", "plan_templates")
|
||
PLAN_SESSION_COLLECTION = os.getenv("PLAN_SESSION_COLLECTION", "plan_sessions")
|
||
EXERCISE_COLLECTION = os.getenv("EXERCISE_COLLECTION", "exercises")
|
||
|
||
# -----------------
|
||
# Modelle
|
||
# -----------------
|
||
class TemplateSection(BaseModel):
|
||
name: str
|
||
target_minutes: int
|
||
must_keywords: List[str] = []
|
||
ideal_keywords: List[str] = []
|
||
supplement_keywords: List[str] = []
|
||
forbid_keywords: List[str] = []
|
||
capability_targets: Dict[str, int] = {}
|
||
|
||
class PlanTemplate(BaseModel):
|
||
id: str = Field(default_factory=lambda: str(uuid4()))
|
||
name: str
|
||
discipline: str
|
||
age_group: str
|
||
target_group: str
|
||
total_minutes: int
|
||
sections: List[TemplateSection] = []
|
||
goals: List[str] = []
|
||
equipment_allowed: List[str] = []
|
||
created_by: str
|
||
version: str
|
||
|
||
class PlanItem(BaseModel):
|
||
exercise_external_id: str
|
||
duration: int
|
||
why: str
|
||
|
||
class PlanSection(BaseModel):
|
||
name: str
|
||
items: List[PlanItem] = []
|
||
minutes: int
|
||
|
||
class Plan(BaseModel):
|
||
id: str = Field(default_factory=lambda: str(uuid4()))
|
||
template_id: Optional[str] = None
|
||
title: str
|
||
discipline: str
|
||
age_group: str
|
||
target_group: str
|
||
total_minutes: int
|
||
sections: List[PlanSection] = []
|
||
goals: List[str] = []
|
||
capability_summary: Dict[str, int] = {}
|
||
novelty_against_last_n: Optional[float] = None
|
||
fingerprint: Optional[str] = None
|
||
created_by: str
|
||
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
||
source: str = "API"
|
||
|
||
class PlanTemplateList(BaseModel):
|
||
items: List[PlanTemplate]
|
||
limit: int
|
||
offset: int
|
||
count: int
|
||
|
||
class PlanList(BaseModel):
|
||
items: List[Plan]
|
||
limit: int
|
||
offset: int
|
||
count: int
|
||
|
||
# -----------------
|
||
# Helpers
|
||
# -----------------
|
||
def _ensure_collection(name: str):
|
||
if not qdrant.collection_exists(name):
|
||
qdrant.recreate_collection(
|
||
collection_name=name,
|
||
vectors_config=VectorParams(size=model.get_sentence_embedding_dimension(), distance=Distance.COSINE),
|
||
)
|
||
|
||
def _norm_list(xs: List[str]) -> List[str]:
|
||
seen, out = set(), []
|
||
for x in xs or []:
|
||
s = str(x).strip()
|
||
k = s.casefold()
|
||
if s and k not in seen:
|
||
seen.add(k)
|
||
out.append(s)
|
||
return sorted(out, key=str.casefold)
|
||
|
||
def _template_embed_text(tpl: PlanTemplate) -> str:
|
||
par
|