kangshifu-board/scripts/verify-product-completion.py

"""Build fresh private staging DBs, then independently verify their actual rows."""
import argparse
from collections import Counter
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
from pathlib import Path
import sqlite3
import subprocess
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'server/voc-library'))
import library_channel
import update_library

spec = importlib.util.spec_from_file_location('recovery', ROOT / 'scripts/recover-product-completion.py')
recovery = importlib.util.module_from_spec(spec)
spec.loader.exec_module(recovery)
WORK, DOCS = recovery.WORK, recovery.DOCS


def read(path):
    return json.loads(Path(path).read_bytes())


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def stage():
    plan = read(WORK / 'batch-plan.json')
    directories = [WORK / b['directory'] for b in plan['batches']]
    assert all(not (d / 'staging.sqlite').exists() for d in directories), 'NEW_STAGING_ONLY'
    for applying in (False, True):
        for directory in directories:
            command = [sys.executable, '-X', 'utf8', '-B', str(ROOT / 'server/voc-library/agent_batch.py'),
                       '--batch', str(directory / 'batch.json'), '--output', str(directory / 'staging.sqlite')]
            if applying:
                command.append('--apply')
            result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=90)
            name = 'staging-build' if applying else 'preflight'
            (directory / (name + '.stdout.json')).write_bytes(result.stdout)
            (directory / (name + '.stderr.txt')).write_bytes(result.stderr)
            recovery.save(directory / (name + '.exit.json'), {'exitCode': result.returncode, 'command': command})
            if result.returncode:
                raise ValueError(name.upper() + '_FAILED:' + directory.name + ':' + result.stderr.decode('utf8', 'replace')[:1000])
            receipt = json.loads(result.stdout)
            assert receipt['ok'] and receipt['activeLibraryModified'] is False and receipt['productionImported'] is False
            print(json.dumps({'batch': directory.name, 'step': name, 'counts': receipt['counts']}, ensure_ascii=False), flush=True)


