`review.py`

146 lines
#!/usr/bin/env python3
"""Historical review only. No authentication, network or apply interface."""
import argparse
import hashlib
import json
from pathlib import Path
import re
import sys

PIN = '0fb91e603d16ff15e1c678931c77f0c43d68e1c4cf7b858760472d5b7bebd449'
LABEL = 'HISTORICAL / PROVISIONAL / VENDOR UNVERIFIED / REVIEW ONLY'


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


def parse(raw):
    return json.loads(raw, object_pairs_hook=strict_object,
                      parse_constant=lambda x: (_ for _ in ()).throw(ValueError('nonfinite JSON')))


def require(condition, message):
    if not condition:
        raise ValueError(message)


def load(bundle, pin):
    raw = (bundle / 'manifest.json').read_bytes()
    require(hashlib.sha256(raw).hexdigest() == pin, 'manifest hash mismatch')
    manifest = parse(raw)
    require(manifest.get('kind') == 'HISTORICAL_SNAPSHOT', 'unsupported snapshot kind')
    n = manifest.get('expected_rows')
    require(type(n) is int and n > 0, 'invalid expected row count')
    require(set(manifest.get('files', {})) == {'scope.json', 'plan.json'}, 'invalid manifest file set')
    datasets = []
    for name in ('scope.json', 'plan.json'):
        raw = (bundle / name).read_bytes()
        require(hashlib.sha256(raw).hexdigest() == manifest['files'][name]['sha256'], name + ' hash mismatch')
        rows = parse(raw)
        require(isinstance(rows, list) and len(rows) == n, name + ' count mismatch')
        keyed = {}
        for row in rows:
            require(isinstance(row, dict), 'invalid row')
            gid = row.get('gid')
            require(isinstance(gid, str) and re.fullmatch(r'gid://shopify/Product/[0-9]+', gid), 'invalid product ID')
            require(gid not in keyed, 'duplicate product ID')
            require(row.get('status') in ('ACTIVE', 'DRAFT', 'ARCHIVED'), 'invalid historical status')
            keyed[gid] = row
        datasets.append(keyed)
    scope, plan = datasets
    require(set(scope) == set(plan), 'scope/plan ID mismatch')
    for gid, row in scope.items():
        require(set(row) == {'gid', 'status', 'old_title', 'old_body'}, 'unexpected scope fields')
        require(all(isinstance(row[x], str) for x in ('old_title', 'old_body')), 'invalid scope text')
        change = plan[gid]
        require(set(change) == {'gid', 'status', 'changes'}, 'unexpected plan fields')
        require(change['status'] == row['status'], 'scope/plan status mismatch')
        require(isinstance(change['changes'], dict) and bool(change['changes']), 'invalid changes')
        for field, pair in change['changes'].items():
            require(field in ('title', 'descriptionHtml'), 'unexpected change field')
            require(isinstance(pair, dict) and set(pair) == {'from', 'to'}, 'invalid change pair')
            require(all(isinstance(x, str) for x in pair.values()), 'invalid change text')
            require(pair['from'] == row['old_title' if field == 'title' else 'old_body'], 'scope/plan before mismatch')
    return manifest, scope, plan


