#!/usr/bin/env python3 """ACP discovery preflight 0.1.1: explicit subset, offline, read-only. MIT.""" import argparse import json from pathlib import Path import sys from urllib.parse import urlsplit VERSION = '0.1.1' MAX_BYTES = 10 * 1024 * 1024 MAX_LINES = 100000 REQUIRED = ('item_id', 'title', 'description', 'url', 'brand', 'seller_name', 'image_url', 'availability', 'price') SUPPORTED = set(REQUIRED) | {'group_id', 'listing_has_variations', 'variant_dict', 'offer_id'} STOCK = {'in_stock', 'out_of_stock', 'pre_order', 'backorder', 'unknown'} PLACEHOLDERS = {'null', 'unknown', 'n/a'} NOT_CHECKED = ['money_format_currency_and_positive_price', 'other_optional_fields', 'product_truth_and_same_product_grouping', 'stable_ids_over_time', 'url_reachability', 'upload_acceptance_and_onboarding', 'geography_shipping_returns_ads_checkout', 'agent_client_integration', 'ignored_variant_dict_content', 'option_name_semantics_and_placeholder_policy_outside_supported_fields', 'price_in_offer_id'] class DuplicateKey(ValueError): pass def object_pairs(pairs): obj = {} for key, value in pairs: if key in obj: raise DuplicateKey('duplicate JSON key') obj[key] = value return obj def bad_constant(value): raise ValueError('non-finite JSON number') def nonempty(value): return isinstance(value, str) and bool(value.strip()) def placeholder(value): return nonempty(value) and value.strip().lower() in PLACEHOLDERS def url_syntax(value): if any(c.isspace() or ord(c) < 32 or c == '\\' for c in value): return False try: u = urlsplit(value) _ = u.port return u.scheme in ('http', 'https') and bool(u.hostname) except ValueError: return False def inspect(raw): report = {'version': VERSION, 'scope': 'discovery_subset_only', 'status': 'input_error', 'lines_read': 0, 'rows_checked': 0, 'signals': [], 'warnings': [], 'not_checked': NOT_CHECKED, 'upload_allowed': False} if len(raw) > MAX_BYTES: report['error'] = 'byte_limit'; return report, 2 try: text = raw.decode('utf-8') except UnicodeDecodeError: report['error'] = 'invalid_utf8'; return report, 2 if text.startswith('\ufeff'): report['error'] = 'utf8_bom_not_supported'; return report, 2 lines = text.split('\n') if lines[-1] == '': lines.pop() if not lines or not any(l.strip() for l in lines): report['error'] = 'no_products'; return report, 2 if len(lines) > MAX_LINES: report['error'] = 'line_limit'; return report, 2 ids, offers, groups = {}, {}, {} def signal(line, code, field=None, first_line=None): event = {'line': line, 'code': code} if field: event['field'] = field if first_line is not None: event['first_line'] = first_line report['signals'].append(event) for number, line in enumerate(lines, 1): report['lines_read'] += 1 if not line.strip(): signal(number, 'blank_line'); continue try: row = json.loads(line, object_pairs_hook=object_pairs, parse_constant=bad_constant) except DuplicateKey: signal(number, 'duplicate_json_key'); continue except (ValueError, RecursionError): signal(number, 'invalid_json'); continue if not isinstance(row, dict): signal(number, 'not_object'); continue report['rows_checked'] += 1 for field in REQUIRED: v = row.get(field) if not nonempty(v): signal(number, 'required_string', field) elif field != 'availability' and placeholder(v): signal(number, 'placeholder', field) stock = row.get('availability') if nonempty(stock) and stock not in STOCK: signal(number, 'availability_enum', 'availability') for field in ('url', 'image_url'): if nonempty(row.get(field)) and not url_syntax(row[field]): signal(number, 'url_syntax', field) item = row.get('item_id') if nonempty(item): if item in ids: signal(number, 'duplicate_item_id', 'item_id', ids[item]) else: ids[item] = number offer = row.get('offer_id') if offer is not None and offer != '': if not nonempty(offer): signal(number, 'offer_id_type', 'offer_id') elif placeholder(offer): signal(number, 'placeholder', 'offer_id') elif offer in offers: signal(number, 'duplicate_offer_id', 'offer_id', offers[offer]) else: offers[offer] = number flag = row.get('listing_has_variations') if flag not in (None, '') and type(flag) is not bool: signal(number, 'boolean_type', 'listing_has_variations') group = row.get('group_id') if group is not None and group != '' and not nonempty(group): signal(number, 'group_id_type', 'group_id') if placeholder(group): signal(number, 'placeholder', 'group_id') options = row.get('variant_dict') if flag is True: group_ok = nonempty(group) and not placeholder(group) and group != item if not group_ok: signal(number, 'variant_group', 'group_id') if not group_ok: if options not in (None, {}): report['warnings'].append({'line': number, 'code': 'variant_dict_ignored'}) continue_options = False else: continue_options = True if continue_options: options_ok = (isinstance(options, dict) and all(nonempty(k) and nonempty(v) for k, v in options.items())) if options == {}: report['warnings'].append({'line': number, 'code': 'no_variant_options'}) if not options_ok: signal(number, 'variant_options', 'variant_dict') options_clean = options_ok if isinstance(options, dict): for value in options.values(): if placeholder(value): signal(number, 'placeholder', 'variant_dict') options_clean = False if options_clean and options: keys = tuple(sorted(options)) combination = tuple(sorted(options.items())) if group not in groups: groups[group] = {'keys': keys, 'combinations': {}} g = groups[group] if g['keys'] != keys: signal(number, 'group_option_keys', 'variant_dict') if combination in g['combinations']: signal(number, 'duplicate_variant_combination', 'variant_dict', g['combinations'][combination]) else: g['combinations'][combination] = number elif options not in (None, {}): report['warnings'].append({'line': number, 'code': 'variant_dict_ignored'}) if set(row) - SUPPORTED: report['warnings'].append({'line': number, 'code': 'optional_fields_not_checked'}) report['status'] = 'signals' if report['signals'] else 'subset_clear' return report, 1 if report['signals'] else 0 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('input', type=Path) args = parser.parse_args() try: with args.input.open('rb') as stream: raw = stream.read(MAX_BYTES + 1) report, code = inspect(raw) except (OSError, ValueError): report = {'version': VERSION, 'scope': 'discovery_subset_only', 'status': 'input_error', 'error': 'cannot_read_input', 'upload_allowed': False, 'not_checked': NOT_CHECKED} code = 2 print(json.dumps(report, ensure_ascii=False, indent=2)) return code if __name__ == '__main__': sys.exit(main())