kangshifu-board/docs/unified-database-migration-20261109/release-r2/package/components/research-runtime/scripts/voc_db.py

"""Versioned, offline VOC corpus. Python 3.9+, SQLite/FTS5, no third-party modules.

Builds ONLY a new database from an immutable archive. Query/get/contract open it
mode=ro and query_only. This is a research library, not the workshop business DB.
"""
import argparse
import base64
import collections
import contextlib
import datetime as dt
import decimal
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import sqlite3
import sys
import tempfile
import time
import uuid
import zipfile

VERSION = 'ksf.voc-library.v2'
HERE = Path(__file__).resolve().parent
PRIMARY = {'products': '底库商品_4635条.jsonl', 'reviews': '底库评价_2360条.jsonl',
           'social': '社媒内容_3814条.jsonl', 'documents': '政策与标准_25份.jsonl',
           'clauses': '政策条款与应用提示_31条.jsonl'}
TABLES = {'products': 'products', 'reviews': 'reviews', 'prices': 'prices',
          'social': 'social_posts', 'documents': 'policy_documents', 'clauses': 'policy_clauses'}


class LibraryError(ValueError):
    pass


def require(test, code):
    if not test:
        raise LibraryError(code)


def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)


def sha(value):
    return hashlib.sha256(value if isinstance(value, bytes) else value.encode('utf-8')).hexdigest()


def file_sha256(stream):
    """Hash bytes in bounded memory; identical on Python 3.9+ without monkey patches."""
    digest = hashlib.sha256()
    for block in iter(lambda: stream.read(1024 * 1024), b''):
        digest.update(block)
    return digest.hexdigest()


def key(*parts):
    return sha(canonical(parts))


def now():
    return dt.datetime.now(dt.timezone.utc).isoformat()


def utc(value):
    try:
        stamp = dt.datetime.fromisoformat(str(value).replace('Z', '+00:00'))
        if stamp.tzinfo is not None:
            return stamp.astimezone(dt.timezone.utc).isoformat()
    except (ValueError, TypeError):
        pass
    return None


def calendar_date(value):
    # Calendar component only: deliberately makes no timezone assumption.
    m = re.match(r'^(\d{4})[-年](\d{1,2})[-月](\d{1,2})', str(value or ''))
    if m:
        try:
            return dt.date(*map(int, m.groups())).isoformat()
        except ValueError:
            pass
    return None


def money(value):
    if value is None or isinstance(value, bool):
        return None
    text = str(value).strip()
    if not re.fullmatch(r'\d+(?:\.\d{1,2})?', text):
        return None  # No '?', ranges, currency strings or coupon guesses.
    amount = decimal.Decimal(text)
    return int(amount * 100) if 0 < amount < 100000000 else None


def family(value):
    return 'taobao' if value in ('taobao', 'taobao_tmall', 'tmall') else value


def identifier(value):
    require(isinstance(value, (str, int)) and not isinstance(value, bool), 'ID_MUST_BE_STRING_OR_INTEGER')
    # JSON integers retain arbitrary precision; never route through float/JS Number.
    result = str(value).strip()
    require(bool(result), 'EMPTY_ID')
    return result


def taxonomy():
    return json.loads((HERE / 'taxonomy.json').read_text(encoding='utf-8'))


def brand_hits(text, tax):
    text = str(text or '').casefold()
    return [b for b in tax['brands'] if any(a.casefold() in text for a in b['aliases'])]


def classify_brand(title, tax, upstream=None):
    hits = brand_hits(title, tax)
    clean = re.sub(r'^(?:\s|【[^】]*】|\[[^\]]*\])*', '', title)
    prefix = [b for b in hits if any(clean.casefold().startswith(a.casefold()) for a in b['aliases'])]
    if upstream:
        direct = [b for b in tax['brands'] if upstream.casefold() in [a.casefold() for a in b['aliases']]]
        if len(direct) == 1 and (not hits or {b['code'] for b in hits} == {direct[0]['code']}):
            return direct[0]['code'], 'source_attribute', None
        return None, 'conflicting_or_unregistered_attribute', 'BRAND_CONFLICT_OR_UNKNOWN'
    if len(hits) == len(prefix) == 1:
        return prefix[0]['code'], 'rule_title_prefix', None
    return None, 'unresolved_title', 'BRAND_AMBIGUOUS' if len(hits) > 1 else 'BRAND_UNCONFIRMED'


def classify_category(title):
    # Conservative business taxonomy; platform catalog codes are NOT this taxonomy.
    text = str(title or '')
    if re.search(r'补邮|邮费|差价|运费|购物袋|餐具', text) and not re.search(r'方便面|拌面|泡面', text):
        return None, 'non_product_link', 'NON_PRODUCT_LINK'
    if re.search(r'拌面酱|拌面汁|葱油酱|调味料|面条酱|拌面专用酱|拌面调料|炸酱罐头', text):
        return 'sauce', 'rule_title_sauce', None
    if re.search(r'挂面|生面条|生鲜面条', text):
        return 'raw_noodle', 'rule_title_raw_noodle', None
    if re.search(r'火腿|午餐肉|烤肠|拌饭|米饭|饮料|蛋糕|饼干', text) and not re.search(r'方便面|拌面|泡面|拉面', text):
        return 'other_food', 'rule_title_other', None
    labels = []
    for code, pattern in [('mianpi', r'面皮|凉皮'), ('reganmian', r'热干面'),
                          ('soup_noodle', r'汤面|汤达人|汤好喝'), ('dry_snack_noodle', r'干脆面|干吃面'),
                          ('fried_noodle', r'炒面'), ('banmian', r'拌面|干拌手擀面')]:
        if re.search(pattern, text):
            labels.append(code)
    # Explicit subtypes take precedence over the generic marketing term 拌面.
    if 'banmian' in labels and any(x in labels for x in ('mianpi', 'reganmian', 'fried_noodle')):
        labels.remove('banmian')
    if len(labels) == 1:
        return labels[0], 'rule_title_explicit', None
    if len(labels) > 1:
        return None, 'mixed_category_title', 'CATEGORY_CONFLICT'
    if re.search(r'方便面|泡面|桶面|杯面|碗面|袋面', text):
        return 'instant_unspecified', 'rule_title_unspecified', None
    return None, 'unresolved_title', 'CATEGORY_UNCONFIRMED'


def insert(c, table, row):
    # table/column names originate from code only, never from a query request.
    c.execute('INSERT INTO ' + table + '(' + ','.join(row) + ') VALUES(' + ','.join('?' for _ in row) + ')', list(row.values()))


