"""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)()