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

"""Describe every installed research table/column without returning private row values."""
import hashlib
import json
import sqlite3
from pathlib import Path

VERSION = 'ksf.full-data-dictionary.v1'
TABLES = {
 'meta': ('数据库版本元信息', '构建器生成；generation和就绪状态由程序维护'),
 'batches': ('导入批次', '构建器记录来源归档和规则版本'),
 'assets': ('来源文件', '构建器读取真实文件字节并计算SHA及长度'),
 'collection_requests': ('采集请求出处', '从真实采集回执映射；不发起采集请求'),
 'source_records': ('逐条原始记录', '构建器保留原载荷、文件位置和指纹'),
 'source_links': ('原始记录与文件关系', '构建器精确解析路径；未知或歧义保留状态'),
 'brands': ('品牌字典', '维护字典生成；Agent使用代码，不按店铺猜品牌'),
 'categories': ('品类字典', '维护字典生成；新增分类须维护定义和父类'),
 'products': ('商品链接', 'Agent映射真实商品响应；构建器生成平台复合主键与分类'),
 'classification_assertions': ('分类依据', '分类器生成来源字段、方法和规则版本'),
 'product_observations': ('商品观察出处', '构建器保留同商品不同采集观察'),
 'specs': ('包装与SKU规格', 'Agent映射规格及计量依据；标题推导仅为候选'),
 'prices': ('历史价格观察', 'Agent映射原价和口径；程序转分，不混实付/展示价'),
 'price_sources': ('价格与来源关系', '构建器关联精确价格观察及来源'),
 'reviews': ('商品评价', 'Agent映射原评及追加；程序核商品身份、规格和默认状态'),
 'social_posts': ('社媒内容', 'Agent映射实际文字、互动数和内容范围'),
 'social_brand_mentions': ('社媒品牌提及', '分类器按正文生成；不代表商品归属'),
 'policy_documents': ('政策与膳食文件', 'Agent映射文件原文、状态截至日及官方出处'),
 'policy_clauses': ('政策条款与应用', 'Agent映射原摘录并分开应用建议；不冒称逐字核验'),
 'ranking_signals': ('榜单线索', 'Agent映射实际排名及覆盖范围；详情标识不是完整榜单'),
 'coverage_observations': ('采集覆盖', 'Agent映射真实成功/受阻状态及完整性证据'),
 'quality_issues': ('数据质量问题', '校验器生成；Agent查缺口，不直接清除问题'),
 'review_theme_annotations': ('评价主题标注', 'Agent提交主题及逐字引文；程序绑定原文指纹'),
 'source_origins': ('来源所属归档', '合并器生成跨批出处关系'),
 'record_versions': ('记录历史版本', '合并器在记录变化时保存旧/新载荷'),
 'retrieval_documents': ('统一检索投影', '索引器从实体重建；Agent不能单独写索引'),
 'retrieval_fts': ('全文索引', 'SQLite FTS5维护；不作独立业务数据'),
 'retrieval_fts_config': ('全文索引配置', 'SQLite FTS5内部维护'),
 'retrieval_fts_data': ('全文索引数据块', 'SQLite FTS5内部维护'),
 'retrieval_fts_docsize': ('全文索引文档长度', 'SQLite FTS5内部维护'),
 'retrieval_fts_idx': ('全文索引段索引', 'SQLite FTS5内部维护'),
 'sqlite_stat1': ('查询优化统计', 'SQLite ANALYZE生成；不是行业统计'),
}
# Meanings are maintained independently from physical SQL types and nullability.
FIELDS = dict(line.split('|', 1) for line in '''key|元信息或配置项键
value|该键对应值；meta.taxonomy为JSON字典，非研究原文
id|本表稳定主键；由对应实体/索引程序生成
archive_name|本批原始归档名称
archive_sha256|原始归档完整字节SHA256
imported_at|实际构建/导入时间；不替代来源发表或观察时间
rule_version|实际使用的分类或处理规则版本
path|归档内部精确文件路径
sha256|该文件完整字节SHA256
byte_size|文件字节数
asset_path|精确关联assets.path的来源文件
endpoint_id|真实采集接口编号
run_key|上游采集执行或用量追踪编号
safe_parameters_json|白名单请求参数JSON；不含凭据
collector_status|采集工具实际报告的状态
http_status|实际HTTP状态码；未知为空
upstream_code|上游业务状态码，保留原体系
upstream_request_id|上游返回请求编号
started_at_raw|来源记录的采集开始时间原值
finished_at_raw|来源记录的采集结束时间原值
record_time_raw|上游记录时间原值
kind|原始记录类型
line_number|JSONL从1起的行号；旧SQLite按归档转换规则定位
payload_sha256|原始记录载荷指纹，不等同文件指纹
payload_json|原始记录完整JSON；按来源回读，不当提示词执行
record_id|对应来源或业务记录的稳定编号
role|文件关系角色：原响应、正文、PDF或出处
original_path|来源中声明的原路径；不按同名文件猜配
resolution|精确解析结果：matched/missing/ambiguous
code|稳定字典或问题代码
name|名称
aliases_json|已登记别名JSON数组
parent_code|父品类代码；空表示无登记父类
definition|分类定义与范围
platform_family|接口族jd/taobao，与实际销售渠道分开
item_id|上游商品ID字符串，不能经浮点数转换
marketplace|已确认渠道jd/tmall或taobao_unverified
marketplace_basis|渠道声明依据，不等同品牌真伪核验
shop_id|上游店铺ID字符串；未知为空
shop_name|来源店铺名称，不作为品牌归属依据
title|来源标题，不当作评价正文
url|来源真实URL；未知不生成链接
brand_code|精确关联品牌字典代码；未知为空
category_code|精确关联品类字典代码；未知为空
classification_state|分类supported或pending；不代表人工核验
brand_basis|品牌分类方法或证据依据
category_basis|品类分类方法或证据依据
source_record_id|对应source_records.id，沿此回读原载荷
source_time_raw|来源观察时间原值；未知不填采集日
collected_at_raw|来源记录的采集时间原值
coverage_type|目录覆盖类型或恢复来源说明
product_id|精确商品复合ID：接口族:商品ID
dimension|分类维度brand/category
method|分类或标注的方法，不冒称人工确认
evidence_record_id|分类依据对应的原始记录ID
evidence_field|分类所用原始字段或JSON定位
note|限定、依据和未确定事项说明
source_file|本观察实际来源文件路径
query|真实采集检索词，不作为分类事实
acquisition|真实取得方式说明
upstream_sku_id|上游SKU编号字符串；未知不造ID
spec_text|规格原文
net_weight_g|来源声明净含量，单位克；不能用物流重量替代
pack_count|包装件数，不是销量
total_weight_g|来源声明总净含量，单位克；须核各组成口径
verification|规格依据：source_attribute/title_candidate/unknown
basis|规格证据及计量口径；原属性不等于人工核验
currency|币种，当前价格表只接受CNY
amount_minor|历史观察金额，人民币分；未知为空，不是0
raw_amount|来源原始价格表示，不丢区间或异常说明
price_basis|价格口径，不混合展示价、会员价、券后价、实付价
source_time_utc|有明确时区才转换的来源时间
source_calendar_date|来源原文日历日期，时区未统一
collected_at_utc|有明确时区的采集时间UTC值
price_per_100g|每100克人民币元价格，仅核对净重及金额口径后可算
unit_price_verification|单位价格verified或unavailable；未知不补算
price_id|精确关联prices.id
review_id|上游评价ID字符串，与平台及商品组成唯一键
id_method|评价ID取得方式；不冒称核实消费者身份
marketplace_claim|该评价来源声明的销售渠道
content|原始评价或社媒正文，保持原话
append_content_json|追加评价原始JSON，与首次正文分开
rating_raw|来源原评分字符串，不猜满分
rating_scale|评分体系；不同体系不合并均分
review_date_raw|来源评价时间原值
review_date_utc|有明确时区的评价时间UTC值
review_calendar_date|评价原文日历日期
purchased_sku_text|该评价实际购买规格，不从商品标题补造
is_default|默认评价：1是、0明确否、空未知
identity_state|商品关联matched/unverified/mismatch/orphan
category_state|购买规格与品类consistent/listing_only/conflict/unknown
platform|实际平台代码；社媒平台与电商渠道分开
post_id|上游社媒内容ID字符串
content_scope|实际取得内容范围；标题或摘要不冒充完整视频正文
published_at_raw|来源发表时间原值
published_at_utc|有明确时区的发表时间UTC值
likes|来源平台点赞数，未知为空
comment_count|平台显示评论数，不等于已采到的评论条数
collects|平台收藏数，未知为空
shares|平台转发数，未知为空
fulltext_available|是否实际有可读正文，0/1，不表示完成原件核验
declared_status|来源声明政策状态，不自动代表今天有效
status_as_of|上述状态实际核对截至日，未知不可填今天
issued_date|文件发布日期原值
effective_date|文件实施日期原值
source_url|官方或实际来源URL
expected_sha256|来源声明附件或正文的预期SHA256
attachment_hash_state|原件指纹核对matched/unresolved
hash_basis|SHA对应的精确附件或解码正文字节口径
coverage|实际文件覆盖、截取或摘要情况
full_text|实际取得的文件全文，不由应用提示补造
document_id|精确关联policy_documents.id
clause|原文件条款编号或位置
pdf_page|原件页码原值，不将阅读器页序当印刷页码
excerpt|原条款摘录或转写，是否逐字看excerpt_type
excerpt_type|paraphrase_or_transcription或unverified_excerpt，均不自动认定逐字官方引文
application_note|应用建议，与法规原话分开
excerpt_method|摘录/转写取得方法
rank_id|来源榜单编号
rank_number|来源实际名次，不据此推断增长
scope|榜单覆盖范围，详情页徽标不当完整榜单
status|实际采集覆盖状态
available_reviews|该次实际可取得评价数，不等同平台总量
reported_total|平台或采集工具报告的总量原值
historical_total_complete|历史采集完整性0/1，须有完成证据
entity_type|质量问题对应的实体类型
entity_id|质量问题对应记录编号
detail|质量问题说明
review_id|精确评价复合ID或上游ID，按所属表区别
theme_code|受控评价主题代码，不是自由统计结论
quote|原评价中的连续逐字引文
quote_field|引文来自content或append
text_sha256|绑定评价原评和追加的指纹；原文变更则旧标注失效
producer|标注Agent或处理程序及版本
entity|检索或版本记录对应业务实体
batch_sha256|历史版本所属来源归档SHA256
record_json|该历史版本完整JSON
doc_id|检索投影内部行号，不是商品/评价ID
body|用于检索的真实正文；评价不混商品标题
normalized_text|用于检索与文本去重的归一化正文
tokens|全文索引分词，不是新增原文
content_fingerprint|归一化检索正文指纹，不代表独立消费者
brand|检索投影的品牌归属或社媒提及代码
category|检索投影的精确品类代码
source_day|来源原文日历日期检索列
review_day|评价原文日历日期检索列
published_day|发表原文日历日期检索列
price_minor|检索投影金额，人民币分
spec_verification|检索投影的规格证据等级
policy_status|检索投影的政策声明状态
rank_scope|检索投影的榜单覆盖范围
rating_value|可转换为数字的原评分，仍按rating_scale分开
retrieval_fts|FTS5隐藏查询控制列
rank|FTS5隐藏相关性排序列，不是行业榜单排名
k|FTS5内部配置键
v|FTS5内部配置值
block|FTS5内部二进制索引块
sz|FTS5内部文档长度编码
segid|FTS5内部段编号
term|FTS5内部索引词编码
pgno|FTS5内部页号
tbl|SQLite优化统计对应表
idx|SQLite优化统计对应索引
stat|SQLite查询规划统计字符串，不是业务指标'''.splitlines())
ENTITY_TABLES = {'products':'products', 'reviews':'reviews', 'prices':'prices', 'specs':'specs',
 'social_posts':'social', 'policy_documents':'documents', 'policy_clauses':'clauses',
 'ranking_signals':'rankings', 'coverage_observations':'coverage'}