class Archive:
    def __init__(self, path):
        self.path = Path(path)
        self.z = zipfile.ZipFile(path)
        self.files = {}
        for info in self.z.infolist():
            if info.is_dir():
                continue
            parts = PurePosixPath(info.filename).parts
            require(not info.filename.startswith('/') and '..' not in parts and '\\' not in info.filename, 'UNSAFE_ARCHIVE_PATH')
            require(info.filename not in self.files, 'DUPLICATE_ARCHIVE_MEMBER')
            self.files[info.filename] = info
        roots = {n.split('/')[0] for n in self.files}
        require(len(roots) == 1, 'ARCHIVE_ROOT_AMBIGUOUS')
        self.root = next(iter(roots)) + '/'
        self.latest = self.root + '01_最新底库/'
        self.cache = {}

    def resolve(self, value):
        if not value:
            return None, 'missing'
        text = str(value).replace('\\', '/').lstrip('./')
        if text in self.files:
            return text, 'matched'
        if '/voc-research/' in text:
            text = 'voc-research/' + text.split('/voc-research/', 1)[1]
        if re.match(r'^[A-Za-z]:/', text):
            return None, 'missing'  # Never guess by basename of an unknown path.
        if text not in self.cache:
            hits = [p for p in self.files if p.endswith('/' + text)]
            self.cache[text] = (hits[0], 'matched') if len(hits) == 1 else (None, 'ambiguous' if hits else 'missing')
        return self.cache[text]


