Supplier Catalog Draft CSV Preparation

Turn a supplied strict supplier CSV into a draft/unpublished product CSV, with validation, independent source-to-output reconciliation and hashes. For 1–25 new products, 1–100 variants and one or two named options. No scraping, store access or import is performed.

Price: USD 0.10 per successfully prepared full batch, including the demo. Invalid or ambiguous input produces no completed-batch item or its USD 0.10 charge; it fails with bounded redacted validation feedback. Platform resource/account fees may still apply. No manual data cleanup, remediation or consultancy is included.

Quick start

To run only the immutable fictional example, use this input. A successful demo is one paid full batch:

{"useDemo": true}

For your own data, first verify that you have rights to process it, it is nonpersonal/nonconfidential, the handles are genuinely new in the intended store, and the declared store currency is correct. Those are your declarations; the Actor cannot check a store. Do not submit credentials, private image URLs or sensitive data. The strict supplier template is different from Shopify's import/export schema.

Here is the complete fictional supplier template to save as supplier.csv. Replace it only with verified data within the contract below:

handle,title,sku,option1_name,option1_value,option2_name,option2_value,price
fictional-sample-mug,Fictional Sample Mug,DEMO-MUG-BLUE,Color,Blue,,,12.50
fictional-sample-mug,Fictional Sample Mug,DEMO-MUG-GREEN,Color,Green,,,12.50

The copyable helpers require Python 3 on Linux/POSIX with safe no-follow/nonblocking file opening and hard-link support, plus trusted parent directories. Unsupported file operations fail closed.

Save the following standalone Python 3 code as prepare_input.py. It needs only the standard library and preserves the raw UTF-8 CSV bytes, including line endings/BOM. Run this command only when both declarations are true:

python3 prepare_input.py --supplier supplier.csv --currency USD --new-products-only --acknowledge-authorized-product-data --output actor-input.json

Paste the resulting JSON into the Actor input or send it as the body of a separately authorized Actor run. The helper does not submit anything. It refuses an existing output path; use a trusted directory. Do not edit/normalize the CSV inside the generated JSON if you want the same input hash.

"""Prepare an exact-byte supplier CSV Actor envelope offline; standard library only."""
import argparse
import json
import os
from pathlib import Path
import re
import stat
MAX_FILE=1024*1024


def read(path):
    if any(type(getattr(os,name,None)) is not int for name in ('O_NOFOLLOW','O_NONBLOCK')):raise ValueError()
    fd=os.open(path,os.O_RDONLY|os.O_NONBLOCK|os.O_NOFOLLOW)
    with os.fdopen(fd,'rb') as handle:
        info=os.fstat(handle.fileno())
        if not stat.S_ISREG(info.st_mode) or info.st_size>MAX_FILE:raise ValueError()
        raw=handle.read(MAX_FILE+1)
    if not raw or len(raw)>MAX_FILE:raise ValueError()
    return raw.decode('utf-8')


class QuietParser(argparse.ArgumentParser):
    def error(self,message):raise ValueError()


