|
@@ -0,0 +1,203 @@
|
|
|
|
|
+"""Validate Agent source mappings, build a new private batch, never modify the active library.
|
|
|
|
|
+
|
|
|
|
|
+Only product/review intake is supported here. Existing archive ingestion remains
|
|
|
|
|
+the route for social and policy material. 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 hashlib.file_digest(f, 'sha256').hexdigest()
|
|
|
|
|
+
|
|
|
|
|
+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):
|
|
|
|
|
+ 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', required=True); p.add_argument('--output', required=True)
|
|
|
|
|
+ p.add_argument('--apply', action='store_true')
|
|
|
|
|
+ args = p.parse_args()
|
|
|
|
|
+ try: print(json.dumps(build_batch(args.batch, args.output, args.apply), 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())
|