384 lines
20 KiB
Python
384 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""Shkor Beauty Assistant — ядро.
|
||
|
||
Универсальный beauty-ассистент магазина Shkor beauty:
|
||
* анализ состояния кожи (self-report + отчёт vision-агента),
|
||
* рекомендация продуктов из каталога каталога витрины,
|
||
* оформление заказа + варианта доставки,
|
||
* инструкция по применению (порядок рутины, частота, предостережения).
|
||
|
||
Протокол: один вход `handle(request: dict) -> dict`, MCP-совместимая форма
|
||
(JSON-RPC tools/call) + простая форма {action, args}. Оболочки (STDIO/HTTP/CLI)
|
||
живут в connector.py — ассистент не знает, какой агент его вызывает.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
CATALOG = json.loads((HERE / "catalog.json").read_text(encoding="utf-8"))
|
||
ORDERS_PATH = HERE / "orders.json"
|
||
PRODUCTS = {p["id"]: p for p in CATALOG["products"]}
|
||
|
||
# ── Анализ кожи ──────────────────────────────────────────────────────────────
|
||
|
||
# ключи — стеммы (корни слов), чтобы ловить любые словоформы: «стянутая/стянутость», «покраснения/покраснение»
|
||
SYMPTOM_MAP = {
|
||
"стянут": "dryness", "сух": "dryness", "шелуш": "dryness", "обезвож": "dehydration", "dry": "dryness",
|
||
"flak": "dryness", "peel": "dryness",
|
||
"покрасн": "redness", "краснот": "redness", "раздраж": "sensitivity", "чувствитель": "sensitivity",
|
||
"sensitive": "sensitivity", "redness": "redness", "red ": "redness", "itch": "redness",
|
||
"блеск": "oiliness", "блест": "oiliness", "жирн": "oiliness", "oily": "oiliness", "shine": "oiliness",
|
||
"чёрных точ": "clogged-pores", "черных точ": "clogged-pores", "blackhead": "clogged-pores",
|
||
"расширенн пор": "clogged-pores", "поры": "clogged-pores", "pores": "clogged-pores",
|
||
"высып": "acne", "акн": "acne", "прыщ": "acne", "breakout": "acne", "угр": "acne",
|
||
"пигмент": "pigmentation", "пятн": "pigmentation", "веснушк": "pigmentation", "dark spot": "pigmentation",
|
||
"морщин": "fine-lines", "wrinkle": "fine-lines", "мимическ": "fine-lines",
|
||
"туск": "dullness", "dull": "dullness", "dullness": "dullness",
|
||
"тёмн круг": "dark-circles", "темн круг": "dark-circles", "dark circle": "dark-circles", "круги под глаз": "dark-circles",
|
||
"отёк": "puffiness", "отек": "puffiness", "puffin": "puffiness", "опух": "puffiness",
|
||
"чёрных угр": "clogged-pores",
|
||
}
|
||
|
||
TYPE_RULES = [
|
||
("sensitive", 2, ["redness", "sensitivity"]),
|
||
("dry", 2, ["dryness"]),
|
||
("oily", 2, ["oiliness", "clogged-pores"]),
|
||
("combination", 0, []), # базовый исход
|
||
]
|
||
|
||
# точные имена concerns для структурированного ввода (symptoms:[...])
|
||
CONCERN_ALIASES = {c: c for c in ["dryness", "dehydration", "sensitivity", "redness", "oiliness",
|
||
"clogged-pores", "acne", "pigmentation", "photoaging", "fine-lines",
|
||
"dullness", "dark-circles", "puffiness", "texture"]}
|
||
|
||
ROUTINE_ORDER = ["cleanse", "toner", "exfoliate", "serum", "mask", "spot", "spf",
|
||
"body-cleanse", "body-care", "hair-wash", "tool"]
|
||
|
||
|
||
def analyze_skin(args: dict) -> dict:
|
||
"""Анализ: self-report (текст/чек-лист) + опциональный vision-отчёт агента-фотографа.
|
||
|
||
args:
|
||
self_report: str — свободный текст жалоб (опционально)
|
||
symptoms: [str] — структурированные симптомы (опционально)
|
||
photo_report: dict/str — отчёт vision-модели по селфи (опционально)
|
||
age: int, climate: str — контекст (опционально)
|
||
"""
|
||
signals: list[str] = []
|
||
src = {"self_report": 0, "symptoms": 0, "photo_report": 0}
|
||
|
||
if args.get("self_report"):
|
||
text = str(args["self_report"]).lower()
|
||
for kw, concern in SYMPTOM_MAP.items():
|
||
if kw in text and concern not in signals:
|
||
signals.append(concern)
|
||
src["self_report"] = len([c for c in signals])
|
||
|
||
for s in args.get("symptoms") or []:
|
||
raw = str(s).lower().strip()
|
||
c = CONCERN_ALIASES.get(raw) or SYMPTOM_MAP.get(raw) or raw
|
||
if c and c not in signals:
|
||
signals.append(c)
|
||
src["symptoms"] += 1
|
||
|
||
photo = args.get("photo_report")
|
||
if photo:
|
||
blob = photo if isinstance(photo, str) else json.dumps(photo, ensure_ascii=False)
|
||
blob = blob.lower()
|
||
for kw, concern in SYMPTOM_MAP.items():
|
||
if kw in blob and concern not in signals:
|
||
signals.append(concern)
|
||
src["photo_report"] = len([c for c in signals]) - src["self_report"] - src["symptoms"]
|
||
|
||
# тип кожи: голосование по правилам
|
||
scores = {"sensitive": 0, "dry": 0, "oily": 0}
|
||
for concern in signals:
|
||
if concern in ("redness", "sensitivity"):
|
||
scores["sensitive"] += 1
|
||
if concern in ("dryness", "dehydration", "fine-lines"):
|
||
scores["dry"] += 1
|
||
if concern in ("oiliness", "clogged-pores", "acne"):
|
||
scores["oily"] += 1
|
||
top = max(scores.values())
|
||
winners = [k for k, v in scores.items() if v == top]
|
||
skin_type = winners[0] if len(winners) == 1 else "combination" # ничья/пусто → честный combination
|
||
|
||
hydration = 5 # 0..10, нейтрально
|
||
if "dryness" in signals:
|
||
hydration -= 3
|
||
if "oiliness" in signals:
|
||
hydration += 2
|
||
if "dehydration" in signals:
|
||
hydration -= 2
|
||
hydration = max(0, min(10, hydration))
|
||
|
||
barrier = "ok"
|
||
if "redness" in signals or "sensitivity" in signals:
|
||
barrier = "compromised"
|
||
if "acne" in signals and barrier == "compromised":
|
||
barrier = "compromised+breakouts"
|
||
|
||
recommendations = [p["id"] for p in rank_products(signals, skin_type)]
|
||
|
||
return {
|
||
"skin_type": skin_type,
|
||
"concerns": signals,
|
||
"hydration_index_0_10": hydration,
|
||
"barrier": barrier,
|
||
"sources": src,
|
||
"routine_focus": routine_focus(signals),
|
||
"recommended_product_ids": recommendations[:5],
|
||
"disclaimer": "Не медицинская консультация. При стойком раздражении/акне — к дерматологу.",
|
||
}
|
||
|
||
|
||
def routine_focus(concerns: list[str]) -> list[str]:
|
||
focus = []
|
||
if "dryness" in concerns or "dehydration" in concerns:
|
||
focus.append("увлажнение: гиалурон + центелла, тонер-эссенция до сыворотки")
|
||
if "sensitivity" in concerns or "redness" in concerns:
|
||
focus.append("успокоение: центелла 77%, без кислот и ретинола первые 2 недели")
|
||
if "acne" in concerns or "clogged-pores" in concerns:
|
||
focus.append("очищение: гидрофильное масло + мягкий BHA, не щёлочь")
|
||
if "pigmentation" in concerns or "photoaging" in concerns:
|
||
focus.append("SPF каждый день + ретинол на ночь (вводить постепенно)")
|
||
if "fine-lines" in concerns:
|
||
focus.append("анти-эйдж: ретинол 2–3 р/нед, коллагеновые маски")
|
||
if "dullness" in concerns:
|
||
focus.append("тон/сияние: пэды с кислотами 2 р/нед, маска-плёнка")
|
||
if not focus:
|
||
focus.append("база: очищение → тонер → сыворотка → крем → SPF")
|
||
return focus
|
||
|
||
|
||
def rank_products(concerns: list[str], skin_type: str) -> list[dict]:
|
||
scored = []
|
||
for p in CATALOG["products"]:
|
||
score = 0
|
||
for c in concerns:
|
||
if c in p["concerns"]:
|
||
score += 3
|
||
if skin_type == "sensitive" and ("retinol" in " ".join(p["actives"]) or "acid" in " ".join(p["actives"])):
|
||
score -= 2
|
||
if skin_type == "dry" and "hyaluronic" in " ".join(p["actives"]):
|
||
score += 2
|
||
if skin_type == "oily" and p["routine_step"] in ("cleanse", "exfoliate"):
|
||
score += 1
|
||
if score > 0:
|
||
scored.append((score, p))
|
||
scored.sort(key=lambda x: (-x[0], x[1]["price"]))
|
||
return [p for _, p in scored] or CATALOG["products"][:3]
|
||
|
||
|
||
# ── Рекомендации / поиск ──────────────────────────────────────────────────────
|
||
|
||
def recommend(args: dict) -> dict:
|
||
concerns = [CONCERN_ALIASES.get(c.lower().strip()) or SYMPTOM_MAP.get(c.lower().strip(), c.lower().strip())
|
||
for c in (args.get("concerns") or [])]
|
||
skin_type = args.get("skin_type") or "combination"
|
||
budget = args.get("budget_max")
|
||
picks = rank_products(concerns, skin_type)
|
||
if budget:
|
||
picks = [p for p in picks if p["price"] <= float(budget)]
|
||
return {
|
||
"matched": len(picks),
|
||
"products": [
|
||
{"id": p["id"], "brand": p["brand"], "title": p["title"], "price": p["price"],
|
||
"why": ", ".join(c for c in concerns if c in p["concerns"]) or "базовый уход"}
|
||
for p in picks[: int(args.get("limit") or 5)]
|
||
],
|
||
}
|
||
|
||
|
||
def search(args: dict) -> dict:
|
||
q = (args.get("query") or "").lower().strip()
|
||
cat = args.get("category")
|
||
out = []
|
||
for p in CATALOG["products"]:
|
||
if cat and p["category"] != cat:
|
||
continue
|
||
if q and not (q in p["title"].lower() or q in p["brand"].lower() or q in p["sub"].lower()
|
||
or any(q in a for a in p["actives"])):
|
||
continue
|
||
out.append({"id": p["id"], "brand": p["brand"], "title": p["title"], "price": p["price"],
|
||
"category": p["category"], "sub": p["sub"]})
|
||
return {"found": len(out), "products": out[: int(args.get("limit") or 20)]}
|
||
|
||
|
||
# ── Заказы и доставка ─────────────────────────────────────────────────────────
|
||
|
||
DELIVERY = {
|
||
"minsk_courier": {"label": "Курьер по Минску", "eta": "сегодня/завтра", "price": 7.0, "free_from": 80.0},
|
||
"europochta": {"label": "Европочта (РБ)", "eta": "2–3 дня", "price": 4.5, "free_from": 80.0},
|
||
"pickup": {"label": "Самовывоз (Минск, пр-т Независимости)", "eta": "через 2 часа", "price": 0.0, "free_from": 0.0},
|
||
}
|
||
|
||
|
||
def _load_orders() -> dict:
|
||
if ORDERS_PATH.exists():
|
||
return json.loads(ORDERS_PATH.read_text(encoding="utf-8"))
|
||
return {}
|
||
|
||
|
||
def _save_orders(orders: dict) -> None:
|
||
ORDERS_PATH.write_text(json.dumps(orders, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
|
||
|
||
def place_order(args: dict) -> dict:
|
||
items = args.get("items") or []
|
||
if not items:
|
||
return {"error": "items обязателен: [{id, qty}]"}
|
||
lines, unknown = [], []
|
||
total = 0.0
|
||
for it in items:
|
||
p = PRODUCTS.get(it.get("id"))
|
||
if not p:
|
||
unknown.append(it.get("id"))
|
||
continue
|
||
qty = int(it.get("qty") or 1)
|
||
lines.append({"id": p["id"], "title": p["title"], "qty": qty, "price": p["price"], "sum": round(p["price"] * qty, 2)})
|
||
total += p["price"] * qty
|
||
if unknown:
|
||
return {"error": f"неизвестные товары: {unknown}", "hint": "tools/call search для поиска id"}
|
||
|
||
delivery_key = args.get("delivery") or "minsk_courier"
|
||
d = DELIVERY.get(delivery_key)
|
||
if not d:
|
||
return {"error": f"нет такой доставки: {delivery_key}", "available": list(DELIVERY)}
|
||
d_price = 0.0 if (d["free_from"] and total >= d["free_from"]) else d["price"]
|
||
|
||
oid = "SB-" + uuid.uuid4().hex[:8].upper()
|
||
order = {
|
||
"order_id": oid,
|
||
"items": lines,
|
||
"subtotal": round(total, 2),
|
||
"delivery": {"key": delivery_key, **d, "charged": d_price},
|
||
"total": round(total + d_price, 2),
|
||
"customer": {"name": args.get("name") or "", "phone": args.get("phone") or "",
|
||
"address": args.get("address") or ""},
|
||
"payment": {"status": "pending", "method": args.get("pay_method") or "cash_on_delivery",
|
||
"pay_link": args.get("pay_link")}, # платёжку подключает агент-хост
|
||
"status": "placed",
|
||
"created": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||
}
|
||
orders = _load_orders()
|
||
orders[oid] = order
|
||
_save_orders(orders)
|
||
return {
|
||
"order_id": oid, "status": "placed", "total": order["total"],
|
||
"delivery_eta": d["eta"], "delivery_price": d_price,
|
||
"next": f"Статус: tools/call check_order {{\"order_id\": \"{oid}\"}}. "
|
||
f"Оплата: pay_method=cash_on_delivery или передайте pay_link от платёжного агента.",
|
||
}
|
||
|
||
|
||
def check_order(args: dict) -> dict:
|
||
oid = args.get("order_id") or ""
|
||
o = _load_orders().get(oid)
|
||
if not o:
|
||
return {"error": f"заказ {oid} не найден", "known": list(_load_orders())[:10]}
|
||
return {"order_id": oid, "status": o["status"], "payment": o["payment"]["status"],
|
||
"total": o["total"], "delivery": o["delivery"]["label"], "eta": o["delivery"]["eta"],
|
||
"items": [{"title": l["title"], "qty": l["qty"]} for l in o["items"]]}
|
||
|
||
|
||
def delivery_options(args: dict) -> dict:
|
||
total = float(args.get("cart_total") or 0)
|
||
return {"options": [
|
||
{"key": k, "label": v["label"], "eta": v["eta"],
|
||
"price": 0.0 if (v["free_from"] and total >= v["free_from"]) else v["price"]}
|
||
for k, v in DELIVERY.items()]}
|
||
|
||
|
||
# ── Инструкции по применению ──────────────────────────────────────────────────
|
||
|
||
INSTRUCTIONS = {
|
||
"cleanse": "Нанести на сухую кожу, помассировать 30–60 сек, смыть тёплой водой. Утром можно мягче.",
|
||
"toner": "После очищения протереть лицо/нанести ладонями, не вытирать — следующий шаг на влажную кожу.",
|
||
"exfoliate": "2–3 раза в неделю, не в один день с ретинолом. При покраснении — прекратить.",
|
||
"serum": "3–4 капли на влажную кожу, дать впитаться 1–2 мин. Ретинол — только на ночь, старт 2–3 р/нед.",
|
||
"mask": "1–2 раза в неделю на очищенную кожу, 20–30 мин, снять плёнку снизу вверх.",
|
||
"spot": "Патчи — 10–15 мин на зону под глазами, после — лёгкий крем.",
|
||
"spf": "Финальный утренний шаг, 2 пальца средства, обновлять каждые 2–3 часа при солнце. Обязателен при ретиноле/кислотах.",
|
||
"body-cleanse": "Ежедневно, мягкой губкой, завершить увлажнением.",
|
||
"body-care": "На влажную кожу рук после каждого мытья.",
|
||
"hair-wash": "Втирать в кожу головы 1–2 мин, смыть; бальзам — только по длине.",
|
||
"tool": "Спонж смочить тёплой водой, использовать с пенкой; промывать и сушить после каждого применения.",
|
||
}
|
||
|
||
|
||
def usage_instructions(args: dict) -> dict:
|
||
ids = args.get("product_ids") or args.get("ids") or []
|
||
if isinstance(ids, str):
|
||
ids = [ids]
|
||
got = []
|
||
for pid in ids:
|
||
p = PRODUCTS.get(pid)
|
||
if p:
|
||
got.append({"id": pid, "title": p["title"], "how": INSTRUCTIONS.get(p["routine_step"], "по инструкции бренда"),
|
||
"step_order": ROUTINE_ORDER.index(p["routine_step"]) + 1 if p["routine_step"] in ROUTINE_ORDER else 99,
|
||
"caution": p.get("caution")})
|
||
if not got:
|
||
return {"error": "укажите product_ids (см. search/recommend)", "known_ids": list(PRODUCTS)}
|
||
got.sort(key=lambda x: x["step_order"])
|
||
plan = " → ".join(g["title"].split()[0] for g in got)
|
||
return {"steps": got, "routine_summary": f"Порядок нанесения: {plan}. Утро: очищение→увлажнение→SPF; вечер: очищение→акт (ретинол/кислоты)→крем.",
|
||
"general": "Новое средство — сначала на сгиб локтя 24ч. Не сочетать ретинол с кислотами в один день."}
|
||
|
||
|
||
# ── Диспетчер инструментов ────────────────────────────────────────────────────
|
||
|
||
TOOLS = {
|
||
"analyze_skin": ("Анализ состояния кожи: self_report/symptoms/photo_report → тип, concerns, фокус рутины",
|
||
["self_report?", "symptoms?", "photo_report?", "age?", "climate?"]),
|
||
"recommend": ("Подбор продуктов: concerns+skin_type+budget_max → карточки", ["concerns?", "skin_type?", "budget_max?", "limit?"]),
|
||
"search": ("Поиск по каталогу", ["query?", "category?", "limit?"]),
|
||
"place_order": ("Оформить заказ", ["items[{id,qty}]", "delivery?", "name?", "phone?", "address?", "pay_method?"]),
|
||
"check_order": ("Статус заказа и доставки", ["order_id"]),
|
||
"delivery_options": ("Варианты доставки и цены", ["cart_total?"]),
|
||
"usage_instructions": ("Инструкции по применению + порядок рутины", ["product_ids"]),
|
||
}
|
||
|
||
|
||
def handle(request: dict) -> dict:
|
||
"""Единая точка входа. Понимает MCP tools/call и простую форму {action, args}."""
|
||
if not isinstance(request, dict):
|
||
return {"error": "request должен быть JSON-объектом"}
|
||
|
||
method = request.get("method")
|
||
if method == "tools/list":
|
||
return {"tools": [{"name": n, "description": d, "args": a} for n, (d, a) in TOOLS.items()]}
|
||
if method == "initialize":
|
||
return {"protocol": "shkor-assistant/1", "tools": list(TOOLS), "shop": CATALOG["shop"]}
|
||
if method == "tools/call":
|
||
params = request.get("params") or {}
|
||
name, arguments = params.get("name"), params.get("arguments") or {}
|
||
else:
|
||
name, arguments = request.get("action"), request.get("args") or request.get("arguments") or {}
|
||
|
||
if name not in TOOLS:
|
||
return {"error": f"неизвестный инструмент {name!r}", "available": list(TOOLS)}
|
||
fn = {"analyze_skin": analyze_skin, "recommend": recommend, "search": search,
|
||
"place_order": place_order, "check_order": check_order,
|
||
"delivery_options": delivery_options, "usage_instructions": usage_instructions}[name]
|
||
try:
|
||
result = fn(arguments if isinstance(arguments, dict) else {})
|
||
except Exception as e: # noqa: BLE001 — ассистент отвечает структурой, не падает
|
||
return {"error": f"{type(e).__name__}: {e}"}
|
||
rid = request.get("id")
|
||
out = {"result": result}
|
||
if rid is not None:
|
||
out["id"] = rid
|
||
return out
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
payload = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {"method": "tools/list"}
|
||
print(json.dumps(handle(payload), ensure_ascii=False, indent=1))
|