feat(products): add shopping suggestions feature
- Add POST /api/products/suggest endpoint that analyzes skin condition and inventory to suggest product types (e.g., 'Salicylic Acid 2% Masque') - Add MCP tool get_shopping_suggestions() for MCP clients - Add 'Suggest' button to Products page in frontend - Add /products/suggest page with suggestion cards - Include product type, key ingredients, target concerns, why_needed, recommended_time, and frequency in suggestions - Fix stock logic: sealed products now count as available inventory - Add legend to clarify ✓ (in stock) vs ✗ (not in stock) markers
This commit is contained in:
parent
389ca5ffdc
commit
40f9a353bb
8 changed files with 583 additions and 2 deletions
|
|
@ -5,6 +5,7 @@ from uuid import UUID, uuid4
|
|||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from google.genai import types as genai_types
|
||||
from pydantic import BaseModel as PydanticBase
|
||||
from pydantic import ValidationError
|
||||
from sqlmodel import Session, SQLModel, col, select
|
||||
|
||||
|
|
@ -19,6 +20,7 @@ from innercontext.models import (
|
|||
ProductPublic,
|
||||
ProductWithInventory,
|
||||
SkinConcern,
|
||||
SkinConditionSnapshot,
|
||||
)
|
||||
from innercontext.models.enums import (
|
||||
AbsorptionSpeed,
|
||||
|
|
@ -176,6 +178,41 @@ class InventoryUpdate(SQLModel):
|
|||
notes: Optional[str] = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shopping suggestion schemas
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ProductSuggestion(PydanticBase):
|
||||
category: str
|
||||
product_type: str
|
||||
key_ingredients: list[str]
|
||||
target_concerns: list[str]
|
||||
why_needed: str
|
||||
recommended_time: str
|
||||
frequency: str
|
||||
|
||||
|
||||
class ShoppingSuggestionResponse(PydanticBase):
|
||||
suggestions: list[ProductSuggestion]
|
||||
reasoning: str
|
||||
|
||||
|
||||
class _ProductSuggestionOut(PydanticBase):
|
||||
category: str
|
||||
product_type: str
|
||||
key_ingredients: list[str]
|
||||
target_concerns: list[str]
|
||||
why_needed: str
|
||||
recommended_time: str
|
||||
frequency: str
|
||||
|
||||
|
||||
class _ShoppingSuggestionsOut(PydanticBase):
|
||||
suggestions: list[_ProductSuggestionOut]
|
||||
reasoning: str
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Product routes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -218,7 +255,9 @@ def list_products(
|
|||
product_ids = [p.id for p in products]
|
||||
inventory_rows = (
|
||||
session.exec(
|
||||
select(ProductInventory).where(col(ProductInventory.product_id).in_(product_ids))
|
||||
select(ProductInventory).where(
|
||||
col(ProductInventory.product_id).in_(product_ids)
|
||||
)
|
||||
).all()
|
||||
if product_ids
|
||||
else []
|
||||
|
|
@ -467,3 +506,134 @@ def create_product_inventory(
|
|||
session.commit()
|
||||
session.refresh(entry)
|
||||
return entry
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shopping suggestion
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
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) -> str:
|
||||
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 = {}
|
||||
for inv in inventory_rows:
|
||||
inv_by_product.setdefault(inv.product_id, []).append(inv)
|
||||
|
||||
products_lines = ["POSIADANE PRODUKTY:"]
|
||||
products_lines.append(
|
||||
" Legenda: [✓] = produkt dostępny (w magazynie), [✗] = brak w magazynie"
|
||||
)
|
||||
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 = len(active_inv) > 0 # any unfinished inventory = in stock
|
||||
stock = "✓" if has_stock else "✗"
|
||||
products_lines.append(
|
||||
f" [{stock}] {p.name} ({p.brand or ''}) - {_ev(p.category)}, "
|
||||
f"targets: {p.targets or []}"
|
||||
)
|
||||
|
||||
return "\n".join(skin_lines) + "\n\n" + "\n".join(products_lines)
|
||||
|
||||
|
||||
_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ę.
|
||||
|
||||
LEGENDA:
|
||||
- [✓] = produkt dostępny w magazynie (nawet jeśli jest zapieczętowany)
|
||||
- [✗] = produkt niedostępny (brak w magazynie, wszystkie opakowania zużyte)
|
||||
|
||||
ZASADY:
|
||||
1. Sugeruj TYLKO typy produktów, NIGDY konkretne marki (np. "Salicylic Acid 2% Masque", nie "La Roche-Posay")
|
||||
2. Produkty oznaczone [✗] to te, których NIE MA w magazynie - możesz je zasugerować
|
||||
3. Produkty oznaczone [✓] są już dostępne - nie sugeruj ich ponownie
|
||||
4. Bierz pod uwagę aktywne problemy skóry (acne, hyperpigmentacja, aging, etc.)
|
||||
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
|
||||
|
||||
Format odpowiedzi - zwróć wyłącznie JSON zgodny z podanym schematem."""
|
||||
|
||||
|
||||
@router.post("/suggest", response_model=ShoppingSuggestionResponse)
|
||||
def suggest_shopping(session: Session = Depends(get_session)):
|
||||
context = _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="products/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 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}")
|
||||
|
||||
return ShoppingSuggestionResponse(
|
||||
suggestions=[ProductSuggestion(**s) for s in parsed.get("suggestions", [])],
|
||||
reasoning=parsed.get("reasoning", ""),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,13 +1,20 @@
|
|||
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,
|
||||
|
|
@ -20,6 +27,36 @@ from innercontext.models import (
|
|||
SkinConditionSnapshot,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pydantic schemas for structured output
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
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
|
||||
price: Optional[float] = None
|
||||
currency: str = "PLN"
|
||||
store: str
|
||||
url: str
|
||||
in_stock: Optional[bool] = None
|
||||
|
||||
|
||||
mcp = FastMCP("innercontext")
|
||||
|
||||
|
||||
|
|
@ -435,3 +472,183 @@ def get_lab_results_for_test(test_code: str) -> list[dict]:
|
|||
.order_by(col(LabResult.collected_at).asc())
|
||||
).all()
|
||||
return [_lab_result_to_dict(r) for r in results]
|
||||
|
||||
|
||||
# ── 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:
|
||||
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]:
|
||||
market_url = os.environ.get("MARKET_SERVICE_URL")
|
||||
if not market_url:
|
||||
return []
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=10.0) as client:
|
||||
resp = client.get(
|
||||
f"{market_url}/search",
|
||||
params={"q": query, "stores": stores or ["rossmann", "dm", "hebe"]},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
return [MarketProduct(**item) for item in data.get("products", [])]
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def search_market_products(
|
||||
query: str,
|
||||
stores: Optional[list[str]] = None,
|
||||
) -> list[dict]:
|
||||
"""Search drug store catalogs for products matching the query.
|
||||
Uses external market service to query Rossmann, DM, Hebe, etc.
|
||||
|
||||
Examples:
|
||||
- query: "salicylic acid serum acne"
|
||||
- stores: ["rossmann", "dm", "hebe"] (optional, queries all by default)"""
|
||||
products = _call_market_service(query, stores)
|
||||
return [p.model_dump(mode="json") for p in products]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue