mockup.py
2.2 kB · python · 65 lines
1from automator.core.api import printful_request2from automator.core.errors import Retry, TaskAborted3from automator.core.models import Mockup, Task4from automator.core.steps import Step56PRINTFUL_MOCKUP_URL = "v2/mockup-tasks"7FORMAT = "png"8MOCKUP_WIDTH_PX = 1000910def choose_variant(task: Task) -> int:11 return task.variants[-1].id1213def allowed(mockup: Mockup, variant_id: int) -> bool:14 if not mockup.variant_ids:15 return True16 return variant_id in mockup.variant_ids1718def fetch_files(task: Task) -> list[dict]:19 files: list[dict] = []20 for placement in task.placements:21 url = next(pf.url for pf in task.printfiles if pf.id == placement.id)22 files.append({23 "placement": placement.name,24 "technique": task.product.technique,25 "print_area_type": "simple",26 "layers": [{"type": "file", "url": url}],27 })28 return files2930def fetch_product(task: Task) -> dict:31 variant_id = choose_variant(task)32 task.metadata["variant_ids"] = [variant_id]33 style_ids = [m.id for m in task.mockups if not m.url and allowed(m, variant_id)]34 if not style_ids:35 raise TaskAborted(f"no mockup style allows variant {variant_id}")36 product = {37 "source": "catalog",38 "catalog_product_id": task.product.id,39 "catalog_variant_ids": [variant_id],40 "mockup_style_ids": style_ids,41 "placements": fetch_files(task),42 }43 if task.product.stitch_colors:44 product["product_options"] = [{"name": "stitch_color", "value": task.stitch_color}]45 return product4647def create_payload(task: Task) -> dict:48 return {49 "format": FORMAT,50 "mockup_width_px": MOCKUP_WIDTH_PX,51 "products": [fetch_product(task)],52 }5354def create_mockup_task(task: Task, payload: dict) -> Task:55 result = printful_request(task, "POST", PRINTFUL_MOCKUP_URL, data=payload)56 if not result.get("data"):57 raise TaskAborted(f"empty mockup task response for {task.key}")58 task.metadata["mockup_generator_id"] = result["data"][0]["id"]59 return task6061def mrly_mockup(task: Task) -> Task:62 payload = create_payload(task)63 task = create_mockup_task(task, payload)64 task.place(Step.PROCESS)65 raise Retry(f"Current: {Step.MOCKUP}. Next: {Step.PROCESS}")