kangshifu-board/scripts/build-data-dictionaries.py

"""Generate complete Skill dictionaries from maintained business definitions and SQLite DDL."""
import hashlib
import json
from pathlib import Path
import re
import sqlite3
import sys

ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'server/voc-library'))
import schema_catalog
import retrieval
from product_insights import ANNOTATION_DDL

PURPOSES=dict(line.split('|',1) for line in '''_User|平台用户账号，员工身份不由昵称创建
_Role|平台角色与授权关系
KfProbe|部署验证记录
KfSource|原文、文件与数据集来源
KfSession|工作坊会话与可信员工/Agent归属
KfInsight|洞察与需求归纳
KfArtifact|阶段成果与报告
KfProject|课堂项目与阶段
KfOpportunity|机会卡片
KfCategory|业务品类字典
KfCategoryRelation|品类之间的结构关系
KfProduct|结构化业务产品
KfEvidence|逐字证据、出处及研究范围
KfEvidenceLink|证据与业务对象关系
KfMarketMetric|市场指标及统计口径
KfAssessment|验证模型与评估结果
KfConcept|新品方案版本
KfStageEvent|项目阶段推进事件
KfImportJob|入库任务及汇总回执
KfImportItem|逐条入库结果及依赖
KfReviewIssue|人工复核问题及状态
KfSupplier|供应商资料
KfMaterial|原料资料
KfProcess|工艺资料
KfStandard|标准与约束依据
KfSupplyQuote|供应报价及时间口径
KfFormulaItem|方案配料及计算基准
KfVocRun|采集执行与用量回执，不执行采集
KfRawXhsNote|小红书原始内容
KfRawXhsComment|小红书原始评论
KfRawDouyinVideo|抖音原始视频文字与指标
KfRawDouyinComment|抖音原始评论
KfRawTaobaoItem|淘宝接口族原始商品
KfRawTaobaoReview|淘宝接口族原始评价
KfRawJdItem|京东原始商品
KfRawJdReview|京东原始评价
KfRawBilibiliPost|哔哩哔哩原始内容
KfRaw1688Offer|1688原始货源
KfSku|业务SKU规格
KfRetailPriceSnapshot|业务零售价格快照
KfProcessStep|工艺步骤与约束
KfQuestionnaireQuestion|问卷问题与字段定义
KfQuestionnaireAnswer|真实问卷答复，保留原话
KsfServiceCredential|服务凭据的授权与到期配置
KsfOperation|幂等操作和回执
KsfUploadSession|文件上传会话
KsfReportPreview|报告预览及内容身份
KsfConfirmationEvent|员工真实确认事件
KsfRuntimeConfig|运行配置、称呼及分组等受控注册表
KsfFileObject|已验证文件及字节'''.splitlines())
EXTRA_LABELS=dict(line.split('|',1) for line in '''objectId|服务生成记录编号
createdAt|服务创建时间
updatedAt|服务更新时间
label|配置显示标签
tokenHash|服务端凭据指纹，非Agent业务数据
scopes|授权作用范围
projectIds|授权项目编号集合
enabled|是否启用
expiresAt|到期时间
ownerId|服务或文件归属编号
requestId|稳定请求追踪编号
operation|操作类型
payloadHash|请求载荷指纹
receipt|操作实际回执
fileName|上传文件名
contentType|文件媒体类型
sizeBytes|文件字节数
sha256|完整文件字节指纹
state|上传状态
fileReceipt|服务验证的文件回执
identity|报告版本与内容绑定身份
file|实际文件引用
reportType|报告类型代码
dataMode|实际取得模式
artifactSubtype|附属成果子类型
employeeId|真实确认员工编号
previewId|已生成预览编号
action|员工确认操作
confirmedAt|实际确认时间
key|受控配置项键
value|受控配置对象
name|文件或业务名称
bytesBase64|文件字节的Base64编码'''.splitlines())
SPECIAL={
 '_User':'平台用户管理；不接受研究资料通用导入', '_Role':'平台角色管理；不接受研究资料通用导入',
 'KfProbe':'运维部署探测，不接受研究资料导入', 'KfSource':'workshop-import内来源登记；文件走受控上传验证',
 'KfSession':'workshop-start', 'KfArtifact':'workshop-save；正式报告report-preview/report-submit及真人确认',
 'KfProject':'project-create/project-progress', 'KfOpportunity':'workshop-save机会卡',
 'KfConcept':'concept-save', 'KfFormulaItem':'concept-save配料', 'KfStageEvent':'project-progress等流程自动生成',
 'KfVocRun':'既有ingest采集回执登记', 'KfImportJob':'workshop-import/ingest自动生成',
 'KfImportItem':'workshop-import/ingest自动生成', 'KfReviewIssue':'审核或噪声反馈服务生成',
 'KfQuestionnaireAnswer':'ksfApplyReviewFeedback(action=record_answer)；当前wrapper无直连命令，须按已部署入口使用，不能冒称可直写',
 'KsfServiceCredential':'管理员配置；Agent不得写凭据值', 'KsfOperation':'服务幂等处理自动生成',
 'KsfUploadSession':'受控文件上传流程生成', 'KsfReportPreview':'report-preview服务生成',
 'KsfConfirmationEvent':'真实员工确认页生成，不代造确认',
 'KsfRuntimeConfig':'employee-name/分组管理等专用服务，或管理员运行配置；不通用导入',
 'KsfFileObject':'受控文件上传服务生成',
}

