process.py

1from automator.core.api import printful_request2from automator.core.errors import Retry, TaskAborted3from automator.core.models import Task4from automator.core.steps import Step56PRINTFUL_POLL_URL = "v2/mockup-tasks?id="7MAX_PENDING_ATTEMPTS = 1089def count_attempt(task: Task) -> int:10    count = task.metadata.get("mockup_pending_count", 0) + 111    task.metadata["mockup_pending_count"] = count12    if count >= MAX_PENDING_ATTEMPTS:13        raise TaskAborted(f"mockup_pending_count {count} for task {task.key}")14    return count1516def fetch_result(task: Task) -> dict:17    generator_id = task.metadata["mockup_generator_id"]18    result = printful_request(task, "GET", f"{PRINTFUL_POLL_URL}{generator_id}")19    if not result.get("data"):20        count = count_attempt(task)21        raise Retry(f"Current: {Step.PROCESS}. Next: {Step.PROCESS} (empty {count}/{MAX_PENDING_ATTEMPTS})")22    data = result["data"][0]23    status = data["status"]24    if status == "completed":25        return data26    if status == "failed":27        raise TaskAborted(f"printful mockup task failed for {task.key}")28    count = count_attempt(task)29    raise Retry(f"Current: {Step.PROCESS}. Next: {Step.PROCESS} ({status} {count}/{MAX_PENDING_ATTEMPTS})")3031def parse_result(task: Task, result: dict) -> Task:32    mockups = {m.id: m for m in task.mockups}33    for variant_data in result["catalog_variant_mockups"]:34        for mockup_data in variant_data["mockups"]:35            style_id = mockup_data["style_id"]36            if style_id not in mockups:37                raise TaskAborted(f"unknown mockup style {style_id} for {task.key}")38            mockups[style_id].url = mockup_data["mockup_url"]39    return task4041def mrly_process(task: Task) -> Task:42    result = fetch_result(task)43    task = parse_result(task, result)44    task.mockups = [m for m in task.mockups if m.url]45    task.mockups = sorted(task.mockups, key=lambda m: (m.is_tile, m.id))46    task.place(Step.FILES)47    return task