Update app.py

app.py

@@ -1,8 +1,7 @@
 import os
 import time
 import json
-import tempfile
-from typing import Dict, Any, List, Generator, Tuple, Optional
+from typing import Dict, Any, List, Generator, Tuple
 
 import requests
 import gradio as gr
@@ -42,241 +41,122 @@ def _log(logs: List[str], msg: str):
     logs.append(f"[{_now()}] {msg}")
 
 
-def _list_shops(logs: List[str]) -> List[Dict[str, Any]]:
-    _log(logs, "Fetching shops: GET /v1/shops.json")
-    shops = _req("GET", "/v1/shops.json")
-    if not isinstance(shops, list):
-        raise RuntimeError("Unexpected shops response")
-    _log(logs, f"Found {len(shops)} shops")
-    return shops
-
-
-def _paginate_blueprints(logs: List[str], per_page: int = 100) -> List[Dict[str, Any]]:
-    out: List[Dict[str, Any]] = []
-    page = 1
-    while True:
-        _log(logs, f"Listing blueprints page={page} per_page={per_page}")
-        data = _req("GET", f"/v1/catalog/blueprints.json?page={page}&per_page={per_page}")
-        if not isinstance(data, list) or not data:
-            break
-        out.extend(data)
-        page += 1
-        if page > 9999:
-            break
-    _log(logs, f"Total blueprints fetched: {len(out)}")
-    return out
-
-
-def _extract_variant_snapshot(variants: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], float]:
-    snapshot = []
-    all_cents: List[int] = []
+def _find_first_valid_pair(logs: List[str]) -> Dict[str, Any]:
+    _log(logs, "Listing blueprints")
+    blueprints = _req("GET", "/v1/catalog/blueprints.json")
 
-    for v in variants:
-        cents_raw = v.get("price")
-        cents = None
-        try:
-            if isinstance(cents_raw, (int, float)):
-                cents = int(cents_raw)
-            elif isinstance(cents_raw, str) and cents_raw.isdigit():
-                cents = int(cents_raw)
-        except Exception:
-            cents = None
-
-        if cents is not None:
-            all_cents.append(cents)
-
-        opts = v.get("options") or {}
-        snapshot.append(
-            {
-                "id": v.get("id"),
-                "sku": v.get("sku"),
-                "size": opts.get("size"),
-                "color": opts.get("color"),
-                "priceCents": cents,
-                "price": round(cents / 100, 2) if cents is not None else None,
-                "placeholders": v.get("placeholders") or [],
-            }
-        )
+    for bp in blueprints:
+        bp_id = str(bp["id"])
+        _log(logs, f"Blueprint {bp_id}: fetching providers")
 
-    min_price = round(min(all_cents) / 100, 2) if all_cents else DEFAULT_BASE_PRICE
-    return snapshot, min_price
+        providers = _req("GET", f"/v1/catalog/blueprints/{bp_id}/print_providers.json")
+        for p in providers:
+            p_id = str(p["id"])
+            _log(logs, f"Blueprint {bp_id} / Provider {p_id}: fetching variants")
 
+            vr = _req(
+                "GET",
+                f"/v1/catalog/blueprints/{bp_id}/print_providers/{p_id}/variants.json?show-out-of-stock=1",
+            )
+            variants = vr.get("variants")
+            if isinstance(variants, list) and variants:
+                _log(logs, f"FOUND {len(variants)} variants")
+
+                return {
+                    "blueprint": bp,
+                    "provider": p,
+                    "variants": variants,
+                    "blueprintDetails": _req("GET", f"/v1/catalog/blueprints/{bp_id}.json"),
+                    "providerDetails": _req("GET", f"/v1/catalog/print_providers/{p_id}.json"),
+                    "shippingInfo": _req(
+                        "GET",
+                        f"/v1/catalog/blueprints/{bp_id}/print_providers/{p_id}/shipping.json",
+                    ),
+                }
 
-def run_phase_a(
-    selected_shop_id: Optional[str],
-    max_templates: int,
-    per_page: int,
-) -> Generator[Tuple[str, str, Optional[str]], None, None]:
-    logs: List[str] = []
-    results: List[Dict[str, Any]] = []
-    download_path: Optional[str] = None
+    raise RuntimeError("No blueprint/provider pair with variants found.")
 
-    def flush():
-        return "\n".join(logs), json.dumps(results[:3], indent=2), download_path
 