def build(archive_path, db_path):
    path, dest = Path(archive_path).resolve(), Path(db_path).resolve()
    require(path.is_file() and path != dest, 'ARCHIVE_NOT_FOUND_OR_BAD_TARGET')
    with path.open('rb') as f:
        digest = file_sha256(f)
    tax = taxonomy()
    generation = key(VERSION, digest, sha((HERE / 'schema.sql').read_bytes()), sha((HERE / 'taxonomy.json').read_bytes()), sha(Path(__file__).read_bytes()), sha((HERE / 'retrieval.py').read_bytes()))
    if dest.exists():
        with readonly(dest) as c:
            require(dict(c.execute('SELECT key,value FROM meta')).get('generation') == generation, 'TARGET_EXISTS_DIFFERENT_GENERATION')
            return {'ok': True, 'operation': 'unchanged', 'generation': generation, 'archiveSha256': digest, 'counts': table_counts(c)}
    dest.parent.mkdir(parents=True, exist_ok=True)
    temp = dest.with_name(dest.name + '.build-' + str(uuid.uuid4()))
    a = Archive(path)
    started = time.perf_counter()
    try:
        c = sqlite3.connect(temp)
        c.executescript((HERE / 'schema.sql').read_text(encoding='utf-8'))
        with c:
            for k, v in {'schema': VERSION, 'generation': generation, 'archive_sha256': digest, 'taxonomy': canonical(tax), 'status': 'building'}.items():
                insert(c, 'meta', {'key': k, 'value': v})
            insert(c, 'batches', {'id': digest, 'archive_name': path.name, 'archive_sha256': digest, 'imported_at': now(), 'rule_version': tax['version']})
            for name, info in a.files.items():
                with a.z.open(name) as f:
                    h = file_sha256(f)
                insert(c, 'assets', {'path': name, 'sha256': h, 'byte_size': info.file_size})
                if name.endswith('.json') and ('/raw/' in name or '/raw-' in name):
                    try:
                        request = json.loads(a.z.read(name))
                    except (ValueError, UnicodeError):
                        continue
                    if isinstance(request, dict) and isinstance(request.get('id'), str) and isinstance(request.get('params'), dict):
                        response = request.get('result', {})
                        data = response.get('data', {}) if isinstance(response, dict) else {}
                        data = data if isinstance(data, dict) else {}
                        safe = {k:v for k,v in request['params'].items() if k in ('keyword','q','page','pageSize','shopId','itemId','tmall','sort','orderType','maxPages','rankId')}
                        insert(c, 'collection_requests', dict(asset_path=name, endpoint_id=request['id'], run_key=request.get('usageTraceId'),
                               safe_parameters_json=canonical(safe), collector_status=str(request.get('status') or 'unknown'),
                               http_status=request.get('httpStatus') if type(request.get('httpStatus')) is int else None,
                               upstream_code=str(data['code']) if data.get('code') is not None else None, upstream_request_id=data.get('requestId'),
                               started_at_raw=request.get('startedAt'), finished_at_raw=request.get('finishedAt'), record_time_raw=data.get('recordTime')))
            for b in tax['brands']:
                insert(c, 'brands', {'code': b['code'], 'name': b['name'], 'aliases_json': canonical(b['aliases'])})
            for cat in tax['categories']:
                insert(c, 'categories', {'code': cat['code'], 'name': cat['name'], 'parent_code': cat['parent'], 'definition': cat['definition']})

            def issue(entity, eid, code, detail):
                c.execute('INSERT OR IGNORE INTO quality_issues VALUES(?,?,?,?,?)', (key(entity, eid, code, detail), entity, eid, code, detail))

            def source(kind, asset, line, value, raw=None):
                rid = key(asset, line)
                insert(c, 'source_records', {'id': rid, 'kind': kind, 'asset_path': asset, 'line_number': line,
                       'payload_sha256': sha(raw if raw is not None else canonical(value)), 'payload_json': canonical(value)})
                for role in ('source_file', 'text_file', 'pdf_file'):
                    if value.get(role):
                        link(rid, role, value[role])
                for p in value.get('provenance', []):
                    if p.get('source_file'):
                        link(rid, 'provenance', p['source_file'])
                return rid

            def link(rid, role, original):
                target, resolution = a.resolve(original)
                c.execute('INSERT OR IGNORE INTO source_links VALUES(?,?,?,?,?)', (rid, role, original, target, resolution))
                if resolution != 'matched':
                    issue('source', rid, 'SOURCE_PATH_' + resolution.upper(), role + ': ' + original)

            def primary(kind):
                asset = a.latest + PRIMARY[kind]
                require(asset in a.files, 'PRIMARY_FILE_MISSING:' + kind)
                for line, raw in enumerate(a.z.read(asset).decode('utf-8-sig').splitlines(), 1):
                    if raw.strip():
                        value = json.loads(raw)
                        require(isinstance(value, dict), 'INVALID_RECORD')
                        yield value, source(kind, asset, line, value, raw)

            legacy = {}
            dbasset = a.latest + '方便面VOC扩充底库.sqlite'
            if dbasset in a.files:
                # Treat supplied SQL/scripts as data, never execute their instructions.
                with tempfile.TemporaryDirectory(prefix='ksf-voc-source-', dir=dest.parent) as td:
                    original = Path(td) / 'source.sqlite'
                    original.write_bytes(a.z.read(dbasset))
                    with readonly(original, check_version=False) as old:
                        for table in ('product_details', 'prices', 'ranking_signals', 'review_coverage', 'partial_catalog'):
                            exists = old.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone()
                            legacy[table] = [(line, json.loads(raw)) for line, raw in old.execute('SELECT row_id,payload FROM ' + table)] if exists else []
            details = collections.defaultdict(list)
            for line, row in legacy.get('product_details', []):
                sid = source('legacy_product_detail', dbasset, line * 10 + 1, row)
                details[(family(row.get('platform')), str(row.get('item_id')))].append((row, sid))
            product_rows = list(primary('products'))
            # Four reviewed listings are absent from the latest product export.
            # Recover only exact referenced IDs from the preserved partial catalog,
            # count these separately, and retain its partial-coverage designation.
            required_review_ids = {(family(r.get('platform')), str(r.get('item_id'))) for r in map(json.loads, a.z.read(a.latest + PRIMARY['reviews']).decode('utf-8-sig').splitlines()) if r}
            known_ids = {(family(r.get('platform')), str(r.get('item_id'))) for r, _ in product_rows}
            for line, row in legacy.get('partial_catalog', []):
                pair = (family(row.get('platform')), str(row.get('item_id')))
                if pair in required_review_ids and pair not in known_ids:
                    sid = source('legacy_partial_catalog', dbasset, line * 10 + 5, row)
                    # The original catalog explicitly identifies this channel;
                    # don't use the taobao_tmall alias alone as proof.
                    market = 'tmall' if '天猫' in str(row.get('shop_name')) else 'taobao_unverified'
                    recovered = {**row, 'marketplace':market, 'marketplace_evidence':'preserved_partial_catalog_explicit_tmall_shop' if market == 'tmall' else 'unverified_partial_catalog', 'coverage_type':'recovered_review_reference_partial_catalog'}
                    product_rows.append((recovered, sid))
                    known_ids.add(pair)
            for row, sid in product_rows:
                fam, iid = family(row.get('platform')), identifier(row['item_id'])
                require(fam in ('jd', 'taobao'), 'UNKNOWN_PRODUCT_PLATFORM')
                pid = fam + ':' + iid
                market = row.get('marketplace')
                require(market in (('jd',) if fam == 'jd' else ('tmall', 'taobao_unverified')), 'MARKETPLACE_FAMILY_CONFLICT')
                title = str(row.get('title') or '')
                detail_rows = details.get((fam, iid), [])
                upstream_brands = {r.get('product', {}).get('brandName') for r, _ in detail_rows if r.get('product', {}).get('brandName')}
                if row.get('brand_name'): upstream_brands.add(row['brand_name'])
                upstream = next(iter(upstream_brands)) if len(upstream_brands) == 1 else None
                brand, bbasis, bissue = classify_brand(title, tax, upstream)
                if len(upstream_brands) > 1:
                    brand, bbasis, bissue = None, 'conflicting_attributes', 'BRAND_CONFLICT'
                category, cbasis, cissue = classify_category(title)
                state = 'supported' if brand and category else 'pending'
                insert(c, 'products', dict(id=pid, platform_family=fam, item_id=iid, marketplace=market, marketplace_basis=str(row.get('marketplace_evidence') or 'collector_marketplace_assertion'),
                       shop_id=str(row.get('shop_id') or '') or None, shop_name=row.get('shop_name'), title=title, url=row.get('url'), brand_code=brand,
                       category_code=category, classification_state=state, brand_basis=bbasis, category_basis=cbasis, source_record_id=sid,
                       source_time_raw=row.get('source_time'), collected_at_raw=row.get('collected_at'), coverage_type=row.get('coverage_type')))
                for dimension, value, basis, evidence, field in [('brand', brand, bbasis, detail_rows[0][1] if upstream and detail_rows else sid, 'brand_name' if row.get('brand_name') else 'product.brandName' if upstream else 'title'),
                                                                ('category', category, cbasis, sid, 'title')]:
                    insert(c, 'classification_assertions', dict(id=key(pid, dimension), product_id=pid, dimension=dimension, value=value, method=basis,
                           rule_version=tax['version'], evidence_record_id=evidence, evidence_field=field, note='来源属性或保守标题规则；不是人工复核，也不是品牌真伪验证。'))
                for code in (bissue, cissue):
                    if code:
                        issue('product', pid, code, title)
                prov = row.get('provenance') or [row]
                seen = set()
                for p in prov:
                    oid = key(pid, {k: p.get(k) for k in ('source_file', 'source_time', 'collected_at', 'query', 'acquisition')})
                    if oid not in seen:
                        insert(c, 'product_observations', dict(id=oid, product_id=pid, source_record_id=sid, source_file=p.get('source_file'), source_time_raw=p.get('source_time'), collected_at_raw=p.get('collected_at'), query=p.get('query'), acquisition=p.get('acquisition')))
                        seen.add(oid)
                matches = list(re.finditer(r'(\d+(?:\.\d+)?)\s*(?:g|克)\s*[*×xX]\s*(\d+)\s*(?:包|袋|桶|碗|盒|杯)', title, re.I))
                match = matches[0] if len(matches) == 1 else None
                insert(c, 'specs', dict(id=key(pid, 'title_spec'), product_id=pid, upstream_sku_id=iid if fam == 'jd' and detail_rows else None,
                       spec_text=match.group(0) if match else None, net_weight_g=float(match[1]) if match and float(match[1]) > 0 else None,
                       pack_count=int(match[2]) if match and int(match[2]) > 0 else None, total_weight_g=float(match[1]) * int(match[2]) if match and float(match[1]) * int(match[2]) > 0 else None,
                       verification='title_candidate' if match else 'unknown', source_record_id=sid, basis='标题声明，未核净含量/赠品/面饼与调料组成；不生成每100g价格。'))
                if row.get('price') is not None:
                    add_price(c, pid, row.get('price'), row.get('price_type') or 'unknown', row, sid, issue)
                if row.get('promotional_price') is not None:
                    issue('product', pid, 'PROMOTIONAL_PRICE_UNRESOLVED', '促销价格口径/适用条件未映射，保留原记录，不混作搜索展示价。')

            for row, sid in primary('reviews'):
                fam, iid, reviewid = family(row.get('platform')), identifier(row['item_id']), identifier(row['review_id'])
                require(fam in ('jd', 'taobao'), 'UNKNOWN_REVIEW_PLATFORM')
                rid, pid = fam + ':' + iid + ':' + reviewid, fam + ':' + iid
                product = c.execute('SELECT marketplace,category_code FROM products WHERE id=?', (pid,)).fetchone()
                identity = 'orphan' if not product else 'mismatch' if row.get('item_identity_match') is False or row.get('observed_item_id') and str(row['observed_item_id']) != iid or product[0] != row.get('marketplace') else 'matched' if row.get('item_identity_match') is True else 'unverified'
                sku = str(row.get('sku') or '')
                scategory, _, sku_issue = classify_category(sku)
                category_state = 'unknown' if not product or not product[1] else 'conflict' if sku_issue == 'CATEGORY_CONFLICT' or scategory and scategory != 'instant_unspecified' and scategory != product[1] else 'consistent' if scategory == product[1] else 'listing_only'
                default = int(row['is_default_text']) if isinstance(row.get('is_default_text'), bool) else None
                insert(c, 'reviews', dict(id=rid, platform_family=fam, item_id=iid, review_id=reviewid, id_method=row.get('review_id_method') or 'unspecified_not_verified',
                       product_id=pid if product else None, marketplace_claim=row.get('marketplace') or 'unknown', content=str(row.get('content') or ''),
                       append_content_json=canonical(row.get('append_content')), rating_raw=str(row['rating']) if row.get('rating') is not None else None,
                       rating_scale=row.get('rating_scale'), review_date_raw=row.get('review_date'), review_date_utc=utc(row.get('review_date')), review_calendar_date=calendar_date(row.get('review_date')),
                       purchased_sku_text=sku, is_default=default, identity_state=identity, category_state=category_state, source_time_raw=row.get('source_time'), collected_at_raw=row.get('collected_at'), source_record_id=sid))
                for code, enabled in [('REVIEW_IDENTITY_' + identity.upper(), identity != 'matched'), ('REVIEW_CATEGORY_' + category_state.upper(), category_state in ('conflict', 'unknown')), ('REVIEW_DEFAULT_UNKNOWN', default is None)]:
                    if enabled:
                        issue('review', rid, code, '不得按标题/昵称模糊关联其他商品或平台。')

            for row, sid in primary('social'):
                platform, postid = str(row['platform']), identifier(row['id'])
                soid = platform + ':' + postid
                counts = {name: row.get(field) if isinstance(row.get(field), int) and not isinstance(row.get(field), bool) and row[field] >= 0 else None for name, field in [('likes','likes'), ('comment_count','comments'), ('collects','collects'), ('shares','shares')]}
                insert(c, 'social_posts', dict(id=soid, platform=platform, post_id=postid, title=row.get('title'), content=row.get('content'), url=row.get('url'),
                       content_scope=row.get('content_scope') or 'unknown', published_at_raw=row.get('published_at'), published_at_utc=utc(row.get('published_at')),
                       collected_at_raw=row.get('collected_at'), fulltext_available=int(bool(str(row.get('content') or '').strip())), source_record_id=sid, **counts))
                for b in brand_hits(str(row.get('title') or '') + '\n' + str(row.get('content') or ''), tax):
                    insert(c, 'social_brand_mentions', dict(post_id=soid, brand_code=b['code'], basis='text_mention_only_not_product_ownership'))

            for row, sid in primary('documents'):
                matches = [p for p in (row.get('pdf_file'), row.get('text_file'), row.get('source_file')) if p]
                hashes = [c.execute('SELECT sha256 FROM assets WHERE path=?', (a.resolve(p)[0],)).fetchone() for p in matches]
                hash_basis = next((a.resolve(p)[0] + '#bytes' for p, h in zip(matches, hashes) if h and h[0] == row.get('sha256')), None)
                if not hash_basis and row.get('source_file'):
                    source_path, _ = a.resolve(row['source_file'])
                    if source_path and source_path.endswith('.json'):
                        doc = json.loads(a.z.read(source_path))
                        # The expansion collector hashes the decoded `text`,
                        # whereas earlier collectors hash a complete JSON/PDF.
                        if isinstance(doc, dict) and isinstance(doc.get('text'), str) and sha(doc['text']) == row.get('sha256'):
                            hash_basis = source_path + '#$.text (UTF-8 decoded text)'
                hash_state = 'matched' if hash_basis else 'unresolved'
                full_text = row.get('full_text') if isinstance(row.get('full_text'), str) else None
                text_path, _ = a.resolve(row.get('text_file'))
                if text_path and full_text is None:
                    full_text = a.z.read(text_path).decode('utf-8-sig')
                elif row.get('source_file') and full_text is None:
                    json_path, _ = a.resolve(row['source_file'])
                    if json_path and json_path.endswith('.json'):
                        document_json = json.loads(a.z.read(json_path))
                        if isinstance(document_json, dict) and isinstance(document_json.get('text'), str):
                            full_text = document_json['text']
                insert(c, 'policy_documents', dict(id=identifier(row['document_id']), title=str(row['title']), declared_status=str(row.get('declared_status') or row.get('status_on_20261104') or 'unknown'), status_as_of=row.get('status_as_of') or ('2026-11-04' if 'status_on_20261104' in row else 'unknown'),
                       issued_date=row.get('issued_date'), effective_date=row.get('effective_date'), source_url=row.get('source_url'), expected_sha256=row.get('sha256'),
                       attachment_hash_state=hash_state, hash_basis=hash_basis, coverage=row.get('coverage'), full_text=full_text, source_record_id=sid))
                if hash_state != 'matched':
                    issue('document', row['document_id'], 'ATTACHMENT_HASH_UNRESOLVED', '需核对声明的 SHA 对应哪份附件；不能声称完整原文已逐字核验。')
            for row, sid in primary('clauses'):
                method = str(row.get('excerpt_method') or '')
                insert(c, 'policy_clauses', dict(id=key(row['document_id'], row['clause']), document_id=row['document_id'], clause=str(row['clause']), pdf_page=str(row['pdf_page']) if row.get('pdf_page') is not None else None,
                       excerpt=str(row.get('source_excerpt') or ''), excerpt_type='paraphrase_or_transcription' if '非逐字' in method or '摘要' in method else 'unverified_excerpt',
                       application_note=row.get('application_note'), excerpt_method=method or None, source_record_id=sid))

            for kind in ('prices', 'ranking_signals', 'review_coverage'):
                for line, row in legacy.get(kind, []):
                    sid = source('legacy_' + kind, dbasset, line * 10 + {'prices': 2, 'ranking_signals': 3, 'review_coverage': 4}[kind], row)
                    pid = str(family(row.get('platform'))) + ':' + str(row.get('item_id'))
                    exists = c.execute('SELECT 1 FROM products WHERE id=?', (pid,)).fetchone()
                    if kind == 'prices':
                        if exists:
                            add_price(c, pid, row.get('price_raw', row.get('price_cny')), str(row.get('price_type') or 'unknown'), row, sid, issue)
                        else:
                            issue('source', sid, 'LEGACY_ORPHAN_PRODUCT', pid)
                    elif kind == 'ranking_signals':
                        insert(c, 'ranking_signals', dict(id=key(sid, kind), product_id=pid if exists else None, rank_id=str(row.get('rank_id') or ''), rank_number=row.get('rank'),
                               title=row.get('rank_title'), scope='detail_page_badge_not_complete_ranking', source_time_raw=row.get('source_time'), source_record_id=sid))
                    else:
                        insert(c, 'coverage_observations', dict(id=key(sid, kind), product_id=pid if exists else None, status=row.get('status'), available_reviews=row.get('available_reviews'),
                               reported_total=str(row.get('reported_total')) if row.get('reported_total') is not None else None, historical_total_complete=int(row.get('historical_total_complete') is True), source_record_id=sid))
            from retrieval import create_index
            create_index(c)
            c.execute("UPDATE meta SET value='ready' WHERE key='status'")
        require(c.execute('PRAGMA integrity_check').fetchone()[0] == 'ok', 'INTEGRITY_CHECK_FAILED')
        require(not c.execute('PRAGMA foreign_key_check').fetchall(), 'FOREIGN_KEY_CHECK_FAILED')
        c.execute('ANALYZE')
        c.commit()
        counts = table_counts(c)
        quality = dict(c.execute('SELECT code,count(*) FROM quality_issues GROUP BY code'))
        c.close()
        os.chmod(temp, 0o600)
        # No overwriting an existing database, including one created concurrently.
        os.link(temp, dest)
        temp.unlink()
        return {'ok': True, 'operation': 'created', 'schema': VERSION, 'generation': generation, 'archiveSha256': digest,
                'counts': counts, 'quality': quality, 'durationSeconds': round(time.perf_counter() - started, 3),
                'productionImported': False, 'note': '已生成本地结构库；部署到授权 Agent 后才能查询，未写生产业务库。'}
    finally:
        a.z.close()
        with contextlib.suppress(Exception):
            c.close()
        if temp.exists():
            temp.unlink()


def add_price(c, pid, raw, basis, row, sid, issue):
    amount = money(raw)
    if amount is None:
        issue('source', sid, 'PRICE_NOT_SINGLE_NUMERIC', str(raw))
    # Repeated observations of the same response are lineage, not new prices.
    basis = {'搜索展示价':'search_listing_snapshot','详情展示价':'detail_listing_snapshot'}.get(basis, basis)
    asset = c.execute("SELECT asset_path FROM source_links WHERE record_id=? AND role='source_file' AND resolution='matched'", (sid,)).fetchone()
    priceid = key(pid, basis, amount if amount is not None else str(raw), row.get('source_time'), asset[0] if asset else row.get('source_file'))
    c.execute('INSERT OR IGNORE INTO prices VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)',
              (priceid, pid, 'CNY', amount, str(raw), basis, row.get('source_time'), utc(row.get('source_time')), calendar_date(row.get('source_time')),
               row.get('collected_at'), utc(row.get('collected_at')), None, 'unavailable', sid))
    c.execute('INSERT OR IGNORE INTO price_sources VALUES(?,?)', (priceid,sid))


@contextlib.contextmanager
def readonly(path, check_version=True):
    from unified_store import Reference, read_research
    if isinstance(path, Reference):
        with read_research(path) as c:
            if check_version:
                values = dict(c.execute('SELECT key,value FROM meta'))
                require(values.get('schema') == VERSION and values.get('status') == 'ready', 'DATABASE_VERSION_OR_STATE_MISMATCH')
            yield c
        return
    p = Path(path).resolve()
    require(p.is_file(), 'DATABASE_NOT_INSTALLED')
    c = sqlite3.connect(p.as_uri() + '?mode=ro&immutable=1', uri=True)
    c.execute('PRAGMA query_only=ON')
    c.execute('PRAGMA trusted_schema=OFF')
    c.row_factory = sqlite3.Row
    try:
        if check_version:
            values = dict(c.execute('SELECT key,value FROM meta'))
            require(values.get('schema') == VERSION and values.get('status') == 'ready', 'DATABASE_VERSION_OR_STATE_MISMATCH')
        yield c
    finally:
        c.close()


def table_counts(c):
    names = ['products','reviews','social_posts','policy_documents','policy_clauses','prices','specs','product_observations','source_records','source_links','assets','collection_requests','quality_issues','ranking_signals','coverage_observations']
    return {t: c.execute('SELECT count(*) FROM ' + t).fetchone()[0] for t in names}


def contract(c):
    meta = dict(c.execute('SELECT key,value FROM meta'))
    from retrieval import catalog
    from schema_catalog import catalog as schema_catalog
    return {'ok': True, 'schema': VERSION, 'generation': meta['generation'], 'readOnly': True, 'databaseWrites': 0,
            'retrieval':catalog(c), 'recommendedCommands':['search','aggregate','facets','get','source'], 'legacyQueryNotice':'旧 query 保留三维模板兼容；一般问题请用 search，按实际需求设置条件，不要求补齐三个维度。',
            'taxonomy': json.loads(meta['taxonomy']), 'entities': list(TABLES), 'queryFields': ['entity','marketplace','brand','category','shopId','itemId','keyword','limit','cursor','from','to','timeBasis','reviewScope','includeDefault','contentScope','documentId','policyStatus'],
            'dataDictionary':schema_catalog(c, detailed=False),
            'limits': {'default': 20, 'max': 100}, 'counts': table_counts(c),
            'rules': ['优先用 search 按问题实际指定的维度过滤；不要求填齐平台/品牌/品类；旧 query 的三维模板仅保留兼容。',
                      '全部条件 AND；不自动包含子类；同族商品精确关联评论；不按标题/昵称跨商品合并。',
                      'reviewScope=sku_confirmed 默认，仅用购买规格明确匹配品类的评价；listing 单独查询并标识规格未证实。',
                      '默认排除默认评价、身份待核/错配、品类冲突；默认文本数量不等于观点数量。',
                      'timeBasis 必填才能用 from/to；calendar 日期含起点、不含终点，来源时区未证实；UTC 不含无时区数据。',
                      'social 的 brand 是文本提及，绝不是商品品牌归属；comment_count 不是已采集评论。',
                      '政策状态仅为 2026-11-04 的声明快照；条款均未自动认定逐字引文。'],
            'capabilities': {'exactScopeQuery': True, 'getCitation': True, 'activeRegistrySupported': True, 'generationGuard': True,
                             'upstreamCollection': False, 'businessDatabaseProjection': False, 'publicBoardConnectionVerified': False}}


def resolve_code(value, rows, code_field='code'):
    if value == 'all':
        return value
    require(isinstance(value, str) and bool(value.strip()), 'EXPLICIT_SCOPE_REQUIRED')
    hits = [r[code_field] for r in rows if value.casefold() in [str(r.get(code_field)).casefold(), str(r.get('name')).casefold(), *[a.casefold() for a in r.get('aliases', [])]]]
    require(len(hits) == 1, 'UNKNOWN_OR_AMBIGUOUS_TAXONOMY:' + value)
    return hits[0]