def write(path,data):
    path.parent.mkdir(parents=True,exist_ok=True)
    path.write_text(json.dumps(data,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')

def main():
    source=ROOT/'server/references/runtime-data.json'; data=json.loads(source.read_text(encoding='utf-8'))
    core=(ROOT/'server/src/core.mjs').read_text(encoding='utf-8')
    protected=set(re.findall(r"'([^']+)'",re.search(r'const serverFields=new Set\(\[(.*?)\]\)',core).group(1)))
    managed={'schemaVersion','externalKey','reviewState','recordState'}
    tables={}
    for table,defs in data['tables'].items():
        assert table in PURPOSES, 'UNDOCUMENTED_TABLE:'+table
        fields={}
        for name,definition in defs.items():
            meaning=definition.get('label') or EXTRA_LABELS.get(name)
            assert meaning, 'UNDOCUMENTED_FIELD:'+table+'.'+name
            item={**definition,'meaning':meaning,'requiredByDefinition':bool(definition.get('required')),
                'allowedInGenericImport':table in data['importClasses'] and name not in protected and definition['type'] in ('String','Number','Boolean','Array','Object','Date','Pointer','File'),
                'writeOwner':'service' if name in protected or table.startswith('Ksf') or table.startswith('_') else 'relation_service' if definition['type']=='Relation' else 'managed_metadata' if name in managed else 'agent_source_mapping',
                'nullMeaning':'未取得/不适用，不用0、空姓名或当前日期代替；必填项缺失须补来源',
                'unit':'g' if name in ('netWeightG','totalNetWeightG') else 'bytes' if name in ('sizeBytes','fileSize') else definition.get('unit')}
            if definition['type'] in ('Object','Array'):item['valueStructure']='开放JSON结构；实体字段已完整列出，内部键按操作合同/真实原载荷，不臆造固定嵌套字段'
            if name=='dataMode':item['allowedValues']=data['reportRules']['dataModes']
            if name=='reviewState':item['importAllowedValues']=['pending','draft']
            if name=='recordState':item['importAllowedValues']=['active','archived']
            fields[name]=item
        tables[table]={'purpose':PURPOSES[table],'readRoute':'workshop-search/workshop-read（受当前授权与noiseContext约束）' if table.startswith('Kf') and not table.startswith('Ksf') else '只提供schema元信息，不通过研究Agent读取内部行值',
            'writeRoute':SPECIAL.get(table,'workshop-import/ingest（以实际allowedClasses为准）'),
            'genericImportDeclared':table in data['importClasses'],'fields':fields}
    dictionary={'schema':'ksf.business-data-dictionary.v1','schemaVersion':data['schemaVersion'],
        'contractRevision':data['contractRevision'],'sourceSha256':hashlib.sha256(source.read_bytes()).hexdigest(),
        'tableCount':len(tables),'fieldCount':sum(len(t['fields']) for t in tables.values()),'tables':tables,
        'liveSchemaCommand':'workshop-schema','authority':'本地维护定义；生产实际表列须workshop-schema核对，未部署不能宣称已读取生产全结构',
        'importAuthority':'workshop-contract实际allowedClasses/字段；本字典不授予写权限','productionVerified':False}
    platform=json.loads((ROOT/'server/references/platform-field-dictionary.json').read_text(encoding='utf-8'))
    physical=json.loads((ROOT/'server/references/production-schema-20261106.json').read_text(encoding='utf-8'))
    dictionary['platformFields']=platform['tables']
    dictionary['platformCommonFields']=platform['commonFields']
    for table in physical['tables']:
        for field in table['fields']:
            definition=tables.get(table['name'],{}).get('fields',{}).get(field['logicalName']) or platform['tables'].get(table['name'],{}).get(field['name']) or platform['commonFields'].get(field['name'])
            assert definition and definition.get('meaning'), 'UNMAPPED_PHYSICAL_FIELD:'+table['name']+'.'+field['name']
            field['meaning']=definition['meaning']
            field['writeOwner']=definition['writeOwner']
            field['allowedInGenericImport']=definition.get('allowedInGenericImport',False)
    dictionary['physicalSchemaSnapshot']=physical
    dictionary['logicalSchemaSnapshot']=json.loads((ROOT/'server/references/production-logical-schema-20261106.json').read_text(encoding='utf-8'))
    dictionary['relationTables']={
        '_Join:'+name+':'+table:{'ownerClass':table,'field':name,'targetClass':definition['targetClass'],
            'meaning':table+'.'+name+'关系存储；关联'+definition['targetClass']+'，由专用业务服务/Parse维护，不直接SQL写入'}
        for table,values in data['tables'].items() for name,definition in values.items() if definition['type']=='Relation'}
    dictionary['physicalFieldCount']=physical['fieldCount']
    dictionary['physicalSnapshotAuthority']='已独立读取的生产元信息快照，仅含列与约束；运行时仍应workshop-schema核差异，不代表可读取内部行值'
    write(ROOT/'server/references/business-data-dictionary.json',dictionary)
    write(ROOT/'skills/ksf-workshop-runtime/references/business-data-dictionary.json',dictionary)
    c=sqlite3.connect(':memory:');c.executescript((ROOT/'server/voc-library/schema.sql').read_text(encoding='utf-8'))
    retrieval.create_index(c);c.executescript(ANNOTATION_DDL)
    c.executescript('CREATE TABLE source_origins(source_record_id TEXT NOT NULL REFERENCES source_records(id),archive_sha256 TEXT NOT NULL,PRIMARY KEY(source_record_id,archive_sha256)); CREATE TABLE record_versions(entity TEXT NOT NULL,record_id TEXT NOT NULL,batch_sha256 TEXT NOT NULL,record_json TEXT NOT NULL,PRIMARY KEY(entity,record_id,batch_sha256)); ANALYZE;')
    research=schema_catalog.catalog(c);assert research['complete'], 'INCOMPLETE_RESEARCH_SEMANTICS'
    research.update(authority='维护DDL生成的完整结构样本；安装状态与实际计数请query_voc.py schema读取',installed=False)
    for t in research['tables']:t['installed']=False;t['recordCount']=None
    write(ROOT/'skills/ksf-voc-library/references/research-data-dictionary.json',research)
    print(json.dumps({'businessTables':len(tables),'businessFields':dictionary['fieldCount'],
        'researchTables':research['tableCount'],'researchFields':sum(len(t['fields']) for t in research['tables'])}))

if __name__=='__main__':main()