Update app.py

app.py

@@ -8,7 +8,6 @@ import requests
 import gradio as gr
 import base64
 import re
-import hashlib
 
 PRINTIFY_BASE = "https://api.printify.com"
 DEFAULT_BASE_PRICE = 0.001
@@ -33,18 +32,6 @@ def _auth_headers() -> Dict[str, str]:
     return {"Authorization": f"Bearer {token}"}
 
 
-def _batch_fingerprint(shop_id: str, blueprint_id: int, provider_id: int, enabled_variant_ids: List[int]) -> str:
-    ids = sorted(int(x) for x in enabled_variant_ids)
-    payload = {
-        "shop_id": str(shop_id),
-        "blueprint_id": int(blueprint_id),
-        "provider_id": int(provider_id),
-        "enabled_variant_ids": ids,
-    }
-    raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
-    return hashlib.sha1(raw).hexdigest()
-
-
 def _parse_retry_after_seconds(r: requests.Response) -> Optional[float]:
     ra = r.headers.get("Retry-After")
     if ra:
@@ -78,15 +65,9 @@ def _req(
     path: str,
     json_body: Optional[Dict[str, Any]] = None,
     logs: Optional[List[str]] = None,
-    extra_headers: Optional[Dict[str, str]] = None,
 ) -> Any:
-    headers = _auth_headers()
-    if extra_headers:
-        for k, v in extra_headers.items():
-            headers[str(k)] = str(v)
-
     kwargs: Dict[str, Any] = {
-        "headers": headers,
+        "headers": _auth_headers(),
         "timeout": 60,
     }
     if json_body is not None:
