Product Price Snapshot Comparison

Compare two supplied product snapshots and get an exact CSV/JSON change report, a review queue, and traceable input hashes. Up to 25 observations per snapshot. No scraping or live-price lookup.

Hosted price: USD 0.10 for one complete report, including demo, zero-row or all-review results. The sole automatic event is apify-default-dataset-item; standard-run platform usage is included. Post-run storage/access charges may apply. Use an explicit USD 0.11 ceiling. Build 0.0.1 passed two fictional private functional tests; these do not establish settled customer billing, revenue or automatic Store QA compatibility.

Run policy: exactly 512 MiB, integer platform timeout 1–300 seconds, limited permissions and no restart or standby. The process still stops after 50 seconds. With authenticated DEVELOPER platform-usage billing, a finite maxTotalChargeUsd strictly above 0.10 is accepted; an omitted or null cap is also accepted within otherwise valid run options and means no finite platform ceiling. With authenticated same-owner USER billing, an explicit finite cap greater than 0.10 and at most 0.11 is required. The rule is identical for demo and own data. Malformed/insufficient caps reject; cap category/value, resource settings and payer must remain unchanged before the single append. Owner Development tests consume platform usage; customer standard-run platform usage is included in the saved event price. Post-run storage/access may cost extra. Automatic owner-paid QA compatibility remains unverified; this candidate does not establish general platform-health compatibility.

Quick start

The default form runs a tiny fictional demonstration. For an API demo, use:

{"useDemo": true}

For your data, set useDemo false, explicitly acknowledge authorized nonpersonal/nonconfidential product data, and paste exact version-1 JSON texts into beforeJson and afterJson. Optional approvedMapJson is a reviewed one-to-one identity map tied to the exact snapshot hashes. Full input rules follow below.

The report contains listed-price changes only for exact scoped identities or your explicit map, subject to variant/seller/quantity/basis/time guards. Missing prices remain unknown; absent rows mean not observed. No automatic identity inference, tax/shipping arithmetic, currency/unit conversion, automatic repricing, HTML output or consultancy review is included.

Minimal own-data snapshot

Save the following as before.json. This example is entirely fictional:

{
  "captured_at": "2026-10-13T11:00:00Z",
  "records": [
    {
      "currency": "USD",
      "identifier": "DEMO-001",
      "identity_namespace": "fictional-catalog",
      "identity_type": "sku",
      "market": "US",
      "observed_at": "2026-10-13T10:00:00Z",
      "price": "10.00",
      "quantity": "1",
      "record_id": "fictional-before-1",
      "seller_id": "fictional-seller",
      "shipping_basis": "excluded",
      "source_url": "https://example.invalid/products/demo-001",
      "stock": "unknown",
      "tax_basis": "exclusive",
      "title": "Fictional sample item",
      "unit": "piece",
      "variant_id": "standard"
    }
  ],
  "schema_version": "1",
  "snapshot_id": "fictional-before"
}

For after.json, use this complete fictional later snapshot:

{
  "captured_at": "2026-10-14T11:00:00Z",
  "records": [
    {
      "currency": "USD",
      "identifier": "DEMO-001",
      "identity_namespace": "fictional-catalog",
      "identity_type": "sku",
      "market": "US",
      "observed_at": "2026-10-14T10:00:00Z",
      "price": "12.50",
      "quantity": "1",
      "record_id": "fictional-after-1",
      "seller_id": "fictional-seller",
      "shipping_basis": "excluded",
      "source_url": "https://example.invalid/products/demo-001",
      "stock": "unknown",
      "tax_basis": "exclusive",
      "title": "Fictional sample item",
      "unit": "piece",
      "variant_id": "standard"
    }
  ],
  "schema_version": "1",
  "snapshot_id": "fictional-after"
}

To create the API/form envelope without changing any snapshot bytes, this minimal Python standard-library recipe writes a new actor-input.json. Run it only after confirming the files contain authorized nonpersonal, nonconfidential product data. Use a nonpersonal team/role label for any map approver. Read the files as bytes; read_text() may alter line endings on some systems.

