| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071 |
- # -*- coding: utf-8 -*-
- """survey.py — ON-1 survey.json 骨架 builder (SOP §1.1; Batch2 债, 2026-07-02 骨架级抽库).
- 诚实边界: §1.1 六项的"读全/坐实"判断是**人工域** (读资料/归属逐源坐实/台数双键核) — 本模块不替代人读,
- 只做三件机器活: ①从 intake_scan 的 disk_manifest 预填骨架 (每层/资料/压缩包起一行, 人工补窗/行/语义)
- ②六键完整性校验 (与 validate_contract --survey 同口径) ③schemas.reconcile_survey 漏读对账的便捷入口。
- 产物契约以既有实跑件为准 (outputs/liangshuiquan/sop/survey.json): 六键 + 可选 note/skipped。
- 纯函数无文件 I/O (json 落盘由调用方/farm_pipeline 做)。
- """
- REQUIRED_KEYS = ('data_layers', 'window_matrix', 'n_machines_claimed', 'n_with_data',
- 'data_attribution', 'materials_read')
- TODO = '【待人工】'
- def survey_skeleton_from_manifest(dman):
- """disk_manifest (scripts/intake_scan.py 产) → survey 骨架: 数据层/资料逐条预填, 人工域标 TODO。
- 预填仅是"认领占位", 不构成"已读": 窗/行数/语义/归属必须人工核后替换 TODO, 否则过不了
- validate_survey_keys (TODO 视为未填)。
- """
- dman = dman or {}
- layers = {}
- for ly in dman.get('layers', []):
- name = ly.get('name') or str(ly.get('path', '?')).rstrip('/').rsplit('/', 1)[-1]
- layers[name] = f"{ly.get('n_data_files', '?')}文件 {TODO}窗/行/列/采样率"
- mats = [m.get('name') or str(m.get('path', '?')).rsplit('/', 1)[-1] + f" {TODO}读后摘要"
- for m in dman.get('materials', [])]
- return {
- 'data_layers': layers or {TODO: '递归列全数据层 (§1.1①, 从最高频层 onboard)'},
- 'window_matrix': {TODO: '逐层 起止日期×机数×采样率 (§1.1④)'},
- 'n_machines_claimed': None, # 全机台表 count 坐实 (§1.1⑤), 禁读前几行假设
- 'n_with_data': TODO,
- 'data_attribution': TODO + ' 逐源坐实, 顶层folder名≠归属 (§1.1③)',
- 'materials_read': mats or [TODO + ' 读全所有非数据资料 (§1.1②)'],
- }
- def validate_survey_keys(survey):
- """六键完整性 (ON-1 退出判据的键位部分): 缺/空/仍含 TODO 占位 = issue。返回 issues list (空=过)。"""
- issues = []
- if not isinstance(survey, dict):
- return ['survey 非 dict']
- for k in REQUIRED_KEYS:
- v = survey.get(k)
- if v in (None, '', [], {}):
- issues.append(f'survey.{k} 缺/空 (§1.1 六键)')
- elif TODO in str(v):
- issues.append(f'survey.{k} 仍含 {TODO} 占位 — 骨架预填≠已读, 人工核后替换')
- for s in (survey.get('skipped') or []):
- if not isinstance(s, dict) or not s.get('reason') or not (s.get('path') or s.get('glob')):
- issues.append(f'survey.skipped 条目缺 path/reason (豁免必显式带理由): {s}')
- return issues
- def build_survey(*, data_layers, window_matrix, n_machines_claimed, n_with_data,
- data_attribution, materials_read, note=None, skipped=None):
- """组装 survey dict + 键校验。返回 (survey, issues); issues 非空时调用方不得落盘当过门件。"""
- survey = {'data_layers': data_layers, 'window_matrix': window_matrix,
- 'n_machines_claimed': n_machines_claimed, 'n_with_data': n_with_data,
- 'data_attribution': data_attribution, 'materials_read': materials_read}
- if note:
- survey['note'] = note
- if skipped:
- survey['skipped'] = skipped
- return survey, validate_survey_keys(survey)
- def reconcile(dman, survey):
- """漏读对账便捷入口 = schemas.reconcile_survey (单一权威源, 此处只转发不复制逻辑)。"""
- from .schemas import reconcile_survey
- return reconcile_survey(dman, survey)
|