-    try:
-        _log(logs, "START")
-        yield flush()
+def _build_product(blob: Dict[str, Any], currency: str, logs: List[str]) -> Dict[str, Any]:
+    variants = blob["variants"]
+
+    snapshot = []
+    for v in variants:
+        cents = v.get("price")
+        cents = int(cents) if isinstance(cents, (int, str)) and str(cents).isdigit() else None
+        snapshot.append({
+            "id": v.get("id"),
+            "sku": v.get("sku"),
+            "size": (v.get("options") or {}).get("size"),
+            "color": (v.get("options") or {}).get("color"),
+            "priceCents": cents,
+            "price": round(cents / 100, 2) if cents is not None else None,
+        })
+
+    _log(logs, f"VARIANT_SNAPSHOT sample={json.dumps(snapshot[:20])}")
+
+    all_cents = [v["priceCents"] for v in snapshot if v["priceCents"] is not None]
+    min_price = round(min(all_cents) / 100, 2) if all_cents else DEFAULT_BASE_PRICE
 
-        shops = _list_shops(logs)
-        if not selected_shop_id:
-            raise RuntimeError("Pick a shop (Atheria) first, then Run.")
-        shop_id = str(selected_shop_id)
+    colors = sorted({v["color"] for v in snapshot if v["color"]})
+    sizes = sorted({v["size"] for v in snapshot if v["size"]})
 
-        hit = None
-        for s in shops:
-            if str(s.get("id")) == shop_id:
-                hit = s
-                break
+    provider = blob["providerDetails"]
+    loc = provider.get("location") or {}
+    loc_str = ", ".join(x for x in [loc.get("city"), loc.get("country")] if x) or "empty"
 
-        if hit:
-            _log(logs, f"Selected shop: title={hit.get('title') or 'untitled'} id={shop_id} channel={hit.get('sales_channel') or 'unknown'}")
-        else:
-            _log(logs, f"Selected shop_id={shop_id} (not found in list_shops response?)")
+    description = (
+        (blob["blueprintDetails"].get("description") or "empty")
+        + f"\n\nFulfilled by {provider.get('title') or provider.get('name')} — {loc_str}"
+    )
 
-        yield flush()
+    return {
+        "name": blob["blueprintDetails"].get("title"),
+        "currency": currency,
+        "price": min_price,
+        "tags": ["Printify", f"Provider: {provider.get('title') or provider.get('name')}"],
+        "options": {
+            "Color": colors,
+            "Size": sizes,
+        },
+        "description": description,
+        "variants": snapshot,
+        "raw": blob,
+    }
+
+
+def run(currency: str) -> Generator[Tuple[str, str], None, None]:
+    logs: List[str] = []
+    result: Dict[str, Any] = {}
 
-        max_templates = int(max_templates or 0)
-        per_page = int(per_page or 100)
+    def flush():
+        return "\n".join(logs), json.dumps(result, indent=2)
 
-        blueprints = _paginate_blueprints(logs, per_page=per_page)
+    try:
+        _log(logs, "START")
         yield flush()
 
-        pairs_scanned = 0
-
-        for bp in blueprints:
-            bp_id = bp.get("id")
-            if bp_id is None:
-                continue
-            bp_id_s = str(bp_id)
-
-            _log(logs, f"Blueprint {bp_id_s}: fetching providers")
-            providers = _req("GET", f"/v1/catalog/blueprints/{bp_id_s}/print_providers.json")
-            if not isinstance(providers, list) or not providers:
-                continue
-
-            for p in providers:
-                p_id = p.get("id")
-                if p_id is None:
-                    continue
-                p_id_s = str(p_id)
-
-                pairs_scanned += 1
-                _log(logs, f"Blueprint {bp_id_s} / Provider {p_id_s}: fetching variants")
-
-                vr = _req(
-                    "GET",
-                    f"/v1/catalog/blueprints/{bp_id_s}/print_providers/{p_id_s}/variants.json?show-out-of-stock=1",
-                )
-                variants = vr.get("variants") if isinstance(vr, dict) else None
-                if not isinstance(variants, list) or not variants:
-                    continue
-
-                _log(logs, f"FOUND {len(variants)} variants (bp={bp_id_s}, provider={p_id_s})")
-
-                blueprint_details = _req("GET", f"/v1/catalog/blueprints/{bp_id_s}.json")
-                provider_details = _req("GET", f"/v1/catalog/print_providers/{p_id_s}.json")
-                shipping_info = _req("GET", f"/v1/catalog/blueprints/{bp_id_s}/print_providers/{p_id_s}/shipping.json")
-
-                variant_snapshot, min_price = _extract_variant_snapshot(variants)
-
-                _log(logs, "VARIANT_SNAPSHOT " + json.dumps({"count": len(variant_snapshot), "sample": variant_snapshot[:20]}))
-
-                provider_loc = provider_details.get("location") or {}
-                provider_name = provider_details.get("title") or provider_details.get("name") or "empty"
-                loc_str = ", ".join([x for x in [provider_loc.get("city"), provider_loc.get("country")] if x]) or "empty"
-                base_desc = blueprint_details.get("description") or bp.get("description") or "empty"
-                description = f"{base_desc}\n\nFulfilled by {provider_name} — {loc_str}"
-
-                colors = sorted({v.get("color") for v in variant_snapshot if v.get("color")})
-                sizes = sorted({v.get("size") for v in variant_snapshot if v.get("size")})
-
-                product_template = {
-                    "shop_id": int(shop_id),
-                    "blueprint_id": int(bp_id_s),
-                    "print_provider_id": int(p_id_s),
-                    "name": blueprint_details.get("title") or bp.get("title") or "empty",
-                    "description": description,
-                    "tags": ["Printify", f"Provider: {provider_name}"],
-                    "options": {"Color": colors, "Size": sizes},
-                    "variants_count": len(variant_snapshot),
-                    "price_min": min_price,
-                    "variants": variant_snapshot,
-                    "shipping": shipping_info,
-                    "raw": {
-                        "blueprint": bp,
-                        "provider": p,
-                        "blueprintDetails": blueprint_details,
-                        "providerDetails": provider_details,
-                    },
-                }
+        blob = _find_first_valid_pair(logs)
+        yield flush()
 