import json
from pathlib import Path
value = {
    "useDemo": False,
    "authorizedProductData": True,
    "beforeJson": Path("before.json").read_bytes().decode("utf-8"),
    "afterJson": Path("after.json").read_bytes().decode("utf-8"),
}
# Optional, only for an explicitly reviewed map bound to these exact bytes:
# value["approvedMapJson"] = Path("approved-map.json").read_bytes().decode("utf-8")
with Path("actor-input.json").open("x", encoding="utf-8", newline="\n") as output:
    output.write(json.dumps(value, ensure_ascii=True, indent=2) + "\n")

The included prepare_input.py provides bounded/no-follow file reads and an explicit acknowledgment switch. Files larger than 262,144 UTF-8 bytes each or more than 25 records per snapshot are rejected by the Actor. See the input contract below for all keys, limits and map semantics.

Retrieve the complete result

The Actor's Output link opens the default dataset items as JSON. Export/download JSON, keeping every field and item; do not flatten, select fields, clean away metadata or export the outer dataset as CSV. There is one complete report item, not one dataset item per product. Save the JSON array as dataset.json.

That one item contains reportJson, reportCsv, manifestJson, a run-bound receipt and delivery key. The strings are the exact report file contents. Copy the standalone helper below to extract_report.py, then run with Python 3.12:

python3 extract_report.py dataset.json new-report-directory

It writes REPORT.json, REPORT.csv and OUTPUT.json into a new local directory and refuses existing paths. No network, credentials or external packages are needed. Use trusted local parent directories. The helper checks hashes, receipt/summary consistency and once-only observation accounting. Hashes are consistency checks, not proof that source observations, mapping approval, permissions or prices are genuine.

CSV is a display derivative with formula-leading cells text-prefixed, including negative values. JSON preserves raw values and is authoritative. Import CSV explicitly as text and verify your chosen spreadsheet; native Excel/Sheets compatibility is not claimed. Review all queued exceptions and supplied evidence before acting.

A failed or interrupted hosted run after append may still be billed. Do not automatically rerun an uncertain result; reconcile the run and dataset first. Code completion counts are not revenue or profit. The standalone local CLI and tests do not establish hosted billing or Store acceptance.

Complete input contract

Version-1 contract

Envelope

UTF-8 JSON object with no duplicate or unsupported keys. useDemo is a required boolean. It defaults to true in the hosted form; API callers must send it explicitly. Demo mode permits authorizedProductData only absent/false and the three text fields only absent/empty. It uses exactly the bundled fictional snapshots.

Own mode requires useDemo: false, authorizedProductData: true, and nonempty beforeJson and afterJson strings. Optional approvedMapJson may be absent or empty. These are JSON texts, not nested objects, URLs, archives, file paths or automatic export imports. Their exact UTF-8 bytes are retained for input hashes, including whitespace and terminal newlines. The acknowledgment means authorized, nonpersonal, nonconfidential product data, without credentials, confidential terms or private/token-bearing links. Use a nonpersonal team/role label for approved_by. Rights and privacy cannot be proved by a parser.

Each snapshot and the optional map is limited to 262,144 UTF-8 bytes. The outer escaped envelope is limited to 4,750,000 bytes. UTF-8 byte limits, JSON nesting (12 levels within each file), uniqueness, timestamp validity and runtime guards are authoritative; the supplied JSON Schemas are aids, not full semantic validation.

Version-1 snapshot

Exactly four keys: schema_version is the string "1"; snapshot_id is nonempty text up to 256 characters; captured_at is a timestamp; records is an array with 0–25 rows.

Every row requires:

• record_id: unique within its snapshot
• identity_type: sku or gtin
• identity_namespace: exact source namespace; GTIN must use gtin; SKU must not
• identifier: exact identifier. GTIN accepts 8, 12, 13 or 14 ASCII digits with a valid check digit
• variant_id, seller_id, unit, market: exact nonempty text
• quantity: positive decimal string
• currency: three uppercase ASCII letters; this is a syntax check, not verification of a currency registry
• price: nonnegative decimal string or JSON null for unknown
• observed_at: timestamp
• source_url: inert HTTP(S) source reference, at most 2,048 characters; no userinfo, whitespace, backslash or malformed port. No URL is fetched, resolved, verified or used as a network destination
• tax_basis: inclusive, exclusive or unknown
• shipping_basis: included, excluded or unknown

