326 lines
11 KiB
Python
326 lines
11 KiB
Python
"""
|
|
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()
|