-                results.append(product_template)
-                yield flush()
-
-                if max_templates and len(results) >= max_templates:
-                    _log(logs, f"Reached max_templates={max_templates}. Stopping.")
-                    break
-
-            if max_templates and len(results) >= max_templates:
-                break
-
-        _log(logs, f"DONE templates={len(results)} pairs_scanned={pairs_scanned}")
-
-        fd, path = tempfile.mkstemp(prefix="phase_a_", suffix=".json")
-        os.close(fd)
-        with open(path, "w", encoding="utf-8") as f:
-            json.dump(
-                {
-                    "shop_id": int(shop_id),
-                    "generated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
-                    "templates": results,
-                },
-                f,
-                indent=2,
-            )
-        download_path = path
+        result = _build_product(blob, currency or "USD", logs)
+        _log(logs, "DONE")
         yield flush()
 
     except Exception as e:
         _log(logs, f"ERROR: {e}")
-        yield "\n".join(logs), "[]", None
-
-
-def fetch_shops_for_dropdown() -> Tuple[str, dict]:
-    logs: List[str] = []
-    shops = _list_shops(logs)
-    choices = []
-    for s in shops:
-        sid = s.get("id")
-        title = s.get("title") or "untitled"
-        chan = s.get("sales_channel") or "unknown"
-        choices.append((f"{title} ({chan}) — {sid}", str(sid)))
-    return "\n".join(logs), gr.update(choices=choices, value=None)
-
-
-with gr.Blocks(title="Printify Phase A Probe") as demo:
-    gr.Markdown("Phase A: fetch shop ID (pick Atheria) + enumerate blueprint/provider templates (variants + placeholders) + download JSON.")
-
-    shops_log = gr.Textbox(label="Shop Logs", lines=6)
-    shop_dropdown = gr.Dropdown(label="Select Shop (pick Atheria)", choices=[], value=None)
-    fetch_shops_btn = gr.Button("Fetch Shops")
-
-    max_templates = gr.Number(label="Max templates (0 = no limit)", value=25, precision=0)
-    per_page = gr.Number(label="Blueprints per page", value=100, precision=0)
+        result = {"error": str(e)}
+        yield flush()
 
-    run_btn = gr.Button("Run Phase A")
 
-    logs = gr.Textbox(label="Logs", lines=16)
-    preview = gr.Textbox(label="Preview (first 3 templates)", lines=16)
-    download = gr.File(label="Download Full Phase A JSON")
+with gr.Blocks(title="Printify Catalog Probe") as demo:
+    gr.Markdown("Extract and normalize ONE Printify blueprint/provider into a structured JSON object.")
 
-    fetch_shops_btn.click(fetch_shops_for_dropdown, inputs=[], outputs=[shops_log, shop_dropdown])
+    currency = gr.Textbox(label="Currency", value="USD")
+    btn = gr.Button("Run")
+    logs = gr.Textbox(label="Logs", lines=18)
+    out = gr.Textbox(label="Output JSON", lines=18)
 
-    run_btn.click(
-        run_phase_a,
-        inputs=[shop_dropdown, max_templates, per_page],
-        outputs=[logs, preview, download],
-    )
+    btn.click(run, inputs=[currency], outputs=[logs, out])
 
-demo.queue().launch(server_name="0.0.0.0", server_port=7860)
+demo.queue(concurrency_count=1).launch()