Mise en forme

quick_deploy_agent.py

@@ -129,6 +129,116 @@ class OFFtoCOICOP(Tool):
 
         return json.dumps({"candidates": c})
 
+# ---- OFFByEAN : robuste + sortie normalisée + step3_inputs ----
+class OFFByEAN(Tool):
+    name = "openfoodfacts_product_by_ean"
+    description = "Open Food Facts /api/v0|v2/product/{ean} (name, brands, categories...)."
+    inputs = {"ean": {"type": "string", "description": "EAN à interroger sur l'API OFF."}}
+    output_type = "string"
+    requirements = ["requests"]  # urllib3 vient via requests
+
+    def forward(self, ean: str) -> str:
+        import re, json, requests
+        from requests.adapters import HTTPAdapter
+        try:
+            from urllib3.util.retry import Retry
+        except Exception:
+            Retry = None  # environnement minimal
+
+        def _to_list(x):
+            if x is None:
+                return []
+            if isinstance(x, list):
+                # stringifier proprement les éléments non-str
+                return [str(t).strip() for t in x if str(t).strip()]
+            if isinstance(x, str):
+                parts = [p.strip() for p in re.split(r"[,\|;]", x) if p.strip()]
+                return parts
+            return [str(x).strip()]
+
+        def _first(*vals):
+            for v in vals:
+                if isinstance(v, str) and v.strip():
+                    return v.strip()
+            return ""
+
+        code = re.sub(r"\D", "", ean or "")
+        if not code:
+            return json.dumps({"ok": False, "status": 0, "code": "", "error": "EAN vide"})
+
+        sess = requests.Session()
+        sess.headers.update({
+            "User-Agent": "insee-coicop-agent/1.0",
+            "Accept": "application/json",
+        })
+        if Retry:
+            retry = Retry(
+                total=3, backoff_factor=0.5,
+                status_forcelist=[429, 500, 502, 503, 504],
+                allowed_methods=frozenset(["GET"]),
+                raise_on_status=False,
+            )
+            sess.mount("https://", HTTPAdapter(max_retries=retry))
+
+        urls = [
+            f"https://world.openfoodfacts.org/api/v0/product/{code}.json",
+            "https://world.openfoodfacts.org/api/v2/product/"
+            f"{code}?lc=fr&fields=code,product_name,product_name_fr,brands,"
+            "categories_tags,categories_tags_fr,ingredients_text,ingredients_text_fr,"
+            "stores,status,status_verbose",
+            f"https://world.openfoodfacts.net/api/v0/product/{code}.json",
+        ]
+
+        last_err = None
+        for u in urls:
+            try:
+                r = sess.get(u, timeout=15)
+                if not r.ok:
+                    last_err = f"HTTP {r.status_code}"
+                    continue
+                data = r.json()
+                product = data.get("product")
+                status = data.get("status", 1 if product else 0)
+                if status == 1 or product:
+                    p = product or {}
+                    # Normalisation stricte des champs
+                    product_name = _first(p.get("product_name_fr"), p.get("product_name"))
+                    categories_tags = (
+                        p.get("categories_tags_fr")
+                        or p.get("categories_tags")
+                        or p.get("categories")
+                    )
+                    categories_tags = _to_list(categories_tags)
+                    ingredients_text = _first(p.get("ingredients_text_fr"), p.get("ingredients_text"))
+                    brands = _first(p.get("brands"), None)
+                    stores = _first(p.get("stores"), None)
+
+                    out = {
+                        "ok": True,
+                        "status": status,
+                        "status_verbose": data.get("status_verbose"),
+                        "code": code,
+                        "used_url": u,
+                        # Champs lisibles directement par le LLM
+                        "product_name": product_name,
+                        "categories_tags": categories_tags,     # toujours list[str]
+                        "ingredients_text": ingredients_text,
+                        "brands": brands,                       # string (OFF est souvent "brand1, brand2")
+                        "brands_list": _to_list(brands),        # list[str] pratique
+                        "stores": stores,
+                        "stores_list": _to_list(stores),
+                        # 🔑 Bloc prêt pour l'étape 3
+                        "step3_inputs": {
+                            "product_name": product_name,
+                            "categories_tags": categories_tags,
+                            "ingredients_text": ingredients_text,
+                        },
+                    }
+                    return json.dumps(out)
+            except Exception as e:
+                last_err = str(e)
+
+        return json.dumps({"ok": False, "status": 0, "code": code, "error": last_err or "not found"})