""" Printful API integration. API Documentation: https://developers.printful.com/docs/ """ from pathlib import Path from typing import Optional import httpx from .base import PODProvider, ProductResult, MockupResult class PrintfulProvider(PODProvider): """Printful print-on-demand provider.""" name = "printful" BASE_URL = "https://api.printful.com" def __init__(self, api_key: str, sandbox: bool = False) -> None: super().__init__(api_key, sandbox) self.client = httpx.AsyncClient( base_url=self.BASE_URL, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, timeout=30.0, ) def _validate_credentials(self) -> None: """Validate Printful API token.""" if not self.api_key: raise ValueError("Printful API token is required") # Note: Printful doesn't have a separate sandbox environment # Test mode is controlled per-store async def create_product( self, design_path: Path, metadata: dict, product_config: dict, ) -> ProductResult: """Create a sync product on Printful.""" try: # 1. Upload the design file file_url = await self.upload_file(design_path) if not file_url: return ProductResult( success=False, error="Failed to upload design file" ) # 2. Build sync product payload sku = product_config.get("sku") colors = product_config.get("colors", ["black"]) sizes = product_config.get("sizes", ["M"]) placement = product_config.get("placement", "front") # Build sync variants sync_variants = [] for color in colors: for size in sizes: sync_variants.append({ "variant_id": self._get_variant_id(sku, color, size), "files": [ { "url": file_url, "type": placement, } ], "retail_price": str(product_config.get("retail_price", "25.00")), }) payload = { "sync_product": { "name": metadata.get("name", "Mycopunk Design"), "thumbnail": file_url, }, "sync_variants": sync_variants, } # 3. Create the product response = await self.client.post("/store/products", json=payload) response.raise_for_status() data = response.json() product_id = str(data.get("result", {}).get("id")) return ProductResult( success=True, product_id=product_id, data=data.get("result"), ) except httpx.HTTPStatusError as e: return ProductResult( success=False, error=f"HTTP {e.response.status_code}: {e.response.text}" ) except Exception as e: return ProductResult(success=False, error=str(e)) async def update_product( self, product_id: str, design_path: Path, metadata: dict, ) -> ProductResult: """Update an existing sync product.""" try: # Upload new design file_url = await self.upload_file(design_path) if not file_url: return ProductResult( success=False, error="Failed to upload design file" ) # Get current product to update variants current = await self.get_product(product_id) if not current: return ProductResult( success=False, error=f"Product {product_id} not found" ) # Update product name/thumbnail payload = { "sync_product": { "name": metadata.get("name"), "thumbnail": file_url, } } response = await self.client.put( f"/store/products/{product_id}", json=payload ) response.raise_for_status() # Update each variant's files for variant in current.get("sync_variants", []): variant_id = variant.get("id") await self.client.put( f"/store/variants/{variant_id}", json={ "files": [{"url": file_url, "type": "front"}] } ) return ProductResult( success=True, product_id=product_id, ) except Exception as e: return ProductResult(success=False, error=str(e)) async def generate_mockup( self, product_id: str, variant: Optional[str] = None, ) -> MockupResult: """Generate product mockup using Printful's mockup generator.""" try: # Get product info product = await self.get_product(product_id) if not product: return MockupResult( success=False, error=f"Product {product_id} not found" ) # Get first variant or specified variant variants = product.get("sync_variants", []) if not variants: return MockupResult( success=False, error="No variants found" ) target_variant = variants[0] if variant: for v in variants: if variant in str(v.get("name", "")).lower(): target_variant = v break # Request mockup generation variant_id = target_variant.get("variant_id") files = target_variant.get("files", []) if not files: return MockupResult( success=False, error="No files attached to variant" ) payload = { "variant_ids": [variant_id], "files": [{"placement": f.get("type"), "image_url": f.get("url")} for f in files], } response = await self.client.post( "/mockup-generator/create-task/variant_ids", json=payload ) response.raise_for_status() data = response.json() task_key = data.get("result", {}).get("task_key") # Poll for result (simplified - in production, use proper polling) import asyncio for _ in range(10): await asyncio.sleep(2) status_response = await self.client.get( f"/mockup-generator/task?task_key={task_key}" ) status_data = status_response.json() if status_data.get("result", {}).get("status") == "completed": mockups = status_data["result"]["mockups"] if mockups: return MockupResult( success=True, image_url=mockups[0]["mockup_url"], ) return MockupResult( success=False, error="Mockup generation timed out" ) except Exception as e: return MockupResult(success=False, error=str(e)) async def get_product(self, product_id: str) -> Optional[dict]: """Get sync product details.""" try: response = await self.client.get(f"/store/products/{product_id}") response.raise_for_status() data = response.json() return data.get("result") except Exception: return None async def list_products(self, limit: int = 100) -> list[dict]: """List all sync products.""" try: response = await self.client.get( "/store/products", params={"limit": limit} ) response.raise_for_status() data = response.json() return data.get("result", []) except Exception: return [] async def get_catalog(self, product_type: Optional[str] = None) -> list[dict]: """Get Printful product catalog.""" try: response = await self.client.get("/products") response.raise_for_status() data = response.json() products = data.get("result", []) if product_type: # Filter by type (approximate matching) type_map = { "tshirt": ["t-shirt", "tee"], "hoodie": ["hoodie", "sweatshirt"], "sticker": ["sticker"], } keywords = type_map.get(product_type, [product_type]) products = [ p for p in products if any(kw in p.get("title", "").lower() for kw in keywords) ] return products except Exception: return [] async def upload_file(self, file_path: Path) -> Optional[str]: """Upload a file to Printful.""" try: # Printful accepts URLs, so we need to either: # 1. Use their file library (POST /files) # 2. Host the file elsewhere and provide URL # For now, use file library with open(file_path, "rb") as f: files = {"file": (file_path.name, f, "image/png")} response = await self.client.post( "/files", files=files, headers={"Content-Type": None}, # Let httpx set multipart ) response.raise_for_status() data = response.json() return data.get("result", {}).get("preview_url") except Exception: return None def _get_variant_id(self, product_id: int, color: str, size: str) -> int: """ Get variant ID for a product+color+size combination. In production, this would query the catalog to find the right variant. For now, returns a placeholder. """ # TODO: Implement proper variant lookup from catalog return 0 async def close(self) -> None: """Close the HTTP client.""" await self.client.aclose()