"""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())