Optional keys only: title (nonempty text), stock (in_stock, out_of_stock, unknown), shipping_amount (decimal string or null). Other text fields are at most 256 characters and reject C0 and DEL controls. No IDs are trimmed, case-folded or normalized. Do not supply names or personal data in product fields.

Decimal strings have 1–12 integer digits, no leading zeros except 0, and optionally 1–6 fractional digits. No negatives, commas, spaces, exponent notation, floats or currency symbols. Prices may be zero. Quantity must be greater than zero. Timestamps are full YYYY-MM-DDTHH:MM:SS, optional 1–6 fractional digits, and Z or an explicit offset within ±14:00; actual calendar and offset validity are checked.

Exact matching and review

Automatic matching requires a unique literal combination of identity type, identity namespace, identifier, variant and seller on both sides. Ambiguous duplicates never receive a fuzzy or fallback match. Before/after variant, seller, quantity string, unit, currency, market, tax basis and shipping basis must match exactly. 1 and 1.0 quantities do not match. Equal unknown tax/shipping basis remains explicitly unknown, with no inferred arithmetic.

After capture must be later than before capture and no more than seven days apart. Each observation must occur at or before its snapshot capture and be no older than 24 hours relative to it. After observation must be later than the paired before observation. These checks compare supplied timestamps; the current clock is not a live-price freshness guarantee. Violations become review items; structurally invalid input rejects the entire run before a report append.

Price deltas use exact decimal arithmetic. Percentages use two decimal places, half-up. A zero baseline leaves percent unavailable. Missing price remains unknown. Stock values are supplied observations; a missing row means only not observed in that file. There is no automatic repricing, exchange-rate conversion, quantity normalization or tax/shipping arithmetic.

Optional explicit map

Exactly these keys: schema_version: "1", approval_id, approved_by, approved_at, before_sha256, after_sha256, mappings. The approval fields are supplied claims, not authentication or legal-signature verification. approved_by must be a nonpersonal role/team label under the data acknowledgment.

SHA256 values must match the exact UTF-8 bytes inside the corresponding snapshot strings. Any whitespace change requires a new hash and reviewed map. Each of at most 25 mapping entries has exactly map_id, before_record_id, after_record_id, reason. All are nonempty text up to 256 characters. Map IDs and record IDs on each side must be unique; referenced rows must exist. Mapping is explicitly labelled approved_map in the report. It may establish an identity relationship but cannot override duplicate identifiers, variant/seller, quantity/unit, currency/market, basis or timestamp guards.

Prepare an envelope without changing bytes

Using Python 3.12 and the included offline helper:

python3 prepare_input.py --before before.json --after after.json --output actor-input.json --acknowledge-authorized-product-data

Add --approved-map approved-map.json only for an explicitly reviewed map. Existing output paths and final-component symlinks are refused. Use trusted local parent directories. The helper only packages exact UTF-8 text; the Actor validates snapshot semantics and map binding. It does not upload or send anything.

Standalone verification and extraction helper