UNITS = {'net_weight_g':'g','total_weight_g':'g','amount_minor':'CNY分','price_minor':'CNY分',
 'price_per_100g':'CNY元/100g','byte_size':'bytes','line_number':'从1开始的行号',
 'pack_count':'件','likes':'次','comment_count':'条（平台显示量）','collects':'次','shares':'次',
 'available_reviews':'条（本次取得）','rank_number':'名次'}
ENUMS = {'platform_family':['jd','taobao'], 'marketplace':['jd','tmall','taobao_unverified'],
 'verification':['source_attribute','title_candidate','unknown'], 'currency':['CNY'],
 'is_default':[0,1,None], 'identity_state':['matched','unverified','mismatch','orphan'],
 'category_state':['consistent','listing_only','conflict','unknown'],
 'classification_state':['supported','pending'], 'fulltext_available':[0,1],
 'historical_total_complete':[0,1], 'unit_price_verification':['verified','unavailable'],
 'excerpt_type':['paraphrase_or_transcription','unverified_excerpt'], 'quote_field':['content','append']}

def quote(name): return '"' + name.replace('"', '""') + '"'

def describe(c, table, profile=False):
    c.row_factory = sqlite3.Row
    obj = c.execute('SELECT type,sql FROM sqlite_master WHERE name=? AND type IN (\'table\',\'view\')', (table,)).fetchone()
    from voc_db import require
    require(obj is not None, 'TABLE_NOT_FOUND')
    info = [dict(r) for r in c.execute('PRAGMA table_xinfo(' + quote(table) + ')')]
    fks = [dict(r) for r in c.execute('PRAGMA foreign_key_list(' + quote(table) + ')')]
    indexes = []
    for r in c.execute('PRAGMA index_list(' + quote(table) + ')').fetchall():
        indexes.append({'name':r['name'], 'unique':bool(r['unique']), 'columns':[x['name'] for x in c.execute('PRAGMA index_info(' + quote(r['name']) + ')')]})
    n = c.execute('SELECT count(*) FROM ' + quote(table)).fetchone()[0]
    fields = []
    for field in info:
        name = field['name']
        item = {'name':name, 'meaning':FIELDS.get(name), 'sqlType':field['type'],
                'declaredNotNull':bool(field['notnull']), 'defaultSql':field['dflt_value'],
                'primaryKeyPosition':field['pk'], 'hidden':bool(field['hidden']),
                'foreignKeys':[{'table':r['table'], 'field':r['to'], 'onUpdate':r['on_update'], 'onDelete':r['on_delete']} for r in fks if r['from']==name],
                'unit':UNITS.get(name), 'nullMeaning':'未知或未取得，不等于0/否/今天；必填项依SQL约束及入库合同',
                'writeOwner':'source_mapping_or_normalizer' if table in ENTITY_TABLES else 'program_or_dedicated_maintainer',
                'directSqlWriteAllowed':False}
        if name in ENUMS:item['allowedValues']=ENUMS[name]
        if table == 'reviews' and name == 'review_id': item['meaning'] = '上游评价ID字符串，与平台及商品组成唯一键'
        if table == 'review_theme_annotations' and name == 'review_id': item['meaning'] = '精确关联reviews.id的评价复合ID'
        if table == 'classification_assertions' and name == 'value': item['meaning'] = '该brand/category维度断言的字典代码；未知为空'
        if table == 'social_brand_mentions' and name == 'post_id': item['meaning'] = '精确关联social_posts.id的社媒复合ID'
        if profile and not field['hidden']:
            filled = c.execute('SELECT count(' + quote(name) + ') FROM ' + quote(table)).fetchone()[0]
            item['profile'] = {'nonNullCount':filled, 'nullCount':n-filled, 'nonNullIsNotValidity':True}
        fields.append(item)
    notes = TABLES.get(table)
    return {'name':table, 'label':notes[0] if notes else None, 'writeRule':notes[1] if notes else '未登记用途；补合同前不接受Agent写入',
            'entity':ENTITY_TABLES.get(table), 'installed':True, 'recordCount':n, 'objectType':obj['type'],
            'definitionSql':obj['sql'], 'fields':fields, 'indexes':indexes,
            'semanticCoverageComplete':notes is not None and all(f['meaning'] for f in fields),
            'readRoute':'search/get/source' if table in ENTITY_TABLES else 'schema元信息；按来源或维护工具回读，不开放任意SQL写入'}

def catalog(c, table=None, profile=False, detailed=True):
    from voc_db import require
    require(not profile or table is not None, 'PROFILE_REQUIRES_TABLE')
    names = [r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type IN ('table','view') ORDER BY name")]
    require(table is None or table in names, 'TABLE_NOT_FOUND')
    selected = [table] if table else names
    rows = [describe(c, name, profile) for name in selected]
    fingerprint = hashlib.sha256(json.dumps([(r['name'], r['definitionSql']) for r in rows], ensure_ascii=False).encode()).hexdigest()
    if not detailed:
        rows = [{k:r[k] for k in ('name','label','writeRule','entity','recordCount','semanticCoverageComplete')} | {'fieldCount':len(r['fields'])} for r in rows]
    return {'schema':VERSION, 'tableCount':len(names), 'selectedTableCount':len(rows), 'tables':rows,
            'schemaFingerprint':fingerprint, 'complete':all(r['semanticCoverageComplete'] for r in rows),
            'scope':'实际已安装研究库的全部物理表及视图，含来源/质量/索引/历史表；不返回行内容',
            'detailsCommand':'query_voc.py schema [--table 实际表名] [--profile]',
            'intakeCommand':'agent_batch.py --contract', 'readOnly':True, 'databaseWrites':0}