def main():
    parser=QuietParser(description=__doc__,allow_abbrev=False)
    parser.add_argument('--supplier',required=True,type=Path)
    parser.add_argument('--currency',required=True)
    parser.add_argument('--output',required=True,type=Path)
    parser.add_argument('--acknowledge-authorized-product-data',required=True,action='store_true',
                        help='Confirm rights to process nonpersonal, nonconfidential supplied product data.')
    parser.add_argument('--new-products-only',required=True,action='store_true',
                        help='Declare independently verified new handles; no existing-product updates.')
    try:
        args=parser.parse_args()
        if re.fullmatch(r'[A-Z]{3}',args.currency) is None:raise ValueError()
        value={'useDemo':False,'authorizedProductData':True,'newProductsOnly':True,
               'currency':args.currency,'supplierCsv':read(args.supplier)}
        data=(json.dumps(value,ensure_ascii=True,allow_nan=False,indent=2)+'\n').encode('ascii')
        if len(data)>6_400_000:raise ValueError()
        fd=os.open(args.output,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
        try:
            with os.fdopen(fd,'wb') as handle:
                if handle.write(data)!=len(data):raise OSError()
                handle.flush();os.fsync(handle.fileno())
        except BaseException:
            args.output.unlink(missing_ok=True)
            raise
        print('Prepared exact UTF-8 supplier input. The Actor still validates and reconciles the batch.')
        return 0
    except (ValueError,OSError,UnicodeError):
        print('Rejected: invalid arguments, unsafe input or existing/unwritable output. No input values logged.')
        return 2


if __name__=='__main__':raise SystemExit(main())

Accepted supplier contract

The Actor accepts a JSON object with only useDemo, authorizedProductData, newProductsOnly, currency and supplierCsv. useDemo is always an explicit JSON boolean. Demo mode accepts absent/empty currency and supplierCsv, and absent/false declarations; any supplied own-data value rejects. The UI defaults to demo, but the runtime does not infer a missing useDemo.

Own-data mode requires exactly all five keys: useDemo false, both declarations literal true, currency a string matching [A-Z]{3}, supplierCsv a nonempty UTF-8 string of at most 1,048,576 encoded bytes. Boolean/string/number substitutes reject. Currency is caller-declared and syntax-checked; no currency verification/conversion is performed. The escaped JSON envelope must fit 6,400,000 bytes. Duplicate keys, extra keys, arrays/nesting, nonfinite numbers and invalid Unicode reject. Source bytes, including BOM/CRLF, are retained when hashing; no trimming or silent repair.

Required CSV headers, in any order and exactly once: handle,title,sku,option1_name,option1_value,option2_name,option2_value,price. Optional headers product_image_url,image_alt_text,variant_image_url must all be present together or all absent. Unknown columns and differing row widths reject. UTF-8 with optional BOM; commas and standard double-quote escaping, LF or CRLF. Bare CR, broken quoting and cell newlines reject.

At least one variant, at most 100 variants and 25 distinct new product handles. Each product has one or two named options; option1 is required, option2 name/value must both be nonempty or both empty. No default-title/no-option products. handle is lowercase ASCII alphanumeric words separated by single hyphens. Title and option names must agree across the same handle. SKU identities are unique batch-wide and option combinations unique per product under NFKC/casefold; values themselves are preserved. Handles must independently be verified new by the caller; the declaration is not a store check.

Price is an explicit positive ASCII decimal <=999999.99 with at most two fractional digits; output pads to two digits. Leading-zero identifiers are retained. Surrounding whitespace, Unicode category-C characters and NFKC-normalized formula prefixes =,+,-,@ block. Character limits: handle100,title255,sku100,each option name/value64,price16; image URLs2048 and alt text512. Nothing is inferred.

Images are optional syntax-only evidence, never fetched. URL policy: ASCII plain HTTPS, a public-looking DNS hostname with dotted alphabetic TLD, optional port443, image path ending jpg/jpeg/png/gif/webp/avif. No credentials, query, fragment, IP literal, local/internal suffix, whitespace, unsafe escapes or prohibited resize suffix. This intentionally rejects some otherwise usable URLs. Alt text requires a product image on that source row; repeated product image URL must have identical alt text within that product; a variant image must also be listed as a product image for that product. Reachability, media content, routability and rights remain unverified.

Successful output groups first-seen products while preserving their source variant order. It emits canonical UTF-8/LF CSV, title only on each product's first row, lowercase false and draft on every row, deduplicated product images with sequential positions and the original variant-image references. Independent reconciliation compares supplied and prepared literals, order, counts, price and visibility before any append. Failure suppresses the batch.

What a successful batch contains

Exactly one dataset item contains preparedCsv, validationJson, reconciliationJson and manifestJson, plus a versioned run-bound receipt. The prepared CSV is canonical UTF-8/LF, with draft status and unpublished values. Validation and independent reconciliation must both pass before that one item is appended. A blank/blocked/mismatched batch is never sold as a completed report.

All 14 base-plus-image output headers were compared with Shopify's current downloadable sample. Shopify's documentation also lists lowercase false and draft as accepted values. This is header/documentation alignment, not actual importer acceptance. Test any prospective import yourself in a suitable review workflow; this product never connects to or imports into a store. Shopify CSV documentation and official sample.

Get the actual prepared files

Open the run's full-item output link and export its default dataset as JSON containing the single complete item; save it as dataset.json. Keep all fields. A dataset CSV export is not the prepared product CSV. Save the standalone helper below as extract_report.py and run:

python3 extract_report.py dataset.json new-prepared-batch

The helper verifies schemas, all artifact hashes, counts and canonical draft/unpublished CSV rules, then saves PREPARED.csv, VALIDATION.json, RECONCILIATION.json and OUTPUT.json. It requires a new destination in a trusted parent directory and never overwrites existing files. Publication is exclusive and atomic per file, not an all-files filesystem transaction. Hashes do not authenticate the producer or prove the source was transformed correctly; the Actor's reconciliation JSON records its independent check, and you should review your original source and output.

"""Verify and extract a Supplier Catalog Draft CSV Preparation dataset export.

Offline, standard-library-only consistency checks. Hashes do not authenticate the
producer, prove that the source was transformed correctly, or establish Shopify
importer acceptance. Independently review source data and all draft CSV values.
"""
from __future__ import annotations

import argparse
import csv
from decimal import Decimal
import hashlib
import io
import ipaddress
import json
import os
from pathlib import Path
import re
import shutil
import stat
import tempfile
import unicodedata
from urllib.parse import unquote, urlsplit

PACKAGE_VERSION = '0.1.0-candidate'
ENGINE_PROFILE = 'bounded-new-products-2026-10-15'
MAX_EXPORT = 9_100_000
MAX_OUTPUT = 1_300_000
MAX_PREPARED = 1_048_576
MAX_JSON = 65_536
MAX_DEPTH = 20
FIELDS = {
    'preparedCsv': 'PREPARED.csv',
    'validationJson': 'VALIDATION.json',
    'reconciliationJson': 'RECONCILIATION.json',
    'manifestJson': 'OUTPUT.json',
}
BASE_HEADERS = ('URL handle', 'Title', 'SKU', 'Option1 name', 'Option1 value',
                'Option2 name', 'Option2 value', 'Price',
                'Published on online store', 'Status')
IMAGE_HEADERS = ('Product image URL', 'Image position', 'Image alt text',
                 'Variant image URL')
SOURCE_NAMES = {'catalog_prep.py', 'image_rules.py', 'reconcile.py', 'safe_io.py',
                'actor_adapter.py'}
SUMMARY_KEYS = {'product_count', 'variant_count', 'image_count'}
VALIDATION_KEYS = {'schema_version', 'validation_profile', 'status', 'issues',
                   'input_sha256', 'output_sha256', 'store_currency',
                   'product_count', 'variant_count', 'image_count',
                   'image_urls_fetched', 'shopify_import_tested',
                   'ordering_policy', 'limitations'}
RECONCILIATION_KEYS = {'status', 'product_count', 'variant_count', 'source_sha256',
                       'prepared_sha256', 'issues', 'shopify_import_tested'}
CORE_LIMITATIONS = [
    'The new-product declaration does not check the live store or existing handles/SKUs.',
    'No inventory, taxes, shipping, descriptions, metafields, collections, or Markets are prepared; supplied image URLs receive syntax checks only.',
    'Draft and unpublished values are emitted; merchant must verify visibility and defaults in Shopify.',
    'Offline checks and independent source QA do not establish Shopify importer acceptance.',
]


def unique(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError('duplicate_json_key')
        result[key] = value
    return result


def decode(raw):
    """Decode bounded-depth UTF-8 JSON; reject duplicate keys and all floats."""
    text = raw.decode('utf-8')
    depth = 0
    quoted = escaped = False
    for char in text:
        if quoted:
            if escaped:
                escaped = False
            elif char == '\\':
                escaped = True
            elif char == '"':
                quoted = False
        elif char == '"':
            quoted = True
        elif char in '[{':
            depth += 1
            if depth > MAX_DEPTH:
                raise ValueError('json_depth')
        elif char in ']}':
            depth -= 1
    def invalid(_):
        raise ValueError('invalid_json_number')
    def integer(value):
        if len(value) > 32:
            raise ValueError('json_integer_limit')
        return int(value)
    return json.loads(text, object_pairs_hook=unique, parse_int=integer,
                      parse_float=invalid, parse_constant=invalid)


def digest(body):
    return hashlib.sha256(body).hexdigest()


def is_hash(value):
    return type(value) is str and re.fullmatch(r'[0-9a-f]{64}', value) is not None


def shape(value, keys):
    if type(value) is not dict or set(value) != set(keys):
        raise ValueError('invalid_object_shape')


def validate_summary(value):
    shape(value, SUMMARY_KEYS)
    if (any(type(number) is not int for number in value.values())
            or not 1 <= value['product_count'] <= 25
            or not value['product_count'] <= value['variant_count'] <= 100
            or not 0 <= value['image_count'] <= value['variant_count']):
        raise ValueError('invalid_summary')
    return value


def valid_image_url(value):
    """The bounded engine's conservative syntax policy; never fetch a URL."""
    if len(value) > 2048 or not value.isascii() or re.search(r'[\s\\]', value):
        return False
    if re.search(r'%(?![0-9a-fA-F]{2})', value):
        return False
    try:
        decoded = unquote(value, errors='strict')
    except UnicodeDecodeError:
        return False
    if any(unicodedata.category(c).startswith('C') for c in decoded) or '\\' in decoded:
        return False
    if re.search(r'%(?:2f|5c|25)', value, re.I):
        return False
    try:
        parts = urlsplit(value)
        host = parts.hostname or ''
        if parts.scheme != 'https' or not host or parts.username is not None or parts.password is not None:
            return False
        if parts.port not in (None, 443) or parts.query or parts.fragment or '?' in value or '#' in value:
            return False
        if not re.fullmatch(r'[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?', host, re.I):
            return False
        labels = host.split('.')
        if len(labels) < 2 or len(host) > 253 or any(not re.fullmatch(r'[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?', p, re.I) for p in labels):
            return False
        if host.endswith(('.localhost', '.local', '.internal')) or not re.fullmatch(r'[a-z]{2,63}', labels[-1], re.I):
            return False
        try:
            ipaddress.ip_address(host)
        except ValueError:
            pass
        else:
            return False
        if not re.search(r'\.(?:jpg|jpeg|png|gif|webp|avif)\Z', parts.path, re.I):
            return False
        if re.search(r'_(?:thumb|small|medium)\.[^.]+\Z', unquote(parts.path), re.I):
            return False
        return True
    except ValueError:
        return False


def identity(value):
    return unicodedata.normalize('NFKC', value).casefold()


def validate_csv(body):
    """Validate generated shape and bounded engine invariants without source I/O."""
    if not 0 < len(body) <= MAX_PREPARED:
        raise ValueError('prepared_size')
    parsed = list(csv.reader(io.StringIO(body.decode('utf-8'), newline=''), strict=True))
    if (not parsed or tuple(parsed[0]) not in (BASE_HEADERS, BASE_HEADERS + IMAGE_HEADERS)
            or not 2 <= len(parsed) <= 101
            or any(len(row) != len(parsed[0]) for row in parsed[1:])):
        raise ValueError('prepared_shape')
    canonical = io.StringIO(newline='')
    csv.writer(canonical, lineterminator='\n').writerows(parsed)
    if canonical.getvalue().encode('utf-8') != body:
        raise ValueError('prepared_encoding_or_quoting')
    images_enabled = len(parsed[0]) == 14
    limits = {'URL handle': 100, 'Title': 255, 'SKU': 100,
              'Option1 name': 64, 'Option1 value': 64,
              'Option2 name': 64, 'Option2 value': 64, 'Price': 16,
              'Product image URL': 2048, 'Image alt text': 512,
              'Variant image URL': 2048}
    products, skus, variants, images, variant_images = {}, set(), set(), {}, {}
    last_handle = None
    for cells in parsed[1:]:
        row = dict(zip(parsed[0], cells))
        for field, value in row.items():
            if (len(value) > limits.get(field, 16)
                    or value != value.strip()
                    or any(unicodedata.category(c).startswith('C') for c in value)
                    or unicodedata.normalize('NFKC', value).lstrip().startswith(('=', '+', '-', '@'))):
                raise ValueError('prepared_cell')
        handle = row['URL handle']
        if (re.fullmatch(r'[a-z0-9]+(?:-[a-z0-9]+)*', handle) is None
                or row['Status'] != 'draft' or row['Published on online store'] != 'false'
                or not row['SKU'] or not row['Option1 name'] or not row['Option1 value']
                or bool(row['Option2 name']) != bool(row['Option2 value'])
                or identity(row['Option1 name']) == 'default title'
                or identity(row['Option1 value']) == 'default title'
                or row['Option2 name'] and identity(row['Option1 name']) == identity(row['Option2 name'])):
            raise ValueError('prepared_invariant')
        first = handle not in products
        details = (row['Option1 name'], row['Option2 name'])
        if first:
            if not row['Title']:
                raise ValueError('missing_first_title')
            products[handle] = details
            images[handle] = []
            variant_images[handle] = []
        elif handle != last_handle or row['Title'] or details != products[handle]:
            raise ValueError('prepared_grouping')
        last_handle = handle
        sku = identity(row['SKU'])
        variant = (handle, identity(row['Option1 value']), identity(row['Option2 value']))
        if sku in skus or variant in variants:
            raise ValueError('duplicate_identity')
        skus.add(sku)
        variants.add(variant)
        price = row['Price']
        if (re.fullmatch(r'(?:0|[1-9][0-9]{0,5})\.[0-9]{2}', price) is None
                or not Decimal('0') < Decimal(price) <= Decimal('999999.99')):
            raise ValueError('prepared_price')
        if images_enabled:
            image = row['Product image URL']
            variant_image = row['Variant image URL']
            if any(value and not valid_image_url(value) for value in (image, variant_image)):
                raise ValueError('image_url')
            if image:
                if image in images[handle] or row['Image position'] != str(len(images[handle]) + 1):
                    raise ValueError('image_ordinal')
                images[handle].append(image)
            elif row['Image position'] or row['Image alt text']:
                raise ValueError('image_dependency')
            if variant_image:
                variant_images[handle].append(variant_image)
    if any(set(variant_images[handle]) - set(images[handle]) for handle in products):
        raise ValueError('variant_image_missing')
    return validate_summary({'product_count': len(products), 'variant_count': len(parsed) - 1,
                             'image_count': sum(map(len, images.values()))})


def verify(raw):
    """Return verified artifact bytes; establish internal consistency, not provenance."""
    if type(raw) is not bytes or not 0 < len(raw) <= MAX_EXPORT:
        raise ValueError('export_size')
    rows = decode(raw)
    if type(rows) is not list or len(rows) != 1:
        raise ValueError('export_count')
    row = rows[0]
    shape(row, {'schema', 'packageVersion', 'runId', 'deliveryKey', 'receipt', *FIELDS})
    if (row['schema'] != 'catalog-draft-delivery/v1' or row['packageVersion'] != PACKAGE_VERSION
            or type(row['runId']) is not str or re.fullmatch(r'[A-Za-z0-9_-]{1,100}', row['runId']) is None
            or any(type(row[key]) is not str for key in FIELDS)):
        raise ValueError('delivery_contract')
    bodies = {name: row[key].encode('utf-8') for key, name in FIELDS.items()}
    if (sum(map(len, bodies.values())) > MAX_OUTPUT
            or len(bodies['PREPARED.csv']) > MAX_PREPARED
            or any(not 0 < len(bodies[name]) <= MAX_JSON for name in ('VALIDATION.json', 'RECONCILIATION.json', 'OUTPUT.json'))):
        raise ValueError('artifact_size')
    manifest = decode(bodies['OUTPUT.json'])
    validation = decode(bodies['VALIDATION.json'])
    reconciliation = decode(bodies['RECONCILIATION.json'])
    receipt = row['receipt']
    shape(manifest, {'schema', 'packageVersion', 'engineProfile', 'status', 'inputSha256',
                     'syntheticDemo', 'currency', 'summary', 'sourceSha256', 'artifacts', 'limitations'})
    shape(receipt, {'schema', 'packageVersion', 'engineProfile', 'manifestSha256',
                    'inputSha256', 'preparedSha256', 'summary', 'currency'})
    shape(validation, VALIDATION_KEYS)
    shape(reconciliation, RECONCILIATION_KEYS)
    if (manifest['schema'] != 'catalog-draft-actor/v1'
            or receipt['schema'] != 'catalog-draft-completion/v1'
            or any(item['packageVersion'] != PACKAGE_VERSION or item['engineProfile'] != ENGINE_PROFILE
                   for item in (manifest, receipt))
            or manifest['status'] != 'prepared_for_review'
            or type(manifest['syntheticDemo']) is not bool
            or type(manifest['currency']) is not str
            or re.fullmatch(r'[A-Z]{3}', manifest['currency']) is None
            or receipt['currency'] != manifest['currency']
            or type(manifest['limitations']) is not list
            or any(type(value) is not str for value in manifest['limitations'])):
        raise ValueError('manifest_contract')
    shape(manifest['sourceSha256'], SOURCE_NAMES)
    if not all(is_hash(value) for value in manifest['sourceSha256'].values()):
        raise ValueError('source_hashes')
    summary = validate_csv(bodies['PREPARED.csv'])
    if any(validate_summary(item['summary']) != summary for item in (manifest, receipt)):
        raise ValueError('summary_binding')
    input_hash, prepared_hash = manifest['inputSha256'], digest(bodies['PREPARED.csv'])
    if (not is_hash(input_hash) or receipt['inputSha256'] != input_hash
            or validation['input_sha256'] != input_hash or reconciliation['source_sha256'] != input_hash
            or receipt['preparedSha256'] != prepared_hash or validation['output_sha256'] != prepared_hash
            or reconciliation['prepared_sha256'] != prepared_hash
            or receipt['manifestSha256'] != digest(bodies['OUTPUT.json'])):
        raise ValueError('hash_binding')
    if row['deliveryKey'] != digest(('catalog-draft/v1:' + row['runId'] + ':' + receipt['manifestSha256']).encode('ascii')):
        raise ValueError('delivery_key')
    if (type(validation['schema_version']) is not int or validation['schema_version'] != 2
            or validation['validation_profile'] != ENGINE_PROFILE
            or validation['status'] != 'prepared_for_review'
            or validation['issues'] != [] or type(validation['issues']) is not list
            or validation['store_currency'] != manifest['currency']
            or validation['image_urls_fetched'] is not False
            or validation['shopify_import_tested'] is not False
            or validation['ordering_policy'] != 'first_seen_products_and_source_rows'
            or validation['limitations'] != CORE_LIMITATIONS
            or any(type(validation[key]) is not int or validation[key] != summary[key] for key in SUMMARY_KEYS)):
        raise ValueError('validation_contract')
    if (reconciliation['status'] != 'matched'
            or type(reconciliation['issues']) is not list or reconciliation['issues'] != []
            or reconciliation['shopify_import_tested'] is not False
            or any(type(reconciliation[key]) is not int or reconciliation[key] != summary[key]
                   for key in ('product_count', 'variant_count'))):
        raise ValueError('reconciliation_contract')
    shape(manifest['artifacts'], {'PREPARED.csv', 'VALIDATION.json', 'RECONCILIATION.json'})
    for name, artifact in manifest['artifacts'].items():
        shape(artifact, {'sha256', 'bytes', 'contentType'})
        content_type = ('text/csv' if name.endswith('.csv') else 'application/json') + '; charset=utf-8'
        if (artifact['sha256'] != digest(bodies[name]) or type(artifact['bytes']) is not int
                or artifact['bytes'] != len(bodies[name]) or artifact['contentType'] != content_type):
            raise ValueError('artifact_binding')
    return bodies


def extract(raw, destination):
    """Stage complete files and hard-link them into a new private directory.

    Use a trusted parent directory whose path entries cannot be replaced by an
    untrusted process. Each file is published atomically without replacement;
    publication of the complete directory is not an atomic transaction. Errors
    remove only this call's published files and its new directory when empty.
    """
    bodies = verify(raw)
    destination = Path(destination)
    if os.path.lexists(destination) or not destination.parent.is_dir():
        raise ValueError('output_exists_or_parent_missing')
    stage = Path(tempfile.mkdtemp(prefix='.catalog-extract-', dir=destination.parent))
    created_identity = None
    published = []
    try:
        for name, body in bodies.items():
            path = stage / name
            with path.open('xb') as handle:
                os.fchmod(handle.fileno(), 0o600)
                if handle.write(body) != len(body):
                    raise OSError('incomplete_write')
                handle.flush()
                os.fsync(handle.fileno())
            if path.read_bytes() != body:
                raise OSError('staged_bytes_mismatch')
        destination.mkdir(mode=0o700)
        info = destination.lstat()
        created_identity = (info.st_dev, info.st_ino)
        for name in bodies:
            info = (stage / name).lstat()
            os.link(stage / name, destination / name, follow_symlinks=False)
            published.append((name, (info.st_dev, info.st_ino)))
    except BaseException:
        if created_identity is not None:
            for name, owned_identity in reversed(published):
                path = destination / name
                try:
                    info = path.lstat()
                except FileNotFoundError:
                    continue
                if (info.st_dev, info.st_ino) == owned_identity:
                    path.unlink()
            try:
                info = destination.lstat()
            except FileNotFoundError:
                pass
            else:
                if (info.st_dev, info.st_ino) == created_identity:
                    destination.rmdir()
        raise
    finally:
        shutil.rmtree(stage)
    return destination


def read_regular_file(path):
    if any(type(getattr(os, name, None)) is not int for name in ('O_NOFOLLOW', 'O_NONBLOCK')):
        raise ValueError('unsupported_safe_file_opening')
    path = Path(path)
    if not stat.S_ISREG(path.lstat().st_mode):
        raise ValueError('input_not_regular')
    flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK
    fd = os.open(path, flags)
    with os.fdopen(fd, 'rb') as handle:
        info = os.fstat(handle.fileno())
        if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_EXPORT:
            raise ValueError('input_not_regular_or_too_large')
        return handle.read(MAX_EXPORT + 1)


class QuietArgumentParser(argparse.ArgumentParser):
    def error(self, message):
        raise ValueError('invalid_arguments')


def main(argv=None):
    parser = QuietArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument('dataset_json', type=Path)
    parser.add_argument('new_directory', type=Path)
    try:
        args = parser.parse_args(argv)
        extract(read_regular_file(args.dataset_json), args.new_directory)
        print('Verified internal hashes, draft CSV and counts; extracted PREPARED.csv, VALIDATION.json, RECONCILIATION.json and OUTPUT.json. This does not authenticate the producer or establish Shopify importer acceptance.')
        return 0
    except (OSError, ValueError, TypeError, KeyError, RecursionError, UnicodeError, csv.Error, NotImplementedError):
        print('Rejected: invalid export, inconsistent artifacts or output failure. No input values logged.')
        return 2


if __name__ == '__main__':
    raise SystemExit(main())

Rejections and costs

Invalid input fails before append. Logs expose at most 20 allowlisted issue codes, a known field and logical record number where available, plus omitted count. For example invalid_price identifies price, duplicate_sku identifies sku, and product_conflict identifies inconsistent rows for one handle. Logical record 2 is the first data record. No submitted cell, URL, engine message or exception text is echoed. Correct the original source; no automatic repair is promised. Envelope errors may give only generic feedback. A timeout after append may already be billed; do not automatically retry.

Use exactly 512 MiB, limited permissions and a 1–300 second integer platform timeout; the process still has a 50-second internal alarm. No restart or standby. Authenticated DEVELOPER runs accept a finite numeric maxTotalChargeUsd strictly above 0.10 or omitted/null ceiling within otherwise valid run options. Omitted/null means no finite platform ceiling. Same-owner USER runs require an explicit finite cap greater than 0.10 and at most 0.11 for all inputs. Cap category/value, resource settings and payer must remain stable before append. There is no demo or test exception. For a supported same-owner USER run, set an explicit USD 0.11 ceiling. The single USD 0.10 item charge does not bound total platform/account fees; review the platform's separate usage and account charges.

Data and scope

Prepared product values are in the dataset; original supplier text remains in platform INPUT storage. No source CSV copy is put in the output, and image URLs are never fetched. Verify input/dataset access, retention and deletion settings before real data. No inventory, shipping, taxes, descriptions, metafields, collections, Markets, existing-product updates or live availability checks are prepared. Optional image syntax checks do not establish reachability, ownership or content.

This guide covers use of the hosted service. No public software license is asserted. Review the prepared CSV before using it; successful preparation does not guarantee acceptance by a store importer.