"""Verify and extract one Product Price Snapshot Comparison dataset JSON export offline."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import stat
import tempfile

MAX_EXPORT = 9_100_000
MAX_OUTPUT = 4_250_000
MAX_ARTIFACT = 2_097_152
MAX_MANIFEST = 65_536
PACKAGE_VERSION = '0.1.1-candidate'
ENGINE_VERSION = '1.0.0'
FIELDS = {'reportJson': 'REPORT.json', 'reportCsv': 'REPORT.csv', 'manifestJson': 'OUTPUT.json'}
SUMMARY_KEYS = {'before_records', 'after_records', 'matched_pairs', 'review_items', 'unmatched_records', 'price_changed_pairs', 'unknown_price_pairs'}


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


def decode(raw):
    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 > 20: raise ValueError()
        elif char in ']}': depth -= 1
    def invalid(_): raise ValueError()
    def integer(value):
        if len(value) > 32: raise ValueError()
        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 hashes(value):
    return (type(value) is dict and set(value) == {'before', 'after', 'approved_map'}
            and all(type(value[key]) is str and re.fullmatch(r'[0-9a-f]{64}', value[key]) for key in ('before', 'after'))
            and (value['approved_map'] is None or type(value['approved_map']) is str
                 and re.fullmatch(r'[0-9a-f]{64}', value['approved_map'])))


def validate_summary(report):
    summary = report['summary']
    if (type(summary) is not dict or set(summary) != SUMMARY_KEYS
            or any(type(v) is not int or not 0 <= v <= 50 for v in summary.values())
            or type(report.get('requires_review')) is not bool): raise ValueError()
    for key, count_key in [('matched', 'matched_pairs'), ('review_queue', 'review_items'), ('unmatched', 'unmatched_records')]:
        if type(report[key]) is not list or len(report[key]) != summary[count_key]: raise ValueError()
    expected = {}
    seen = {'before': [], 'after': []}
    for side in seen:
        records = report['inputs'][side]['snapshot']['records']
        if (type(records) is not list or len(records) > 25
                or len(records) != summary[side + '_records']): raise ValueError()
        expected[side] = [row['record_id'] for row in records]
        if (any(type(value) is not str for value in expected[side])
                or len(expected[side]) != len(set(expected[side]))): raise ValueError()
    pairs = summary['matched_pairs']
    if (pairs > min(summary['before_records'], summary['after_records'])
            or summary['price_changed_pairs'] + summary['unknown_price_pairs'] > pairs
            or summary['review_items'] + summary['unmatched_records'] > sum(len(v) for v in expected.values()) - 2 * pairs): raise ValueError()
    for category in ('matched', 'review_queue', 'unmatched'):
        for entry in report[category]:
            if type(entry) is not dict: raise ValueError()
            paired = 'before_record_id' in entry or 'after_record_id' in entry
            if category == 'matched':
                if not paired or entry.get('price_change') not in ('increased', 'decreased', 'unchanged', 'unknown'): raise ValueError()
            if category == 'unmatched' and paired: raise ValueError()
            if paired:
                if ('record_id' in entry or 'side' in entry
                        or any(type(entry.get(side + '_record_id')) is not str for side in seen)): raise ValueError()
                seen['before'].append(entry['before_record_id'])
                seen['after'].append(entry['after_record_id'])
            else:
                if entry.get('side') not in seen or type(entry.get('record_id')) is not str: raise ValueError()
                seen[entry['side']].append(entry['record_id'])
    if any(sorted(seen[side]) != sorted(expected[side]) for side in seen): raise ValueError()
    if (summary['price_changed_pairs'] != sum(x['price_change'] in ('increased', 'decreased') for x in report['matched'])
            or summary['unknown_price_pairs'] != sum(x['price_change'] == 'unknown' for x in report['matched'])
            or report['requires_review'] != bool(report['review_queue'])): raise ValueError()
    return summary


def verify(raw):
    if type(raw) is not bytes or len(raw) > MAX_EXPORT: raise ValueError()
    rows = decode(raw)
    if type(rows) is not list or len(rows) != 1 or type(rows[0]) is not dict: raise ValueError()
    row = rows[0]
    if (set(row) != {'schema', 'packageVersion', 'runId', 'deliveryKey', 'receipt', *FIELDS}
            or row.get('schema') != 'product-price-delivery/v1'
            or type(row.get('runId')) is not str or re.fullmatch(r'[A-Za-z0-9_-]{1,100}', row['runId']) is None
            or any(type(row.get(key)) is not str for key in FIELDS)): raise ValueError()
    bodies = {name: row[key].encode('utf-8') for key, name in FIELDS.items()}
    if (sum(map(len, bodies.values())) > MAX_OUTPUT
            or len(bodies['OUTPUT.json']) > MAX_MANIFEST
            or any(len(bodies[k]) > MAX_ARTIFACT for k in ('REPORT.json', 'REPORT.csv'))): raise ValueError()
    manifest, report, receipt = decode(bodies['OUTPUT.json']), decode(bodies['REPORT.json']), row['receipt']
    if any(type(value) is not dict for value in (manifest, report, receipt)): raise ValueError()
    if (set(receipt) != {'schema', 'packageVersion', 'engineVersion', 'manifestSha256', 'reportSha256', 'inputSha256', 'requiresReview', 'summary'}
            or set(manifest) != {'schema', 'packageVersion', 'engineVersion', 'status', 'inputSha256', 'syntheticDemo', 'summary', 'requiresReview', 'sourceSha256', 'artifacts', 'limitations'}
            or row['packageVersion'] != PACKAGE_VERSION or receipt['packageVersion'] != PACKAGE_VERSION
            or manifest['packageVersion'] != PACKAGE_VERSION or receipt['engineVersion'] != ENGINE_VERSION
            or manifest['engineVersion'] != ENGINE_VERSION or report.get('tool_version') != ENGINE_VERSION
            or report.get('schema_version') != '1' or manifest['status'] != 'report_prepared'
            or type(manifest['syntheticDemo']) is not bool): raise ValueError()
    summary = validate_summary(report)
    if (receipt['summary'] != summary or manifest['summary'] != summary
            or any(type(v['summary'].get(k)) is not int for v in (manifest, receipt) for k in SUMMARY_KEYS)
            or type(receipt['requiresReview']) is not bool or type(manifest['requiresReview']) is not bool
            or receipt['requiresReview'] != report['requires_review'] or manifest['requiresReview'] != report['requires_review']): raise ValueError()
    inputs = {key: report['inputs'][key]['sha256'] if report['inputs'][key] is not None else None for key in ('before', 'after', 'approved_map')}
    if (manifest['schema'] != 'product-price-actor/v1' or receipt['schema'] != 'product-price-completion/v1'
            or receipt['manifestSha256'] != digest(bodies['OUTPUT.json']) or receipt['reportSha256'] != digest(bodies['REPORT.json'])
            or not hashes(inputs) or receipt['inputSha256'] != inputs or manifest['inputSha256'] != inputs): raise ValueError()
    if row['deliveryKey'] != digest(('product-price/v1:' + row['runId'] + ':' + receipt['manifestSha256']).encode('ascii')): raise ValueError()
    if set(manifest['artifacts']) != {'REPORT.json', 'REPORT.csv'}: raise ValueError()
    for name, content_type in [('REPORT.json', 'application/json; charset=utf-8'), ('REPORT.csv', 'text/csv; charset=utf-8')]:
        artifact = manifest['artifacts'][name]
        if (type(artifact) is not dict or set(artifact) != {'sha256', 'bytes', 'contentType'}
                or 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()
    return bodies


def extract(raw, destination):
    bodies = verify(raw)
    destination = Path(destination)
    if os.path.lexists(destination) or not destination.parent.is_dir(): raise ValueError()
    stage = Path(tempfile.mkdtemp(prefix='.price-extract-', dir=destination.parent))
    try:
        for name, body in bodies.items():
            path = stage / name
            with path.open('xb') as handle:
                if handle.write(body) != len(body): raise OSError()
            path.chmod(0o600)
            if path.read_bytes() != body: raise OSError()
        destination.mkdir(mode=0o700)
        try: os.rename(stage, destination)
        except OSError:
            destination.rmdir()
            raise
    finally:
        if stage.exists(): shutil.rmtree(stage)
    return destination


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


def main():
    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()
        fd = os.open(args.dataset_json, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
        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()
            raw = handle.read(MAX_EXPORT + 1)
        extract(raw, args.new_directory)
        print('Verified hashes and row accounting; extracted REPORT.json, REPORT.csv and OUTPUT.json.')
        return 0
    except (OSError, ValueError, TypeError, KeyError, RecursionError, UnicodeError):
        print('Rejected: invalid export, inconsistent report or output failure. No input values logged.')
        return 2


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

Data handling and operations

All original supplied product evidence is retained in the JSON report and echoed into CSV evidence columns. Keep sources, access, recipients and retention within your permissions. The acknowledgment cannot detect secrets or verify rights. No submitted URL is fetched. The only hosted network operations are authenticated, fixed current-run Apify metadata/storage calls. Diagnostics are fixed labels without submitted content or tokens.

See the source package's COMPLETION-CONTRACT.md and RELEASE-GATES.md for fail-closed runtime rules and remaining release checks. The source archive includes schemas, helpers and exact file hashes. There is no promise of manual correction, business advice or response-time SLA in this API tool.