267 lines
8.7 KiB
Python
267 lines
8.7 KiB
Python
"""Tool definitions — Admin-facing actions the KI can plan.
|
|
|
|
Each tool: name, description, JSON Schema for args, and a handler that is only
|
|
ever called from the `execute` endpoint after the user confirmed the Card.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.events import event_bus
|
|
from core.settings_service import set_setting
|
|
|
|
from apps.ai_core.tools import ToolSpec, register_tool
|
|
from apps.catalog.models import Category, Product
|
|
from apps.catalog.projector import project_category, project_product
|
|
|
|
|
|
# ---- settings.update ------------------------------------------------
|
|
|
|
|
|
def _handler_settings_update(args: dict, db: Session) -> dict:
|
|
key = args["key"]
|
|
value = args["value"]
|
|
set_setting(db, key, value)
|
|
return {"key": key, "value": value}
|
|
|
|
|
|
SETTINGS_UPDATE = ToolSpec(
|
|
name="settings.update",
|
|
description="Update a shop-wide setting (e.g. shop name, currency, support email).",
|
|
args_schema={
|
|
"type": "object",
|
|
"required": ["key", "value"],
|
|
"properties": {
|
|
"key": {
|
|
"type": "string",
|
|
"description": "Setting key, e.g. 'core.shop_name'.",
|
|
},
|
|
"value": {
|
|
"description": "New value (string / number / boolean).",
|
|
},
|
|
},
|
|
},
|
|
handler=_handler_settings_update,
|
|
examples=[
|
|
{"key": "core.shop_name", "value": "TEST123"},
|
|
{"key": "core.support_email", "value": "help@example.com"},
|
|
],
|
|
)
|
|
|
|
|
|
# ---- catalog.product.create ----------------------------------------
|
|
|
|
|
|
def _coalesce(value, default):
|
|
"""Return default if value is None or missing; otherwise the value."""
|
|
return default if value is None else value
|
|
|
|
|
|
def _handler_product_create(args: dict, db: Session) -> dict:
|
|
if db.query(Product).filter_by(sku=args["sku"]).first():
|
|
raise ValueError(f"SKU already exists: {args['sku']}")
|
|
name_de = _coalesce(args.get("name_de"), "")
|
|
name_en = _coalesce(args.get("name_en"), name_de)
|
|
desc_de = _coalesce(args.get("description_de"), "")
|
|
desc_en = _coalesce(args.get("description_en"), desc_de)
|
|
category_id = args.get("category_id")
|
|
if category_id in ("", 0):
|
|
category_id = None
|
|
p = Product(
|
|
sku=args["sku"],
|
|
name={"de": name_de, "en": name_en},
|
|
description={"de": desc_de, "en": desc_en},
|
|
price=float(args["price"]),
|
|
currency=_coalesce(args.get("currency"), "EUR") or "EUR",
|
|
stock=int(_coalesce(args.get("stock"), 0) or 0),
|
|
active=bool(_coalesce(args.get("active"), True)),
|
|
image_url=_coalesce(args.get("image_url"), "") or "",
|
|
category_id=category_id,
|
|
attributes=_coalesce(args.get("attributes"), {}) or {},
|
|
)
|
|
db.add(p)
|
|
db.commit()
|
|
db.refresh(p)
|
|
project_product(db, p.id)
|
|
event_bus.publish("product.created", {"id": p.id, "sku": p.sku}, db=db)
|
|
return {"id": p.id, "sku": p.sku}
|
|
|
|
|
|
PRODUCT_CREATE = ToolSpec(
|
|
name="catalog.product.create",
|
|
description="Create a new product in the catalog.",
|
|
args_schema={
|
|
"type": "object",
|
|
"required": ["sku", "name_de", "price"],
|
|
"properties": {
|
|
"sku": {"type": "string"},
|
|
"name_de": {"type": "string"},
|
|
"name_en": {"type": "string"},
|
|
"description_de": {"type": "string"},
|
|
"description_en": {"type": "string"},
|
|
"price": {"type": "number"},
|
|
"currency": {"type": "string", "default": "EUR"},
|
|
"stock": {"type": "integer", "default": 0},
|
|
"active": {"type": "boolean", "default": True},
|
|
"image_url": {"type": "string"},
|
|
"category_id": {"type": "integer"},
|
|
"attributes": {"type": "object"},
|
|
},
|
|
},
|
|
handler=_handler_product_create,
|
|
examples=[
|
|
{
|
|
"sku": "TS-GREEN-M",
|
|
"name_de": "Grünes T-Shirt",
|
|
"name_en": "Green T-Shirt",
|
|
"price": 19.90,
|
|
"stock": 42,
|
|
"attributes": {"color": "green", "size": "M"},
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
# ---- catalog.product.update ----------------------------------------
|
|
|
|
|
|
def _handler_product_update(args: dict, db: Session) -> dict:
|
|
pid = int(args["id"])
|
|
p = db.get(Product, pid)
|
|
if not p:
|
|
raise ValueError(f"Product {pid} not found")
|
|
if "name_de" in args or "name_en" in args:
|
|
p.name = {
|
|
"de": args.get("name_de", p.name.get("de", "")),
|
|
"en": args.get("name_en", p.name.get("en", "")),
|
|
}
|
|
if "description_de" in args or "description_en" in args:
|
|
p.description = {
|
|
"de": args.get("description_de", p.description.get("de", "")),
|
|
"en": args.get("description_en", p.description.get("en", "")),
|
|
}
|
|
for f in ("price", "currency", "stock", "active", "image_url", "category_id"):
|
|
if f in args:
|
|
setattr(p, f, args[f])
|
|
if "attributes" in args:
|
|
p.attributes = args["attributes"]
|
|
db.commit()
|
|
db.refresh(p)
|
|
project_product(db, p.id)
|
|
event_bus.publish("product.updated", {"id": p.id}, db=db)
|
|
return {"id": p.id, "sku": p.sku}
|
|
|
|
|
|
PRODUCT_UPDATE = ToolSpec(
|
|
name="catalog.product.update",
|
|
description="Update fields of an existing product.",
|
|
args_schema={
|
|
"type": "object",
|
|
"required": ["id"],
|
|
"properties": {
|
|
"id": {"type": "integer"},
|
|
"name_de": {"type": "string"},
|
|
"name_en": {"type": "string"},
|
|
"description_de": {"type": "string"},
|
|
"description_en": {"type": "string"},
|
|
"price": {"type": "number"},
|
|
"stock": {"type": "integer"},
|
|
"active": {"type": "boolean"},
|
|
"image_url": {"type": "string"},
|
|
"category_id": {"type": "integer"},
|
|
"attributes": {"type": "object"},
|
|
},
|
|
},
|
|
handler=_handler_product_update,
|
|
examples=[{"id": 5, "price": 24.90, "stock": 10}],
|
|
)
|
|
|
|
|
|
# ---- catalog.category.create --------------------------------------
|
|
|
|
|
|
def _handler_category_create(args: dict, db: Session) -> dict:
|
|
if db.query(Category).filter_by(slug=args["slug"]).first():
|
|
raise ValueError(f"Slug exists: {args['slug']}")
|
|
c = Category(
|
|
slug=args["slug"],
|
|
name={"de": args.get("name_de", ""), "en": args.get("name_en", args.get("name_de", ""))},
|
|
parent_id=args.get("parent_id"),
|
|
sort_order=int(args.get("sort_order", 0)),
|
|
)
|
|
db.add(c)
|
|
db.commit()
|
|
db.refresh(c)
|
|
project_category(db, c.id)
|
|
event_bus.publish("category.created", {"id": c.id}, db=db)
|
|
return {"id": c.id, "slug": c.slug}
|
|
|
|
|
|
CATEGORY_CREATE = ToolSpec(
|
|
name="catalog.category.create",
|
|
description="Create a new category.",
|
|
args_schema={
|
|
"type": "object",
|
|
"required": ["slug", "name_de"],
|
|
"properties": {
|
|
"slug": {"type": "string"},
|
|
"name_de": {"type": "string"},
|
|
"name_en": {"type": "string"},
|
|
"parent_id": {"type": "integer"},
|
|
"sort_order": {"type": "integer"},
|
|
},
|
|
},
|
|
handler=_handler_category_create,
|
|
examples=[{"slug": "accessoires", "name_de": "Accessoires", "name_en": "Accessories"}],
|
|
)
|
|
|
|
|
|
# ---- orders.update_status -----------------------------------------
|
|
|
|
|
|
def _handler_order_status(args: dict, db: Session) -> dict:
|
|
from apps.orders.models import Order, OrderStatusHistory
|
|
|
|
order_id = int(args["id"])
|
|
o = db.get(Order, order_id)
|
|
if not o:
|
|
raise ValueError(f"Order {order_id} not found")
|
|
new_status = args["status"]
|
|
note = _coalesce(args.get("note"), "")
|
|
o.status = new_status
|
|
db.add(OrderStatusHistory(order_id=order_id, status=new_status, note=note))
|
|
db.commit()
|
|
event_bus.publish("order.status_changed", {"id": order_id, "status": new_status}, db=db)
|
|
return {"id": order_id, "status": new_status}
|
|
|
|
|
|
ORDER_STATUS = ToolSpec(
|
|
name="orders.update_status",
|
|
description="Change the status of an order (e.g. paid, packed, shipped, delivered, cancelled).",
|
|
args_schema={
|
|
"type": "object",
|
|
"required": ["id", "status"],
|
|
"properties": {
|
|
"id": {"type": "integer", "description": "Order ID"},
|
|
"status": {
|
|
"type": "string",
|
|
"description": "New status: paid, packed, shipped, delivered, cancelled",
|
|
},
|
|
"note": {"type": "string", "description": "Optional note"},
|
|
},
|
|
},
|
|
handler=_handler_order_status,
|
|
examples=[{"id": 1, "status": "shipped", "note": "DHL versendet"}],
|
|
)
|
|
|
|
|
|
# ---- registration -------------------------------------------------
|
|
|
|
|
|
ALL_TOOLS = [SETTINGS_UPDATE, PRODUCT_CREATE, PRODUCT_UPDATE, CATEGORY_CREATE, ORDER_STATUS]
|
|
|
|
|
|
def register_all() -> None:
|
|
for t in ALL_TOOLS:
|
|
register_tool(t)
|