kangshifu-board/scripts/audit-industry-library.py

"""Inventory the installed industry corpus without collection or SQL writes.

Python 3.9+, standard library only. Output must be a new directory, outside the
source registry. The optional board request is a read-only summary projection.
"""
import argparse
import datetime as dt
import json
from pathlib import Path
import sys
import urllib.parse
import urllib.request

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


def fingerprint(path):
    with Path(path).open('rb') as stream:
        return voc_db.file_sha256(stream)


def read_board(url):
    parsed = urllib.parse.urlsplit(url)
    voc_db.require(parsed.scheme in ('https', 'http') and bool(parsed.hostname) and
                   not parsed.username and not parsed.password and not parsed.query and not parsed.fragment,
                   'INVALID_BOARD_URL')
    req = urllib.request.Request(url, data=b'{"op":"summary"}',
                                 headers={'Content-Type': 'application/json', 'Accept': 'application/json'}, method='POST')
    with urllib.request.urlopen(req, timeout=20) as response:
        body = response.read(8 * 1024 * 1024 + 1)
    voc_db.require(len(body) <= 8 * 1024 * 1024, 'BOARD_RESPONSE_TOO_LARGE')
    return json.loads(body)


def audit(db=None, registry=None, project=None, board_url=None, board_reader=read_board):
    voc_db.require(bool(db) != bool(registry), 'ONE_DATABASE_INPUT_REQUIRED')
    project = Path(project or ROOT / 'docs/industry-voc-monitoring-20261106').resolve()
    scope_bytes = (project / 'scope.json').read_bytes()
    tasks_bytes = (project / 'tasks.json').read_bytes()
    scope = json.loads(scope_bytes)
    tasks = json.loads(tasks_bytes)
    voc_db.require(scope.get('schema') == 'ksf.industry-monitoring-scope.v1', 'UNSUPPORTED_PROJECT_SCOPE')
    voc_db.require(tasks.get('schema') == 'ksf.industry-monitoring-tasks.v1', 'UNSUPPORTED_TASK_PLAN')
    pointer_path = Path(registry).resolve() / 'current.json' if registry else None
    pointer_bytes = pointer_path.read_bytes() if pointer_path else None
    pointer = json.loads(pointer_bytes) if pointer_bytes else None
    if pointer_path:
        db = (pointer_path.parent / pointer['file']).resolve()
        voc_db.require(db.is_relative_to(pointer_path.parent), 'INVALID_LIBRARY_POINTER')
    db = Path(db).resolve()
    voc_db.require(db.is_file(), 'DATABASE_NOT_INSTALLED')
    voc_db.require(not any(Path(str(db) + suffix).exists() for suffix in ('-wal', '-journal')), 'LIVE_DATABASE_NOT_IMMUTABLE')
    before = fingerprint(db)
    if pointer and pointer.get('databaseSha256'):
        voc_db.require(pointer['databaseSha256'] == before, 'REGISTRY_DATABASE_SHA_MISMATCH')
    with voc_db.readonly(db) as c:
        inventory = library_audit.inventory(c, scope)
    after = fingerprint(db)
    voc_db.require(after == before, 'SOURCE_DATABASE_CHANGED_DURING_AUDIT')
    if pointer_path:
        voc_db.require(pointer_path.read_bytes() == pointer_bytes, 'REGISTRY_CHANGED_DURING_AUDIT')
        voc_db.require(pointer.get('generation') == inventory['generation'], 'REGISTRY_GENERATION_MISMATCH')
    report = {
        'ok': True, 'schema': library_audit.VERSION, 'generatedAt': dt.datetime.now(dt.timezone.utc).isoformat(),
        'readOnly': True, 'databaseWrites': 0, 'collectionCalls': 0, 'productionWrites': 0,
        'database': {'path': str(db), 'bytes': db.stat().st_size, 'sha256Before': before, 'sha256After': after,
                     'unchanged': True, 'registry': str(pointer_path.parent) if pointer_path else None,
                     'pointerSha256': voc_db.sha(pointer_bytes) if pointer_bytes else None},
        'project': {'path': str(project), 'scopeSha256': voc_db.sha(scope_bytes), 'tasksSha256': voc_db.sha(tasks_bytes)},
        'inventory': inventory,
        'boardComparison': {'status': 'not_requested', 'browserVerified': False},
    }
    if board_url:
        try:
            report['boardComparison'] = {**library_audit.compare_board(report, board_reader(board_url)),
                                         'url': board_url, 'checkedAt': dt.datetime.now(dt.timezone.utc).isoformat()}
        except Exception as error:
            report['boardComparison'] = {'status': 'failed', 'url': board_url,
                                         'error': str(error) if isinstance(error, voc_db.LibraryError) else type(error).__name__,
                                         'browserVerified': False}
    return report, tasks, scope