def verify():
    plan = read(WORK / 'batch-plan.json')
    baseline = read(ROOT / 'work/industry-monitoring-20261107-r1/snapshot.json')
    old_review_ids = {r['record_id'] for r in baseline['entities']['reviews']}
    expected_specs = {r['productId']: r['specText'] for r in read(DOCS / 'spec-recovery-plan.json')['candidates']}
    expected_reviews = {r['recordId']: r for r in read(DOCS / 'review-recovery-audit.json')['missingMeaningful']}
    total, verified_specs, verified_reviews, reports = Counter(), {}, {}, []
    current_specs = {r['record_id']: r for r in baseline['entities']['specs']}
    # Context products cause the frozen importer to regenerate title-spec rows.
    # Prove these are duplicates of existing rows, not additional discoveries.
    old_database = ROOT / 'work/library-preview/registry/versions/7e0df42395d9d2ea39dd5b537d17ae35ac910cc29721ed92a680b7fe6b25c92c.sqlite'
    with sqlite3.connect(old_database.resolve().as_uri() + '?mode=ro', uri=True) as old:
        old.row_factory = sqlite3.Row
        old_specs = {r['id']: dict(r) for r in old.execute('SELECT * FROM specs')}
    auto_specs, live_contexts = [], []
    for part in plan['batches']:
        directory = WORK / part['directory']
        batch = read(directory / 'batch.json')
        build = read(directory / 'staging-build.stdout.json')
        assert sha((directory / 'staging.sqlite').read_bytes()) == build['databaseSha256']
        assert sha((directory / 'staging.source.zip').read_bytes()) == build['archiveSha256']
        meta = update_library.verified(directory / 'staging.sqlite')
        assert meta['generation'] == build['generation']
        packed, archive_sha = library_channel.pack(directory / 'batch.json')
        # Persist the exact channel archive fingerprint for dry/apply/status reconciliation.
        (directory / 'channel-batch.zip').write_bytes(packed)
        counts = {}
        review_adapters = {}
        if part.get('reviews'):
            review_adapters = {r['itemId'] + ':' + r['reviewId']: r for r in read(directory / 'raw/parsed/reviews.json')['records']}
        with sqlite3.connect((directory / 'staging.sqlite').resolve().as_uri() + '?mode=ro', uri=True) as c:
            c.row_factory = sqlite3.Row
            for table in ('products', 'reviews', 'specs', 'prices'):
                counts[table] = c.execute('SELECT count(*) FROM ' + table).fetchone()[0]
                total[table] += counts[table]
            assert counts['products'] == part['products'] and counts['specs'] == part['specs'] + part['products']
            assert counts['reviews'] == part.get('reviews', 0) and counts['prices'] == 0
            assert {r[0] for r in c.execute('SELECT id FROM products')} == set(part['productIds'])
            for row in c.execute('SELECT * FROM specs'):
                pid = row['product_id']
                if row['verification'] != 'source_attribute':
                    current = current_specs[row['id']]
                    assert all(row[k] == current[k] for k in ('product_id', 'spec_text', 'net_weight_g', 'pack_count', 'total_weight_g'))
                    assert row['verification'] == current['spec_verification']
                    prior = old_specs.get(row['id'])
                    if prior and prior['source_record_id'] == current['source_record_id']:
                        assert all(row[k] == prior[k] for k in row.keys() if k != 'source_record_id')
                        origin = 'SAME_EXISTING_SOURCE_RECORD_AND_ALL_BUSINESS_FIELDS'
                    else:
                        path = WORK / 'current-spec-details' / (row['id'] + '.json')
                        if path.exists():
                            detail = read(path)
                        else:
                            detail = recovery.links.api({'op': 'detail', 'entity': 'specs', 'id': row['id'], 'generation': plan['generation']})
                            recovery.save(path, detail)
                        assert detail['generation'] == plan['generation']
                        actual = detail['record']
                        assert all(row[k] == actual[k] for k in row.keys() if k != 'source_record_id')
                        live_contexts.append(row['id'])
                        origin = 'CURRENT_GENERATION_FULL_RECORD_READBACK'
                    auto_specs.append({'recordId': row['id'], 'productId': pid, 'verification': row['verification'],
                                       'existingSourceRecordId': current['source_record_id'], 'duplicateProof': origin})
                    continue
                assert row['spec_text'] == expected_specs[pid]
                assert row['verification'] == 'source_attribute'
                assert all(row[f] is None for f in ('net_weight_g', 'pack_count', 'total_weight_g'))
                adapter = read(directory / ('raw/parsed/' + pid.replace(':', '-') + '.json'))
                original = adapter['original']
                raw = (directory / original['path']).read_bytes()
                actual = recovery.select_spec(raw, pid.split(':')[1], original['sha256'])
                assert actual['payload'] == adapter['literal'] and actual['specText'] == row['spec_text']
                assert original['observedAt'].startswith('2026-11-06')
                assert pid not in verified_specs
                verified_specs[pid] = row['id']
            for row in c.execute('SELECT * FROM reviews'):
                rid = row['id']
                assert rid not in old_review_ids and rid not in verified_reviews
                expected = expected_reviews[rid]
                mapped = review_adapters[':'.join(rid.split(':')[1:])]
                original_bytes = (directory / mapped['origin']['path']).read_bytes()
                assert sha(original_bytes) == mapped['origin']['sha256']
                original = recovery.unbound.at(json.loads(original_bytes), mapped['origin']['pointer'])
                actual_append = original.get('afterComment') if rid.startswith('jd:') else original.get('appendList') or original.get('append')
                assert mapped['appendContent'] == actual_append
                assert row['content'] == expected['content']
                assert row['is_default'] is None and row['rating_scale'] is None
                assert row['product_id'] == rid.rsplit(':', 1)[0]
                append = json.loads(row['append_content_json']) if row['append_content_json'] else None
                assert append == actual_append
                assert row['review_date_raw'] == mapped['reviewDate']
                assert row['rating_raw'] == str(mapped['rating'])
                assert row['purchased_sku_text'] == mapped['purchasedSku']
                verified_reviews[rid] = {'reviewDateRaw': row['review_date_raw'], 'ratingRaw': row['rating_raw'],
                                         'sourceRecordId': row['source_record_id'], 'hasAppend': bool(append)}
            # Product context must carry no new observation timestamp or fabricated price.
            assert c.execute('SELECT count(*) FROM products WHERE source_time_raw IS NOT NULL').fetchone()[0] == 0
        report = {'batch': directory.name, 'ok': True, 'counts': counts, 'stagingGeneration': build['generation'],
                  'batchManifestSha256': sha((directory / 'batch.json').read_bytes()),
                  'channelArchiveSha256': archive_sha, 'channelArchiveBytes': len(packed),
                  'channelArchiveExpandedBytes': len((directory / 'batch.json').read_bytes()) + sum((directory / p).stat().st_size for p in
                    {s['path'] for s in batch['sources']} | {a['path'] for a in batch.get('attachments', [])}),
                  'databaseSha256': build['databaseSha256'], 'sourceArchiveSha256': build['archiveSha256']}
        recovery.save(directory / 'readback.json', report)
        reports.append(report)
    assert set(verified_specs) == set(expected_specs)
    assert set(verified_reviews) == set(expected_reviews)
    summary = recovery.links.api({'op': 'summary'})
    assert summary['generation'] == plan['generation'], 'PRODUCTION_GENERATION_CHANGED'
    recovery.save(DOCS / 'generated-title-spec-duplicates.json', {'generation': plan['generation'], 'rows': auto_specs,
                  'count': len(auto_specs), 'fullLiveReads': len(live_contexts), 'newResearchFacts': 0})
    result = {'ok': True, 'checkedAt': datetime.now(timezone.utc).isoformat(), 'baselineGeneration': plan['generation'],
              'counts': dict(total), 'uniqueProductsWithVerifiedSpecs': len(verified_specs),
              'explicitNewSpecRecords': len(verified_specs), 'generatedExistingTitleSpecs': len(auto_specs),
              'expectedProductionAddedSpecs': len(verified_specs), 'expectedProductionAddedReviews': len(verified_reviews),
              'verifiedSpecRecordIds': verified_specs, 'verifiedReviewRecords': verified_reviews, 'batches': reports,
              'productionGenerationUnchanged': True, 'productionCountsAfterLocalStaging': summary['counts'],
              'numericWeightFieldsInferredInNewAttributes': 0, 'defaultFlagsInvented': 0, 'paidCalls': 0,
              'databaseWrites': 'new-private-staging-only', 'productionImported': False}
    recovery.save(DOCS / 'staging-readback.json', result)
    print(json.dumps({k: v for k, v in result.items() if k not in ('verifiedSpecRecordIds', 'verifiedReviewRecords', 'batches')}, ensure_ascii=False))


if __name__ == '__main__':
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('command', choices=['stage', 'verify'])
    (stage if p.parse_args().command == 'stage' else verify)()