@@ -448,49 +429,6 @@ def _log_catalog_variants(logs: List[str], product_info: Dict[str, Any], limit:
             break
 
 
-def _find_existing_product_by_fp(
-    logs: List[str],
-    shop_id: str,
-    fp: str,
-) -> Optional[Dict[str, Any]]:
-    page = 1
-    while True:
-        resp = _req(
-            "GET",
-            f"/v1/shops/{shop_id}/products.json?page={page}&limit=50",
-            logs=logs,
-        )
-        if isinstance(resp, dict):
-            items = resp.get("data") or resp.get("products") or resp.get("items")
-            if isinstance(items, list):
-                products = items
-            else:
-                products = []
-        elif isinstance(resp, list):
-            products = resp
-        else:
-            products = []
-
-        if not products:
-            break
-
-        for p in products:
-            if not isinstance(p, dict):
-                continue
-            desc = p.get("description") or ""
-            if isinstance(desc, str) and f"BATCH_FP={fp}" in desc:
-                _log(
-                    logs,
-                    f"FOUND_EXISTING_PRODUCT_FOR_FP shop_id={shop_id} product_id={p.get('id')} batch_fp={fp}",
-                )
-                return p
-
-        page += 1
-
-    _log(logs, f"NO_EXISTING_PRODUCT_FOR_FP shop_id={shop_id} batch_fp={fp}")
-    return None
-
-
 def _create_product_all_variants_with_grid(
     logs: List[str],
     shop_id: str,
@@ -528,29 +466,6 @@ def _create_product_all_variants_with_grid(
             except Exception:
                 continue
 
-    all_ids_for_fp = []
-    for v in all_variants:
-        if not isinstance(v, dict):
-            continue
-        vid = v.get("id")
-        if vid is None:
-            continue
-        try:
-            all_ids_for_fp.append(int(vid))
-        except Exception:
-            continue
-
-    fp = _batch_fingerprint(str(shop_id), int(bp_id), int(provider_id), sorted(all_ids_for_fp))
-
-    existing = _find_existing_product_by_fp(logs, shop_id, fp)
-    if isinstance(existing, dict) and existing.get("id"):
-        _log(
-            logs,
-            f"PHASE_B_REUSE_EXISTING_PRODUCT shop_id={shop_id} blueprint_id={bp_id} "
-            f"provider_id={provider_id} existing_product_id={existing.get('id')} batch_fp={fp}",
-        )
-        return existing
-
     img_id = upload.get("id")
     img_w = float(upload.get("width") or 0)
     img_h = float(upload.get("height") or 0)
@@ -636,7 +551,7 @@ def _create_product_all_variants_with_grid(
     provider_name = provider_details.get("title") or provider_details.get("name") or str(provider_id)
 
     title = (product_info.get("name") or "Printify Grid Test") + f" — {provider_name}"
-    description = (product_info.get("description") or "") + f"\n\nBATCH_FP={fp}"
+    description = product_info.get("description") or ""
 
     payload = {
         "title": title,
@@ -650,8 +565,7 @@ def _create_product_all_variants_with_grid(
     _log(
         logs,
         f"PHASE_B_CREATE shop_id={shop_id} blueprint_id={bp_id} provider_id={provider_id} "
-        f"variants_payload_total={len(variants_payload)} enabled={enabled_count} print_areas_payload={len(print_areas_payload)} "
-        f"idempotency_key={fp}",
+        f"variants_payload_total={len(variants_payload)} enabled={enabled_count} print_areas_payload={len(print_areas_payload)}",
     )
 
     created = _req(
@@ -667,126 +581,6 @@ def _create_product_all_variants_with_grid(
     return created
 
 
-def _update_product_stitch_cumulative_with_grid(
-    logs: List[str],
-    shop_id: str,
-    product_id: str,
-    all_variants: List[Dict[str, Any]],
-    enabled_ids: set,
-    upload: Dict[str, Any],
-    title: Optional[str],
-    description: Optional[str],
-) -> Dict[str, Any]:
-    img_id = upload.get("id")
-    img_w = float(upload.get("width") or 0)
-    img_h = float(upload.get("height") or 0)
-
-    variants_payload = []
-    print_areas_payload = []
-
-    enabled_count = 0
-    for v in all_variants:
-        if not isinstance(v, dict):
-            continue
-        vid = v.get("id")
-        if vid is None:
-            continue
-        try:
-            vid_i = int(vid)
-        except Exception:
-            continue
-        is_on = vid_i in enabled_ids
-        if is_on:
-            enabled_count += 1
-        variants_payload.append({
-            "id": vid_i,
-            "price": 1,
-            "is_enabled": bool(is_on),
-        })
-
-    for v in all_variants:
-        if not isinstance(v, dict):
-            continue
-        vid = v.get("id")
-        if vid is None:
-            continue
-        try:
-            vid_i = int(vid)
-        except Exception:
-            continue
-        if vid_i not in enabled_ids:
-            continue
-
-        placeholders = v.get("placeholders") or []
-        if not isinstance(placeholders, list) or not placeholders:
-            continue
-
-        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
-            scale = _scale_fill(pwf, phf, img_w, img_h)
-            ph_payload.append({
-                "position": pos,
-                "images": [{
-                    "id": img_id,
-                    "x": 0.5,
-                    "y": 0.5,
-                    "scale": scale,
-                    "angle": 0,
-                }],
-            })
-
-        if ph_payload:
-            print_areas_payload.append({
-                "variant_ids": [vid_i],
-                "placeholders": ph_payload,
-            })
-
-    if not variants_payload:
-        raise RuntimeError("No variants payload could be built for stitch update.")
-    if enabled_count > 0 and not print_areas_payload:
-        raise RuntimeError("Enabled variants exist but no print_areas payload could be generated (stitch update).")
-
-    payload: Dict[str, Any] = {
-        "variants": variants_payload,
-        "print_areas": print_areas_payload,
-    }
-    if isinstance(title, str) and title:
-        payload["title"] = title
-    if isinstance(description, str) and description:
-        payload["description"] = description
-
-    _log(
-        logs,
-        f"PHASE_B_STITCH_UPDATE product_id={product_id} "
-        f"variants_payload_total={len(variants_payload)} enabled={enabled_count} "
-        f"print_areas_payload={len(print_areas_payload)}",
-    )
-
-    upd = _req(
-        "PUT",
-        f"/v1/shops/{shop_id}/products/{product_id}.json",
-        json_body=payload,
-        logs=logs,
-    )
-    if not isinstance(upd, dict):
-        raise RuntimeError(f"Unexpected update response: {str(upd)[:500]}")
-
-    _log(logs, f"PHASE_B_STITCH_UPDATE_DONE product_id={product_id}")
-    return upd
-
-
 def _get_product(logs: List[str], shop_id: str, product_id: str) -> Dict[str, Any]:
     prod = _req("GET", f"/v1/shops/{shop_id}/products/{product_id}.json", logs=logs)
     if not isinstance(prod, dict):
@@ -853,16 +647,10 @@ def _update_prices_to_cost_plus_margin(
     if not payload_variants:
         raise RuntimeError("No variant prices could be computed from costs.")
 
-    upd_body: Dict[str, Any] = {"variants": payload_variants}
-    if isinstance(product.get("title"), str):
-        upd_body["title"] = product.get("title")
-    if isinstance(product.get("description"), str):
-        upd_body["description"] = product.get("description")
-
     upd = _req(
         "PUT",
         f"/v1/shops/{shop_id}/products/{product_id}.json",
-        json_body=upd_body,
+        json_body={"variants": payload_variants},
         logs=logs,
     )
     if not isinstance(upd, dict):
@@ -1045,116 +833,53 @@ def phase_b(currency: str) -> Generator[Tuple[str, str], None, None]:
 
             else:
                 total_chunks = (total_variants + CHUNK_TARGET_VARIANTS - 1) // CHUNK_TARGET_VARIANTS
-
-                enabled_ids: set = set()
                 chunk_index = 0
-
-                first_chunk = variants_all[0:CHUNK_TARGET_VARIANTS]
-                for v in first_chunk:
-                    if not isinstance(v, dict):
-                        continue
-                    vid = v.get("id")
-                    if vid is None:
-                        continue
-                    try:
-                        enabled_ids.add(int(vid))
-                    except Exception:
-                        continue
-
-                p_info_first = dict(product_info_full)
-                p_info_first["variants"] = first_chunk
-                p_info_first["options"] = _rebuild_options_for_variants(first_chunk)
-                p_info_first["_allVariants"] = variants_all
-                p_info_first["_enabledVariantIds"] = sorted(list(enabled_ids))
-
-                _log(
-                    logs,
-                    f"PROVIDER_CHUNK_CREATE provider_id={provider_id} name={provider_name} "
-                    f"chunk_index={chunk_index} total_chunks={total_chunks} "
-                    f"offset={0} size={len(first_chunk)} total_variants={total_variants}",
-                )
-                yield flush()
-
-                created = _create_product_all_variants_with_grid(
-                    logs,
-                    shop_id=shop_id,
-                    blob=blob,
-                    product_info=p_info_first,
-                    upload=upload,
-                )
-                created_id = str(created.get("id") or "")
-                if not created_id:
-                    raise RuntimeError("Created product response missing id.")
-                yield flush()
-
-                prod0 = _get_product(logs, shop_id, created_id)
-                yield flush()
-
-                provider_runs.append(
-                    {
-                        "providerId": provider_id,
-                        "providerName": provider_name,
-                        "chunkIndex": chunk_index,
-                        "totalChunks": total_chunks,
-                        "catalogVariantCountTotal": total_variants,
-                        "catalogVariantCountChunk": len(first_chunk),
-                        "shopVariantCountAfterCreate": len(prod0.get("variants") or []),
-                        "shopVariantCountFinal": len(prod0.get("variants") or []),
-                        "productId": created_id,
-                        "priceUpdateResponse": {},
-                    }
-                )
-
-                offset = CHUNK_TARGET_VARIANTS
-                chunk_index = 1
-
+                offset = 0
                 while offset < total_variants:
                     chunk = variants_all[offset: offset + CHUNK_TARGET_VARIANTS]
-                    for v in chunk:
-                        if not isinstance(v, dict):
-                            continue
-                        vid = v.get("id")
-                        if vid is None:
-                            continue
-                        try:
-                            enabled_ids.add(int(vid))
-                        except Exception:
-                            continue
+                    p_info_chunk = dict(product_info_full)
+                    p_info_chunk["variants"] = chunk
+                    p_info_chunk["options"] = _rebuild_options_for_variants(chunk)
+                    p_info_chunk["_allVariants"] = variants_all
+                    p_info_chunk["_enabledVariantIds"] = [int(v.get("id")) for v in chunk if v.get("id") is not None]
 
                     _log(
                         logs,
-                        f"PROVIDER_CHUNK_STITCH provider_id={provider_id} name={provider_name} "
+                        f"PROVIDER_CHUNK provider_id={provider_id} name={provider_name} "
                         f"chunk_index={chunk_index} total_chunks={total_chunks} "
-                        f"offset={offset} size={len(chunk)} total_variants={total_variants} product_id={created_id}",
+                        f"offset={offset} size={len(chunk)} total_variants={total_variants}",
                     )
                     yield flush()
 
-                    current_prod = _get_product(logs, shop_id, created_id)
-                    yield flush()
-
-                    upd_stitch = _update_product_stitch_cumulative_with_grid(
+                    created = _create_product_all_variants_with_grid(
                         logs,
                         shop_id=shop_id,
-                        product_id=created_id,
-                        all_variants=variants_all,
-                        enabled_ids=enabled_ids,
+                        blob=blob,
+                        product_info=p_info_chunk,
                         upload=upload,
-                        title=current_prod.get("title"),
-                        description=current_prod.get("description"),
                     )
+                    created_id = str(created.get("id") or "")
+                    if not created_id:
+                        raise RuntimeError("Created product response missing id.")
                     yield flush()
 
-                    prod_u = _get_product(logs, shop_id, created_id)
+                    prod1 = _get_product(logs, shop_id, created_id)
                     yield flush()
 
                     _log(
                         logs,
                         f"COMPARE_COUNTS provider={provider_id} chunk_index={chunk_index} "
                         f"catalog_chunk={len(chunk)} "
-                        f"shop={len(prod_u.get('variants') or [])}",
+                        f"shop={len(prod1.get('variants') or [])}",
                     )
                     yield flush()
 
+                    upd = _update_prices_to_cost_plus_margin(logs, shop_id, created_id, prod1)
+                    yield flush()
+
+                    prod2 = _get_product(logs, shop_id, created_id)
+                    yield flush()
+
                     provider_runs.append(
                         {
                             "providerId": provider_id,
@@ -1163,40 +888,16 @@ def phase_b(currency: str) -> Generator[Tuple[str, str], None, None]:
                             "totalChunks": total_chunks,
                             "catalogVariantCountTotal": total_variants,
                             "catalogVariantCountChunk": len(chunk),
-                            "shopVariantCountAfterCreate": len(prod_u.get("variants") or []),
-                            "shopVariantCountFinal": len(prod_u.get("variants") or []),
+                            "shopVariantCountAfterCreate": len(prod1.get("variants") or []),
+                            "shopVariantCountFinal": len(prod2.get("variants") or []),
                             "productId": created_id,
-                            "priceUpdateResponse": upd_stitch,
+                            "priceUpdateResponse": upd,
                         }
                     )
 
                     offset += CHUNK_TARGET_VARIANTS
                     chunk_index += 1
 
-                prod_before_price = _get_product(logs, shop_id, created_id)
-                yield flush()
-
-                upd_prices = _update_prices_to_cost_plus_margin(logs, shop_id, created_id, prod_before_price)
-                yield flush()
-
-                prod_after_price = _get_product(logs, shop_id, created_id)
-                yield flush()
-
-                provider_runs.append(
-                    {
-                        "providerId": provider_id,
-                        "providerName": provider_name,
-                        "chunkIndex": "PRICE_UPDATE",
-                        "totalChunks": total_chunks,
-                        "catalogVariantCountTotal": total_variants,
-                        "catalogVariantCountChunk": 0,
-                        "shopVariantCountAfterCreate": len(prod_before_price.get("variants") or []),
-                        "shopVariantCountFinal": len(prod_after_price.get("variants") or []),
-                        "productId": created_id,
-                        "priceUpdateResponse": upd_prices,
-                    }
-                )
-
         result["providerRuns"] = provider_runs
         _log(logs, "PHASE_B_DONE")
         yield flush()