|
@@ -0,0 +1,766 @@
|
|
|
|
|
+"""Versioned, offline VOC corpus. Python 3.11+, SQLite, 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 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 = hashlib.file_digest(f, 'sha256').hexdigest()
|
|
|
|
|
+ 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 = hashlib.file_digest(f, 'sha256').hexdigest()
|
|
|
|
|
+ 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')}
|
|
|
|
|
+ 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, '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 = None
|
|
|
|
|
+ text_path, _ = a.resolve(row.get('text_file'))
|
|
|
|
|
+ if text_path:
|
|
|
|
|
+ full_text = a.z.read(text_path).decode('utf-8-sig')
|
|
|
|
|
+ elif row.get('source_file'):
|
|
|
|
|
+ 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('status_on_20261104') or 'unknown'), status_as_of='2026-11-04',
|
|
|
|
|
+ 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):
|
|
|
|
|
+ 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
|
|
|
|
|
+ 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'],
|
|
|
|
|
+ '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, 'upstreamCollection': False, 'businessDatabaseProjection': False, 'publicBoardConnected': 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'],))]
|
|
|
|
|
+ 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 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','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('--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')
|
|
|
|
|
+ args = parser.parse_args()
|
|
|
|
|
+ try:
|
|
|
|
|
+ if args.command == 'build':
|
|
|
|
|
+ require(not args.registry, 'BUILD_REJECTS_ACTIVE_REGISTRY')
|
|
|
|
|
+ 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')
|
|
|
|
|
+ with readonly(registry_database(args.registry) if args.registry else args.db) as c:
|
|
|
|
|
+ if args.command == 'contract':
|
|
|
|
|
+ result = contract(c)
|
|
|
|
|
+ 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)}
|
|
|
|
|
+ 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:
|
|
|
|
|
+ print(json.dumps({'ok':False, 'error':{'code':str(error)}, 'readOnly':args.command != 'build', 'databaseWrites':0 if args.command != 'build' else None}, ensure_ascii=False), file=sys.stderr)
|
|
|
|
|
+ return 1
|
|
|
|
|
+ return 0
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+if __name__ == '__main__':
|
|
|
|
|
+ sys.exit(main())
|