kangshifu-board/server/voc-library/agent_batch.py

"""Validate Agent source mappings, build a new private batch, never modify the active library.

V2 supports every domain entity and generates provenance/index/quality tables.
V1 product/review batches remain compatible. No network, credentials or paid calls.
"""
import argparse
import collections
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import sqlite3
import sys
import tempfile
import zipfile
import voc_db
from product_insights import ANNOTATION_DDL, THEMES, text_hash

SCHEMA = 'ksf.voc-agent-batch.v1'
PRODUCT_FIELDS = {'itemId': 'item_id', 'title': 'title', 'url': 'url', 'shopId': 'shop_id', 'shopName': 'shop_name',
                  'price': 'price', 'priceBasis': 'price_type', 'sourceTime': 'source_time'}
REVIEW_FIELDS = {'itemId': 'item_id', 'reviewId': 'review_id', 'content': 'content', 'appendContent': 'append_content',
                 'rating': 'rating', 'ratingScale': 'rating_scale', 'reviewDate': 'review_date', 'purchasedSku': 'sku',
                 'isDefault': 'is_default_text', 'sourceTime': 'source_time'}
SOURCE_FIELDS = {'id', 'path', 'sha256', 'platform', 'marketplace', 'requestItemId', 'observedAt', 'collectedAt', 'url'}

def require(value, code):
    voc_db.require(value, code)

def digest(p):
    with Path(p).open('rb') as f: return voc_db.file_sha256(f)

def pointer(value, path):
    require(isinstance(path, str) and (not path or path.startswith('/')), 'INVALID_JSON_POINTER')
    for token in path.split('/')[1:]:
        token = token.replace('~1', '/').replace('~0', '~')
        if isinstance(value, list):
            require(token.isdigit() and int(token) < len(value), 'SOURCE_POINTER_NOT_FOUND')
            value = value[int(token)]
        else:
            require(isinstance(value, dict) and token in value, 'SOURCE_POINTER_NOT_FOUND')
            value = value[token]
    return value

def identifier(value):
    require(isinstance(value, str) and value.strip() == value and 0 < len(value) <= 200 and not any(ord(c) < 32 for c in value), 'ID_MUST_BE_NONEMPTY_STRING')
    return value

