diff --git a/backend/innercontext/api/products.py b/backend/innercontext/api/products.py index 24ada36..ac99681 100644 --- a/backend/innercontext/api/products.py +++ b/backend/innercontext/api/products.py @@ -147,15 +147,15 @@ class ProductParseResponse(SQLModel): class AIActiveIngredient(ActiveIngredient): # Gemini API rejects int-enum values in response_schema; override with plain int. - strength_level: Optional[int] = None - irritation_potential: Optional[int] = None + strength_level: Optional[int] = None # type: ignore[assignment] + irritation_potential: Optional[int] = None # type: ignore[assignment] class ProductParseLLMResponse(ProductParseResponse): # Gemini response schema currently requires enum values to be strings. # Strength fields are numeric in our domain (1-3), so keep them as ints here # and convert via ProductParseResponse validation afterward. - actives: Optional[list[AIActiveIngredient]] = None + actives: Optional[list[AIActiveIngredient]] = None # type: ignore[assignment] class InventoryCreate(SQLModel): @@ -571,9 +571,30 @@ def _build_shopping_context(session: Session) -> str: active_inv = [i for i in inv_by_product.get(p.id, []) if i.finished_at is None] has_stock = len(active_inv) > 0 # any unfinished inventory = in stock stock = "✓" if has_stock else "✗" + + actives = [] + for a in p.actives or []: + name = a.get("name") if isinstance(a, dict) else getattr(a, "name", None) + if name: + actives.append(name) + actives_str = f", actives: {actives}" if actives else "" + + ep = p.product_effect_profile + if isinstance(ep, dict): + effects = {k.replace("_strength", ""): v for k, v in ep.items() if v >= 3} + else: + effects = { + k.replace("_strength", ""): v + for k, v in ep.model_dump().items() + if v >= 3 + } + effects_str = f", effects: {effects}" if effects else "" + + targets = [_ev(t) for t in (p.targets or [])] + products_lines.append( f" [{stock}] {p.name} ({p.brand or ''}) - {_ev(p.category)}, " - f"targets: {p.targets or []}" + f"targets: {targets}{actives_str}{effects_str}" ) return "\n".join(skin_lines) + "\n\n" + "\n".join(products_lines) @@ -596,7 +617,8 @@ ZASADY: 5. Sugeruj realistyczną częstotliwość użycia (dzienna, 2-3x tygodniowo, etc.) 6. Zachowaj kolejność warstw: cleanse → toner → serum → moisturizer → SPF 7. Jeśli użytkownik ma uszkodzoną barierę, unikaj silnych eksfoliantów i retinoidów -8. Odpowiadaj w języku polskim +8. Zwracaj uwagę na ewentualne konflikty polecanych składników z tymi, które użytkownik już posiada (np. nie polecaj peptydów miedziowych jeśli użytkownik nadużywa kwasów) +9. Odpowiadaj w języku polskim Format odpowiedzi - zwróć wyłącznie JSON zgodny z podanym schematem.""" diff --git a/backend/innercontext/api/skincare.py b/backend/innercontext/api/skincare.py index edf11dd..31407db 100644 --- a/backend/innercontext/api/skincare.py +++ b/backend/innercontext/api/skincare.py @@ -167,9 +167,7 @@ async def analyze_skin_photos( ) ) - image_summary = ( - f"{len(photos)} image(s): {', '.join((p.content_type or 'unknown') for p in photos)}" - ) + image_summary = f"{len(photos)} image(s): {', '.join((p.content_type or 'unknown') for p in photos)}" response = call_gemini( endpoint="skincare/analyze-photos", contents=parts, diff --git a/backend/innercontext/mcp_server.py b/backend/innercontext/mcp_server.py index d2c1235..4de7c79 100644 --- a/backend/innercontext/mcp_server.py +++ b/backend/innercontext/mcp_server.py @@ -1,20 +1,16 @@ from __future__ import annotations -import json import os from datetime import date, datetime, timedelta from typing import Optional from uuid import UUID import httpx -from fastapi import HTTPException from fastmcp import FastMCP -from google.genai import types as genai_types from pydantic import BaseModel from sqlmodel import Session, col, select from db import engine -from innercontext.llm import call_gemini from innercontext.models import ( GroomingSchedule, LabResult, @@ -32,21 +28,6 @@ from innercontext.models import ( # --------------------------------------------------------------------------- -class ProductSuggestionOut(BaseModel): - category: str - product_type: str - key_ingredients: list[str] - target_concerns: list[str] - why_needed: str - recommended_time: str - frequency: str - - -class ShoppingSuggestionsOut(BaseModel): - suggestions: list[ProductSuggestionOut] - reasoning: str - - class MarketProduct(BaseModel): name: str brand: str @@ -477,149 +458,6 @@ def get_lab_results_for_test(test_code: str) -> list[dict]: # ── Shopping assistant ──────────────────────────────────────────────────────── -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_shopping_context(session: Session) -> tuple[str, list[dict]]: - snapshot = session.exec( - select(SkinConditionSnapshot).order_by( - col(SkinConditionSnapshot.snapshot_date).desc() - ) - ).first() - - skin_lines = ["STAN SKÓRY:"] - if snapshot: - skin_lines.append(f" Data: {snapshot.snapshot_date}") - skin_lines.append(f" Ogólny stan: {_ev(snapshot.overall_state)}") - skin_lines.append(f" Typ skóry: {_ev(snapshot.skin_type)}") - skin_lines.append(f" Nawilżenie: {snapshot.hydration_level}/5") - skin_lines.append(f" Wrażliwość: {snapshot.sensitivity_level}/5") - skin_lines.append(f" Bariera: {_ev(snapshot.barrier_state)}") - concerns = ", ".join(_ev(c) for c in (snapshot.active_concerns or [])) - skin_lines.append(f" Aktywne problemy: {concerns or 'brak'}") - if snapshot.priorities: - skin_lines.append(f" Priorytety: {', '.join(snapshot.priorities)}") - else: - skin_lines.append(" (brak danych)") - - stmt = select(Product).where(col(Product.is_tool).is_(False)) - products = session.exec(stmt).all() - - 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) - - products_data = [] - for p in products: - if p.is_medication: - continue - active_inv = [i for i in inv_by_product.get(p.id, []) if i.finished_at is None] - has_stock = any(i.is_opened and i.finished_at is None for i in active_inv) - products_data.append( - { - "id": str(p.id), - "name": p.name, - "brand": p.brand or "", - "category": _ev(p.category), - "targets": p.targets or [], - "actives": [ - { - "name": a.get("name") if isinstance(a, dict) else a.name, - "percent": a.get("percent") if isinstance(a, dict) else None, - } - for a in (p.actives or []) - ], - "has_inventory": has_stock, - } - ) - - products_lines = ["POSIADANE PRODUKTY:"] - for p in products_data: - stock = "✓" if p["has_inventory"] else "✗" - products_lines.append( - f" [{stock}] {p['name']} ({p['brand']}) - {p['category']}, " - f"targets: {p['targets']}" - ) - - return "\n".join(skin_lines) + "\n\n" + "\n".join(products_lines), products_data - - -_SHOPPING_SYSTEM_PROMPT = """Jesteś asystentem zakupowym w dziedzinie pielęgnacji skóry. -Twoim zadaniem jest przeanalizować stan skóry użytkownika oraz produkty, które już posiada, -a następnie zasugerować TYPY produktów (bez marek), które mogłyby uzupełnić ich rutynę. - -ZASADY: -0. Sugeruj tylko wtedy, gdy jest realna potrzeba - nie zwracaj stałej liczby produktów -1. Sugeruj TYLKO typy produktów, NIGDY konkretne marki (np. "Salicylic Acid 2% Masque", nie "La Roche-Posay") -2. Koncentruj się na produktach, których użytkownik NIE posiada w swoim inventarzu -3. Bierz pod uwagę aktywne problemy skóry (acne, hyperpigmentacja, aging, etc.) -4. Sugeruj realistyczną częstotliwość użycia (dzienna, 2-3x tygodniowo, etc.) -5. Zachowaj kolejność warstw: cleanse → toner → serum → moisturizer → SPF -6. Jeśli użytkownik ma uszkodzoną barierę, unikaj silnych eksfoliantów i retinoidów -7. Odpowiadaj w języku polskim - -Format odpowiedzi - zwróć wyłącznie JSON zgodny z podanym schematem.""" - - -@mcp.tool() -def get_shopping_suggestions() -> dict: - """Analyze skin condition and inventory to suggest product types that could fill gaps. - Returns generic product suggestions (e.g., 'Salicylic Acid 2% Masque'), not specific brands. - """ - with Session(engine) as session: - context, products = _build_shopping_context(session) - - prompt = ( - f"Na podstawie poniższych danych przeanalizuj, jakie TYPY produktów " - f"mogłyby uzupełnić rutynę pielęgnacyjną użytkownika.\n\n" - f"{context}\n\n" - f"Zwróć wyłącznie JSON zgodny ze schematem." - ) - - response = call_gemini( - endpoint="shopping/suggest", - contents=prompt, - config=genai_types.GenerateContentConfig( - system_instruction=_SHOPPING_SYSTEM_PROMPT, - response_mime_type="application/json", - response_schema=ShoppingSuggestionsOut, - max_output_tokens=4096, - temperature=0.4, - ), - user_input=prompt, - ) - - raw = response.text - if not raw: - raise HTTPException(status_code=502, detail="LLM returned empty response") - - try: - parsed = json.loads(raw) - except json.JSONDecodeError as e: - raise HTTPException(status_code=502, detail=f"Invalid JSON from LLM: {e}") - - return { - "suggestions": parsed.get("suggestions", []), - "reasoning": parsed.get("reasoning", ""), - } - - def _call_market_service( query: str, stores: Optional[list[str]] = None ) -> list[MarketProduct]: