innercontext/backend/innercontext/api/skincare.py

import json
from datetime import date
from typing import Optional
from uuid import UUID, uuid4

from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
from google.genai import types as genai_types
from pydantic import ValidationError
from sqlmodel import Session, SQLModel, select

from db import get_session
from innercontext.api.utils import get_or_404
from innercontext.llm import get_gemini_client
from innercontext.models import (
    SkinConditionSnapshot,
    SkinConditionSnapshotBase,
    SkinConditionSnapshotPublic,
)
from innercontext.models.enums import (
    BarrierState,
    OverallSkinState,
    SkinConcern,
    SkinTexture,
    SkinType,
)

router = APIRouter()


# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------


class SnapshotCreate(SkinConditionSnapshotBase):
    pass


class SnapshotUpdate(SQLModel):
    snapshot_date: Optional[date] = None
    overall_state: Optional[OverallSkinState] = None
    skin_type: Optional[SkinType] = None
    texture: Optional[SkinTexture] = None

    hydration_level: Optional[int] = None
    sebum_tzone: Optional[int] = None
    sebum_cheeks: Optional[int] = None
    sensitivity_level: Optional[int] = None

    barrier_state: Optional[BarrierState] = None

    active_concerns: Optional[list[SkinConcern]] = None
    risks: Optional[list[str]] = None
    priorities: Optional[list[str]] = None
    notes: Optional[str] = None


class SkinPhotoAnalysisResponse(SQLModel):
    overall_state: Optional[OverallSkinState] = None
    skin_type: Optional[SkinType] = None
    texture: Optional[SkinTexture] = None
    hydration_level: Optional[int] = None
    sebum_tzone: Optional[int] = None
    sebum_cheeks: Optional[int] = None
    sensitivity_level: Optional[int] = None
    barrier_state: Optional[BarrierState] = None
    active_concerns: Optional[list[SkinConcern]] = None
    risks: Optional[list[str]] = None
    priorities: Optional[list[str]] = None
    notes: Optional[str] = None


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _skin_photo_system_prompt() -> str:
    return """\
You are a dermatology-trained skin assessment AI. Analyze the provided photo(s) of a person's
skin and return a structured JSON assessment.

RULES:
- Return ONLY raw JSON — no markdown fences, no explanation.
- Omit any field you cannot confidently determine from the photos. Do not guess.
- All enum values must exactly match the allowed strings listed below.
- Numeric metrics use a 1–5 scale (1 = minimal, 5 = maximal).
- risks and priorities: short English phrases, max 10 words each.
- notes: 2–4 sentence paragraph describing key observations.

ENUM VALUES:
overall_state: "excellent" | "good" | "fair" | "poor"
skin_type: "dry" | "oily" | "combination" | "sensitive" | "normal" | "acne_prone"
texture: "smooth" | "rough" | "flaky" | "bumpy"
barrier_state: "intact" | "mildly_compromised" | "compromised"
active_concerns: "acne" | "rosacea" | "hyperpigmentation" | "aging" | "dehydration" |
  "redness" | "damaged_barrier" | "pore_visibility" | "uneven_texture" | "sebum_excess"

METRICS (int 1–5, omit if not assessable):
hydration_level: 1=very dehydrated/dull → 5=plump/luminous
sebum_tzone: 1=very dry T-zone → 5=very oily T-zone
sebum_cheeks: 1=very dry cheeks → 5=very oily cheeks
sensitivity_level: 1=no visible signs → 5=severe redness/reactivity

OUTPUT (all fields optional):
{"overall_state":…, "skin_type":…, "texture":…, "hydration_level":…,
 "sebum_tzone":…, "sebum_cheeks":…, "sensitivity_level":…,
 "barrier_state":…, "active_concerns":[…], "risks":[…], "priorities":[…], "notes":…}
"""


# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------


MAX_IMAGE_BYTES = 5 * 1024 * 1024  # 5 MB


@router.post("/analyze-photos", response_model=SkinPhotoAnalysisResponse)
async def analyze_skin_photos(
    photos: list[UploadFile] = File(...),
) -> SkinPhotoAnalysisResponse:
    if not (1 <= len(photos) <= 3):
        raise HTTPException(status_code=422, detail="Send between 1 and 3 photos.")

    client, model = get_gemini_client()

    allowed = {"image/jpeg", "image/png", "image/webp"}
    parts: list[genai_types.Part] = []
    for photo in photos:
        if photo.content_type not in allowed:
            raise HTTPException(status_code=422, detail=f"Unsupported type: {photo.content_type}")
        data = await photo.read()
        if len(data) > MAX_IMAGE_BYTES:
            raise HTTPException(status_code=413, detail=f"{photo.filename} exceeds 5 MB.")
        parts.append(genai_types.Part.from_bytes(data=data, mime_type=photo.content_type))
    parts.append(
        genai_types.Part.from_text(
            text="Analyze the skin condition visible in the above photo(s) and return the JSON assessment."
        )
    )

    try:
        response = client.models.generate_content(
            model=model,
            contents=parts,
            config=genai_types.GenerateContentConfig(
                system_instruction=_skin_photo_system_prompt(),
                response_mime_type="application/json",
                max_output_tokens=2048,
                temperature=0.0,
            ),
        )
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Gemini API error: {e}")

    try:
        parsed = json.loads(response.text)
    except json.JSONDecodeError as e:
        raise HTTPException(status_code=502, detail=f"LLM returned invalid JSON: {e}")

    try:
        return SkinPhotoAnalysisResponse.model_validate(parsed)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())


@router.get("", response_model=list[SkinConditionSnapshotPublic])
def list_snapshots(
    from_date: Optional[date] = None,
    to_date: Optional[date] = None,
    overall_state: Optional[OverallSkinState] = None,
    session: Session = Depends(get_session),
):
    stmt = select(SkinConditionSnapshot)
    if from_date is not None:
        stmt = stmt.where(SkinConditionSnapshot.snapshot_date >= from_date)
    if to_date is not None:
        stmt = stmt.where(SkinConditionSnapshot.snapshot_date <= to_date)
    if overall_state is not None:
        stmt = stmt.where(SkinConditionSnapshot.overall_state == overall_state)
    return session.exec(stmt).all()


@router.post("", response_model=SkinConditionSnapshotPublic, status_code=201)
def create_snapshot(data: SnapshotCreate, session: Session = Depends(get_session)):
    snapshot = SkinConditionSnapshot(id=uuid4(), **data.model_dump())
    session.add(snapshot)
    session.commit()
    session.refresh(snapshot)
    return snapshot


@router.get("/{snapshot_id}", response_model=SkinConditionSnapshotPublic)
def get_snapshot(snapshot_id: UUID, session: Session = Depends(get_session)):
    return get_or_404(session, SkinConditionSnapshot, snapshot_id)


@router.patch("/{snapshot_id}", response_model=SkinConditionSnapshotPublic)
def update_snapshot(
    snapshot_id: UUID,
    data: SnapshotUpdate,
    session: Session = Depends(get_session),
):
    snapshot = get_or_404(session, SkinConditionSnapshot, snapshot_id)
    for key, value in data.model_dump(exclude_unset=True).items():
        setattr(snapshot, key, value)
    session.add(snapshot)
    session.commit()
    session.refresh(snapshot)
    return snapshot


@router.delete("/{snapshot_id}", status_code=204)
def delete_snapshot(snapshot_id: UUID, session: Session = Depends(get_session)):
    snapshot = get_or_404(session, SkinConditionSnapshot, snapshot_id)
    session.delete(snapshot)
    session.commit()