def load(batch_path):
    path = Path(batch_path).absolute()
    require(path.is_file() and not path.is_symlink(), 'BATCH_NOT_FOUND')
    require(path.stat().st_size <= 20000000, 'BATCH_MANIFEST_TOO_LARGE')
    batch = json.loads(path.read_text(encoding='utf-8-sig'))
    require(isinstance(batch, dict) and not set(batch) - {'schema', 'batchId', 'producer', 'sources', 'products', 'reviews', 'themes'}, 'INVALID_BATCH_FIELDS')
    require(batch.get('schema') == SCHEMA, 'UNSUPPORTED_BATCH_SCHEMA')
    identifier(batch.get('batchId'))
    require(isinstance(batch.get('producer'), str) and 0 < len(batch['producer']) <= 200, 'PRODUCER_REQUIRED')
    require(isinstance(batch.get('sources'), list) and 0 < len(batch['sources']) <= 1000, 'SOURCES_REQUIRED')
    sources, assets = {}, {}
    for source in batch['sources']:
        require(isinstance(source, dict) and not set(source) - SOURCE_FIELDS, 'INVALID_SOURCE_FIELDS')
        sid = identifier(source.get('id'))
        require(sid not in sources, 'DUPLICATE_SOURCE_ID')
        rel = source.get('path')
        require(isinstance(rel, str) and '\\' not in rel and ':' not in rel and not PurePosixPath(rel).is_absolute()
                and '..' not in PurePosixPath(rel).parts and rel.startswith('raw/'), 'INVALID_SOURCE_PATH')
        p = path.parent / rel
        require(not any(x.is_symlink() for x in (p, *p.parents)) and p.resolve().is_relative_to(path.parent.resolve()) and p.is_file(), 'SOURCE_MISSING_OR_REDIRECTED')
        require(p.stat().st_size <= 50000000, 'SOURCE_TOO_LARGE')
        raw = p.read_bytes()
        require(hashlib.sha256(raw).hexdigest() == source.get('sha256'), 'SOURCE_HASH_MISMATCH')
        require(source.get('platform') in ('jd', 'taobao') and source.get('marketplace') in
                (('jd',) if source['platform'] == 'jd' else ('tmall', 'taobao_unverified')), 'SOURCE_PLATFORM_MISMATCH')
        for field in ('observedAt', 'collectedAt', 'url'):
            require(source.get(field) is None or isinstance(source[field], str), 'INVALID_SOURCE_METADATA')
        if source.get('requestItemId') is not None: identifier(source['requestItemId'])
        sources[sid] = (source, json.loads(raw.decode('utf-8-sig')))
        require(rel not in assets or assets[rel][1] == source['sha256'], 'SOURCE_PATH_HASH_CONFLICT')
        assets[rel] = (p, source['sha256'])
    result = {'products': [], 'reviews': [], 'social': [], 'documents': [], 'clauses': []}
    identifiers = collections.defaultdict(set)
    for entity, mapping, mandatory in [('products', PRODUCT_FIELDS, {'itemId', 'title'}), ('reviews', REVIEW_FIELDS, {'itemId', 'reviewId', 'content'})]:
        rows = batch.get(entity, [])
        require(isinstance(rows, list) and len(rows) <= 100000, 'INVALID_BATCH_ROWS')
        for record in rows:
            require(isinstance(record, dict) and set(record) == {'sourceId', 'pointer', 'fields'}, 'INVALID_RECORD_MAPPING')
            require(record['sourceId'] in sources and isinstance(record['fields'], dict) and mandatory <= set(record['fields'])
                    and not set(record['fields']) - set(mapping), 'INVALID_FIELD_MAPPING')
            source, payload = sources[record['sourceId']]
            original = pointer(payload, record['pointer'])
            values = {name: pointer(original, loc) for name, loc in record['fields'].items()}
            iid = identifier(values['itemId'])
            require(all(value is None or isinstance(value, (str, int, float, bool)) or name == 'appendContent' for name, value in values.items()), 'INVALID_MAPPED_FIELD_TYPE')
            require('isDefault' not in values or values['isDefault'] is None or type(values['isDefault']) is bool, 'DEFAULT_STATE_MUST_BE_BOOLEAN_OR_NULL')
            require('appendContent' not in values or values['appendContent'] is None or isinstance(values['appendContent'], (str, dict, list)), 'INVALID_APPEND_CONTENT')
            for field in ('title', 'content', 'purchasedSku', 'ratingScale', 'reviewDate', 'sourceTime', 'priceBasis', 'url', 'shopName'):
                require(field not in values or values[field] is None or isinstance(values[field], str), 'TEXT_FIELD_REQUIRED')
            if entity == 'products':
                require(isinstance(values['title'], str) and bool(values['title'].strip()), 'PRODUCT_TITLE_REQUIRED')
                key = source['platform'] + ':' + iid
                require(source.get('requestItemId') in (None, iid), 'PRODUCT_REQUEST_ID_MISMATCH')
            else:
                key = source['platform'] + ':' + iid + ':' + identifier(values['reviewId'])
                require(isinstance(values['content'], str), 'REVIEW_CONTENT_REQUIRED')
            require(key not in identifiers[entity], 'DUPLICATE_RECORD_ID')
            identifiers[entity].add(key)
            row = {mapping[k]: v for k, v in values.items()}
            row.update(platform=source['platform'], marketplace=source['marketplace'], source_file=source['path'],
                       source_time=values.get('sourceTime', source.get('observedAt')), collected_at=source.get('collectedAt'),
                       source_locator=record['pointer'], source_field_map=record['fields'], producer=batch['producer'])
            if entity == 'reviews':
                row.update(observed_item_id=source.get('requestItemId'),
                           item_identity_match=(iid == source['requestItemId']) if source.get('requestItemId') else None,
                           review_id_method='source_field')
            result[entity].append(row)
    require(result['products'], 'PRODUCT_CONTEXT_REQUIRED')
    require(all(r['platform'] + ':' + r['item_id'] in identifiers['products'] for r in result['reviews']), 'REVIEW_PRODUCT_CONTEXT_MISSING')
    themes = batch.get('themes', [])
    require(isinstance(themes, list) and len(themes) <= 100000, 'INVALID_THEMES')
    from product_insights import append_text
    by_review = {r['platform'] + ':' + r['item_id'] + ':' + r['review_id']: r for r in result['reviews']}
    seen = set()
    for tag in themes:
        require(isinstance(tag, dict) and set(tag) == {'reviewId', 'theme', 'quote', 'quoteField'}, 'INVALID_THEME_FIELDS')
        require(tag['reviewId'] in by_review and tag['theme'] in THEMES and tag['quoteField'] in ('content', 'append')
                and isinstance(tag['quote'], str) and 0 < len(tag['quote']) <= 500, 'INVALID_THEME_EVIDENCE')
        key = tag['reviewId'], tag['theme']
        require(key not in seen, 'DUPLICATE_REVIEW_THEME')
        seen.add(key)
        review = by_review[tag['reviewId']]
        original = review['content'] if tag['quoteField'] == 'content' else append_text(voc_db.canonical(review.get('append_content')))
        require(tag['quote'] in original, 'THEME_QUOTE_NOT_VERBATIM')
    return path, batch, result, assets