def write_report(out, report, tasks, scope):
    out = Path(out).resolve()
    db = Path(report['database']['path']).resolve()
    registry = report['database']['registry']
    voc_db.require(not out.exists(), 'OUTPUT_DIRECTORY_ALREADY_EXISTS')
    voc_db.require(not (registry and out.is_relative_to(Path(registry))), 'OUTPUT_INSIDE_SOURCE_REGISTRY')
    voc_db.require(not db.is_relative_to(out), 'OUTPUT_CONTAINS_SOURCE_DATABASE')
    # mkdir(exist_ok=False) also prevents a concurrent report from being overwritten.
    out.mkdir(parents=True, mode=0o700, exist_ok=False)
    data = {
        'audit.json': json.dumps(report, ensure_ascii=False, indent=2) + '\n',
        'tasks.json': json.dumps(tasks, ensure_ascii=False, indent=2) + '\n',
        'scope.json': json.dumps(scope, ensure_ascii=False, indent=2) + '\n',
        '监督台账.md': library_audit.markdown_report(report, tasks),
        '监督台账.html': library_audit.html_report(report, tasks),
    }
    for name, text in data.items():
        path = out / name
        with path.open('x', encoding='utf-8', newline='\n') as stream:
            stream.write(text)
        path.chmod(0o600)
    manifest = ''.join(fingerprint(out / name) + '  ' + name + '\n' for name in sorted(data))
    with (out / 'FILES.sha256').open('x', encoding='utf-8', newline='\n') as stream:
        stream.write(manifest)
    (out / 'FILES.sha256').chmod(0o600)
    return out


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument('--db')
    source.add_argument('--registry')
    parser.add_argument('--project', default=str(ROOT / 'docs/industry-voc-monitoring-20261106'))
    parser.add_argument('--compare-board', help='Public read-only summary URL, no credentials/query string')
    parser.add_argument('--out', required=True, help='New non-existing report directory')
    args = parser.parse_args()
    try:
        voc_db.require(not Path(args.out).exists(), 'OUTPUT_DIRECTORY_ALREADY_EXISTS')
        report, tasks, scope = audit(args.db, args.registry, args.project, args.compare_board)
        out = write_report(args.out, report, tasks, scope)
        board = report['boardComparison']
        comparison_passed = board.get('status') == 'compared' and board['generationMatches'] and board['allCountsMatch']
        print(json.dumps({'ok': True, 'report': str(out / '监督台账.html'),
                          'generation': report['inventory']['generation'], 'counts': report['inventory']['counts'],
                          'sourceUnchanged': True, 'databaseWrites': 0, 'collectionCalls': 0,
                          'boardComparisonPassed': comparison_passed if args.compare_board else None,
                          'physicalShaMatches': board.get('physicalShaMatches')}, ensure_ascii=False))
        return 0 if not args.compare_board or comparison_passed else 2
    except Exception as error:
        print(json.dumps({'ok': False, 'error': {'code': str(error) if isinstance(error, voc_db.LibraryError)
                                             else type(error).__name__}, 'databaseWrites': 0}, ensure_ascii=False), file=sys.stderr)
        return 1


if __name__ == '__main__':
    raise SystemExit(main())