files.py

1from automator.core.api import check_errors, shopify_request2from automator.core.errors import Retry, TaskAborted3from automator.core.models import Mockup, Task4from automator.core.steps import Step56BATCH = 2578MUTATION = """9mutation fileCreate($files: [FileCreateInput!]!) {10    fileCreate(files: $files) {11        files {12            alt13            fileStatus14            id15            ... on MediaImage {16                image {17                    url18                }19            }20        }21        userErrors {22            field23            message24        }25    }26}27"""2829def filename(mockup: Mockup) -> str:30    return f"{mockup.name}.{mockup.extension}"3132def create_file(mockup: Mockup) -> dict:33    return {34        "alt": mockup.alt,35        "contentType": "IMAGE",36        "duplicateResolutionMode": "REPLACE",37        "filename": filename(mockup),38        "originalSource": mockup.url,39    }4041def send_batch(task: Task, batch: list[Mockup]) -> list[dict]:42    variables = {"files": [create_file(mockup) for mockup in batch]}43    result = shopify_request(task, MUTATION, variables)44    return check_errors(result, "fileCreate")["files"]4546def match(mockup: Mockup, files: list[dict]) -> str:47    for file in files:48        if file.get("alt") == mockup.alt:49            return file["id"]50    for file in files:51        source = (file.get("image") or {}).get("url") or ""52        if mockup.name in source:53            return file["id"]54    raise TaskAborted(f"no shopify file for {mockup.alt}")5556def send_files(task: Task) -> Task:57    for start in range(0, len(task.mockups), BATCH):58        batch = task.mockups[start:start + BATCH]59        files = send_batch(task, batch)60        for mockup in batch:61            mockup.shopify_id = match(mockup, files)62    return task6364def mrly_files(task: Task) -> Task:65    task = send_files(task)66    task.place(Step.STATUS)67    raise Retry(f"Current: {Step.FILES}. Next: {Step.STATUS}")