Update app.py

app.py

@@ -2,7 +2,7 @@ import os
 import time
 import json
 import tempfile
-from typing import Dict, Any, List, Generator, Tuple
+from typing import Dict, Any, List, Generator, Tuple, Optional
 
 import requests
 import gradio as gr
@@ -10,6 +10,8 @@ import gradio as gr
 PRINTIFY_BASE = "https://api.printify.com"
 DEFAULT_BASE_PRICE = 0.00
 
+GRID_FILENAME = "grid.png"
+
 
 def _now() -> str:
     return time.strftime("%H:%M:%S")
@@ -26,11 +28,12 @@ def _auth_headers() -> Dict[str, str]:
     return {"Authorization": f"Bearer {token}"}
 
 
-def _req(method: str, path: str) -> Any:
+def _req(method: str, path: str, json_body: Optional[Dict[str, Any]] = None) -> Any:
     r = requests.request(
         method,
         f"{PRINTIFY_BASE}{path}",
-        headers=_auth_headers(),
+        headers={**_auth_headers(), "Content-Type": "application/json"} if json_body is not None else _auth_headers(),
+        json=json_body,
         timeout=60,
     )
     if r.status_code >= 400:
@@ -42,6 +45,41 @@ def _log(logs: List[str], msg: str):
     logs.append(f"[{_now()}] {msg}")
 
 
+def _as_list(resp: Any) -> List[Any]:
+    if isinstance(resp, list):
+        return resp
+    if isinstance(resp, dict):
+        for k in ("data", "items", "uploads", "results"):
+            v = resp.get(k)
+            if isinstance(v, list):
+                return v
+    return []
+
+
+def _space_id_parts() -> Tuple[str, str]:
+    owner = os.environ.get("HF_SPACE_OWNER") or os.environ.get("SPACE_OWNER") or ""
+    name = os.environ.get("HF_SPACE_NAME") or os.environ.get("SPACE_NAME") or ""
+    if owner and name:
+        return owner, name
+
+    sid = os.environ.get("SPACE_ID") or os.environ.get("HF_SPACE_ID") or ""
+    if sid and "/" in sid:
+        a, b = sid.split("/", 1)
+        if a and b:
+            return a, b
+
+    # last-resort: hard error with actionable message
+    raise RuntimeError("Missing HF space identity. Set HF_SPACE_OWNER and HF_SPACE_NAME (or SPACE_ID=owner/name).")
+
+
+def _hf_asset_url(rel_path: str) -> str:
+    owner, name = _space_id_parts()
+    rel_path = (rel_path or "").lstrip("/")
+    if not rel_path:
+        raise RuntimeError("Empty asset path for HF URL.")
+    return f"https://huggingface.co/spaces/{owner}/{name}/resolve/main/{rel_path}?download=1"
+
+
 def _find_first_valid_pair(logs: List[str]) -> Dict[str, Any]:
     _log(logs, "Listing blueprints")
     blueprints = _req("GET", "/v1/catalog/blueprints.json")
@@ -177,6 +215,137 @@ def _build_product(blob: Dict[str, Any], currency: str, logs: List[str]) -> Dict
     }
 
 
+def _pick_shop_id(shops_resp: Any) -> str:
+    shops = _as_list(shops_resp)
+    for s in shops:
+        if isinstance(s, dict) and s.get("id"):
+            return str(s["id"])
+    raise RuntimeError("No shops found on /v1/shops.json for this token.")
+
+
+def _upload_grid_image(logs: List[str]) -> Dict[str, Any]:
+    url = _hf_asset_url(GRID_FILENAME)
+    _log(logs, f"GRID_URL {url}")
+
+    # Upload-by-URL (Printify fetches it)
+    resp = _req(
+        "POST",
+        "/v1/uploads/images.json",
+        json_body={"file_name": GRID_FILENAME, "url": url},
+    )
+
+    if not isinstance(resp, dict) or not resp.get("id"):
+        raise RuntimeError(f"Unexpected upload response shape: {str(resp)[:500]}")
+
+    _log(
+        logs,
+        f"GRID_UPLOAD id={resp.get('id')} file={resp.get('file_name') or resp.get('name')} w={resp.get('width')} h={resp.get('height')}",
+    )
+    return resp
+
+
+def _scale_fill(ph_w: float, ph_h: float, img_w: float, img_h: float) -> float:
+    # Printify scale is defined relative to placeholder width, so:
+    # scale=1 means image width == placeholder width.
+    # For "fill to placeholder" (cover), increase scale if height would underfill.
+    if ph_w <= 0 or ph_h <= 0 or img_w <= 0 or img_h <= 0:
+        return 1.0
+    s = (ph_h * img_w) / (ph_w * img_h)
+    if s < 1.0:
+        s = 1.0
+    return float(s)
+
+
+def _phase_b_create_one_product(
+    logs: List[str],
+    shop_id: str,
+    blob: Dict[str, Any],
+    upload: Dict[str, Any],
+    currency: str,
+) -> Dict[str, Any]:
+    bp_id = str((blob.get("blueprint") or {}).get("id"))
+    provider_id = str((blob.get("provider") or {}).get("id"))
+
+    variants = blob.get("variants") or []
+    if not isinstance(variants, list) or not variants:
+        raise RuntimeError("No variants in blob.")
+
+    # Minimal: one variant only for the first Phase B test.
+    v = variants[0]
+    variant_id = v.get("id")
+    if variant_id is None:
+        raise RuntimeError("First variant missing id.")
+
+    placeholders = v.get("placeholders")
+    if not isinstance(placeholders, list):
+        placeholders = []
+
+    img_id = upload.get("id")
+    img_w = upload.get("width")
+    img_h = upload.get("height")
+    try:
+        img_w = float(img_w) if img_w is not None else 0.0
+        img_h = float(img_h) if img_h is not None else 0.0
+    except Exception:
+        img_w, img_h = 0.0, 0.0
+
+    ph_payload = []
+    for ph in placeholders:
+        if not isinstance(ph, dict):
+            continue
+        pos = ph.get("position")
+        pw = ph.get("width")
+        phh = ph.get("height")
+        if not pos or pw is None or phh is None:
+            continue
+        try:
+            pwf = float(pw)
+            phf = float(phh)
+        except Exception:
+            continue
+        s = _scale_fill(pwf, phf, img_w, img_h)
+        ph_payload.append({
+            "position": pos,
+            "images": [{
+                "id": img_id,
+                "x": 0.5,
+                "y": 0.5,
+                "scale": s,
+                "angle": 0,
+            }],
+        })
+        _log(logs, f"FILL_PLACEHOLDER pos={pos} ph={int(pwf)}x{int(phf)} img={int(img_w)}x{int(img_h)} scale={round(s, 6)}")
+
+    if not ph_payload:
+        raise RuntimeError("No placeholders found on chosen variant; cannot place image.")
+
+    title = (blob.get("blueprintDetails") or {}).get("title") or "Printify Test Product"
+    description = (blob.get("blueprintDetails") or {}).get("description") or ""
+
+    payload = {
+        "title": title,
+        "description": description,
+        "blueprint_id": int(bp_id),
+        "print_provider_id": int(provider_id),
+        "variants": [{
+            "id": int(variant_id),
+            "price": 0,  # keep minimal for test; adjust later
+            "is_enabled": True,
+        }],
+        "print_areas": [{
+            "variant_ids": [int(variant_id)],
+            "placeholders": ph_payload,
+        }],
+    }
+
+    _log(logs, f"PHASE_B_CREATE shop_id={shop_id} blueprint_id={bp_id} provider_id={provider_id} variant_id={variant_id}")
+    created = _req("POST", f"/v1/shops/{shop_id}/products.json", json_body=payload)
+    if not isinstance(created, dict):
+        raise RuntimeError(f"Unexpected create response: {str(created)[:500]}")
+    _log(logs, f"PHASE_B_CREATED product_id={created.get('id')}")
+    return created
+
+
 def run(currency: str) -> Generator[Tuple[str, str], None, None]:
     logs: List[str] = []
     result: Dict[str, Any] = {}
