kangshifu-board/scripts/audit-voc-generations.py

"""Compare immutable registry versions without importing deployed code or reading secrets.

Only the new --out file is written. Review prose, account data and raw source
payloads are excluded; selected quality flags, IDs and fingerprints are reported.
Uses the Python 3.9 standard library.
"""
import argparse
import collections
import hashlib
import json
import os
from pathlib import Path
import re
import sqlite3


QUALITY_FIELDS = ('is_default', 'identity_state', 'category_state', 'rating_scale')
REVIEW_FIELDS = QUALITY_FIELDS + ('product_id', 'source_record_id', 'source_time_raw')
MODULES = ('voc_db.py', 'update_library.py', 'schema_catalog.py',
           'library_receiver.py', 'board_library.py', 'retrieval.py', 'product_insights.py')


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


def digest(path):
    h = hashlib.sha256()
    with path.open('rb') as f:
        for block in iter(lambda: f.read(1024 * 1024), b''):
            h.update(block)
    return h.hexdigest()


def value_digest(value):
    return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True,
                                     separators=(',', ':')).encode()).hexdigest()


def plain(path):
    path = Path(path).absolute()
    require(not any(p.is_symlink() for p in (path, *path.parents)), 'REDIRECTED_PATH')
    return path


def qualified(row, strict=False):
    return (row['is_default'] == 0 and row['identity_state'] == 'matched'
            and row['category_state'] is not None
            and row['category_state'] != 'conflict'
            and (not strict or row['category_state'] == 'consistent'))