def analyze(manifest, scope, plan):
    rows = []
    counts = {'PROVISIONAL_CANDIDATE': 0, 'HELD_AMBIGUOUS': 0, 'EXCLUDED_NO_TITLE_SIGNALS': 0}
    for gid, row in scope.items():
        title = row['old_title']
        bespoke = bool(re.search(r'\bbespoke\b', title, re.I))
        digital = bool(re.search(r'\bdigital(?:ly)?\b', title, re.I))
        has_mylar = bool(re.search(r'\bmylar\b', title + '\n' + row['old_body'], re.I))
        if bespoke and digital and has_mylar:
            disposition, reason = 'PROVISIONAL_CANDIDATE', 'Both explicit title signals; actual vendor and digital method unverified.'
        elif bespoke or digital:
            disposition, reason = 'HELD_AMBIGUOUS', 'Missing a required title signal or Mylar text; no scope inference from body.'
        else:
            disposition, reason = 'EXCLUDED_NO_TITLE_SIGNALS', 'Neither required title signal; exclusion from this historical proposal only.'
        counts[disposition] += 1
        item = {'gid': gid, 'historical_status': row['status'], 'historical_title': title,
                'title_signals': {'bespoke': bespoke, 'digital_or_digitally': digital},
                'vendor_identity': 'UNVERIFIED', 'digital_method': 'UNVERIFIED',
                'apply_eligible': False, 'disposition': disposition, 'reason': reason}
        if disposition == 'PROVISIONAL_CANDIDATE':
            review = {}
            for field, old in [('title', title), ('descriptionHtml', row['old_body'])]:
                new = re.sub(r'\bmylar\b', lambda m: 'Metallic' if m[0][0].isupper() else 'metallic', old, flags=re.I)
                if new != old:
                    review[field] = {'before': old, 'after_provisional': new}
            item['review_comparison'] = review
            item['historical_broad_plan_fields'] = sorted(plan[gid]['changes'])
        rows.append(item)
    return {'schema_version': 1, 'label': LABEL, 'source_ticket': 'TK-11785',
            'snapshot_manifest': manifest, 'inventory_freshness': 'UNVERIFIED',
            'snapshot_completeness': {'expected': manifest['expected_rows'], 'reviewed': len(rows),
                                      'scope': 'These supplied historical files only; live inventory coverage unverified.'},
            'counts': counts, 'apply_eligible_count': 0,
            'limitations': ['Title words are provisional signals, never vendor or manufacturing evidence.',
                            'No current inventory, freshness, publication or authorization claim.',
                            'Fresh authoritative vendor/digital export required for operational review; exact approval required only for later catalog mutations.',
                            'Comparisons replace whole-word Mylar only; no product updates are provided.'],
            'rows': rows}


def markdown(report):
    lines = ['# ' + LABEL, '', 'TK-11785 — historical review; no row is apply eligible.', '',
             'Snapshot rows: ' + str(report['snapshot_completeness']['reviewed']),
             'Counts: ' + json.dumps(report['counts'], sort_keys=True), '']
    for limitation in report['limitations']:
        lines.extend(['- ' + limitation])
    for row in report['rows']:
        lines.extend(['', '## ' + row['gid'], '', row['disposition'] + ' — VENDOR UNVERIFIED', '',
                      '    ' + json.dumps(row['historical_title'], ensure_ascii=False), '', row['reason']])
        for field, values in row.get('review_comparison', {}).items():
            lines.extend(['', field + ' (historical before → provisional after):', '',
                          '    BEFORE ' + json.dumps(values['before'], ensure_ascii=False),
                          '    AFTER  ' + json.dumps(values['after_provisional'], ensure_ascii=False)])
    return '\n'.join(lines) + '\n'


def main():
    parser = argparse.ArgumentParser(description=LABEL)
    parser.add_argument('--bundle', type=Path, default=Path(__file__).resolve().parent / 'snapshots')
    parser.add_argument('--manifest-sha256', default=PIN, help='independently trusted pin; changing it does not establish provenance')
    parser.add_argument('--format', choices=('json', 'markdown'), default='json')
    args = parser.parse_args()
    try:
        report = analyze(*load(args.bundle, args.manifest_sha256))
    except (OSError, ValueError, TypeError, KeyError, AttributeError) as error:
        print(json.dumps({'label': LABEL, 'error': str(error), 'apply_eligible_count': 0}), file=sys.stderr)
        return 2
    print(markdown(report) if args.format == 'markdown' else json.dumps(report, indent=2, ensure_ascii=False))
    return 0


if __name__ == '__main__':
    sys.exit(main())