def query_plan(c, request):
    require(isinstance(request, dict), 'INVALID_QUERY')
    allowed = {'entity','marketplace','brand','category','shopId','itemId','keyword','limit','cursor','from','to','timeBasis','reviewScope','includeDefault','contentScope','documentId','policyStatus'}
    require(not set(request) - allowed, 'UNKNOWN_QUERY_FIELD')
    entity = request.get('entity')
    require(entity in TABLES, 'UNKNOWN_ENTITY')
    size = request.get('limit', 20)
    require(type(size) is int and 1 <= size <= 100, 'INVALID_LIMIT')
    tax = json.loads(c.execute("SELECT value FROM meta WHERE key='taxonomy'").fetchone()[0])
    normalized = dict(request)
    conditions, values = [], []
    entity_id = 'p.id' if entity == 'products' else 'r.id' if entity == 'reviews' else 'v.id'
    scope_conditions, scope_values = [], []
    excluded = {}
    if entity in ('products','reviews','prices'):
        market = request.get('marketplace')
        require(market == 'all' or market in tax['marketplaceAliases'], 'MARKETPLACE_REQUIRED_OR_UNKNOWN')
        market = 'all' if market == 'all' else tax['marketplaceAliases'][market]
        normalized.update(marketplace=market, brand=resolve_code(request.get('brand'), tax['brands']), category=resolve_code(request.get('category'), tax['categories']))
        for col, val in [('marketplace', market), ('brand_code', normalized['brand']), ('category_code', normalized['category'])]:
            if val != 'all':
                conditions.append('p.' + col + '=?')
                values.append(val)
        conditions.append("p.classification_state='supported'")
        for field, column in [('shopId','shop_id'),('itemId','item_id')]:
            if field in request:
                conditions.append('p.' + column + '=?')
                values.append(identifier(request[field]))
        if 'keyword' in request:
            require(isinstance(request['keyword'], str) and 0 < len(request['keyword']) <= 100, 'INVALID_KEYWORD')
            conditions.append('instr(p.title,?)>0')
            values.append(request['keyword'])
        scope_conditions, scope_values = conditions[:], values[:]
        scoped_products = c.execute('SELECT count(*) FROM products p WHERE ' + ' AND '.join(conditions), values).fetchone()[0]
        excluded['scopeProducts'] = scoped_products
        excluded['unclassifiedProductsInMarketplace'] = c.execute("SELECT count(*) FROM products WHERE classification_state='pending'" + (' AND marketplace=?' if market != 'all' else ''), [market] if market != 'all' else []).fetchone()[0]
        if entity == 'products':
            require(not set(request) & {'reviewScope','includeDefault','contentScope','documentId','policyStatus','from','to','timeBasis'}, 'FILTER_NOT_SUPPORTED_FOR_ENTITY')
            from_sql = 'products p'
            select = 'p.*'
        elif entity == 'reviews':
            require(not set(request) & {'contentScope','documentId','policyStatus'}, 'FILTER_NOT_SUPPORTED_FOR_ENTITY')
            scope = request.get('reviewScope', 'sku_confirmed')
            require(scope in ('sku_confirmed','listing'), 'INVALID_REVIEW_SCOPE')
            require(type(request.get('includeDefault', False)) is bool, 'INVALID_INCLUDE_DEFAULT')
            normalized.update(reviewScope=scope, includeDefault=request.get('includeDefault', False))
            base = ' FROM reviews r JOIN products p ON p.id=r.product_id WHERE ' + ' AND '.join(conditions)
            excluded['reviewCountsBeforeQualityFilters'] = {str(x[0]): x[1] for x in c.execute('SELECT r.identity_state || ":" || r.category_state || ":default=" || coalesce(r.is_default,"unknown"),count(*)' + base + ' GROUP BY 1', values)}
            conditions.append("r.identity_state='matched'")
            conditions.append("r.category_state='consistent'" if scope == 'sku_confirmed' else "r.category_state IN ('consistent','listing_only')")
            if not normalized['includeDefault']:
                conditions.append('r.is_default=0')
            from_sql = 'products p JOIN reviews r ON r.product_id=p.id'
            select = 'r.*,p.marketplace,p.brand_code,p.category_code,p.title AS product_title,p.brand_basis,p.category_basis'
        else:
            require(not set(request) & {'reviewScope','includeDefault','contentScope','documentId','policyStatus'}, 'FILTER_NOT_SUPPORTED_FOR_ENTITY')
            conditions.append('v.amount_minor IS NOT NULL')
            from_sql = 'products p JOIN prices v ON v.product_id=p.id'
            select = 'v.*,p.marketplace,p.brand_code,p.category_code,p.title AS product_title,p.brand_basis,p.category_basis'
    elif entity == 'social':
        require(not set(request) & {'category','shopId','itemId','reviewScope','includeDefault','documentId','policyStatus'}, 'FILTER_NOT_SUPPORTED_FOR_ENTITY')
        platform = request.get('marketplace')
        require(platform in ('bilibili','douyin','xiaohongshu','all'), 'SOCIAL_PLATFORM_REQUIRED')
        if platform != 'all':
            conditions.append('v.platform=?'); values.append(platform)
        scope = request.get('contentScope', 'packaged_noodles')
        require(scope in ('packaged_noodles','adjacent_noodle_food','unknown','all'), 'UNKNOWN_CONTENT_SCOPE')
        normalized['contentScope'] = scope
        if scope != 'all':
            conditions.append('v.content_scope=?'); values.append(scope)
        if 'brand' in request:
            brand = resolve_code(request['brand'], tax['brands'])
            normalized['brand'] = brand
            if brand != 'all':
                conditions.append('EXISTS(SELECT 1 FROM social_brand_mentions m WHERE m.post_id=v.id AND m.brand_code=?)'); values.append(brand)
        if 'keyword' in request:
            require(isinstance(request['keyword'], str) and 0 < len(request['keyword']) <= 100, 'INVALID_KEYWORD')
            conditions.append("instr(coalesce(v.title,'') || coalesce(v.content,''),?)>0"); values.append(request['keyword'])
        from_sql, select = 'social_posts v', 'v.*'
    else:
        require(not set(request) & {'marketplace','brand','category','shopId','itemId','reviewScope','includeDefault','contentScope','from','to','timeBasis'}, 'FILTER_NOT_SUPPORTED_FOR_ENTITY')
        from_sql, select = ('policy_documents v', 'v.*') if entity == 'documents' else ('policy_clauses v JOIN policy_documents d ON d.id=v.document_id', 'v.*,d.declared_status,d.status_as_of,d.source_url')
        if 'documentId' in request:
            conditions.append(('v.id' if entity == 'documents' else 'v.document_id') + '=?'); values.append(identifier(request['documentId']))
        if 'policyStatus' in request:
            require(request['policyStatus'] in {r[0] for r in c.execute('SELECT DISTINCT declared_status FROM policy_documents')}, 'UNKNOWN_POLICY_STATUS')
            conditions.append(('v' if entity == 'documents' else 'd') + '.declared_status=?'); values.append(request['policyStatus'])
        if 'keyword' in request:
            require(isinstance(request['keyword'], str) and 0 < len(request['keyword']) <= 100, 'INVALID_KEYWORD')
            conditions.append(('instr(v.title,?)>0' if entity == 'documents' else 'instr(v.excerpt || coalesce(v.application_note,\'\'),?)>0')); values.append(request['keyword'])
    if set(request) & {'from','to','timeBasis'}:
        time_fields = {'reviews': {'review_calendar_date':'r.review_calendar_date','review_utc':'r.review_date_utc'},
                       'prices': {'source_calendar_date':'v.source_calendar_date','source_utc':'v.source_time_utc'},
                       'social': {'published_utc':'v.published_at_utc'}}
        basis = request.get('timeBasis')
        require(basis in time_fields.get(entity, {}) and bool(set(request) & {'from','to'}), 'EXPLICIT_SUPPORTED_TIME_BASIS_REQUIRED')
        column = time_fields[entity][basis]
        before = ' AND '.join(conditions) if conditions else '1=1'
        excluded['unknownTimeExcluded'] = c.execute('SELECT count(*) FROM ' + from_sql + ' WHERE ' + before + ' AND ' + column + ' IS NULL', values).fetchone()[0]
        normalized_times = {}
        for field, op in [('from','>='),('to','<')]:
            if field in request:
                value = request[field]
                require(isinstance(value, str), 'INVALID_TIME')
                if basis.endswith('calendar_date'):
                    require(bool(re.fullmatch(r'\d{4}-\d{2}-\d{2}', value)) and calendar_date(value) == value, 'INVALID_DATE')
                else:
                    value = utc(value); require(value is not None, 'TIMEZONE_REQUIRED')
                normalized_times[field] = value
                conditions.append(column + op + '?'); values.append(value)
        require(not ('from' in normalized_times and 'to' in normalized_times) or normalized_times['from'] < normalized_times['to'], 'INVALID_TIME_RANGE')
        normalized.update(normalized_times)
    where = ' AND '.join(conditions) if conditions else '1=1'
    total = c.execute('SELECT count(*) FROM ' + from_sql + ' WHERE ' + where, values).fetchone()[0]
    generation = c.execute("SELECT value FROM meta WHERE key='generation'").fetchone()[0]
    context = key(generation, {k:v for k,v in normalized.items() if k not in ('cursor','limit')})
    if request.get('cursor'):
        try:
            cursor = json.loads(base64.urlsafe_b64decode(request['cursor']))
            require(set(cursor) == {'context','after'} and cursor['context'] == context and isinstance(cursor['after'], str), 'CURSOR_SCOPE_MISMATCH')
        except (ValueError, TypeError, KeyError):
            raise LibraryError('CURSOR_SCOPE_MISMATCH') from None
        where += ' AND ' + entity_id + '>?'; values.append(cursor['after'])
    sql = 'SELECT ' + select + ' FROM ' + from_sql + ' WHERE ' + where + ' ORDER BY ' + entity_id + ' LIMIT ?'
    return sql, values + [size + 1], normalized, context, total, excluded


