sysops
233 lines
7.3 KiB
Python
233 lines
7.3 KiB
Python
"""Packaging agent — parse Herman-opdrachten en genereer designs."""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
from app.config import settings
|
||
from app.services import packaging_nas, projects
|
||
|
||
PACKAGING_KEYWORDS = (
|
||
"maak verpakking",
|
||
"maak een verpakking",
|
||
"genereer verpakking",
|
||
"ontwerp verpakking",
|
||
"packaging:",
|
||
"packaging ",
|
||
"/packaging",
|
||
"maak packaging",
|
||
"maak package",
|
||
"stanstekening",
|
||
"verpakkingsontwerp",
|
||
)
|
||
|
||
TYPE_ALIASES: list[tuple[str, str]] = [
|
||
("folding box", "folding_box"),
|
||
("folding_box", "folding_box"),
|
||
("sluitdoos", "folding_box"),
|
||
("doos", "folding_box"),
|
||
("banderole", "wrap"),
|
||
("wrap", "wrap"),
|
||
("rond label", "round_label"),
|
||
("round label", "round_label"),
|
||
("round_label", "round_label"),
|
||
("sleeve", "sleeve"),
|
||
("huls", "sleeve"),
|
||
("pouch", "pouch"),
|
||
("zak", "pouch"),
|
||
("tray", "tray"),
|
||
("schaal", "tray"),
|
||
]
|
||
|
||
DEFAULT_ELEMENTS = {
|
||
"barcode": True,
|
||
"logo_area": True,
|
||
"fold_lines": True,
|
||
"cut_lines": True,
|
||
"nutrition_panel": False,
|
||
"ingredients": True,
|
||
"halal_badge": False,
|
||
"window": False,
|
||
"qr_code": False,
|
||
"glue_tabs": False,
|
||
"bleed": True,
|
||
"dimensions": True,
|
||
}
|
||
|
||
|
||
def wants_packaging(raw: str) -> bool:
|
||
t = (raw or "").strip().lower()
|
||
return any(k in t for k in PACKAGING_KEYWORDS)
|
||
|
||
|
||
def extract_packaging_body(raw: str) -> str:
|
||
t = raw.strip()
|
||
lower = t.lower()
|
||
for k in PACKAGING_KEYWORDS:
|
||
if lower.startswith(k):
|
||
rest = t[len(k) :].strip(" :,-")
|
||
if rest:
|
||
return rest
|
||
for k in PACKAGING_KEYWORDS:
|
||
if k in lower:
|
||
idx = lower.index(k) + len(k)
|
||
rest = t[idx:].strip(" :,-")
|
||
if rest:
|
||
return rest
|
||
return t
|
||
|
||
|
||
def _detect_type(text: str) -> str:
|
||
lower = text.lower()
|
||
for alias, ptype in TYPE_ALIASES:
|
||
if alias in lower:
|
||
return ptype
|
||
return "folding_box"
|
||
|
||
|
||
def _detect_dimensions(text: str) -> tuple[float, float, float]:
|
||
m = re.search(r"(\d{2,4})\s*[x×]\s*(\d{2,4})(?:\s*[x×]\s*(\d{1,4}))?", text, re.I)
|
||
if m:
|
||
w, h = float(m.group(1)), float(m.group(2))
|
||
d = float(m.group(3)) if m.group(3) else (40.0 if _detect_type(text) == "folding_box" else 20.0)
|
||
return w, h, d
|
||
return 120.0, 80.0, 40.0
|
||
|
||
|
||
def _detect_project_id(text: str) -> int | None:
|
||
m = re.search(r"project\s*#?\s*(\d+)", text, re.I)
|
||
if m:
|
||
return int(m.group(1))
|
||
m = re.search(r"\bproject\s+(\d+)\b", text, re.I)
|
||
return int(m.group(1)) if m else None
|
||
|
||
|
||
def _detect_product_name(text: str) -> str:
|
||
m = re.search(r'voor\s+["\']?([^"\']+?)["\']?(?:\s+project|\s*$|,)', text, re.I)
|
||
if m:
|
||
return m.group(1).strip()[:80]
|
||
m = re.search(r'product\s*[:=]\s*["\']?([^"\']+)["\']?', text, re.I)
|
||
if m:
|
||
return m.group(1).strip()[:80]
|
||
cleaned = text
|
||
for alias, _ in TYPE_ALIASES:
|
||
cleaned = re.sub(re.escape(alias), "", cleaned, flags=re.I)
|
||
cleaned = re.sub(r"\d{2,4}\s*[x×]\s*\d{2,4}(?:\s*[x×]\s*\d{1,4})?", "", cleaned, flags=re.I)
|
||
cleaned = re.sub(r"project\s*#?\s*\d+", "", cleaned, flags=re.I)
|
||
for kw in ("halal-badge", "halal badge", "voedingswaarden", "nutrition", "qr-code", "qr code", "venster", "window"):
|
||
cleaned = re.sub(re.escape(kw), "", cleaned, flags=re.I)
|
||
cleaned = cleaned.strip(" ,:-")
|
||
return (cleaned[:80] or "Foodlinkk Product")
|
||
|
||
|
||
def parse_packaging_request(message: str) -> dict[str, Any]:
|
||
body = extract_packaging_body(message)
|
||
lower = body.lower()
|
||
ptype = _detect_type(body)
|
||
w, h, d = _detect_dimensions(body)
|
||
product = _detect_product_name(body)
|
||
project_id = _detect_project_id(message) or _detect_project_id(body)
|
||
|
||
elements = dict(DEFAULT_ELEMENTS)
|
||
if any(k in lower for k in ("halal", "halal-badge", "halal badge")):
|
||
elements["halal_badge"] = True
|
||
if any(k in lower for k in ("voedingswaarden", "nutrition")):
|
||
elements["nutrition_panel"] = True
|
||
if any(k in lower for k in ("qr", "qrcode")):
|
||
elements["qr_code"] = True
|
||
if "venster" in lower or "window" in lower:
|
||
elements["window"] = True
|
||
|
||
return {
|
||
"type": ptype,
|
||
"width_mm": w,
|
||
"height_mm": h,
|
||
"depth_mm": d,
|
||
"bleed_mm": 3,
|
||
"design_name": product,
|
||
"barcode_value": "8710000000012",
|
||
"elements": elements,
|
||
"text": {
|
||
"product_name": product,
|
||
"tagline": "Premium halal kant-en-klaar",
|
||
"subtitle": "",
|
||
"ingredients": "",
|
||
"origin": "Geproduceerd in NL",
|
||
"best_before": "Ten minste houdbaar tot: zie verpakking",
|
||
},
|
||
"brand": {},
|
||
"project_id": project_id,
|
||
"created_by": "packaging",
|
||
}
|
||
|
||
|
||
async def _download_bytes(packaging_id: str, fmt: str) -> bytes | None:
|
||
try:
|
||
async with httpx.AsyncClient(timeout=120.0) as client:
|
||
r = await client.get(
|
||
f"{settings.TOOLS_API_URL.rstrip('/')}/packaging/download/{packaging_id}",
|
||
params={"format": fmt},
|
||
)
|
||
if r.status_code < 400:
|
||
return r.content
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
async def generate_from_message(message: str) -> dict[str, Any]:
|
||
"""Genereer packaging design en exporteer naar NAS; retour voor Herman-reply."""
|
||
spec = parse_packaging_request(message)
|
||
|
||
if not spec.get("project_id"):
|
||
proj = projects.create_project(
|
||
f"Packaging · {spec['design_name'][:48]}",
|
||
description=f"Aangemaakt door packaging agent via Herman\n\nOpdracht: {message[:500]}",
|
||
project_type="packaging",
|
||
ensure_nas=True,
|
||
created_by="packaging",
|
||
)
|
||
spec["project_id"] = proj.get("id")
|
||
|
||
async with httpx.AsyncClient(timeout=120.0) as client:
|
||
r = await client.post(
|
||
f"{settings.TOOLS_API_URL.rstrip('/')}/packaging/generate",
|
||
json=spec,
|
||
)
|
||
r.raise_for_status()
|
||
result = r.json()
|
||
|
||
packaging_id = result.get("id", "")
|
||
cockpit_project_id = result.get("cockpit_project_id") or spec.get("project_id")
|
||
svg = result.get("svg") or ""
|
||
saved_spec = result.get("spec") or spec
|
||
|
||
nas_info: dict[str, Any] = {}
|
||
if packaging_id and cockpit_project_id and svg:
|
||
png_b = await _download_bytes(packaging_id, "png")
|
||
pdf_b = await _download_bytes(packaging_id, "pdf")
|
||
try:
|
||
nas_info = packaging_nas.export_packaging_files(
|
||
packaging_id, int(cockpit_project_id), svg, saved_spec, png_b, pdf_b
|
||
)
|
||
except Exception as exc:
|
||
nas_info = {"ok": False, "error": str(exc)}
|
||
|
||
studio_url = f"/packaging?project_id={cockpit_project_id}"
|
||
pdf_url = f"/api/packaging/download/{packaging_id}?format=pdf"
|
||
|
||
return {
|
||
"packaging_id": packaging_id,
|
||
"cockpit_project_id": cockpit_project_id,
|
||
"design_name": saved_spec.get("design_name") or spec.get("design_name"),
|
||
"type": saved_spec.get("type"),
|
||
"dimensions": f"{saved_spec.get('width_mm')}×{saved_spec.get('height_mm')}×{saved_spec.get('depth_mm')} mm",
|
||
"studio_url": studio_url,
|
||
"pdf_url": pdf_url,
|
||
"nas": nas_info,
|
||
"spec": saved_spec,
|
||
"original_message": message,
|
||
}
|