def build_batch(batch_path, output, apply=False):
    probe = json.loads(Path(batch_path).read_text(encoding='utf-8-sig'))
    from intake_contract import VERSION
    if isinstance(probe, dict) and probe.get('schema') == VERSION:
        from structured_batch import build
        return build(batch_path, output, apply)
    path, batch, rows, assets = load(batch_path)
    target = Path(output).absolute()
    require(target.suffix == '.sqlite' and target != path and not target.exists() and not target.is_symlink(), 'NEW_DATABASE_TARGET_REQUIRED')
    require(not any(p.is_symlink() for p in target.parents), 'REDIRECTED_TARGET')
    archive = target.with_suffix('.source.zip')
    require(not archive.exists() and not archive.is_symlink(), 'SOURCE_ARCHIVE_ALREADY_EXISTS')
    receipt = {'ok': True, 'readOnly': not apply, 'applied': False, 'schema': SCHEMA, 'batchId': batch['batchId'],
               'counts': {k: len(v) for k, v in rows.items()}, 'sourceFilesVerified': len(assets),
               'themeAnnotations': len(batch.get('themes', [])), 'networkRequests': 0, 'productionImported': False,
               'databaseWrites': 0, 'activeLibraryModified': False}
    if not apply: return receipt
    target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    created = []
    with tempfile.TemporaryDirectory(prefix='.agent-batch-', dir=target.parent) as temp:
        temp = Path(temp)
        packed, staged = temp / 'source.zip', temp / 'batch.sqlite'
        with zipfile.ZipFile(packed, 'x', zipfile.ZIP_DEFLATED) as z:
            def write(name, body):
                info = zipfile.ZipInfo('agent-batch/' + name, (2026, 1, 1, 0, 0, 0))
                info.external_attr = 0o100600 << 16
                z.writestr(info, body, compress_type=zipfile.ZIP_DEFLATED)
            for entity, filename in voc_db.PRIMARY.items():
                write('01_最新底库/' + filename, '\n'.join(voc_db.canonical(r) for r in rows[entity]))
            manifest = path.read_bytes()
            require(json.loads(manifest.decode('utf-8-sig')) == batch, 'BATCH_CHANGED_DURING_BUILD')
            write('batch.json', manifest)
            for rel, (p, expected) in sorted(assets.items()):
                raw = p.read_bytes()
                require(hashlib.sha256(raw).hexdigest() == expected, 'SOURCE_CHANGED_DURING_BUILD')
                write('02_采集档案/' + rel, raw)
        built = voc_db.build(packed, staged)
        if batch.get('themes'):
            c = sqlite3.connect(staged); c.row_factory = sqlite3.Row
            try:
                c.executescript(ANNOTATION_DDL)
                for tag in batch['themes']:
                    r = c.execute('SELECT * FROM reviews WHERE id=?', (tag['reviewId'],)).fetchone()
                    voc_db.insert(c, 'review_theme_annotations', {'id': voc_db.key(tag, batch['producer']), 'review_id': r['id'],
                        'product_id': r['product_id'], 'theme_code': tag['theme'], 'quote': tag['quote'], 'quote_field': tag['quoteField'],
                        'text_sha256': text_hash(r), 'method': 'agent', 'producer': batch['producer']})
                generation = voc_db.key(built['generation'], digest(Path(__file__)), digest(Path(__file__).with_name('product_insights.py')), batch['themes'])
                c.execute("UPDATE meta SET value=? WHERE key='generation'", (generation,)); c.commit()
            finally: c.close()
        from update_library import verified
        meta = verified(staged)
        try:
            for src, dest in ((packed, archive), (staged, target)):
                os.chmod(src, 0o600); os.link(src, dest); created.append((dest, digest(src)))
        except Exception:
            for p, h in reversed(created):
                if p.is_file() and digest(p) == h: p.unlink()
            raise
    return {**receipt, 'readOnly': False, 'applied': True, 'databaseWrites': 'staging-only', 'generation': meta['generation'],
            'database': str(target), 'databaseSha256': digest(target), 'sourceArchive': str(archive), 'archiveSha256': digest(archive)}

def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--batch'); p.add_argument('--output')
    p.add_argument('--contract', action='store_true', help='Describe all supported entity mappings without a database')
    p.add_argument('--apply', action='store_true')
    args = p.parse_args()
    try:
        require(not args.contract or not (args.batch or args.output or args.apply), 'CONTRACT_REJECTS_WRITE_OPTIONS')
        if args.contract:
            from intake_contract import contract
            result = contract()
        else:
            require(args.batch and args.output, 'BATCH_AND_OUTPUT_REQUIRED')
            result = build_batch(args.batch, args.output, args.apply)
        print(json.dumps(result, ensure_ascii=False, indent=2))
    except Exception as e:
        code = str(e) if isinstance(e, voc_db.LibraryError) else 'AGENT_BATCH_FAILED'
        print(json.dumps({'ok': False, 'error': {'code': code}, 'productionImported': False}, ensure_ascii=False), file=sys.stderr)
        return 1
    return 0

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