@@ -192,27 +361,7 @@ def run(currency: str) -> Generator[Tuple[str, str], None, None]:
         _log(logs, f"SHOP_LIST {json.dumps(shops)}")
         yield flush()
 
-        uploads = _req("GET", "/v1/uploads.json?limit=1")
-        latest = uploads[0] if isinstance(uploads, list) and uploads else None
-
-        if latest:
-            _log(
-                logs,
-                f"LATEST_UPLOAD id={latest.get('id')} "
-                f"file={latest.get('file_name')} "
-                f"w={latest.get('width')} h={latest.get('height')}"
-            )
-            result["latestUpload"] = {
-                "id": latest.get("id"),
-                "file_name": latest.get("file_name"),
-                "width": latest.get("width"),
-                "height": latest.get("height"),
-            }
-        else:
-            _log(logs, "LATEST_UPLOAD none found")
-
-        yield flush()
-
+        # (keep your original shops temp file write exactly as-is)
         fd, path = tempfile.mkstemp(prefix="shops_", suffix=".json")
         os.close(fd)
         with open(path, "w", encoding="utf-8") as f:
@@ -248,32 +397,26 @@ def phase_b(currency: str) -> Generator[Tuple[str, str], None, None]:
         _log(logs, f"SHOP_LIST {json.dumps(shops)}")
         yield flush()
 
-        uploads = _req("GET", "/v1/uploads.json?limit=1")
-        latest = uploads[0] if isinstance(uploads, list) and uploads else None
-
-        if latest:
-            _log(
-                logs,
-                f"LATEST_UPLOAD id={latest.get('id')} "
-                f"file={latest.get('file_name')} "
-                f"w={latest.get('width')} h={latest.get('height')}"
-            )
-            result["latestUpload"] = {
-                "id": latest.get("id"),
-                "file_name": latest.get("file_name"),
-                "width": latest.get("width"),
-                "height": latest.get("height"),
-            }
-        else:
-            _log(logs, "LATEST_UPLOAD none found")
+        shop_id = _pick_shop_id(shops)
+        _log(logs, f"SHOP_ID {shop_id}")
+        yield flush()
 
+        upload = _upload_grid_image(logs)
+        result["gridUpload"] = {
+            "id": upload.get("id"),
+            "file_name": upload.get("file_name") or upload.get("name"),
+            "width": upload.get("width"),
+            "height": upload.get("height"),
+            "url": upload.get("preview_url") or upload.get("url"),
+        }
         yield flush()
 
         blob = _find_first_valid_pair(logs)
         yield flush()
 
-        result = _build_product(blob, currency or "USD", logs)
-        _log(logs, "PHASE_B_READY")
+        created = _phase_b_create_one_product(logs, shop_id, blob, upload, currency or "USD")
+        result["createdProduct"] = created
+        _log(logs, "PHASE_B_DONE")
         yield flush()
 
     except Exception as e:
@@ -294,4 +437,4 @@ with gr.Blocks(title="Printify Catalog Probe") as demo:
     btn.click(run, inputs=[currency], outputs=[logs, out])
     btn_b.click(phase_b, inputs=[currency], outputs=[logs, out])
 
-demo.queue().launch()
+demo.queue