kangshifu-board/scripts/verify-expanded-review-intake.py

"""Rebuild the transferred source packages and compare every review field."""
import json
from pathlib import Path
import sqlite3
import sys
import tempfile
import zipfile

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


def main():
    root = ROOT / 'work/product-reviews/round2/intake'
    report = json.loads((root / 'verification.json').read_text(encoding='utf-8'))
    verified = 0
    products = set()
    packages = []
    fields = 'id,product_id,content,append_content_json,rating_raw,rating_scale,review_date_raw,purchased_sku_text,is_default,identity_state,source_time_raw'
    for ordinal, batch in enumerate(report['batches'], 1):
        folder = root / ('batch-' + str(ordinal).zfill(2))
        archive = folder / 'channel-source.zip'
        assert agent_batch.digest(archive) == batch['channelArchiveSha256']
        before = {p.name: agent_batch.digest(p) for p in [archive, folder / 'staging.sqlite', folder / 'batch.json']}
        with tempfile.TemporaryDirectory(prefix='review-r2-source-check-', dir=ROOT / 'work/product-reviews/round2') as td:
            target = Path(td)
            with zipfile.ZipFile(archive) as z:
                assert z.testzip() is None
                for info in z.infolist():
                    p = target / info.filename
                    assert p.resolve().is_relative_to(target.resolve())
                    p.parent.mkdir(parents=True, exist_ok=True)
                    p.write_bytes(z.read(info))
            dry = structured_batch.build(target / 'batch.json', target / 'rebuilt.sqlite')
            structured_batch.build(target / 'batch.json', target / 'rebuilt.sqlite', True)
            update_library.verified(target / 'rebuilt.sqlite')
            with voc_db.readonly(folder / 'staging.sqlite') as old, voc_db.readonly(target / 'rebuilt.sqlite') as new:
                prior = [tuple(r) for r in old.execute('SELECT ' + fields + ' FROM reviews ORDER BY id')]
                rebuilt = [tuple(r) for r in new.execute('SELECT ' + fields + ' FROM reviews ORDER BY id')]
                assert prior == rebuilt and len(rebuilt) == batch['reviews']
                assert all(row[8] is None and row[9] == 'matched' for row in rebuilt)
                for pid, n in batch['reviewCounts'].items():
                    data = board_library.run({'db': str(target / 'rebuilt.sqlite')}, {'op': 'browse',
                        'entity': 'reviews', 'quality': 'audit', 'filters': {'productId': pid}, 'limit': 100})
                    assert data['totalMatched'] == n and len(data['records']) == n
                    assert all(r['product_id'] == pid for r in data['records'])
                    products.add(pid)
            verified += len(rebuilt)
            packages.append({'batchId': batch['batchId'], 'reviews': len(rebuilt),
                'sourceFilesVerified': dry['sourceFilesVerified'], 'channelArchiveSha256': batch['channelArchiveSha256']})
        assert before == {p.name: agent_batch.digest(p) for p in [archive, folder / 'staging.sqlite', folder / 'batch.json']}
    assert verified == report['selectedReviews'] and len(products) == report['sourceVerifiedProducts']
    result = {'ok': True, 'verifiedReviews': verified, 'verifiedProducts': len(products),
        'allTransferSourcesRebuilt': True, 'allReviewFieldsMatched': True, 'sourceFilesUnchanged': True,
        'exactProductBrowseVerified': True, 'unknownDefaultStatesPreserved': True,
        'productionWrites': 0, 'packages': packages}
    with (root / 'source-rebuild-verification.json').open('x', encoding='utf-8', newline='\n') as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
        f.write('\n')
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()