963 lines
33 KiB
Python
963 lines
33 KiB
Python
import json
|
|
from datetime import date, timedelta
|
|
from typing import Optional
|
|
from uuid import UUID, uuid4
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from google.genai import types as genai_types
|
|
from pydantic import BaseModel as PydanticBase
|
|
from sqlmodel import Field, Session, SQLModel, col, select
|
|
|
|
from db import get_session
|
|
from innercontext.api.utils import get_or_404
|
|
from innercontext.api.llm_context import build_user_profile_context
|
|
from innercontext.api.product_llm_tools import (
|
|
PRODUCT_DETAILS_FUNCTION_DECLARATION,
|
|
_extract_requested_product_ids as _shared_extract_requested_product_ids,
|
|
build_last_used_on_by_product,
|
|
build_product_details_tool_handler,
|
|
)
|
|
from innercontext.llm import (
|
|
call_gemini,
|
|
call_gemini_with_function_tools,
|
|
get_creative_config,
|
|
)
|
|
from innercontext.models import (
|
|
GroomingSchedule,
|
|
Product,
|
|
ProductInventory,
|
|
Routine,
|
|
RoutineStep,
|
|
SkinConditionSnapshot,
|
|
)
|
|
from innercontext.models.enums import GroomingAction, PartOfDay
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class RoutineCreate(SQLModel):
|
|
routine_date: date
|
|
part_of_day: PartOfDay
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class RoutineUpdate(SQLModel):
|
|
routine_date: Optional[date] = None
|
|
part_of_day: Optional[PartOfDay] = None
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class RoutineStepCreate(SQLModel):
|
|
product_id: Optional[UUID] = None
|
|
order_index: int
|
|
action_type: Optional[GroomingAction] = None
|
|
action_notes: Optional[str] = None
|
|
dose: Optional[str] = None
|
|
region: Optional[str] = None
|
|
|
|
|
|
class RoutineStepUpdate(SQLModel):
|
|
product_id: Optional[UUID] = None
|
|
order_index: Optional[int] = None
|
|
action_type: Optional[GroomingAction] = None
|
|
action_notes: Optional[str] = None
|
|
dose: Optional[str] = None
|
|
region: Optional[str] = None
|
|
|
|
|
|
class GroomingScheduleCreate(SQLModel):
|
|
day_of_week: int
|
|
action: GroomingAction
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class GroomingScheduleUpdate(SQLModel):
|
|
day_of_week: Optional[int] = None
|
|
action: Optional[GroomingAction] = None
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class SuggestedStep(SQLModel):
|
|
product_id: Optional[UUID] = None
|
|
action_type: Optional[GroomingAction] = None
|
|
action_notes: Optional[str] = None
|
|
region: Optional[str] = None
|
|
why_this_step: Optional[str] = None
|
|
optional: Optional[bool] = None
|
|
|
|
|
|
class SuggestRoutineRequest(SQLModel):
|
|
routine_date: date
|
|
part_of_day: PartOfDay
|
|
notes: Optional[str] = None
|
|
include_minoxidil_beard: bool = False
|
|
leaving_home: Optional[bool] = None
|
|
|
|
|
|
class RoutineSuggestionSummary(SQLModel):
|
|
primary_goal: str = ""
|
|
constraints_applied: list[str] = Field(default_factory=list)
|
|
confidence: float = 0.0
|
|
|
|
|
|
class RoutineSuggestion(SQLModel):
|
|
steps: list[SuggestedStep]
|
|
reasoning: str
|
|
summary: Optional[RoutineSuggestionSummary] = None
|
|
|
|
|
|
class SuggestBatchRequest(SQLModel):
|
|
from_date: date
|
|
to_date: date
|
|
notes: Optional[str] = None
|
|
include_minoxidil_beard: bool = False
|
|
minimize_products: Optional[bool] = None
|
|
|
|
|
|
class DayPlan(SQLModel):
|
|
date: date
|
|
am_steps: list[SuggestedStep]
|
|
pm_steps: list[SuggestedStep]
|
|
reasoning: str
|
|
|
|
|
|
class BatchSuggestion(SQLModel):
|
|
days: list[DayPlan]
|
|
overall_reasoning: str
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pydantic schemas for Gemini structured output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _SingleStepOut(PydanticBase):
|
|
product_id: Optional[str] = None
|
|
action_type: Optional[GroomingAction] = None
|
|
region: Optional[str] = None
|
|
action_notes: Optional[str] = None
|
|
why_this_step: Optional[str] = None
|
|
optional: Optional[bool] = None
|
|
|
|
|
|
class _BatchStepOut(PydanticBase):
|
|
product_id: Optional[str] = None
|
|
action_type: Optional[GroomingAction] = None
|
|
region: Optional[str] = None
|
|
action_notes: Optional[str] = None
|
|
|
|
|
|
class _SummaryOut(PydanticBase):
|
|
primary_goal: str
|
|
constraints_applied: list[str]
|
|
confidence: float
|
|
|
|
|
|
class _SuggestionOut(PydanticBase):
|
|
steps: list[_SingleStepOut]
|
|
reasoning: str
|
|
summary: _SummaryOut
|
|
|
|
|
|
class _DayPlanOut(PydanticBase):
|
|
date: str
|
|
am_steps: list[_BatchStepOut]
|
|
pm_steps: list[_BatchStepOut]
|
|
reasoning: str
|
|
|
|
|
|
class _BatchOut(PydanticBase):
|
|
days: list[_DayPlanOut]
|
|
overall_reasoning: str
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Prompt helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_DAY_NAMES = [
|
|
"poniedziałek",
|
|
"wtorek",
|
|
"środa",
|
|
"czwartek",
|
|
"piątek",
|
|
"sobota",
|
|
"niedziela",
|
|
]
|
|
|
|
|
|
def _contains_minoxidil_text(value: Optional[str]) -> bool:
|
|
if not value:
|
|
return False
|
|
text = value.lower()
|
|
return "minoxidil" in text or "minoksydyl" in text
|
|
|
|
|
|
def _is_minoxidil_product(product: Product) -> bool:
|
|
if _contains_minoxidil_text(product.name):
|
|
return True
|
|
if _contains_minoxidil_text(product.brand):
|
|
return True
|
|
if _contains_minoxidil_text(product.line_name):
|
|
return True
|
|
if any(_contains_minoxidil_text(i) for i in (product.inci or [])):
|
|
return True
|
|
|
|
actives = product.actives or []
|
|
for a in actives:
|
|
if isinstance(a, dict):
|
|
if _contains_minoxidil_text(str(a.get("name", ""))):
|
|
return True
|
|
continue
|
|
if _contains_minoxidil_text(a.name):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _ev(v: object) -> str:
|
|
if v is None:
|
|
return ""
|
|
value = getattr(v, "value", None)
|
|
if isinstance(value, str):
|
|
return value
|
|
return str(v)
|
|
|
|
|
|
def _build_skin_context(session: Session) -> str:
|
|
snapshot = session.exec(
|
|
select(SkinConditionSnapshot).order_by(
|
|
col(SkinConditionSnapshot.snapshot_date).desc()
|
|
)
|
|
).first()
|
|
if snapshot is None:
|
|
return "SKIN CONDITION: no data\n"
|
|
ev = _ev
|
|
return (
|
|
f"SKIN CONDITION (snapshot from {snapshot.snapshot_date}):\n"
|
|
f" Overall state: {ev(snapshot.overall_state)}\n"
|
|
f" Hydration: {snapshot.hydration_level}/5\n"
|
|
f" Barrier: {ev(snapshot.barrier_state)}\n"
|
|
f" Active concerns: {', '.join(ev(c) for c in (snapshot.active_concerns or []))}\n"
|
|
f" Priorities: {', '.join(snapshot.priorities or [])}\n"
|
|
f" Notes: {snapshot.notes or 'none'}\n"
|
|
)
|
|
|
|
|
|
def _build_grooming_context(
|
|
session: Session, weekdays: Optional[list[int]] = None
|
|
) -> str:
|
|
entries = session.exec(
|
|
select(GroomingSchedule).order_by(col(GroomingSchedule.day_of_week))
|
|
).all()
|
|
if not entries:
|
|
return "GROOMING SCHEDULE: none\n"
|
|
lines = ["GROOMING SCHEDULE:"]
|
|
for e in entries:
|
|
if weekdays is not None and e.day_of_week not in weekdays:
|
|
continue
|
|
day_name = (
|
|
_DAY_NAMES[e.day_of_week] if 0 <= e.day_of_week <= 6 else str(e.day_of_week)
|
|
)
|
|
lines.append(
|
|
f" {day_name}: {_ev(e.action)}" + (f" ({e.notes})" if e.notes else "")
|
|
)
|
|
if len(lines) == 1:
|
|
lines.append(" (no entries for specified days)")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _build_recent_history(session: Session) -> str:
|
|
cutoff = date.today() - timedelta(days=7)
|
|
routines = session.exec(
|
|
select(Routine)
|
|
.where(Routine.routine_date >= cutoff)
|
|
.order_by(col(Routine.routine_date).desc())
|
|
).all()
|
|
if not routines:
|
|
return "RECENT ROUTINES: none\n"
|
|
lines = ["RECENT ROUTINES:"]
|
|
for r in routines:
|
|
steps = session.exec(
|
|
select(RoutineStep)
|
|
.where(RoutineStep.routine_id == r.id)
|
|
.order_by(col(RoutineStep.order_index))
|
|
).all()
|
|
step_names = []
|
|
for s in steps:
|
|
if s.product_id:
|
|
p = session.get(Product, s.product_id)
|
|
if p:
|
|
short_id = str(p.id)[:8]
|
|
step_names.append(f"{_ev(p.category)} [{short_id}]")
|
|
else:
|
|
step_names.append(f"unknown [{str(s.product_id)[:8]}]")
|
|
elif s.action_type:
|
|
step_names.append(f"action: {_ev(s.action_type)}")
|
|
lines.append(
|
|
f" {r.routine_date} {_ev(r.part_of_day).upper()}: {', '.join(step_names)}"
|
|
)
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _build_products_context(
|
|
session: Session,
|
|
time_filter: Optional[str] = None,
|
|
reference_date: Optional[date] = None,
|
|
) -> str:
|
|
products = _get_available_products(session, time_filter=time_filter)
|
|
product_ids = [p.id for p in products]
|
|
inventory_rows = (
|
|
session.exec(
|
|
select(ProductInventory).where(
|
|
col(ProductInventory.product_id).in_(product_ids)
|
|
)
|
|
).all()
|
|
if product_ids
|
|
else []
|
|
)
|
|
inv_by_product: dict[UUID, list[ProductInventory]] = {}
|
|
for inv in inventory_rows:
|
|
inv_by_product.setdefault(inv.product_id, []).append(inv)
|
|
|
|
recent_usage_counts: dict[UUID, int] = {}
|
|
if reference_date is not None:
|
|
cutoff = reference_date - timedelta(days=7)
|
|
recent_usage = session.exec(
|
|
select(RoutineStep.product_id)
|
|
.join(Routine)
|
|
.where(col(Routine.routine_date) > cutoff)
|
|
.where(col(Routine.routine_date) <= reference_date)
|
|
).all()
|
|
for pid in recent_usage:
|
|
if pid:
|
|
recent_usage_counts[pid] = recent_usage_counts.get(pid, 0) + 1
|
|
|
|
lines = ["AVAILABLE PRODUCTS:"]
|
|
for p in products:
|
|
p.inventory = inv_by_product.get(p.id, [])
|
|
ctx = p.to_llm_context()
|
|
entry = (
|
|
f' - id={ctx["id"]} name="{ctx["name"]}" brand="{ctx["brand"]}"'
|
|
f" category={ctx.get('category', '')} recommended_time={ctx.get('recommended_time', '')}"
|
|
f" leave_on={ctx.get('leave_on', '')}"
|
|
f" targets={ctx.get('targets', [])}"
|
|
)
|
|
active_names = _extract_active_names(p)
|
|
if active_names:
|
|
entry += f" actives={active_names}"
|
|
|
|
active_inventory = [inv for inv in p.inventory if inv.finished_at is None]
|
|
open_inventory = [inv for inv in active_inventory if inv.is_opened]
|
|
sealed_inventory = [inv for inv in active_inventory if not inv.is_opened]
|
|
entry += (
|
|
" inventory_status={"
|
|
f"active:{len(active_inventory)},opened:{len(open_inventory)},sealed:{len(sealed_inventory)}"
|
|
"}"
|
|
)
|
|
if open_inventory:
|
|
expiry_dates = sorted(
|
|
inv.expiry_date.isoformat() for inv in open_inventory if inv.expiry_date
|
|
)
|
|
if expiry_dates:
|
|
entry += f" nearest_open_expiry={expiry_dates[0]}"
|
|
if p.pao_months is not None:
|
|
pao_deadlines = sorted(
|
|
(inv.opened_at + timedelta(days=30 * p.pao_months)).isoformat()
|
|
for inv in open_inventory
|
|
if inv.opened_at
|
|
)
|
|
if pao_deadlines:
|
|
entry += f" nearest_open_pao_deadline={pao_deadlines[0]}"
|
|
if p.pao_months is not None:
|
|
entry += f" pao_months={p.pao_months}"
|
|
profile = ctx.get("effect_profile", {})
|
|
if profile:
|
|
notable = {k: v for k, v in profile.items() if v and v > 0}
|
|
if notable:
|
|
entry += f" effects={notable}"
|
|
if ctx.get("context_rules"):
|
|
entry += f" context_rules={ctx['context_rules']}"
|
|
safety = ctx.get("safety") or {}
|
|
if isinstance(safety, dict):
|
|
not_safe = {k: v for k, v in safety.items() if v is False}
|
|
if not_safe:
|
|
entry += f" safety_alerts={not_safe}"
|
|
if ctx.get("min_interval_hours"):
|
|
entry += f" min_interval_hours={ctx['min_interval_hours']}"
|
|
if ctx.get("max_frequency_per_week"):
|
|
entry += f" max_frequency_per_week={ctx['max_frequency_per_week']}"
|
|
usage_count = recent_usage_counts.get(p.id, 0)
|
|
entry += f" used_in_last_7_days={usage_count}"
|
|
lines.append(entry)
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _get_available_products(
|
|
session: Session,
|
|
time_filter: Optional[str] = None,
|
|
) -> list[Product]:
|
|
stmt = select(Product).where(col(Product.is_tool).is_(False))
|
|
products = session.exec(stmt).all()
|
|
result: list[Product] = []
|
|
for p in products:
|
|
if p.is_medication and not _is_minoxidil_product(p):
|
|
continue
|
|
if time_filter and _ev(p.recommended_time) not in (time_filter, "both"):
|
|
continue
|
|
result.append(p)
|
|
return result
|
|
|
|
|
|
def _extract_active_names(product: Product) -> list[str]:
|
|
names: list[str] = []
|
|
for a in product.actives or []:
|
|
if isinstance(a, dict):
|
|
name = str(a.get("name") or "").strip()
|
|
else:
|
|
name = str(getattr(a, "name", "") or "").strip()
|
|
if not name:
|
|
continue
|
|
if name in names:
|
|
continue
|
|
names.append(name)
|
|
if len(names) >= 12:
|
|
break
|
|
return names
|
|
|
|
|
|
def _extract_requested_product_ids(
|
|
args: dict[str, object], max_ids: int = 8
|
|
) -> list[str]:
|
|
return _shared_extract_requested_product_ids(args, max_ids=max_ids)
|
|
|
|
|
|
def _build_objectives_context(include_minoxidil_beard: bool) -> str:
|
|
if include_minoxidil_beard:
|
|
return (
|
|
"USER OBJECTIVES:\n"
|
|
" - Priority: improve beard and mustache density\n"
|
|
" - If a product with minoxidil is available, include it adhering strictly to safety rules\n"
|
|
)
|
|
return ""
|
|
|
|
|
|
def _build_day_context(leaving_home: Optional[bool]) -> str:
|
|
if leaving_home is None:
|
|
return ""
|
|
val = "yes" if leaving_home else "no"
|
|
return f"DAY CONTEXT:\n Leaving home: {val}\n"
|
|
|
|
|
|
_ROUTINES_SYSTEM_PROMPT = """\
|
|
Jesteś ekspertem planowania pielęgnacji.
|
|
|
|
CEL:
|
|
Twórz realistyczne, bezpieczne i krótkie rutyny o wysokiej zgodności z danymi wejściowymi.
|
|
|
|
PRIORYTETY DECYZYJNE (od najwyższego):
|
|
1) Bezpieczeństwo (brak realnego ryzyka klinicznego)
|
|
2) Cel terapeutyczny użytkownika
|
|
3) Reguły częstotliwości i odstępów
|
|
4) Zarządzanie inwentarzem
|
|
5) Prostota
|
|
|
|
> Cel terapeutyczny oznacza maksymalizację realnego efektu klinicznego,
|
|
> nie tylko zgodność z deklarowanymi targetami produktu.
|
|
|
|
WYMAGANIA ODPOWIEDZI:
|
|
- Zwracaj wyłącznie poprawny JSON (bez markdown, bez komentarzy, bez preambuły).
|
|
- Trzymaj się dokładnie przekazanego schematu odpowiedzi.
|
|
- Nie używaj żadnych pól spoza schematu.
|
|
- Nie twórz produktów spoza listy wejściowej.
|
|
- Jeśli nie da się bezpiecznie dodać kroku, pomiń go zamiast zgadywać.
|
|
|
|
ZASADY PLANOWANIA:
|
|
- Kolejność warstw: cleanser -> toner -> essence -> serum -> moisturizer -> [SPF dla AM].
|
|
- Respektuj: context_rules, min_interval_hours, max_frequency_per_week.
|
|
- Zarządzanie inwentarzem:
|
|
- Preferuj produkty już otwarte (miękka preferencja).
|
|
- Unikaj funkcjonalnej redundancji (np. wielokrotne źródła panthenolu, ceramidów lub niacynamidu w tej samej rutynie),
|
|
chyba że istnieje wyraźne uzasadnienie terapeutyczne.
|
|
- Maksymalnie 2 serum w rutynie.
|
|
Jeśli 2: jedno jako główny aktywny bodziec, drugie wyłącznie wspierające.
|
|
- Dla mildly_compromised nie eliminuj automatycznie umiarkowanych aktywnych;
|
|
decyzję opieraj na effect_profile (irritation_risk, barrier_disruption_risk) i regułach bezpieczeństwa.
|
|
- Nie zwiększaj intensywności terapii (retinoid/kwasy) dzień po dniu,
|
|
jeśli brak wyraźnej poprawy stanu skóry lub brak wskazań klinicznych.
|
|
- Nie łącz retinoidów i kwasów w tej samej rutynie ani tego samego dnia (dla planu wielodniowego).
|
|
- W AM zawsze uwzględnij SPF, jeśli kompatybilny produkt SPF istnieje na liście.
|
|
Wybór filtra (na podstawie KONTEKST DNIA):
|
|
- "Wyjście z domu: tak" → najwyższy współczynnik dostępny w nazwie (SPF50+, SPF50, SPF30);
|
|
- "Wyjście z domu: nie" → SPF30 wystarczy; wyższy dopuszczalny jeśli brak SPF30;
|
|
- brak KONTEKST DNIA → wybierz najwyższy dostępny.
|
|
- Dla minoksydylu (jeśli celem jest zarost i produkt jest dostępny): ustaw adekwatny region
|
|
broda/wąsy i nie naruszaj ograniczeń bezpieczeństwa.
|
|
- Preferuj 4-7 kroków na pojedynczą rutynę; unikaj zbędnych duplikatów aktywnych.
|
|
- Jeśli krok to produkt: podaj poprawny UUID z listy.
|
|
- Jeśli krok to czynność pielęgnacyjna: product_id = null. Dozwolone akcje są ściśle określone w schemacie (action_type).
|
|
- Nie zwracaj "pustych" kroków: każdy krok musi mieć product_id albo action_type.
|
|
- Pole region uzupełniaj tylko gdy ma znaczenie kliniczne/praktyczne (np. broda, wąsy, okolica oczu, szyja).
|
|
Dla standardowych kroków pielęgnacji całej twarzy pozostaw region puste.
|
|
- Nie podawaj dawek ani ilości produktu (np. "1 pompa", "2 krople", "pea-size").
|
|
|
|
JAK ROZWIĄZYWAĆ KONFLIKTY:
|
|
- Bezpieczeństwo > wszystko.
|
|
- Jeśli MODE=travel: logistyka podróży > różnorodność terapeutyczna.
|
|
- W MODE=travel odejdź od minimalizacji produktów tylko gdy wymaga tego bezpieczeństwo
|
|
lub bez dodatkowego produktu nie da się osiągnąć głównego celu terapeutycznego.
|
|
- Jeśli MODE=standard i bezpieczeństwo jest zachowane, preferuj różnorodność terapeutyczną.
|
|
- Przy niepełnych danych wybierz wariant konserwatywny.
|
|
"""
|
|
|
|
|
|
_ROUTINES_SINGLE_EXTRA = """\
|
|
DODATKOWE WYMAGANIA DLA TRYBU JEDNEJ RUTYNY:
|
|
- Każdy krok powinien mieć zwięzłe why_this_step (maks. jedno zdanie).
|
|
- Pole optional ustawiaj na true tylko dla kroków niekrytycznych.
|
|
- Uzupełnij summary:
|
|
- primary_goal: główny cel tej rutyny,
|
|
- constraints_applied: lista kluczowych ograniczeń zastosowanych przy planowaniu,
|
|
- confidence: liczba 0-1.
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routine routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("")
|
|
def list_routines(
|
|
from_date: Optional[date] = None,
|
|
to_date: Optional[date] = None,
|
|
part_of_day: Optional[PartOfDay] = None,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
stmt = select(Routine)
|
|
if from_date is not None:
|
|
stmt = stmt.where(Routine.routine_date >= from_date)
|
|
if to_date is not None:
|
|
stmt = stmt.where(Routine.routine_date <= to_date)
|
|
if part_of_day is not None:
|
|
stmt = stmt.where(Routine.part_of_day == part_of_day)
|
|
routines = session.exec(stmt).all()
|
|
|
|
routine_ids = [r.id for r in routines]
|
|
steps_by_routine: dict = {}
|
|
if routine_ids:
|
|
all_steps = session.exec(
|
|
select(RoutineStep).where(col(RoutineStep.routine_id).in_(routine_ids))
|
|
).all()
|
|
for step in all_steps:
|
|
steps_by_routine.setdefault(step.routine_id, []).append(step)
|
|
|
|
result = []
|
|
for r in routines:
|
|
data = r.model_dump(mode="json")
|
|
data["steps"] = [
|
|
s.model_dump(mode="json") for s in steps_by_routine.get(r.id, [])
|
|
]
|
|
result.append(data)
|
|
return result
|
|
|
|
|
|
@router.post("", response_model=Routine, status_code=201)
|
|
def create_routine(data: RoutineCreate, session: Session = Depends(get_session)):
|
|
routine = Routine(id=uuid4(), **data.model_dump())
|
|
session.add(routine)
|
|
session.commit()
|
|
session.refresh(routine)
|
|
return routine
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AI suggestion endpoints (must appear BEFORE /{routine_id})
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/suggest", response_model=RoutineSuggestion)
|
|
def suggest_routine(
|
|
data: SuggestRoutineRequest,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
weekday = data.routine_date.weekday()
|
|
skin_ctx = _build_skin_context(session)
|
|
profile_ctx = build_user_profile_context(session, reference_date=data.routine_date)
|
|
grooming_ctx = _build_grooming_context(session, weekdays=[weekday])
|
|
history_ctx = _build_recent_history(session)
|
|
day_ctx = _build_day_context(data.leaving_home)
|
|
products_ctx = _build_products_context(
|
|
session, time_filter=data.part_of_day.value, reference_date=data.routine_date
|
|
)
|
|
available_products = _get_available_products(
|
|
session,
|
|
time_filter=data.part_of_day.value,
|
|
)
|
|
last_used_on_by_product = build_last_used_on_by_product(
|
|
session,
|
|
product_ids=[p.id for p in available_products],
|
|
)
|
|
objectives_ctx = _build_objectives_context(data.include_minoxidil_beard)
|
|
|
|
mode_line = "MODE: standard"
|
|
notes_line = f"USER CONTEXT: {data.notes}\n" if data.notes else ""
|
|
day_name = _DAY_NAMES[weekday]
|
|
|
|
prompt = (
|
|
f"Zaproponuj rutynę pielęgnacyjną {data.part_of_day.value.upper()} "
|
|
f"na {data.routine_date} ({day_name}).\n\n"
|
|
f"{mode_line}\n"
|
|
"INPUT DATA:\n"
|
|
f"{profile_ctx}\n{skin_ctx}\n{grooming_ctx}\n{history_ctx}\n{day_ctx}\n{products_ctx}\n{objectives_ctx}"
|
|
"\nNARZEDZIA:\n"
|
|
"- Masz dostep do funkcji: get_product_details.\n"
|
|
"- Wywoluj narzedzia tylko, gdy potrzebujesz detali do decyzji klinicznej/bezpieczenstwa.\n"
|
|
"- Staraj sie grupowac zapytania: podawaj wszystkie potrzebne UUID w jednym wywolaniu narzedzia.\n"
|
|
"- Nie zgaduj detali skladu i zasad bezpieczenstwa; jesli potrzebujesz szczegolow, wywolaj odpowiednie narzedzie.\n"
|
|
f"{notes_line}"
|
|
f"{_ROUTINES_SINGLE_EXTRA}\n"
|
|
"Zwróć JSON zgodny ze schematem."
|
|
)
|
|
|
|
config = get_creative_config(
|
|
system_instruction=_ROUTINES_SYSTEM_PROMPT,
|
|
response_schema=_SuggestionOut,
|
|
max_output_tokens=4096,
|
|
).model_copy(
|
|
update={
|
|
"tools": [
|
|
genai_types.Tool(
|
|
function_declarations=[
|
|
PRODUCT_DETAILS_FUNCTION_DECLARATION,
|
|
],
|
|
)
|
|
],
|
|
"tool_config": genai_types.ToolConfig(
|
|
function_calling_config=genai_types.FunctionCallingConfig(
|
|
mode=genai_types.FunctionCallingConfigMode.AUTO,
|
|
)
|
|
),
|
|
}
|
|
)
|
|
|
|
function_handlers = {
|
|
"get_product_details": build_product_details_tool_handler(
|
|
available_products,
|
|
last_used_on_by_product=last_used_on_by_product,
|
|
),
|
|
}
|
|
|
|
try:
|
|
response = call_gemini_with_function_tools(
|
|
endpoint="routines/suggest",
|
|
contents=prompt,
|
|
config=config,
|
|
function_handlers=function_handlers,
|
|
user_input=prompt,
|
|
max_tool_roundtrips=3,
|
|
)
|
|
except HTTPException as exc:
|
|
if (
|
|
exc.status_code != 502
|
|
or str(exc.detail) != "Gemini requested too many function calls"
|
|
):
|
|
raise
|
|
|
|
conservative_prompt = (
|
|
f"{prompt}\n\n"
|
|
"TRYB AWARYJNY (KONSERWATYWNY):\n"
|
|
"- Osiagnieto limit wywolan narzedzi.\n"
|
|
"- Nie wywoluj narzedzi ponownie.\n"
|
|
"- Zaproponuj maksymalnie konserwatywna, bezpieczna rutyne na podstawie dostepnych juz danych,"
|
|
" preferujac lagodne produkty wspierajace bariere i fotoprotekcje.\n"
|
|
"- Gdy masz watpliwosci, pomijaj ryzykowne aktywne kroki.\n"
|
|
)
|
|
response = call_gemini(
|
|
endpoint="routines/suggest",
|
|
contents=conservative_prompt,
|
|
config=get_creative_config(
|
|
system_instruction=_ROUTINES_SYSTEM_PROMPT,
|
|
response_schema=_SuggestionOut,
|
|
max_output_tokens=4096,
|
|
),
|
|
user_input=conservative_prompt,
|
|
tool_trace={
|
|
"mode": "fallback_conservative",
|
|
"reason": "max_tool_roundtrips_exceeded",
|
|
},
|
|
)
|
|
|
|
raw = response.text
|
|
if not raw:
|
|
raise HTTPException(status_code=502, detail="LLM returned an empty response")
|
|
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except json.JSONDecodeError as e:
|
|
raise HTTPException(status_code=502, detail=f"LLM returned invalid JSON: {e}")
|
|
|
|
steps = [
|
|
SuggestedStep(
|
|
product_id=UUID(s["product_id"]) if s.get("product_id") else None,
|
|
action_type=s.get("action_type") or None,
|
|
action_notes=s.get("action_notes"),
|
|
region=s.get("region"),
|
|
why_this_step=s.get("why_this_step"),
|
|
optional=s.get("optional"),
|
|
)
|
|
for s in parsed.get("steps", [])
|
|
]
|
|
|
|
summary_raw = parsed.get("summary") or {}
|
|
confidence_raw = summary_raw.get("confidence", 0)
|
|
try:
|
|
confidence = float(confidence_raw)
|
|
except (TypeError, ValueError):
|
|
confidence = 0.0
|
|
confidence = max(0.0, min(1.0, confidence))
|
|
constraints_applied = summary_raw.get("constraints_applied") or []
|
|
if not isinstance(constraints_applied, list):
|
|
constraints_applied = []
|
|
|
|
summary = RoutineSuggestionSummary(
|
|
primary_goal=str(summary_raw.get("primary_goal") or ""),
|
|
constraints_applied=[str(x) for x in constraints_applied],
|
|
confidence=confidence,
|
|
)
|
|
|
|
return RoutineSuggestion(
|
|
steps=steps,
|
|
reasoning=parsed.get("reasoning", ""),
|
|
summary=summary,
|
|
)
|
|
|
|
|
|
@router.post("/suggest-batch", response_model=BatchSuggestion)
|
|
def suggest_batch(
|
|
data: SuggestBatchRequest,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
delta = (data.to_date - data.from_date).days + 1
|
|
if delta > 14:
|
|
raise HTTPException(
|
|
status_code=400, detail="Date range must not exceed 14 days."
|
|
)
|
|
if data.from_date > data.to_date:
|
|
raise HTTPException(status_code=400, detail="from_date must be <= to_date.")
|
|
|
|
weekdays = list(
|
|
{(data.from_date + timedelta(days=i)).weekday() for i in range(delta)}
|
|
)
|
|
profile_ctx = build_user_profile_context(session, reference_date=data.from_date)
|
|
skin_ctx = _build_skin_context(session)
|
|
grooming_ctx = _build_grooming_context(session, weekdays=weekdays)
|
|
history_ctx = _build_recent_history(session)
|
|
products_ctx = _build_products_context(session, reference_date=data.from_date)
|
|
objectives_ctx = _build_objectives_context(data.include_minoxidil_beard)
|
|
|
|
date_range_lines = []
|
|
for i in range(delta):
|
|
d = data.from_date + timedelta(days=i)
|
|
date_range_lines.append(f" {d} ({_DAY_NAMES[d.weekday()]})")
|
|
dates_str = "\n".join(date_range_lines)
|
|
|
|
notes_line = f"USER CONTEXT: {data.notes}\n" if data.notes else ""
|
|
mode_line = "MODE: travel" if data.minimize_products else "MODE: standard"
|
|
minimize_line = (
|
|
"\nCONSTRAINTS (TRAVEL MODE):\n"
|
|
"- To tryb podróżny: minimalizuj liczbę unikalnych produktów w całym planie (wszystkie dni, AM+PM).\n"
|
|
"- Preferuj reużycie tych samych produktów między dniami, jeśli bezpieczeństwo i główny cel terapeutyczny są zachowane.\n"
|
|
"- Dodaj nowy produkt tylko gdy to konieczne dla bezpieczeństwa albo realizacji głównego celu terapeutycznego.\n"
|
|
if data.minimize_products
|
|
else ""
|
|
)
|
|
|
|
prompt = (
|
|
f"Zaproponuj plan pielęgnacji AM + PM dla każdego dnia z zakresu:\n{dates_str}\n\n{mode_line}\n"
|
|
"INPUT DATA:\n"
|
|
f"{profile_ctx}\n{skin_ctx}\n{grooming_ctx}\n{history_ctx}\n{products_ctx}\n{objectives_ctx}"
|
|
f"{notes_line}{minimize_line}"
|
|
"\nZwróć JSON zgodny ze schematem."
|
|
)
|
|
|
|
response = call_gemini(
|
|
endpoint="routines/suggest-batch",
|
|
contents=prompt,
|
|
config=get_creative_config(
|
|
system_instruction=_ROUTINES_SYSTEM_PROMPT,
|
|
response_schema=_BatchOut,
|
|
max_output_tokens=8192,
|
|
),
|
|
user_input=prompt,
|
|
)
|
|
|
|
raw = response.text
|
|
if not raw:
|
|
raise HTTPException(status_code=502, detail="LLM returned an empty response")
|
|
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except json.JSONDecodeError as e:
|
|
raise HTTPException(status_code=502, detail=f"LLM returned invalid JSON: {e}")
|
|
|
|
def _parse_steps(raw_steps: list) -> list[SuggestedStep]:
|
|
result = []
|
|
for s in raw_steps:
|
|
result.append(
|
|
SuggestedStep(
|
|
product_id=UUID(s["product_id"]) if s.get("product_id") else None,
|
|
action_type=s.get("action_type") or None,
|
|
action_notes=s.get("action_notes"),
|
|
region=s.get("region"),
|
|
why_this_step=s.get("why_this_step"),
|
|
optional=s.get("optional"),
|
|
)
|
|
)
|
|
return result
|
|
|
|
days = []
|
|
for day_raw in parsed.get("days", []):
|
|
try:
|
|
day_date = date.fromisoformat(day_raw["date"])
|
|
except (KeyError, ValueError):
|
|
continue
|
|
days.append(
|
|
DayPlan(
|
|
date=day_date,
|
|
am_steps=_parse_steps(day_raw.get("am_steps", [])),
|
|
pm_steps=_parse_steps(day_raw.get("pm_steps", [])),
|
|
reasoning=day_raw.get("reasoning", ""),
|
|
)
|
|
)
|
|
|
|
return BatchSuggestion(
|
|
days=days, overall_reasoning=parsed.get("overall_reasoning", "")
|
|
)
|
|
|
|
|
|
# Grooming-schedule GET must appear before /{routine_id} to avoid being shadowed
|
|
@router.get("/grooming-schedule", response_model=list[GroomingSchedule])
|
|
def list_grooming_schedule(session: Session = Depends(get_session)):
|
|
return session.exec(select(GroomingSchedule)).all()
|
|
|
|
|
|
@router.get("/{routine_id}")
|
|
def get_routine(routine_id: UUID, session: Session = Depends(get_session)):
|
|
routine = get_or_404(session, Routine, routine_id)
|
|
steps = session.exec(
|
|
select(RoutineStep).where(RoutineStep.routine_id == routine_id)
|
|
).all()
|
|
data = routine.model_dump(mode="json")
|
|
data["steps"] = [step.model_dump(mode="json") for step in steps]
|
|
return data
|
|
|
|
|
|
@router.patch("/{routine_id}", response_model=Routine)
|
|
def update_routine(
|
|
routine_id: UUID,
|
|
data: RoutineUpdate,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
routine = get_or_404(session, Routine, routine_id)
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(routine, key, value)
|
|
session.add(routine)
|
|
session.commit()
|
|
session.refresh(routine)
|
|
return routine
|
|
|
|
|
|
@router.delete("/{routine_id}", status_code=204)
|
|
def delete_routine(routine_id: UUID, session: Session = Depends(get_session)):
|
|
routine = get_or_404(session, Routine, routine_id)
|
|
session.delete(routine)
|
|
session.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RoutineStep sub-routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/{routine_id}/steps", response_model=RoutineStep, status_code=201)
|
|
def add_step(
|
|
routine_id: UUID,
|
|
data: RoutineStepCreate,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
get_or_404(session, Routine, routine_id)
|
|
step = RoutineStep(id=uuid4(), routine_id=routine_id, **data.model_dump())
|
|
session.add(step)
|
|
session.commit()
|
|
session.refresh(step)
|
|
return step
|
|
|
|
|
|
@router.patch("/steps/{step_id}", response_model=RoutineStep)
|
|
def update_step(
|
|
step_id: UUID,
|
|
data: RoutineStepUpdate,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
step = get_or_404(session, RoutineStep, step_id)
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(step, key, value)
|
|
session.add(step)
|
|
session.commit()
|
|
session.refresh(step)
|
|
return step
|
|
|
|
|
|
@router.delete("/steps/{step_id}", status_code=204)
|
|
def delete_step(step_id: UUID, session: Session = Depends(get_session)):
|
|
step = get_or_404(session, RoutineStep, step_id)
|
|
session.delete(step)
|
|
session.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GroomingSchedule routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/grooming-schedule", response_model=GroomingSchedule, status_code=201)
|
|
def create_grooming_schedule(
|
|
data: GroomingScheduleCreate, session: Session = Depends(get_session)
|
|
):
|
|
entry = GroomingSchedule(id=uuid4(), **data.model_dump())
|
|
session.add(entry)
|
|
session.commit()
|
|
session.refresh(entry)
|
|
return entry
|
|
|
|
|
|
@router.patch("/grooming-schedule/{entry_id}", response_model=GroomingSchedule)
|
|
def update_grooming_schedule(
|
|
entry_id: UUID,
|
|
data: GroomingScheduleUpdate,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
entry = get_or_404(session, GroomingSchedule, entry_id)
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(entry, key, value)
|
|
session.add(entry)
|
|
session.commit()
|
|
session.refresh(entry)
|
|
return entry
|
|
|
|
|
|
@router.delete("/grooming-schedule/{entry_id}", status_code=204)
|
|
def delete_grooming_schedule(entry_id: UUID, session: Session = Depends(get_session)):
|
|
entry = get_or_404(session, GroomingSchedule, entry_id)
|
|
session.delete(entry)
|
|
session.commit()
|