sync.py

1import time2from automator.core.api import printful_request3from automator.core.errors import Retry, TaskAborted4from automator.core.models import Task, Variant5from automator.core.steps import Step67PRINTFUL_SYNC_URL = "sync/variant/"8VISIBLE = False9DELAY = 0.610BATCH = 301112def fetch_files(task: Task) -> list[dict]:13    files: list[dict] = []14    for placement in task.placements:15        printfile = next(pf for pf in task.printfiles if pf.id == placement.id)16        files.append({17            "type": placement.name,18            "url": printfile.url,19            "filename": f"{printfile.name}.png",20            "visible": VISIBLE,21        })22    return files2324def create_payload(task: Task, variant: Variant) -> dict:25    return {26        "variant_id": variant.id,27        "retail_price": variant.cost,28        "sku": variant.name,29        "is_ignored": False,30        "files": fetch_files(task),31        "options": [{"id": "stitch_color", "value": task.stitch_color}],32    }3334def sync_variant(task: Task, variant: Variant) -> Task:35    payload = create_payload(task, variant)36    result = printful_request(37        task,38        "PUT",39        f"{PRINTFUL_SYNC_URL}{variant.printful_id}",40        data=payload,41    )42    if not result:43        raise Retry(f"Current: {Step.SYNC}. Next: {Step.SYNC}")44    sync_variant_data = result["result"]["sync_variant"]45    if not sync_variant_data["synced"]:46        raise TaskAborted(f"variant {variant.name} not synced")47    if sync_variant_data["is_ignored"]:48        raise TaskAborted(f"variant {variant.name} ignored")49    variant.synced = True50    return task5152def unsynced_variants(task: Task) -> list[Variant]:53    return [v for v in task.variants if not v.synced]5455def process_batch(task: Task) -> Task:56    batch = unsynced_variants(task)[:BATCH]57    for i, variant in enumerate(batch):58        task = sync_variant(task, variant)59        if i < len(batch) - 1:60            time.sleep(DELAY)61    return task6263def mrly_sync(task: Task) -> Task:64    task = process_batch(task)65    if unsynced_variants(task):66        task.place(Step.SYNC)67        raise Retry(f"Current: {Step.SYNC}. Next: {Step.SYNC}")68    task.place(Step.PUBLISH)69    return task