def snapshot(path, product_id):
    before = digest(path)
    c = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True)
    c.row_factory = sqlite3.Row
    try:
        c.execute('PRAGMA query_only=ON')
        meta = dict(c.execute('SELECT key,value FROM meta'))
        require(meta.get('generation') == path.stem, 'GENERATION_FILENAME_MISMATCH')
        products = {r['id']: dict(r) for r in c.execute('SELECT * FROM products')}
        reviews = {r['id']: dict(r) for r in c.execute('SELECT * FROM reviews')}
        strict = {rid: r for rid, r in reviews.items()
                  if (lambda p: p.get('marketplace') == 'jd'
                      and p.get('brand_code') == 'ksf' and p.get('category_code') == 'banmian')
                  (products.get(r['product_id'], {}))}
        tables = {r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='table'")}
        indexed = None
        if 'retrieval_documents' in tables:
            indexed = dict(c.execute("""SELECT
              count(*) AS audit,
              sum(is_default=0 AND identity_state='matched' AND category_state!='conflict') AS analysis,
              sum(platform='jd' AND brand='ksf' AND category='banmian') AS strictAudit,
              sum(platform='jd' AND brand='ksf' AND category='banmian'
                  AND is_default=0 AND identity_state='matched' AND category_state='consistent') AS strictAnalysis
              FROM retrieval_documents WHERE entity='reviews'""").fetchone())
        selected = []
        for rid, row in reviews.items():
            if row['product_id'] != product_id:
                continue
            source = c.execute('SELECT kind,payload_json,payload_sha256 FROM source_records WHERE id=?',
                               (row['source_record_id'],)).fetchone()
            payload = json.loads(source['payload_json']) if source else None
            flag = payload.get('is_default_text') if isinstance(payload, dict) else None
            selected.append({'id': rid, **{k: row.get(k) for k in REVIEW_FIELDS},
                             'contentSha256': value_digest(row['content']),
                             'purchasedSkuSha256': value_digest(row.get('purchased_sku_text')),
                             'sourceKind': source['kind'] if source else None,
                             'sourcePayloadSha256': source['payload_sha256'] if source else None,
                             'normalizedSourceDefault': flag if type(flag) is bool else None,
                             'normalizedSourceDefaultIsBoolean': type(flag) is bool})
        report = {'generation': meta['generation'], 'databaseSha256': before,
                  'counts': {'products': len(products), 'reviews': len(reviews)},
                  'reviews': {'analysis': sum(qualified(r) for r in reviews.values()),
                              'strictAudit': len(strict),
                              'strictAnalysis': sum(qualified(r, True) for r in strict.values())},
                  'retrievalIndexCounts': indexed,
                  'selectedProductReviews': sorted(selected, key=lambda r: r['id']),
                  'selectedProductDefaultGroups': dict(collections.Counter(
                      str(r['is_default']) for r in reviews.values() if r['product_id'] == product_id)),
                  'mergeDecisionsTablePresent': 'merge_decisions' in tables}
        if 'merge_decisions' in tables:
            report['selectedReviewDecisions'] = [dict(r) for r in c.execute('''SELECT
                record_id,decision,reason_codes_json,prior_source_record_id,
                incoming_source_record_id,selected_source_record_id,batch_sha256
                FROM merge_decisions WHERE entity='reviews' AND record_id IN
                (SELECT id FROM reviews WHERE product_id=?) ORDER BY record_id''', (product_id,))]
    finally:
        c.close()
    require(digest(path) == before, 'VERSION_FILE_CHANGED')
    report['sha256Unchanged'] = True
    return report, products, reviews


def changes(old, new, fields):
    common = old.keys() & new.keys()
    modified, lost = [], []
    for rid in sorted(common):
        a, b = old[rid], new[rid]
        if a == b:
            continue
        diff = {k: {'before': a.get(k), 'after': b.get(k)} for k in fields if a.get(k) != b.get(k)}
        modified.append({'id': rid, 'fields': diff,
                         'beforeRowSha256': value_digest(a), 'afterRowSha256': value_digest(b)})
        missing = [k for k in fields if a.get(k) is not None and a.get(k) != ''
                   and (b.get(k) is None or b.get(k) == '')]
        if missing:
            lost.append({'id': rid, 'fields': missing})
    return {'added': len(new.keys() - old.keys()), 'removed': len(old.keys() - new.keys()),
            'changed': len(modified), 'knownToUnknown': lost[:100],
            'knownToUnknownCount': len(lost), 'changes': modified[:100],
            'changesTruncated': len(modified) > 100, 'knownToUnknownTruncated': len(lost) > 100}


def audit(registry, generations, expected_generation, product_id, module_dirs=()):
    registry = plain(registry)
    raw = (registry / 'current.json').read_bytes()
    pointer = json.loads(raw)
    require(re.fullmatch('[0-9a-f]{64}', expected_generation), 'INVALID_GENERATION')
    require(pointer.get('generation') == expected_generation, 'LIBRARY_GENERATION_CHANGED')
    require(pointer.get('file') == 'versions/' + expected_generation + '.sqlite',
            'UNEXPECTED_CURRENT_VERSION_PATH')
    require(bool(generations), 'GENERATIONS_REQUIRED')
    require(len(generations) == len(set(generations)), 'DUPLICATE_GENERATION')
    require(generations[-1] == expected_generation, 'CURRENT_VERSION_MUST_BE_LAST')
    reports, transitions = [], []
    previous = None
    for generation in generations:
        require(re.fullmatch('[0-9a-f]{64}', generation), 'INVALID_GENERATION')
        path = plain(registry / 'versions' / (generation + '.sqlite'))
        require(path.is_file(), 'VERSION_NOT_FOUND:' + generation)
        report, products, reviews = snapshot(path, product_id)
        if previous:
            old_report, old_products, old_reviews = previous
            transitions.append({'from': old_report['generation'], 'to': generation,
                                'products': changes(old_products, products, ('marketplace', 'shop_id',
                                    'brand_code', 'category_code', 'brand_basis', 'category_basis')),
                                'reviews': changes(old_reviews, reviews, REVIEW_FIELDS)})
        reports.append(report)
        previous = report, products, reviews
    modules = []
    for folder in module_dirs:
        folder = plain(folder)
        modules.append({'directory': str(folder), 'files': {
            name: digest(plain(folder / name)) if (folder / name).is_file() else None for name in MODULES}})
    require((registry / 'current.json').read_bytes() == raw, 'CONCURRENT_POINTER_CHANGE')
    return {'schema': 'ksf.voc-generation-audit.v1', 'ok': True, 'readOnly': True,
            'databaseWrites': 0, 'networkRequests': 0, 'sourcePayloadsReturned': False,
            'reviewProseReturned': False, 'filesWritten': 'new-report-only',
            'currentGeneration': expected_generation, 'pointerSha256': hashlib.sha256(raw).hexdigest(),
            'pointerUnchanged': True, 'productId': product_id,
            'versions': reports, 'transitions': transitions, 'moduleDirectories': modules}


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--registry', type=Path, required=True)
    p.add_argument('--generation', action='append', required=True)
    p.add_argument('--expected-generation', required=True)
    p.add_argument('--product-id', default='jd:100005702440')
    p.add_argument('--module-dir', type=Path, action='append', default=[])
    p.add_argument('--out', type=Path, required=True)
    a = p.parse_args()
    out = plain(a.out)
    require(not out.exists() and out.parent.is_dir(), 'NEW_REPORT_IN_EXISTING_DIRECTORY_REQUIRED')
    report = audit(a.registry, a.generation, a.expected_generation, a.product_id, a.module_dir)
    fd = os.open(str(out), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    with os.fdopen(fd, 'w', encoding='utf-8', newline='\n') as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
        f.write('\n')
    print(json.dumps({'ok': True, 'out': str(out), 'sha256': digest(out),
                      'databaseWrites': 0, 'networkRequests': 0,
                      'versionCounts': [{'generation': v['generation'], **v['reviews']} for v in report['versions']]},
                     ensure_ascii=False))


if __name__ == '__main__':
    main()