def citation(c, sid, expand=False):
    s = c.execute('SELECT id,kind,asset_path,line_number,payload_sha256 FROM source_records WHERE id=?', (sid,)).fetchone()
    require(s is not None, 'SOURCE_NOT_FOUND')
    result = dict(s)
    result['locatorType'] = 'sqlite_legacy_row_encoded' if s['kind'].startswith('legacy_') else 'jsonl_line_1_based'
    if s['kind'].startswith('legacy_'):
        result['legacyRowId'] = s['line_number'] // 10
        result['legacyTable'] = {'legacy_product_detail':'product_details','legacy_prices':'prices','legacy_ranking_signals':'ranking_signals','legacy_review_coverage':'review_coverage','legacy_partial_catalog':'partial_catalog'}[s['kind']]
    result['assetSha256'] = c.execute('SELECT sha256 FROM assets WHERE path=?', (s['asset_path'],)).fetchone()[0]
    result['archiveSha256'] = c.execute("SELECT value FROM meta WHERE key='archive_sha256'").fetchone()[0]
    if c.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='source_origins'").fetchone():
        origins=[r[0] for r in c.execute('SELECT archive_sha256 FROM source_origins WHERE source_record_id=? ORDER BY archive_sha256',(sid,))]
        if origins:
            result['archiveSha256']=origins[0]
            result['archiveSha256s']=origins
    result['upstreamSourceCount'] = c.execute('SELECT count(*) FROM source_links WHERE record_id=?', (sid,)).fetchone()[0]
    result['upstreamSources'] = [dict(r) for r in c.execute('SELECT l.role,l.original_path,l.asset_path,l.resolution,a.sha256 FROM source_links l LEFT JOIN assets a ON a.path=l.asset_path WHERE l.record_id=? ORDER BY CASE WHEN l.role=\'source_file\' THEN 0 ELSE 1 END,l.role,l.original_path' + ('' if expand else ' LIMIT 3'), (sid,))]
    for link in result['upstreamSources']:
        request = c.execute('SELECT endpoint_id,run_key,safe_parameters_json,collector_status,http_status,upstream_code,upstream_request_id,started_at_raw,finished_at_raw,record_time_raw FROM collection_requests WHERE asset_path=?', (link['asset_path'],)).fetchone()
        if request:
            link['collectionRequest'] = dict(request)
    result['upstreamSourcesTruncated'] = len(result['upstreamSources']) < result['upstreamSourceCount']
    return result


def decorate(c, row, entity):
    result = dict(row)
    result['citation'] = citation(c, row['source_record_id'])
    if entity == 'products':
        result['classification'] = [dict(r) for r in c.execute('SELECT dimension,value,method,rule_version,evidence_record_id,evidence_field,note FROM classification_assertions WHERE product_id=? ORDER BY dimension', (row['id'],))]
        for assertion in result['classification']:
            dimension=assertion['dimension']
            assertion['selectedForCurrentRecord']=assertion['value']==row[dimension+'_code'] and assertion['method']==row[dimension+'_basis']
        result['specs'] = [dict(r) for r in c.execute('SELECT upstream_sku_id,spec_text,net_weight_g,pack_count,total_weight_g,verification,basis,source_record_id FROM specs WHERE product_id=?', (row['id'],))]
    if c.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_decisions'").fetchone():
        decisions=[dict(r) for r in c.execute('SELECT decision,reason_codes_json,batch_sha256,prior_source_record_id,incoming_source_record_id,selected_source_record_id,policy_version FROM merge_decisions WHERE entity=? AND record_id=? ORDER BY id LIMIT 11',(entity,row['id']))]
        for decision in decisions:decision['reasons']=json.loads(decision.pop('reason_codes_json'))
        result['mergeHistory']=decisions[:10];result['mergeHistoryTruncated']=len(decisions)>10
    if entity in ('reviews', 'prices') and row['product_id']:
        product = c.execute('SELECT source_record_id FROM products WHERE id=?', (row['product_id'],)).fetchone()
        result['productCitation'] = citation(c, product[0])
    return result


