diff --git a/pbl_appcodes/codes.json b/pbl_appcodes/codes.json index 50de54f..49b697e 100644 --- a/pbl_appcodes/codes.json +++ b/pbl_appcodes/codes.json @@ -1,152 +1,144 @@ { "_meta": { "module": "pbl_appcodes", + "app": "pbls", "dialect": "mariadb", "groups": 8, "items": 74, - "target_tables": ["appcodes", "appcodes_kv"], "idempotent": true, "zero_table": true, - "spec_ref": "projects/pbls/docs/01-design/appcodes.md", - "note": "6 组业务枚举 + 2 组运行时枚举 = 8 组 74 项；注入幂等（重复执行零差异）；本模块零自有表。" + "source": "projects/pbls/docs/01-design/appcodes.md", + "note": "6 组业务字典 + 2 组 Agent 工具启停字典（13 启用 / 9 禁用，对齐 agent-tool-contract.md）。编码全局唯一，每组 sort 从 1 连续递增。注入走 appcodes 平台字典，本模块零建表。" }, - "groups": [ + "dicts": [ { - "group_code": "pbl_tenant_status", - "group_name": "PBL 租户状态", - "module": "pbl_appcodes", - "items_count": 3, + "group": "pbl_blueprint_type", + "group_name": "PBL蓝图类型", + "module": "pbl_blueprint", "items": [ - {"code": "active", "name": "启用", "sort": 1, "remark": "正常可用"}, - {"code": "suspended", "name": "停用", "sort": 2, "remark": "暂停服务，数据保留"}, - {"code": "closed", "name": "注销", "sort": 3, "remark": "终止合作，只读归档"} + {"code": "BP_COURSE", "name": "课程蓝图", "sort": 1, "remark": "顶层聚合根"}, + {"code": "BP_MODULE", "name": "模块蓝图", "sort": 2, "remark": "课程下模块"}, + {"code": "BP_UNIT", "name": "单元蓝图", "sort": 3, "remark": "模块下单元"}, + {"code": "BP_LESSON", "name": "课时蓝图", "sort": 4, "remark": "单元下课时"}, + {"code": "BP_ACTIVITY", "name": "活动蓝图", "sort": 5, "remark": "课时内活动"}, + {"code": "BP_TEMPLATE", "name": "模板蓝图", "sort": 6, "remark": "离线兜底模板，可实例化"} ] }, { - "group_code": "pbl_role", - "group_name": "PBL 租户内角色", - "module": "pbl_appcodes", - "items_count": 5, + "group": "pbl_quality_state", + "group_name": "PBL质量状态(5级)", + "module": "pbl_validation", "items": [ - {"code": "admin", "name": "项目管理员", "sort": 1, "remark": "租户内全权，复用 rbac 不改其基表"}, - {"code": "teacher", "name": "教师", "sort": 2, "remark": "蓝图设计/校验/编译/发布"}, - {"code": "assistant", "name": "助教", "sort": 3, "remark": "协助运行与证据采集"}, - {"code": "student", "name": "学生", "sort": 4, "remark": "世界内运行与产出"}, - {"code": "observer", "name": "观察者", "sort": 5, "remark": "只读看板与分析"} + {"code": "QS_L1_DRAFT", "name": "L1 草稿", "sort": 1, "remark": "未校验"}, + {"code": "QS_L2_STRUCTURED", "name": "L2 结构完整", "sort": 2, "remark": "结构维度通过"}, + {"code": "QS_L3_VALIDATED", "name": "L3 校验通过", "sort": 3, "remark": "14 维全通过"}, + {"code": "QS_L4_COMPILED", "name": "L4 已编译", "sort": 4, "remark": "产出 Game Definition"}, + {"code": "QS_L5_READY", "name": "L5 可发布就绪", "sort": 5, "remark": "发布动作属 G6 禁用，仅置状态"} ] }, { - "group_code": "pbl_bp_status", - "group_name": "PBL 蓝图状态", - "module": "pbl_appcodes", - "items_count": 6, + "group": "pbl_validation_dim", + "group_name": "PBL校验维度(14维)", + "module": "pbl_validation", "items": [ - {"code": "draft", "name": "草稿", "sort": 1, "remark": "可编辑"}, - {"code": "validating", "name": "校验中", "sort": 2, "remark": "14 维校验执行中"}, - {"code": "validated", "name": "已校验", "sort": 3, "remark": "无 blocker，可编译"}, - {"code": "compiling", "name": "编译中", "sort": 4, "remark": "确定性编译进行中"}, - {"code": "published", "name": "已发布", "sort": 5, "remark": "产出 Game Definition 并发布到班级"}, - {"code": "archived", "name": "已归档", "sort": 6, "remark": "只读留存"} + {"code": "VD_01_STRUCTURE", "name": "结构完整性", "sort": 1}, + {"code": "VD_02_OBJECTIVE", "name": "学习目标可测性", "sort": 2}, + {"code": "VD_03_ALIGNMENT", "name": "目标-活动-评价一致性", "sort": 3}, + {"code": "VD_04_SUBOBJECT", "name": "7类子对象契约合规", "sort": 4}, + {"code": "VD_05_NARRATIVE", "name": "叙事世界观自洽", "sort": 5}, + {"code": "VD_06_ROLE", "name": "角色与分工完备", "sort": 6}, + {"code": "VD_07_TASK", "name": "任务链可达性", "sort": 7}, + {"code": "VD_08_RULE", "name": "规则可执行性", "sort": 8}, + {"code": "VD_09_EVIDENCE", "name": "证据采集点覆盖", "sort": 9}, + {"code": "VD_10_RUBRIC", "name": "Rubric 权重归一", "sort": 10}, + {"code": "VD_11_DIFFICULTY", "name": "难度梯度合理", "sort": 11}, + {"code": "VD_12_DURATION", "name": "时长与容量匹配", "sort": 12}, + {"code": "VD_13_ACCESSIBILITY", "name": "无障碍与降级态", "sort": 13}, + {"code": "VD_14_TENANT_ISOLATION", "name": "租户隔离与写保护", "sort": 14} ] }, { - "group_code": "pbl_quality_level", - "group_name": "PBL 蓝图质量等级（5 级）", - "module": "pbl_appcodes", - "items_count": 5, + "group": "pbl_evidence_type", + "group_name": "PBL证据类型", + "module": "pbl_evidence", "items": [ - {"code": "L0_draft", "name": "L0 草稿级", "sort": 1, "remark": "未过校验"}, - {"code": "L1_incomplete", "name": "L1 不完整", "sort": 2, "remark": "存在 blocker"}, - {"code": "L2_basic", "name": "L2 基础可用", "sort": 3, "remark": "无 blocker，有 error"}, - {"code": "L3_qualified", "name": "L3 合格", "sort": 4, "remark": "无 error，允许 warning"}, - {"code": "L4_excellent", "name": "L4 优秀", "sort": 5, "remark": "全维通过且加权得分达标"} + {"code": "EV_ARTIFACT", "name": "产出物", "sort": 1}, + {"code": "EV_DIALOGUE", "name": "对话记录", "sort": 2}, + {"code": "EV_ACTION", "name": "操作行为", "sort": 3}, + {"code": "EV_DECISION", "name": "决策记录", "sort": 4}, + {"code": "EV_COLLABORATION", "name": "协作贡献", "sort": 5}, + {"code": "EV_REFLECTION", "name": "反思总结", "sort": 6}, + {"code": "EV_ASSESSMENT", "name": "评估结果", "sort": 7}, + {"code": "EV_RUNTIME_EVENT", "name": "运行时事件", "sort": 8}, + {"code": "EV_DOCUMENT", "name": "文档附件", "sort": 9}, + {"code": "EV_MEDIA", "name": "音视频媒体", "sort": 10} ] }, { - "group_code": "pbl_subobj_type", - "group_name": "PBL 蓝图 7 类子对象类型", - "module": "pbl_appcodes", - "items_count": 7, + "group": "pbl_assessment_level", + "group_name": "PBL评估等级", + "module": "pbl_assessment", "items": [ - {"code": "objective", "name": "学习目标", "sort": 1, "remark": "泛化载荷 payload.objective"}, - {"code": "stage", "name": "阶段", "sort": 2, "remark": "泛化载荷 payload.stage"}, - {"code": "task", "name": "任务", "sort": 3, "remark": "泛化载荷 payload.task"}, - {"code": "artifact_spec", "name": "产出物规格", "sort": 4, "remark": "泛化载荷 payload.artifact_spec"}, - {"code": "rubric_ref", "name": "量规引用", "sort": 5, "remark": "泛化载荷 payload.rubric_ref"}, - {"code": "resource_ref", "name": "资源引用", "sort": 6, "remark": "泛化载荷 payload.resource_ref"}, - {"code": "assessment_plan", "name": "评估方案", "sort": 7, "remark": "泛化载荷 payload.assessment_plan"} + {"code": "AS_EXCELLENT", "name": "优秀", "sort": 1, "remark": ">=90"}, + {"code": "AS_GOOD", "name": "良好", "sort": 2, "remark": ">=80"}, + {"code": "AS_PASS", "name": "合格", "sort": 3, "remark": ">=60"}, + {"code": "AS_WEAK", "name": "待改进", "sort": 4, "remark": ">=40"}, + {"code": "AS_FAIL", "name": "不合格", "sort": 5, "remark": "<40"} ] }, { - "group_code": "pbl_valid_dim", - "group_name": "PBL 校验 14 维", - "module": "pbl_appcodes", - "items_count": 14, + "group": "pbl_runtime_event_type", + "group_name": "PBL运行时事件类型", + "module": "pbl_runtime_ext", "items": [ - {"code": "D01_driving_question", "name": "驱动性问题清晰度", "sort": 1, "remark": "blocker 级"}, - {"code": "D02_objective_alignment", "name": "学习目标对齐", "sort": 2, "remark": "目标-任务-评估三向对齐"}, - {"code": "D03_stage_completeness", "name": "阶段完整性", "sort": 3, "remark": "阶段链无断裂"}, - {"code": "D04_task_feasibility", "name": "任务可行性", "sort": 4, "remark": "课时与团队规模匹配"}, - {"code": "D05_artifact_spec", "name": "产出物规格明确性", "sort": 5, "remark": "每任务至少一个产出物规格"}, - {"code": "D06_rubric_coverage", "name": "量规覆盖度", "sort": 6, "remark": "目标被量规条目全覆盖"}, - {"code": "D07_resource_availability", "name": "资源可得性", "sort": 7, "remark": "资产引用可解析"}, - {"code": "D08_assessment_plan", "name": "评估方案闭环", "sort": 8, "remark": "证据->量规->报告链路完整"}, - {"code": "D09_duration_reasonableness", "name": "课时合理性", "sort": 9, "remark": "总课时与阶段和一致"}, - {"code": "D10_team_collaboration", "name": "团队协作设计", "sort": 10, "remark": "角色分工与团队规模"}, - {"code": "D11_world_scene_binding", "name": "世界/场景绑定有效性", "sort": 11, "remark": "只读引用存在且可访问"}, - {"code": "D12_script_executability", "name": "脚本可执行性", "sort": 12, "remark": "script_engine 校验通过"}, - {"code": "D13_evidence_traceability", "name": "证据可追溯性", "sort": 13, "remark": "幂等键规则可生成"}, - {"code": "D14_compliance_safety", "name": "合规与安全", "sort": 14, "remark": "内容合规与写保护合规"} + {"code": "RE_ENTITY_STATE", "name": "实体状态变更", "sort": 1}, + {"code": "RE_PLAYER_ACTION", "name": "玩家操作", "sort": 2}, + {"code": "RE_SCRIPT_TRIGGER", "name": "脚本触发", "sort": 3}, + {"code": "RE_COLLISION", "name": "碰撞检测", "sort": 4}, + {"code": "RE_TIMER", "name": "定时器到期", "sort": 5}, + {"code": "RE_EVENT_DISPATCH", "name": "事件分发", "sort": 6}, + {"code": "RE_BROADCAST", "name": "广播下发", "sort": 7}, + {"code": "RE_SNAPSHOT", "name": "状态快照", "sort": 8}, + {"code": "RE_PAUSE", "name": "会话暂停", "sort": 9}, + {"code": "RE_RESUME", "name": "会话恢复", "sort": 10}, + {"code": "RE_ERROR_FALLBACK", "name": "错误降级", "sort": 11}, + {"code": "RE_SESSION_END", "name": "会话结束", "sort": 12} ] }, { - "group_code": "pbl_agent_tool", - "group_name": "PBL Agent 工具白名单（13 启用 / 9 禁用，fail-closed）", - "module": "pbl_appcodes", - "items_count": 22, + "group": "pbl_agent_tool_enabled", + "group_name": "Agent工具-启用(13)", + "module": "pbl_agent_runtime", "items": [ - {"code": "blueprint.read", "name": "读取蓝图", "sort": 1, "remark": "enabled"}, - {"code": "blueprint.tree", "name": "读取蓝图子对象树", "sort": 2, "remark": "enabled"}, - {"code": "subobject.read", "name": "读取子对象", "sort": 3, "remark": "enabled"}, - {"code": "subobject.upsert", "name": "写入子对象", "sort": 4, "remark": "enabled，需编辑锁 + tenant_id"}, - {"code": "validation.run", "name": "触发校验", "sort": 5, "remark": "enabled"}, - {"code": "validation.findings", "name": "读取校验发现", "sort": 6, "remark": "enabled"}, - {"code": "compile.submit", "name": "提交编译", "sort": 7, "remark": "enabled，前置校验必须 passed"}, - {"code": "compile.status", "name": "查询编译状态", "sort": 8, "remark": "enabled"}, - {"code": "evidence.collect", "name": "采集证据", "sort": 9, "remark": "enabled，幂等键强制"}, - {"code": "evidence.query", "name": "查询证据", "sort": 10, "remark": "enabled"}, - {"code": "assessment.score", "name": "Rubric 加权评分", "sort": 11, "remark": "enabled"}, - {"code": "assessment.report", "name": "生成评估报告", "sort": 12, "remark": "enabled"}, - {"code": "analytics.snapshot_read", "name": "读取分析快照", "sort": 13, "remark": "enabled，只读匿名聚合"}, - {"code": "blueprint.delete", "name": "删除蓝图", "sort": 14, "remark": "disabled，fail-closed 拒绝"}, - {"code": "blueprint.publish", "name": "发布蓝图", "sort": 15, "remark": "disabled，需人工拍板"}, - {"code": "tenant.create", "name": "创建租户", "sort": 16, "remark": "disabled，治理域人工操作"}, - {"code": "rbac.role_grant", "name": "授予角色", "sort": 17, "remark": "disabled，rbac 写保护"}, - {"code": "world.write", "name": "写入世界", "sort": 18, "remark": "disabled，复用域零改表"}, - {"code": "scene.write", "name": "写入场景", "sort": 19, "remark": "disabled，复用域零改表"}, - {"code": "entity.write", "name": "写入实体", "sort": 20, "remark": "disabled，复用域零改表"}, - {"code": "kdb.write", "name": "写入知识库", "sort": 21, "remark": "disabled，M7 零写入"}, - {"code": "runtime.broadcast_force", "name": "强制广播", "sort": 22, "remark": "disabled，广播仅由服务端权威时序触发"} + {"code": "AT_BLUEPRINT_READ", "name": "蓝图读取", "sort": 1, "status": "enabled"}, + {"code": "AT_BLUEPRINT_WRITE", "name": "蓝图写入", "sort": 2, "status": "enabled"}, + {"code": "AT_SUBOBJECT_WRITE", "name": "子对象写入", "sort": 3, "status": "enabled"}, + {"code": "AT_VALIDATE", "name": "蓝图校验", "sort": 4, "status": "enabled"}, + {"code": "AT_COMPILE", "name": "编译GameDefinition", "sort": 5, "status": "enabled"}, + {"code": "AT_EVIDENCE_COLLECT", "name": "证据采集", "sort": 6, "status": "enabled"}, + {"code": "AT_ASSESS", "name": "评估打分", "sort": 7, "status": "enabled"}, + {"code": "AT_KDB_QUERY", "name": "KDB只读检索", "sort": 8, "status": "enabled"}, + {"code": "AT_DOMAIN_QUERY", "name": "基础域只读查询", "sort": 9, "status": "enabled"}, + {"code": "AT_RUNTIME_QUERY", "name": "运行时状态查询", "sort": 10, "status": "enabled"}, + {"code": "AT_TEMPLATE_QUERY", "name": "模板检索", "sort": 11, "status": "enabled"}, + {"code": "AT_ANALYTICS_QUERY", "name": "分析统计查询", "sort": 12, "status": "enabled"}, + {"code": "AT_GOVERNANCE_SUBMIT", "name": "治理审批提交", "sort": 13, "status": "enabled"} ] }, { - "group_code": "pbl_rel_artifact_kind", - "group_name": "PBL 关系类型与产出物种类", - "module": "pbl_appcodes", - "items_count": 12, + "group": "pbl_agent_tool_disabled", + "group_name": "Agent工具-禁用(9,G6范围纪律)", + "module": "pbl_agent_runtime", "items": [ - {"code": "rel.depends_on", "name": "依赖", "sort": 1, "remark": "子对象前置依赖"}, - {"code": "rel.produces", "name": "产出", "sort": 2, "remark": "任务产出产出物规格"}, - {"code": "rel.assesses", "name": "评估", "sort": 3, "remark": "量规引用评估学习目标"}, - {"code": "rel.references", "name": "引用", "sort": 4, "remark": "只读引用复用域对象"}, - {"code": "rel.contains", "name": "包含", "sort": 5, "remark": "树形父子包含"}, - {"code": "artifact.doc", "name": "文档", "sort": 6, "remark": "产出物种类"}, - {"code": "artifact.image", "name": "图片", "sort": 7, "remark": "产出物种类"}, - {"code": "artifact.model", "name": "三维模型", "sort": 8, "remark": "产出物种类，glb/gltf"}, - {"code": "artifact.audio", "name": "音频", "sort": 9, "remark": "产出物种类"}, - {"code": "artifact.video", "name": "视频", "sort": 10, "remark": "产出物种类"}, - {"code": "artifact.code", "name": "代码/脚本", "sort": 11, "remark": "产出物种类"}, - {"code": "artifact.log", "name": "运行日志", "sort": 12, "remark": "产出物种类，运行时自动采集"} + {"code": "AT_PBL_PUBLISH", "name": "蓝图发布(禁用)", "sort": 1, "status": "disabled"}, + {"code": "AT_MARKETPLACE", "name": "Marketplace(禁用)", "sort": 2, "status": "disabled"}, + {"code": "AT_SIM_RUN", "name": "sim.*(禁用)", "sort": 3, "status": "disabled"}, + {"code": "AT_KDB_WRITE", "name": "kdb.write(禁用)", "sort": 4, "status": "disabled"}, + {"code": "AT_RESEARCH_WRITE", "name": "research.write(禁用)", "sort": 5, "status": "disabled"}, + {"code": "AT_3D_AUTHOR", "name": "3D创作(禁用)", "sort": 6, "status": "disabled"}, + {"code": "AT_FRAME_SYNC", "name": "帧同步CRDT(禁用)", "sort": 7, "status": "disabled"}, + {"code": "AT_AGENT_AUTONOMY", "name": "全自主Agent(禁用)", "sort": 8, "status": "disabled"}, + {"code": "AT_SSO_ADMIN", "name": "SSO管理(禁用)", "sort": 9, "status": "disabled"} ] } ] diff --git a/pbl_appcodes/seed.py b/pbl_appcodes/seed.py index 77b458c..4ec16fd 100644 --- a/pbl_appcodes/seed.py +++ b/pbl_appcodes/seed.py @@ -1,203 +1,137 @@ # -*- coding: utf-8 -*- -"""pbl_appcodes 枚举幂等注入（6+2 组 74 项，零自有表）。 +"""pbl_appcodes.seed —— 枚举字典幂等注入（零表，走平台 appcodes 字典） -目标表：appcodes（组）/ appcodes_kv（项）——均为平台基础模块 appbase 既有表， -本模块只写入 pbl_* 前缀的组与项，不改动既有组，不新增表。 +设计约束（docs/01-design/appcodes.md）： + * 本模块 zero_table：不建任何业务表，字典项落在平台 appcodes 字典表内； + * 幂等：同一 (group, code) 已存在则跳过（exists 判定），重复执行结果一致； + * 8 组 74 项，与 codes.json 完全一致（由 self_check.py 断言）； + * 由 apps/pbls/build.sh 在部署阶段调用：python -m pbl_appcodes.seed + * fail-closed：平台字典接口不可用时抛错退出，不静默跳过。 -幂等策略（build.sh 可重复执行零差异）： - 1. 按 group_code 查 appcodes：存在则比对 name/module，不一致才 UPDATE，一致则跳过 - 2. 按 (group_code, code) 查 appcodes_kv：存在则比对 name/sort/remark，一致跳过 - 3. 全程不 DELETE（避免误删平台既有字典） - 4. 返回统计 {groups_inserted, groups_updated, groups_skipped, items_*}，重复执行应全为 skipped - -租户维度：枚举字典为平台级（tenant_id='__system__'），业务数据租户隔离由各模块 CRUD 保证。 +对外契约： + load_codes() -> dict 读取 codes.json + iter_items(doc) -> generator 展平为 (group, group_name, item) + inject(dry_run=False) -> dict 执行注入，返回统计 + main() -> int CLI 入口（0 成功 / 1 失败） """ import json import os +import sys -from pbl_common.kernel import ( - PblError, - current_tenant_id, - db, - fail, - now_str, - write_audit, -) +HERE = os.path.dirname(os.path.abspath(__file__)) +CODES_JSON = os.path.join(HERE, "codes.json") -CODES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'codes.json') -SYSTEM_TENANT = '__system__' -GROUP_TABLE = 'appcodes' -ITEM_TABLE = 'appcodes_kv' - -EXPECTED_GROUPS = 8 -EXPECTED_ITEMS = 74 +#: 平台字典表（appcodes 基础模块所有，本模块只写字典项，不建表） +DICT_TABLE = "app_codes" def load_codes(path=None): - """读取 codes.json（注入与自检共用同一事实源）。""" - target = path or CODES_FILE - if not os.path.exists(target): - fail('PBL_E_PARAM_INVALID', detail={'reason': 'codes.json 不存在: %s' % target}) - with open(target, 'r', encoding='utf-8') as fp: - return json.load(fp) + """读取 codes.json；文件缺失或不可解析直接抛错（fail-closed）。""" + p = path or CODES_JSON + if not os.path.isfile(p): + raise RuntimeError("codes.json 不存在: %s" % p) + with open(p, "r", encoding="utf-8") as f: + doc = json.load(f) + if not isinstance(doc, dict) or "dicts" not in doc: + raise RuntimeError("codes.json 结构非法：缺少 dicts 段") + return doc -def iter_groups(data): - """产出组字典序列（兼容 groups 为数组/字典两种写法）。""" - groups = data.get('groups') - if isinstance(groups, list): - for grp in groups: - if isinstance(grp, dict): - yield grp - return - if isinstance(groups, dict): - for key, grp in groups.items(): - if isinstance(grp, dict): - grp.setdefault('group_code', key) - yield grp +def iter_items(doc): + """展平字典：yield (group, group_name, item_dict)。""" + for d in doc.get("dicts") or []: + group = d.get("group") + group_name = d.get("group_name") + for it in d.get("items") or []: + yield group, group_name, it -def _diff_fields(old_row, new_fields): - """比对字段差异，返回需要更新的字段子集（无差异返回空 dict）。""" - changed = {} - for key, val in new_fields.items(): - old_val = old_row.get(key) - if old_val is None and val is None: - continue - if str(old_val if old_val is not None else '') != str(val if val is not None else ''): - changed[key] = val - return changed - - -def seed_group(adp, grp, stats): - """注入单个枚举组到 appcodes（存在即比对，无差异跳过）。""" - group_code = grp.get('group_code') - if not group_code: - fail('PBL_E_PARAM_INVALID', detail={'reason': '组缺少 group_code'}) - - fields = { - 'code': group_code, - 'name': grp.get('group_name') or group_code, - 'module': grp.get('module') or 'pbl_appcodes', - 'tenant_id': SYSTEM_TENANT, - 'remark': grp.get('remark') or 'PBL Agent OS 枚举（%s）' % group_code, - } - existing = adp.get(GROUP_TABLE, {'code': group_code}, tenant_id=SYSTEM_TENANT) - if existing: - changed = _diff_fields(existing, fields) - if changed: - changed['update_time'] = now_str() - adp.update(GROUP_TABLE, {'code': group_code}, changed, tenant_id=SYSTEM_TENANT) - stats['groups_updated'] += 1 - else: - stats['groups_skipped'] += 1 - return group_code - - row = dict(fields) - row['create_time'] = now_str() - row['update_time'] = row['create_time'] - adp.insert(GROUP_TABLE, row, tenant_id=SYSTEM_TENANT) - stats['groups_inserted'] += 1 - return group_code - - -def seed_items(adp, group_code, items, stats): - """注入组内枚举项到 appcodes_kv（按 (group_code, code) 幂等）。""" - for item in items or []: - code = item.get('code') - if not code: - fail('PBL_E_PARAM_INVALID', detail={'reason': '组 %s 存在缺 code 的项' % group_code}) - fields = { - 'group_code': group_code, - 'code': code, - 'name': item.get('name') or code, - 'sort': int(item.get('sort') or 0), - 'remark': item.get('remark') or '', - 'enabled': 1, - 'tenant_id': SYSTEM_TENANT, - } - existing = adp.get(ITEM_TABLE, {'group_code': group_code, 'code': code}, - tenant_id=SYSTEM_TENANT) - if existing: - changed = _diff_fields(existing, fields) - if changed: - changed['update_time'] = now_str() - adp.update(ITEM_TABLE, {'group_code': group_code, 'code': code}, changed, - tenant_id=SYSTEM_TENANT) - stats['items_updated'] += 1 - else: - stats['items_skipped'] += 1 - continue - - row = dict(fields) - row['create_time'] = now_str() - row['update_time'] = row['create_time'] - adp.insert(ITEM_TABLE, row, tenant_id=SYSTEM_TENANT) - stats['items_inserted'] += 1 - - -def new_stats(): - """初始化注入统计。""" - return { - 'groups_inserted': 0, 'groups_updated': 0, 'groups_skipped': 0, - 'items_inserted': 0, 'items_updated': 0, 'items_skipped': 0, - 'groups_total': 0, 'items_total': 0, - } - - -def seed_all(adapter=None, path=None, audit=True): - """全量幂等注入 8 组 74 项，返回统计 dict。 - - 重复执行时 inserted/updated 应全为 0、skipped 等于总量（零差异校验依据）。 - """ - data = load_codes(path) - adp = adapter or db('pbl_appcodes') - stats = new_stats() - - for grp in iter_groups(data): - group_code = seed_group(adp, grp, stats) - seed_items(adp, group_code, grp.get('items'), stats) - stats['groups_total'] += 1 - stats['items_total'] += len(grp.get('items') or []) - - if stats['groups_total'] != EXPECTED_GROUPS: - fail('PBL_E_PARAM_INVALID', - detail={'reason': '组数契约不符', 'expect': EXPECTED_GROUPS, 'got': stats['groups_total']}) - if stats['items_total'] != EXPECTED_ITEMS: - fail('PBL_E_PARAM_INVALID', - detail={'reason': '项数契约不符', 'expect': EXPECTED_ITEMS, 'got': stats['items_total']}) - - if audit: - try: - write_audit('seed', object_type='pbl_appcodes', object_code='codes.json', - result='success', detail=stats, tenant_id=SYSTEM_TENANT, adapter=adp) - except PblError: - pass # 审计降级不阻断种子注入 - - stats['idempotent'] = (stats['groups_inserted'] == 0 and stats['groups_updated'] == 0 - and stats['items_inserted'] == 0 and stats['items_updated'] == 0) - return stats - - -def list_group_codes(adapter=None): - """返回 codes.json 中全部组编码（供 build.sh 校验与 api 下拉）。""" - data = load_codes() - return [grp.get('group_code') for grp in iter_groups(data)] - - -def get_items(group_code, adapter=None): - """按组编码取枚举项（优先读库，库不可用时回落 codes.json 离线兜底）。""" - adp = adapter or db('pbl_appcodes') +def _get_sor(): + """取 sqlor 句柄；库名只来自 ServerEnv().get_module_dbname()，禁止硬编码。""" try: - rows = adp.query(ITEM_TABLE, where={'group_code': group_code}, - order_by='sort', limit=200, tenant_id=SYSTEM_TENANT) - if rows: - return rows - except PblError: - pass - data = load_codes() - for grp in iter_groups(data): - if grp.get('group_code') == group_code: - return list(grp.get('items') or []) - return [] + from pbl_common.kernel import get_dbname + dbname = get_dbname("pbl_appcodes") + except Exception: + try: + from appbase.serverenv import ServerEnv + dbname = ServerEnv().get_module_dbname("pbl_appcodes") + except Exception as e: + raise RuntimeError("无法解析 pbl_appcodes 库名（ServerEnv 不可用）: %s" % e) + import sqlor + return sqlor.sor(dbname) if hasattr(sqlor, "sor") else sqlor.Sor(dbname) + + +def _already_exists(sor, group, code): + """幂等判定：同 (group, code) 已存在则返回 True。""" + rows = sor.R(DICT_TABLE, {"group_name": group, "code": code}, limit=1) + return bool(rows) + + +def inject(dry_run=False, verbose=True): + """执行幂等注入。 + + 返回 {'total': n, 'inserted': i, 'skipped': s, 'failed': f, 'dry_run': bool} + dry_run=True 时只统计不落库（供 build.sh 预检）。 + """ + doc = load_codes() + stat = {"total": 0, "inserted": 0, "skipped": 0, "failed": 0, "dry_run": bool(dry_run)} + sor = None if dry_run else _get_sor() + + for group, group_name, it in iter_items(doc): + stat["total"] += 1 + code = it.get("code") + record = { + "group_name": group, + "group_label": group_name, + "code": code, + "name": it.get("name"), + "sort": it.get("sort"), + "module": (doc.get("_meta") or {}).get("module") or "pbl_appcodes", + "status": it.get("status") or "enabled", + "remark": it.get("remark") or "", + } + if dry_run: + stat["inserted"] += 1 + if verbose: + print("[DRY] %s.%s sort=%s" % (group, code, record["sort"])) + continue + try: + if _already_exists(sor, group, code): + stat["skipped"] += 1 + if verbose: + print("[SKIP] already exists: %s.%s" % (group, code)) + continue + sor.C(DICT_TABLE, record) + stat["inserted"] += 1 + if verbose: + print("[ADD ] %s.%s sort=%s" % (group, code, record["sort"])) + except Exception as e: + stat["failed"] += 1 + print("[FAIL] %s.%s -> %s" % (group, code, e), file=sys.stderr) + + return stat + + +def main(argv=None): + argv = list(argv if argv is not None else sys.argv[1:]) + dry = "--dry-run" in argv + quiet = "--quiet" in argv + try: + stat = inject(dry_run=dry, verbose=not quiet) + except Exception as e: + print("SEED pbl_appcodes: FAIL (%s)" % e, file=sys.stderr) + return 1 + print("SEED pbl_appcodes: total=%(total)d inserted=%(inserted)d " + "skipped=%(skipped)d failed=%(failed)d dry_run=%(dry_run)s" % stat) + if stat["failed"]: + return 1 + if stat["total"] != 74: + print("SEED pbl_appcodes: FAIL 项数不等于 74", file=sys.stderr) + return 1 + print("SEED pbl_appcodes: PASS 74/74") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pbl_appcodes/self_check.py b/pbl_appcodes/self_check.py index bb8fe4f..d28ec7b 100644 --- a/pbl_appcodes/self_check.py +++ b/pbl_appcodes/self_check.py @@ -1,181 +1,178 @@ # -*- coding: utf-8 -*- -"""pbl_appcodes 数量契约自检（可独立运行，不依赖数据库）。 +"""pbl_appcodes.self_check —— 枚举字典契约自检（可独立运行） -核验项（对齐 docs/01-design/appcodes.md 与 QC 退回意见 #15）： - 1. _meta.groups == 8、_meta.items == 74、_meta.dialect == 'mariadb' - 2. 实际组数 == _meta.groups；实际项数合计 == _meta.items - 3. 每组声明 items 数 == 该组实际项数 - 4. 全部编码（group_code 与 item code）全局唯一 - 5. 每组内 sort 从 1 开始连续递增（无跳号、无重复） - 6. 每项必含 code/name/sort 三字段且类型合法 +运行方式（无需数据库、无需平台环境）： + cd modules/pbl_appcodes && python -m pbl_appcodes.self_check + 或 python modules/pbl_appcodes/pbl_appcodes/self_check.py -运行： - python -m pbl_appcodes.self_check - python modules/pbl_appcodes/pbl_appcodes/self_check.py -输出（成功）： +期望输出末行： SELF_CHECK pbl_appcodes: PASS 74/74 -退出码：0=PASS，1=FAIL（build.sh 据此 fail-fast）。 + +校验项（对齐 docs/01-design/appcodes.md 与 QC 退回意见 #11/#12）： + C1 codes.json 可解析，_meta.groups == 8、_meta.items == 74、_meta.dialect == 'mariadb' + C2 实际组数 == _meta.groups，实际项数合计 == _meta.items + C3 编码全局唯一（跨组不得重复） + C4 每组 sort 从 1 连续递增到 n（无跳号、无重复、无 0/负数） + C5 每项必备字段 code/name/sort 齐全且类型正确，code 形如大写标识符 + C6 组名唯一、group 字段为合法标识符 + C7 Agent 工具启停契约：enabled 组 13 项、disabled 组 9 项，status 与组语义一致 + C8 零表约束：本模块不得声明任何 CREATE TABLE（读 seed.py 源码断言） + +退出码：全部通过 0；任一失败 1（fail-closed，供 build.sh / CI 直接判定）。 """ import json import os +import re import sys -CODES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'codes.json') +HERE = os.path.dirname(os.path.abspath(__file__)) +CODES_JSON = os.path.join(HERE, "codes.json") +SEED_PY = os.path.join(HERE, "seed.py") -EXPECTED_GROUPS = 8 -EXPECTED_ITEMS = 74 -EXPECTED_DIALECT = 'mariadb' -REQUIRED_ITEM_FIELDS = ('code', 'name', 'sort') +EXPECT_GROUPS = 8 +EXPECT_ITEMS = 74 +EXPECT_DIALECT = "mariadb" +EXPECT_ENABLED = 13 +EXPECT_DISABLED = 9 +GROUP_ENABLED = "pbl_agent_tool_enabled" +GROUP_DISABLED = "pbl_agent_tool_disabled" +CODE_RE = re.compile(r"^[A-Z][A-Z0-9_]{1,63}$") +IDENT_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") + +_results = [] -def load_codes(path=None): - """读取 codes.json；文件缺失或非法 JSON 直接抛错（fail-fast，不静默）。""" - target = path or CODES_FILE - if not os.path.exists(target): - raise IOError('codes.json 不存在: %s' % target) - with open(target, 'r', encoding='utf-8') as fp: - return json.load(fp) +def _check(name, ok, detail=""): + _results.append((name, bool(ok), detail)) + return bool(ok) -def check_meta(data, errors): - """校验 _meta 段声明值。""" - meta = data.get('_meta') or {} - if meta.get('dialect') != EXPECTED_DIALECT: - errors.append('_meta.dialect 应为 %s，实际 %r' % (EXPECTED_DIALECT, meta.get('dialect'))) - if meta.get('groups') != EXPECTED_GROUPS: - errors.append('_meta.groups 应为 %d，实际 %r' % (EXPECTED_GROUPS, meta.get('groups'))) - if meta.get('items') != EXPECTED_ITEMS: - errors.append('_meta.items 应为 %d，实际 %r' % (EXPECTED_ITEMS, meta.get('items'))) - return meta +def load_codes(): + if not os.path.isfile(CODES_JSON): + raise SystemExit("FATAL: codes.json 不存在: %s" % CODES_JSON) + with open(CODES_JSON, "r", encoding="utf-8") as f: + return json.load(f) -def iter_groups(data): - """产出 (group_code, group_dict) 序列，跳过 _meta 段。""" - groups = data.get('groups') - if isinstance(groups, list): - for grp in groups: - if isinstance(grp, dict): - yield grp.get('group_code') or grp.get('code') or '', grp - return - if isinstance(groups, dict): - for key, grp in groups.items(): - if isinstance(grp, dict): - yield grp.get('group_code') or key, grp - return - # 兼容平铺结构：顶层除 _meta 外每个键即一组 - for key, val in data.items(): - if key.startswith('_'): - continue - if isinstance(val, dict): - yield val.get('group_code') or key, val - elif isinstance(val, list): - yield key, {'group_code': key, 'items': val} +def run_checks(): + doc = load_codes() + meta = doc.get("_meta") or {} + dicts = doc.get("dicts") or [] + # C1 _meta 契约 + _check("C1.meta.groups==8", meta.get("groups") == EXPECT_GROUPS, + "actual=%r" % (meta.get("groups"),)) + _check("C1.meta.items==74", meta.get("items") == EXPECT_ITEMS, + "actual=%r" % (meta.get("items"),)) + _check("C1.meta.dialect==mariadb", meta.get("dialect") == EXPECT_DIALECT, + "actual=%r" % (meta.get("dialect"),)) + _check("C1.meta.idempotent", meta.get("idempotent") is True) + _check("C1.meta.zero_table", meta.get("zero_table") is True) -def check_group(group_code, grp, errors, seen_group_codes, seen_item_codes): - """校验单组：组编码唯一、声明项数与实际一致、项字段齐全、sort 连续、项编码唯一。""" - if not group_code: - errors.append('存在缺少 group_code 的组') - elif group_code in seen_group_codes: - errors.append('组编码重复: %s' % group_code) + # C2 实际数量 + actual_groups = len(dicts) + actual_items = sum(len(d.get("items") or []) for d in dicts) + _check("C2.actual_groups==8", actual_groups == EXPECT_GROUPS, "actual=%d" % actual_groups) + _check("C2.actual_items==74", actual_items == EXPECT_ITEMS, "actual=%d" % actual_items) + + # C3 编码全局唯一 + seen = {} + dup = [] + for d in dicts: + for it in (d.get("items") or []): + c = it.get("code") + if c in seen: + dup.append("%s(%s/%s)" % (c, seen[c], d.get("group"))) + else: + seen[c] = d.get("group") + _check("C3.code_globally_unique", not dup, "dup=%s" % dup) + + # C4 每组 sort 1..n 连续 + bad_sort = [] + for d in dicts: + items = d.get("items") or [] + sorts = [it.get("sort") for it in items] + expect = list(range(1, len(items) + 1)) + if sorts != expect: + bad_sort.append("%s:%s!=%s" % (d.get("group"), sorts, expect)) + _check("C4.sort_continuous_1..n", not bad_sort, "bad=%s" % bad_sort) + + # C5 项字段完备 + code 形态 + bad_item = [] + for d in dicts: + for it in (d.get("items") or []): + if not isinstance(it.get("code"), str) or not CODE_RE.match(it["code"]): + bad_item.append("code:%r@%s" % (it.get("code"), d.get("group"))) + if not isinstance(it.get("name"), str) or not it["name"].strip(): + bad_item.append("name:%r@%s" % (it.get("name"), d.get("group"))) + if not isinstance(it.get("sort"), int) or isinstance(it.get("sort"), bool): + bad_item.append("sort:%r@%s" % (it.get("sort"), d.get("group"))) + _check("C5.item_fields_valid", not bad_item, "bad=%s" % bad_item[:5]) + + # C6 组标识唯一且合法 + groups = [d.get("group") for d in dicts] + bad_group = [g for g in groups if not isinstance(g, str) or not IDENT_RE.match(g)] + _check("C6.group_ident_valid", not bad_group, "bad=%s" % bad_group) + _check("C6.group_unique", len(set(groups)) == len(groups), "groups=%s" % groups) + names = [d.get("group_name") for d in dicts] + _check("C6.group_name_unique", len(set(names)) == len(names)) + _check("C6.group_name_nonempty", all(isinstance(n, str) and n.strip() for n in names)) + + # C7 Agent 工具启停契约 13/9 + by_group = {d.get("group"): (d.get("items") or []) for d in dicts} + en = by_group.get(GROUP_ENABLED, []) + dis = by_group.get(GROUP_DISABLED, []) + _check("C7.enabled_count==13", len(en) == EXPECT_ENABLED, "actual=%d" % len(en)) + _check("C7.disabled_count==9", len(dis) == EXPECT_DISABLED, "actual=%d" % len(dis)) + _check("C7.enabled_status_all_enabled", + all(it.get("status") == "enabled" for it in en) if en else False) + _check("C7.disabled_status_all_disabled", + all(it.get("status") == "disabled" for it in dis) if dis else False) + _check("C7.tool_total==22", len(en) + len(dis) == 22, + "actual=%d" % (len(en) + len(dis))) + + # C8 零表约束：seed.py 不得含 CREATE TABLE + ddl_hit = [] + if os.path.isfile(SEED_PY): + with open(SEED_PY, "r", encoding="utf-8") as f: + src = f.read() + for kw in ("CREATE TABLE", "create table", "ALTER TABLE", "DROP TABLE"): + if kw in src: + ddl_hit.append(kw) + _check("C8.seed_has_inject_entry", "def inject" in src or "def seed" in src, + "seed.py 缺少注入入口函数") + _check("C8.seed_idempotent_guard", + ("already" in src.lower()) or ("exists" in src.lower()) or ("ON DUPLICATE" in src), + "seed.py 未见幂等判定") else: - seen_group_codes.add(group_code) + _check("C8.seed_py_exists", False, "seed.py 不存在") + _check("C8.zero_table_no_ddl", not ddl_hit, "hit=%s" % ddl_hit) - items = grp.get('items') or [] - if not isinstance(items, list): - errors.append('组 %s 的 items 不是数组' % group_code) - return 0 - - declared = grp.get('items_count') - if declared is not None and declared != len(items): - errors.append('组 %s 声明 items_count=%s 与实际 %d 不一致' % (group_code, declared, len(items))) - - sorts = [] - for idx, item in enumerate(items): - if not isinstance(item, dict): - errors.append('组 %s 第 %d 项不是对象' % (group_code, idx)) - continue - for field in REQUIRED_ITEM_FIELDS: - if field not in item: - errors.append('组 %s 第 %d 项缺字段 %s' % (group_code, idx, field)) - code = item.get('code') - if not isinstance(code, str) or not code: - errors.append('组 %s 第 %d 项 code 非法: %r' % (group_code, idx, code)) - elif code in seen_item_codes: - errors.append('项编码重复: %s（组 %s）' % (code, group_code)) - else: - seen_item_codes.add(code) - if not isinstance(item.get('name'), str) or not item.get('name'): - errors.append('组 %s 项 %s 的 name 非法' % (group_code, code)) - sort_val = item.get('sort') - if not isinstance(sort_val, int) or isinstance(sort_val, bool): - errors.append('组 %s 项 %s 的 sort 非整数: %r' % (group_code, code, sort_val)) - else: - sorts.append(sort_val) - - check_sort_continuous(group_code, sorts, errors) - return len(items) + return actual_items -def check_sort_continuous(group_code, sorts, errors): - """sort 必须为 1..n 连续（允许输入乱序，排序后逐位比对）。""" - if not sorts: - return - ordered = sorted(sorts) - if len(set(ordered)) != len(ordered): - errors.append('%s sort 存在重复值' % group_code) - return - expected = list(range(1, len(ordered) + 1)) - if ordered != expected: - errors.append('%s sort 不连续：期望 %s，实际 %s' % (group_code, expected, ordered)) +def main(): + total_items = run_checks() + failed = [r for r in _results if not r[1]] + print("=" * 68) + print("SELF_CHECK pbl_appcodes —— 枚举字典契约自检") + print("codes.json: %s" % CODES_JSON) + print("-" * 68) + for name, ok, detail in _results: + flag = "PASS" if ok else "FAIL" + line = "[%s] %s" % (flag, name) + if not ok and detail: + line += " -> %s" % detail + print(line) + print("-" * 68) + print("检查项: %d 通过 / %d 失败；字典项总数: %d" % (len(_results) - len(failed), len(failed), total_items)) + if failed: + print("SELF_CHECK pbl_appcodes: FAIL %d/%d" % (total_items - len(failed), total_items)) + return 1 + print("SELF_CHECK pbl_appcodes: PASS %d/%d" % (total_items, total_items)) + return 0 -def run_self_check(path=None, verbose=False): - """执行全部核验，返回 (ok, total_items, errors)。""" - errors = [] - try: - data = load_codes(path) - except Exception as exc: - return False, 0, ['codes.json 读取失败: %s' % exc] - - check_meta(data, errors) - - seen_group_codes = set() - seen_item_codes = set() - group_count = 0 - item_total = 0 - for group_code, grp in iter_groups(data): - group_count += 1 - item_total += check_group(group_code, grp, errors, seen_group_codes, seen_item_codes) - if verbose: - print(' group %-28s items=%d' % (group_code, len(grp.get('items') or []))) - - if group_count != EXPECTED_GROUPS: - errors.append('实际组数 %d != 契约 %d' % (group_count, EXPECTED_GROUPS)) - if item_total != EXPECTED_ITEMS: - errors.append('实际项数 %d != 契约 %d' % (item_total, EXPECTED_ITEMS)) - - return (len(errors) == 0), item_total, errors - - -def main(argv=None): - argv = list(argv if argv is not None else sys.argv[1:]) - verbose = '-v' in argv or '--verbose' in argv - path = None - for arg in argv: - if arg.endswith('.json'): - path = arg - - ok, total, errors = run_self_check(path=path, verbose=verbose) - if ok: - print('SELF_CHECK pbl_appcodes: PASS %d/%d' % (total, EXPECTED_ITEMS)) - return 0 - print('SELF_CHECK pbl_appcodes: FAIL %d/%d' % (total, EXPECTED_ITEMS)) - for msg in errors: - print(' - %s' % msg) - return 1 - - -if __name__ == '__main__': +if __name__ == "__main__": sys.exit(main())