def query(c, request, explain=False):
    started = time.perf_counter()
    sql, values, normalized, context, total, excluded = query_plan(c, request)
    rows = c.execute(sql, values).fetchall()
    size = normalized.get('limit', 20)
    more = len(rows) > size
    rows = rows[:size]
    cursor = base64.urlsafe_b64encode(canonical({'context':context, 'after':rows[-1]['id']}).encode()).decode() if more else None
    result = {'ok':True, 'schema':VERSION, 'readOnly':True, 'databaseWrites':0, 'effectiveFilters':{k:v for k,v in normalized.items() if k != 'cursor'},
              'scopeCombination':'AND', 'totalMatched':total, 'records':[decorate(c, row, normalized['entity']) for row in rows], 'nextCursor':cursor,
              'coverage':excluded, 'classificationNotice':'品牌/品类含保守标题规则，并非全部人工复核；未知/冲突排除，详见 basis 与证据。',
              'elapsedMs':round((time.perf_counter()-started)*1000,3)}
    if explain:
        result['queryPlan'] = [r['detail'] for r in c.execute('EXPLAIN QUERY PLAN ' + sql, values)]
    return result


def get_record(c, entity, recordid):
    from retrieval import ENTITIES
    require(entity in ENTITIES, 'UNKNOWN_ENTITY')
    row = c.execute('SELECT * FROM ' + ENTITIES[entity] + ' WHERE id=?', (recordid,)).fetchone()
    require(row is not None, 'RECORD_NOT_FOUND')
    return {'ok':True, 'schema':VERSION, 'readOnly':True, 'databaseWrites':0, 'record':decorate(c, row, entity)}


def get_source(c, recordid):
    row = c.execute('SELECT payload_json FROM source_records WHERE id=?', (recordid,)).fetchone()
    require(row is not None, 'SOURCE_NOT_FOUND')
    return {'ok':True, 'schema':VERSION, 'readOnly':True, 'databaseWrites':0,
            'citation':citation(c, recordid, expand=True), 'sourcePayload':json.loads(row[0])}


def registry_database(root):
    """Resolve the same atomic active pointer used by the browser service."""
    root=Path(root).resolve()
    pointer=json.loads((root/'current.json').read_text(encoding='utf-8'))
    db=(root/pointer['file']).resolve()
    require(db.is_relative_to(root) and db.is_file(),'INVALID_LIBRARY_POINTER')
    return db


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('command', choices=['build','contract','schema','query','search','aggregate','facets','get','source','verify'])
    parser.add_argument('--db', default=os.environ.get('KSF_VOC_DB', '/opt/data/ksf-voc-library/library.sqlite'))
    parser.add_argument('--registry', default=os.environ.get('KSF_LIBRARY_REGISTRY'))
    parser.add_argument('--generation', help='Expected database generation from contract; fail if the active library changed')
    parser.add_argument('--archive')
    parser.add_argument('--input')
    from retrieval import ENTITIES
    parser.add_argument('--entity', choices=list(ENTITIES))
    parser.add_argument('--id')
    parser.add_argument('--receipt')
    parser.add_argument('--explain', action='store_true')
    parser.add_argument('--table', help='Exact installed table for schema inspection')
    parser.add_argument('--profile', action='store_true', help='Count non-null fields of the selected schema table; no row contents')
    args = parser.parse_args()
    try:
        require(args.command == 'schema' or not (args.table or args.profile), 'SCHEMA_OPTIONS_REQUIRE_SCHEMA_COMMAND')
        if args.command == 'build':
            require(not args.registry, 'BUILD_REJECTS_ACTIVE_REGISTRY')
            require(not args.generation, 'BUILD_REJECTS_GENERATION')
            require(args.archive, 'ARCHIVE_REQUIRED')
            result = build(args.archive, args.db)
        else:
            require(not args.archive and not args.receipt, 'READ_ONLY_COMMAND_REJECTS_WRITE_OPTIONS')
            require(not (args.registry and '--db' in sys.argv), 'AMBIGUOUS_DATABASE_LOCATION')
            import unified_store
            location = unified_store.research_reference() if unified_store.configured() and '--db' not in sys.argv else (registry_database(args.registry) if args.registry else args.db)
            with readonly(location) as c:
                generation = c.execute("SELECT value FROM meta WHERE key='generation'").fetchone()[0]
                require(args.generation is None or args.generation == generation, 'LIBRARY_GENERATION_CHANGED')
                if args.command == 'contract':
                    result = contract(c)
                    root = Path(args.registry) if args.registry else None
                    unified = isinstance(location, unified_store.Reference)
                    result['publicationBinding'] = {'mode':'parse-postgresql' if unified else ('registry' if root else 'snapshot'),
                        'newGenerationsSupported':bool(unified or root), 'registryWritableByCurrentProcess':False if unified else bool(root and os.access(root, os.W_OK)),
                        'currentGeneration':generation, 'taskAuthorizationStillRequired':True,
                        'note':'研究底料与业务成果共用企业数据库，按原接口发布新代际' if unified else '新数据可按已授权任务通过update_library.py合并新代际；快照模式须先由运维绑定实际发布位置'}
                elif args.command == 'schema':
                    from schema_catalog import catalog as schema_catalog
                    result = {'ok':True, 'generation':generation, **schema_catalog(c, args.table, args.profile)}
                elif args.command in ('query','search','aggregate','facets'):
                    require(args.input, 'INPUT_REQUIRED')
                    request = json.loads(sys.stdin.read() if args.input == '-' else Path(args.input).read_text(encoding='utf-8-sig'))
                    if args.command=='query':result = query(c, request, args.explain)
                    else:
                        from retrieval import search, aggregate
                        result = search(c,request,args.explain) if args.command=='search' else aggregate(c,request,facets=args.command=='facets')
                elif args.command == 'get':
                    require(args.entity and args.id, 'ENTITY_AND_ID_REQUIRED')
                    result = get_record(c, args.entity, args.id)
                elif args.command == 'source':
                    require(args.id, 'SOURCE_ID_REQUIRED')
                    result = get_source(c, args.id)
                else:
                    require(c.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' and not c.execute('PRAGMA foreign_key_check').fetchall(), 'DATABASE_CHECK_FAILED')
                    result = {'ok':True, 'schema':VERSION, 'readOnly':True, 'databaseWrites':0, 'counts':table_counts(c)}
                result['generation'] = generation
        if args.command == 'build' and args.receipt:
            # An operator may create a new receipt; never overwrite arbitrary files.
            with Path(args.receipt).open('x', encoding='utf-8') as f:
                f.write(canonical(result) + '\n')
        print(json.dumps(result, ensure_ascii=False, indent=2))
    except (LibraryError, sqlite3.Error, ValueError, OSError, zipfile.BadZipFile) as error:
        code = str(error)
        if os.environ.get('KSF_UNIFIED_DB_CONFIG') and not __import__('re').fullmatch('[A-Z][A-Z0-9_:.-]{0,150}', code):
            code = 'UNIFIED_READ_FAILED'
        print(json.dumps({'ok':False, 'error':{'code':code}, 'readOnly':args.command != 'build', 'databaseWrites':0 if args.command != 'build' else None}, ensure_ascii=False), file=sys.stderr)
        return 1
    except Exception:
        print(json.dumps({'ok':False, 'error':{'code':'UNIFIED_READ_FAILED'}, 'readOnly':args.command != 'build'}), file=sys.stderr)
        return 1
    return 0


if __name__ == '__main__':
    sys.exit(main())