diff --git a/pbl_appcodes/codes.json b/pbl_appcodes/codes.json index a946512..50de54f 100644 --- a/pbl_appcodes/codes.json +++ b/pbl_appcodes/codes.json @@ -4,150 +4,149 @@ "dialect": "mariadb", "groups": 8, "items": 74, + "target_tables": ["appcodes", "appcodes_kv"], "idempotent": true, "zero_table": true, - "code_type_column": "varchar(32)", - "source_of_truth": "projects/pbls/docs/01-design/appcodes.md", - "contract": "8 组 / 74 项 / mariadb 方言；组编码唯一、项编码组内唯一、sort 自 1 连续", - "self_check": "python -m pbl_appcodes.self_check → SELF_CHECK pbl_appcodes: PASS 8/8" + "spec_ref": "projects/pbls/docs/01-design/appcodes.md", + "note": "6 组业务枚举 + 2 组运行时枚举 = 8 组 74 项；注入幂等（重复执行零差异）；本模块零自有表。" }, "groups": [ { - "code": "pbl_blueprint_type", - "name": "蓝图类型", - "module": "pbl_blueprint", - "sort": 1, + "group_code": "pbl_tenant_status", + "group_name": "PBL 租户状态", + "module": "pbl_appcodes", + "items_count": 3, "items": [ - { "code": "course", "name": "课程型蓝图", "sort": 1 }, - { "code": "project", "name": "项目型蓝图", "sort": 2 }, - { "code": "game", "name": "游戏型蓝图", "sort": 3 }, - { "code": "sandbox", "name": "沙盒型蓝图", "sort": 4 }, - { "code": "assessment", "name": "测评型蓝图", "sort": 5 }, - { "code": "hybrid", "name": "混合型蓝图", "sort": 6 } + {"code": "active", "name": "启用", "sort": 1, "remark": "正常可用"}, + {"code": "suspended", "name": "停用", "sort": 2, "remark": "暂停服务，数据保留"}, + {"code": "closed", "name": "注销", "sort": 3, "remark": "终止合作，只读归档"} ] }, { - "code": "pbl_blueprint_status", - "name": "蓝图状态", - "module": "pbl_blueprint", - "sort": 2, + "group_code": "pbl_role", + "group_name": "PBL 租户内角色", + "module": "pbl_appcodes", + "items_count": 5, "items": [ - { "code": "draft", "name": "草稿", "sort": 1 }, - { "code": "validating", "name": "校验中", "sort": 2 }, - { "code": "validated", "name": "校验通过", "sort": 3 }, - { "code": "rejected", "name": "校验驳回", "sort": 4 }, - { "code": "compiled", "name": "已编译", "sort": 5 }, - { "code": "published", "name": "已发布", "sort": 6 }, - { "code": "archived", "name": "已归档", "sort": 7 } + {"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": "pbl_quality_level", - "name": "质量等级（5 级）", - "module": "pbl_validation", - "sort": 3, + "group_code": "pbl_bp_status", + "group_name": "PBL 蓝图状态", + "module": "pbl_appcodes", + "items_count": 6, "items": [ - { "code": "L1_broken", "name": "L1 不可用", "sort": 1 }, - { "code": "L2_incomplete", "name": "L2 不完整", "sort": 2 }, - { "code": "L3_consistent", "name": "L3 一致可编译", "sort": 3 }, - { "code": "L4_coherent", "name": "L4 教学连贯", "sort": 4 }, - { "code": "L5_exemplary", "name": "L5 示范级", "sort": 5 } + {"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": "pbl_validation_dim", - "name": "校验维度（14 维）", - "module": "pbl_validation", - "sort": 4, + "group_code": "pbl_quality_level", + "group_name": "PBL 蓝图质量等级（5 级）", + "module": "pbl_appcodes", + "items_count": 5, "items": [ - { "code": "D01_schema", "name": "D01 结构完整性", "sort": 1 }, - { "code": "D02_reference", "name": "D02 引用完整性", "sort": 2 }, - { "code": "D03_type", "name": "D03 类型一致性", "sort": 3 }, - { "code": "D04_cycle", "name": "D04 循环依赖", "sort": 4 }, - { "code": "D05_naming", "name": "D05 命名规范", "sort": 5 }, - { "code": "D06_cardinality", "name": "D06 基数约束", "sort": 6 }, - { "code": "D07_goal_alignment", "name": "D07 目标对齐", "sort": 7 }, - { "code": "D08_assessment_coverage", "name": "D08 评估覆盖", "sort": 8 }, - { "code": "D09_evidence_traceability", "name": "D09 证据可追溯", "sort": 9 }, - { "code": "D10_tool_safety", "name": "D10 工具安全", "sort": 10 }, - { "code": "D11_tenant_isolation", "name": "D11 租户隔离", "sort": 11 }, - { "code": "D12_version_compat", "name": "D12 版本兼容", "sort": 12 }, - { "code": "D13_resource_bound", "name": "D13 资源边界", "sort": 13 }, - { "code": "D14_i18n_complete", "name": "D14 多语言完整", "sort": 14 } + {"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": "pbl_subobject_type", - "name": "子对象类型（7 类泛化契约）", - "module": "pbl_blueprint", - "sort": 5, + "group_code": "pbl_subobj_type", + "group_name": "PBL 蓝图 7 类子对象类型", + "module": "pbl_appcodes", + "items_count": 7, "items": [ - { "code": "goal", "name": "学习目标", "sort": 1 }, - { "code": "stage", "name": "阶段", "sort": 2 }, - { "code": "task", "name": "任务", "sort": 3 }, - { "code": "role", "name": "角色", "sort": 4 }, - { "code": "rule", "name": "规则", "sort": 5 }, - { "code": "rubric", "name": "评分量规", "sort": 6 }, - { "code": "resource", "name": "资源", "sort": 7 } + {"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": "pbl_agent_tool", - "name": "Agent 工具（22 个：13 启用 / 9 禁用，fail-closed）", - "module": "pbl_agent_runtime", - "sort": 6, + "group_code": "pbl_valid_dim", + "group_name": "PBL 校验 14 维", + "module": "pbl_appcodes", + "items_count": 14, "items": [ - { "code": "blueprint_read", "name": "蓝图读取", "sort": 1, "enabled": 1 }, - { "code": "blueprint_write", "name": "蓝图写入", "sort": 2, "enabled": 1 }, - { "code": "subobject_read", "name": "子对象读取", "sort": 3, "enabled": 1 }, - { "code": "subobject_write", "name": "子对象写入", "sort": 4, "enabled": 1 }, - { "code": "validate_run", "name": "执行校验", "sort": 5, "enabled": 1 }, - { "code": "validate_report", "name": "校验报告", "sort": 6, "enabled": 1 }, - { "code": "compile_run", "name": "执行编译", "sort": 7, "enabled": 1 }, - { "code": "evidence_collect", "name": "证据采集", "sort": 8, "enabled": 1 }, - { "code": "evidence_query", "name": "证据查询", "sort": 9, "enabled": 1 }, - { "code": "assess_run", "name": "执行评估", "sort": 10, "enabled": 1 }, - { "code": "assess_report", "name": "评估报告", "sort": 11, "enabled": 1 }, - { "code": "kdb_query", "name": "知识库只读查询", "sort": 12, "enabled": 1 }, - { "code": "runtime_event", "name": "运行时事件写入", "sort": 13, "enabled": 1 }, - { "code": "shell_exec", "name": "Shell 执行（禁用）", "sort": 14, "enabled": 0 }, - { "code": "sql_raw", "name": "裸 SQL 执行（禁用）", "sort": 15, "enabled": 0 }, - { "code": "file_write_abs", "name": "绝对路径写文件（禁用）", "sort": 16, "enabled": 0 }, - { "code": "net_request", "name": "外网请求（禁用）", "sort": 17, "enabled": 0 }, - { "code": "rbac_modify", "name": "权限修改（禁用）", "sort": 18, "enabled": 0 }, - { "code": "tenant_switch", "name": "租户切换（禁用）", "sort": 19, "enabled": 0 }, - { "code": "kdb_write", "name": "知识库写入（禁用）", "sort": 20, "enabled": 0 }, - { "code": "base_table_write", "name": "引用基表写入（禁用）", "sort": 21, "enabled": 0 }, - { "code": "model_invoke_paid", "name": "付费模型调用（禁用）", "sort": 22, "enabled": 0 } + {"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": "pbl_evidence_type", - "name": "证据类型", - "module": "pbl_evidence", - "sort": 7, + "group_code": "pbl_agent_tool", + "group_name": "PBL Agent 工具白名单（13 启用 / 9 禁用，fail-closed）", + "module": "pbl_appcodes", + "items_count": 22, "items": [ - { "code": "artifact", "name": "产出物", "sort": 1 }, - { "code": "dialogue", "name": "对话记录", "sort": 2 }, - { "code": "action_log", "name": "操作日志", "sort": 3 }, - { "code": "code_snippet", "name": "代码片段", "sort": 4 }, - { "code": "document", "name": "文档", "sort": 5 }, - { "code": "media", "name": "多媒体", "sort": 6 }, - { "code": "peer_review", "name": "同伴互评", "sort": 7 }, - { "code": "teacher_note", "name": "教师批注", "sort": 8 } + {"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": "pbl_assessment_rubric_level", - "name": "量规评分等级", - "module": "pbl_assessment", - "sort": 8, + "group_code": "pbl_rel_artifact_kind", + "group_name": "PBL 关系类型与产出物种类", + "module": "pbl_appcodes", + "items_count": 12, "items": [ - { "code": "R1_beginning", "name": "R1 起步", "sort": 1 }, - { "code": "R2_developing", "name": "R2 发展中", "sort": 2 }, - { "code": "R3_proficient", "name": "R3 熟练", "sort": 3 }, - { "code": "R4_advanced", "name": "R4 进阶", "sort": 4 }, - { "code": "R5_exemplary", "name": "R5 卓越", "sort": 5 } + {"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": "产出物种类，运行时自动采集"} ] } ] diff --git a/pbl_appcodes/init.py b/pbl_appcodes/init.py index 6093648..78b2aca 100644 --- a/pbl_appcodes/init.py +++ b/pbl_appcodes/init.py @@ -1,131 +1,71 @@ # -*- coding: utf-8 -*- -""" -pbl_appcodes.init —— 模块挂载入口（load_pbl_appcodes） +"""pbl_appcodes 模块挂载入口（load_pbl_appcodes）。 -挂载职责： - 1. 零表模块：ensure_tables() 返回空列表（不建自有表） - 2. 注册 API 契约到 ServerEnv（get_codes/is_valid/assert_valid/list_groups） - 3. 幂等注入编码种子（8 组 74 项）——注入失败不阻断启动，降级 stderr - 4. 注册 RBAC 权限路径 +三处同步注册（module-development-spec）： + 1. 本文件 load_pbl_appcodes() + 2. apps/pbls/app/pbls.py LOAD_ORDER 含 'pbl_appcodes'（Wave0 第二位，紧随 pbl_common） + 3. apps/pbls/conf/config.json load_order / module_dbname 含 pbl_appcodes + +零自有表：只向平台既有 appcodes / appcodes_kv 注入 8 组 74 项枚举（幂等）。 +挂载时执行数量契约自检（self_check），不通过直接抛错 fail-fast，避免脏字典进入运行期。 """ -import os -import sys - -from pbl_appcodes import api as appcodes_api -from pbl_appcodes.seed import seed, ensure_tables, seed_summary +from pbl_appcodes.seed import ( + EXPECTED_GROUPS, + EXPECTED_ITEMS, + get_items, + list_group_codes, + seed_all, +) +from pbl_appcodes.self_check import run_self_check MODULE_NAME = 'pbl_appcodes' +MODULE_TABLES = () # 零自有表 +MODULE_VERSION = '1.0.0' -CONTRACT_FUNCS = ( - 'get_codes', 'get_code_name', 'is_valid', 'assert_valid', - 'list_groups', 'health', 'self_check', -) - -RBAC_PATHS = ( - '/api/pbl_appcodes_list.dspy', - '/api/pbl_appcodes_inject.dspy', -) +_loaded = False -def _register_contract(env): - contract = {} - for name in CONTRACT_FUNCS: - fn = getattr(appcodes_api, name, None) - if fn is not None: - contract[name] = fn - # 种子注入也作为契约暴露（部署脚本/运维接口调用） - contract['seed'] = seed - contract['ensure_tables'] = ensure_tables - try: - env.pbl_appcodes = contract - except Exception: # noqa: BLE001 - try: - setattr(env, 'modules', getattr(env, 'modules', {})) - env.modules[MODULE_NAME] = contract - except Exception: # noqa: BLE001 - pass - return contract +def load_pbl_appcodes(auto_seed=True): + """挂载 pbl_appcodes：先跑数量契约自检，再幂等注入枚举。 - -def _register_rbac(env): - registered = [] - try: - perm = getattr(env, 'register_perm', None) or getattr(env, 'add_perm', None) - if callable(perm): - for path in RBAC_PATHS: - try: - perm('logined', path) - registered.append(path) - except Exception: # noqa: BLE001 - pass - except Exception: # noqa: BLE001 - pass - return registered - - -def load_pbl_appcodes(env=None): + auto_seed=False 时只注册查询能力（供单测/离线场景），不触库。 """ - 挂载 pbl_appcodes。 + global _loaded + if _loaded: + return True - env 为 None 时自取 ServerEnv()（兼容单模块测试）。 - 返回 {'module','tables','contract','rbac','seed','ok'} - """ - if env is None: + # ---- 1. 数量契约自检（8 组 74 项 / mariadb 方言 / 编码唯一 / sort 连续）---- + ok_flag, total, errors = run_self_check() + if not ok_flag: + raise AssertionError('pbl_appcodes 契约自检失败: %s' % '; '.join(errors)) + print('SELF_CHECK pbl_appcodes: PASS %d/%d' % (total, EXPECTED_ITEMS)) + + # ---- 2. 注册到 ServerEnv，供其余模块与 api 层取枚举 ---- + from appbase.serverenv import ServerEnv + env = ServerEnv() + env.pbl_appcodes = { + 'version': MODULE_VERSION, + 'groups': EXPECTED_GROUPS, + 'items': EXPECTED_ITEMS, + 'group_codes': list_group_codes(), + 'get_items': get_items, + 'seed_all': seed_all, + 'self_check': run_self_check, + } + + # ---- 3. 幂等注入（build.sh 亦会调用，重复执行零差异）---- + if auto_seed: try: - from ahserver.serverenv import ServerEnv - env = ServerEnv() - except Exception: # noqa: BLE001 - env = None + stats = seed_all() + env.pbl_appcodes['last_seed_stats'] = stats + except Exception as exc: # 库未就绪时降级：保留离线 codes.json 兜底查询 + env.pbl_appcodes['last_seed_stats'] = {'skipped': True, 'reason': str(exc)} - result = {'module': MODULE_NAME, 'tables': [], 'contract': [], - 'rbac': [], 'seed': None, 'ok': True} + _loaded = True + return True - # 1) 零表契约 - try: - result['tables'] = ensure_tables(module=MODULE_NAME) or [] - except Exception as e: # noqa: BLE001 - sys.stderr.write('[pbl_appcodes] ensure_tables failed: %s\n' % e) - result['tables'] = [] - # 2) 注册契约 - if env is not None: - contract = _register_contract(env) - result['contract'] = sorted(contract.keys()) - result['rbac'] = _register_rbac(env) - - # 3) 自检（数量契约 8/74）——不通过只告警，不阻断启动 - try: - sc = appcodes_api.self_check(verbose=False) - result['self_check'] = {'passed': sc['passed'], 'total': sc['total'], - 'groups': sc['groups'], 'items': sc['items'], - 'ok': sc['ok']} - if not sc['ok']: - sys.stderr.write('[pbl_appcodes] SELF_CHECK FAIL %d/%d\n' - % (sc['passed'], sc['total'])) - except Exception as e: # noqa: BLE001 - sys.stderr.write('[pbl_appcodes] self_check error: %s\n' % e) - - # 4) 幂等注入种子（PBLS_SKIP_SEED=1 可跳过，供离线/无库环境启动） - if os.environ.get('PBLS_SKIP_SEED', '').strip() in ('1', 'true', 'yes'): - result['seed'] = {'skipped_by_env': True} - else: - try: - tenant = os.environ.get('PBLS_SEED_TENANT', '*') - res = seed(tenant_id=tenant, dry_run=False) - result['seed'] = {'ok': res.get('ok'), 'groups': res.get('groups'), - 'items': res.get('items'), - 'inserted': res.get('inserted'), - 'skipped': res.get('skipped'), - 'failed': res.get('failed')} - sys.stdout.write('[pbl_appcodes] %s\n' % seed_summary(res)) - if not res.get('ok'): - result['ok'] = False - except Exception as e: # noqa: BLE001 - # 注入失败不阻断启动：常量回落仍可提供编码（get_codes source='constant'） - sys.stderr.write('[pbl_appcodes] seed failed (fallback to constant): %s\n' % e) - result['seed'] = {'ok': False, 'error': str(e), 'fallback': 'constant'} - - sys.stdout.write('[pbl_appcodes] loaded groups=%d contract=%d\n' - % (len(result['contract']), len(result['contract']))) - return result +# 别名：兼容 load_{module} 与 init 两种解析路径 +load_module = load_pbl_appcodes +init = load_pbl_appcodes diff --git a/pbl_appcodes/seed.py b/pbl_appcodes/seed.py index 55b2e4c..77b458c 100644 --- a/pbl_appcodes/seed.py +++ b/pbl_appcodes/seed.py @@ -1,213 +1,203 @@ # -*- coding: utf-8 -*- -""" -pbl_appcodes.seed —— 幂等编码注入（零自有表，写平台 appcodes 表） +"""pbl_appcodes 枚举幂等注入（6+2 组 74 项，零自有表）。 -幂等策略：先 COUNT 判存在 → 存在则 skip（内容变更才 update）→ 不存在则 insert。 -返回 per_group 统计：{'inserted':n,'skipped':n,'updated':n,'failed':n} +目标表：appcodes（组）/ appcodes_kv（项）——均为平台基础模块 appbase 既有表， +本模块只写入 pbl_* 前缀的组与项，不改动既有组，不新增表。 -铁律： - - 零表模块：ensure_tables() 返回空列表，不建任何自有表 - - tenant_id 打头：注入必须指定租户（'*' 表示全局种子） - - 写保护：只写 appcodes 表，绝不触碰 rbac/world/scene/entity/scense/ - scense_runtime/script_engine 基表 - - dry_run 支持：只统计不落库，供部署前预演 +幂等策略（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 保证。 """ -import datetime -import hashlib +import json import os -import sys -from pbl_appcodes.codes_data import ( - ALL_GROUPS, GROUP_COUNT, ITEM_COUNT, group_items, find_group, counts, +from pbl_common.kernel import ( + PblError, + current_tenant_id, + db, + fail, + now_str, + write_audit, ) -CODE_TABLE = os.environ.get('PBLS_CODE_TABLE', 'appcodes') -CODE_MODULE = os.environ.get('PBLS_CODE_MODULE', 'pbl') -GLOBAL_TENANT = '*' +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 -def ensure_tables(conn=None, module='pbl_appcodes'): - """ - 零表模块契约：不建任何自有表，恒返回空列表。 - （编码写入平台既有 appcodes 表，由 appbase/rbac 侧建表） - """ - return [] +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) -def _now(): - return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') +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 _checksum(text): - return hashlib.sha256((text or '').encode('utf-8')).hexdigest()[:32] +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 _item_payload(group_code, item): - """项 → appcodes 表行（不含 tenant_id，由调用方注入）""" - from pbl_common.serialize import dumps - extra = {k: v for k, v in item.items() if k not in ('code', 'name', 'sort')} - return { - 'code_type': group_code, - 'code': item.get('code'), - 'name': item.get('name'), - 'sort': item.get('sort'), - 'module': CODE_MODULE, - 'remark': dumps(extra) if extra else None, +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 _exists(conn, tenant_id, code_type, code): - """幂等判定：COUNT 该 (tenant, code_type, code) 是否已存在""" - from pbl_common.dbutil import query_one - sql = ('SELECT COUNT(1) AS cnt FROM `%s` ' - 'WHERE tenant_id = %%s AND code_type = %%s AND code = %%s' % CODE_TABLE) - row = query_one(sql, [tenant_id, code_type, code], conn=None) - return int((row or {}).get('cnt', 0)) > 0 - - -def _insert_row(conn, tenant_id, row): - from pbl_common.dbutil import execute - data = dict(row) - data['tenant_id'] = tenant_id - data.setdefault('created_at', _now()) - data.setdefault('updated_at', _now()) - cols = list(data.keys()) - sql = 'INSERT INTO `%s` (%s) VALUES (%s)' % ( - CODE_TABLE, - ', '.join('`%s`' % c for c in cols), - ', '.join(['%s'] * len(cols)), - ) - return execute(sql, [data[c] for c in cols], conn=conn) - - -def _update_row(conn, tenant_id, code_type, code, row): - from pbl_common.dbutil import execute - data = dict(row) - data.pop('tenant_id', None) - data['updated_at'] = _now() - cols = list(data.keys()) - sql = 'UPDATE `%s` SET %s WHERE tenant_id = %%s AND code_type = %%s AND code = %%s' % ( - CODE_TABLE, ', '.join('`%s` = %%s' % c for c in cols)) - args = [data[c] for c in cols] + [tenant_id, code_type, code] - return execute(sql, args, conn=conn) - - -def seed_group(group_code, tenant_id=GLOBAL_TENANT, conn=None, dry_run=False): - """ - 注入单个编码组（幂等）。 - 返回 {'group','tenant_id','inserted','skipped','updated','failed','dry_run','errors'} - """ - stat = {'group': group_code, 'tenant_id': tenant_id, 'inserted': 0, - 'skipped': 0, 'updated': 0, 'failed': 0, 'dry_run': bool(dry_run), - 'errors': []} - group = find_group(group_code) - if group is None: - stat['failed'] = 1 - stat['errors'].append('未登记的编码组：%s' % group_code) - return stat - - items = group_items(group) - for item in items: +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: - stat['failed'] += 1 - stat['errors'].append('%s 存在无 code 的项' % group_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 = _item_payload(group_code, item) - try: - if dry_run: - stat['inserted'] += 1 - continue - if _exists(conn, tenant_id, group_code, code): - # 已存在：内容一致则 skip，不一致则 update（幂等且可演进） - stat['skipped'] += 1 - else: - _insert_row(conn, tenant_id, row) - stat['inserted'] += 1 - except Exception as e: # noqa: BLE001 - stat['failed'] += 1 - stat['errors'].append('%s.%s: %s' % (group_code, code, e)) - return stat + + 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 seed(tenant_id=GLOBAL_TENANT, groups=None, dry_run=False, conn=None): - """ - 注入全部（或指定）编码组，幂等。 - - tenant_id : 目标租户，'*' 表示全局种子 - groups : 组编码列表，None = 全部 8 组 - dry_run : True 只统计不落库 - conn : 可选外部连接（多组共用一个事务） - - 返回 {'ok','tenant_id','dry_run','groups','items','per_group':[...], - 'inserted','skipped','updated','failed'} - """ - from pbl_common.tenant import normalize_tenant - from pbl_common.audit import write_audit - - tid = tenant_id if tenant_id == GLOBAL_TENANT else normalize_tenant(tenant_id) - target_codes = list(groups) if groups else [g.get('code') for g in ALL_GROUPS] - - per_group = [] - totals = {'inserted': 0, 'skipped': 0, 'updated': 0, 'failed': 0} - - own_conn = conn is None and not dry_run - if own_conn: - from pbl_common.dbutil import get_conn - conn = get_conn(module='pbl_appcodes') - - try: - for gcode in target_codes: - st = seed_group(gcode, tid, conn=conn, dry_run=dry_run) - per_group.append(st) - for k in totals: - totals[k] += st.get(k, 0) - if own_conn and conn is not None: - if totals['failed']: - conn.rollback() - else: - conn.commit() - finally: - if own_conn and conn is not None: - try: - conn.close() - except Exception: # noqa: BLE001 - pass - - result = { - 'ok': totals['failed'] == 0, - 'tenant_id': tid, - 'dry_run': bool(dry_run), - 'groups': len(per_group), - 'items': sum(len(group_items(find_group(g) or {})) for g in target_codes), - 'per_group': per_group, - 'code_table': CODE_TABLE, +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, } - result.update(totals) - if not dry_run: + +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', 'appcodes', None, tenant_id=tid, - result='success' if result['ok'] else 'fail', - detail={'groups': len(per_group), 'inserted': totals['inserted'], - 'skipped': totals['skipped'], 'failed': totals['failed']}, - module='pbl_appcodes') - except Exception: # noqa: BLE001 - pass + write_audit('seed', object_type='pbl_appcodes', object_code='codes.json', + result='success', detail=stats, tenant_id=SYSTEM_TENANT, adapter=adp) + except PblError: + pass # 审计降级不阻断种子注入 - return result + stats['idempotent'] = (stats['groups_inserted'] == 0 and stats['groups_updated'] == 0 + and stats['items_inserted'] == 0 and stats['items_updated'] == 0) + return stats -def dry_run_seed(tenant_id=GLOBAL_TENANT, groups=None): - """预演注入（不落库），返回与 seed 同结构统计""" - return seed(tenant_id=tenant_id, groups=groups, dry_run=True) +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 seed_summary(result): - """把 seed() 结果压成一行文本（build.sh 日志用）""" - return ('appcodes seed tenant=%s groups=%s items=%s ' - 'inserted=%s skipped=%s updated=%s failed=%s dry_run=%s' - % (result.get('tenant_id'), result.get('groups'), result.get('items'), - result.get('inserted'), result.get('skipped'), result.get('updated'), - result.get('failed'), result.get('dry_run'))) +def get_items(group_code, adapter=None): + """按组编码取枚举项（优先读库，库不可用时回落 codes.json 离线兜底）。""" + adp = adapter or db('pbl_appcodes') + 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 [] diff --git a/pbl_appcodes/self_check.py b/pbl_appcodes/self_check.py index 7998670..bb8fe4f 100644 --- a/pbl_appcodes/self_check.py +++ b/pbl_appcodes/self_check.py @@ -1,166 +1,181 @@ # -*- coding: utf-8 -*- -"""pbl_appcodes 自检脚本（QC 硬门禁 #15 证据件，真实落库可运行） +"""pbl_appcodes 数量契约自检（可独立运行，不依赖数据库）。 -输出机器可判读结论行：SELF_CHECK pbl_appcodes: PASS 6/6 / FAIL n/6 (failed: ...) -契约来源：pbls_spec.json appcodes{groups:8,items:74,idempotent:true} + docs/01-design/appcodes.md + data-model.md(dialect=mariadb, code varchar(32)) -运行：python3 modules/pbl_appcodes/pbl_appcodes/self_check.py [-q] | python3 -m pbl_appcodes.self_check -退出码 0=通过 1=有失败项。零表、零写入、零网络，只读 codes.json，幂等可重复。 +核验项（对齐 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 三字段且类型合法 + +运行： + 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）。 """ -from __future__ import print_function -import io, json, os, sys -HERE = os.path.dirname(os.path.abspath(__file__)) -CODES_JSON = os.path.join(HERE, 'codes.json') -EXPECT_GROUPS = 8 -EXPECT_ITEMS = 74 -EXPECT_DIALECT = 'mariadb' -EXPECT_CODE_TYPE_MAXLEN = 32 -EXPECT_ITEM_FIELDS = ('code', 'name', 'sort') +import json +import os +import sys -def _load_codes(path=CODES_JSON): - with io.open(path, 'r', encoding='utf-8') as f: - return json.load(f) +CODES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'codes.json') -def _extract_groups(doc): - """兼容三种落盘形态，统一返回 [(group_name, code_type, [item,...]), ...]""" - out = [] - if not isinstance(doc, dict): - return out - gs = doc.get('groups') - if isinstance(gs, list): - for g in gs: - if not isinstance(g, dict): - continue - name = g.get('group') or g.get('name') or g.get('code_type') or '' - ctype = g.get('code_type') or name - items = g.get('items') - if not isinstance(items, list): - items = [] - out.append((name, ctype, items)) - return out - if isinstance(gs, dict): - for k, v in gs.items(): - if isinstance(v, list): - items, ctype = v, k - elif isinstance(v, dict): - items = v.get('items') if isinstance(v.get('items'), list) else [] - ctype = v.get('code_type') or k - else: - items, ctype = [], k - out.append((k, ctype, items)) - return out - for k, v in doc.items(): - if k.startswith('_'): +EXPECTED_GROUPS = 8 +EXPECTED_ITEMS = 74 +EXPECTED_DIALECT = 'mariadb' +REQUIRED_ITEM_FIELDS = ('code', 'name', 'sort') + + +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_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 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(v, list): - out.append((k, k, v)) - elif isinstance(v, dict) and isinstance(v.get('items'), list): - out.append((k, v.get('code_type') or k, v['items'])) - return out + if isinstance(val, dict): + yield val.get('group_code') or key, val + elif isinstance(val, list): + yield key, {'group_code': key, 'items': val} -def _meta(doc): - m = doc.get('_meta') if isinstance(doc, dict) else None - return m if isinstance(m, dict) else {} -def t_meta_contract(doc, groups): - m = _meta(doc); errs = [] - if m.get('groups') != EXPECT_GROUPS: errs.append('_meta.groups=%r 期望 %d' % (m.get('groups'), EXPECT_GROUPS)) - if m.get('items') != EXPECT_ITEMS: errs.append('_meta.items=%r 期望 %d' % (m.get('items'), EXPECT_ITEMS)) - if str(m.get('dialect') or '').lower() != EXPECT_DIALECT: errs.append('_meta.dialect=%r 期望 %s' % (m.get('dialect'), EXPECT_DIALECT)) - rg = len(groups); ri = sum(len(g[2]) for g in groups) - if m.get('groups') != rg: errs.append('_meta.groups=%r 与实际组数 %d 不符' % (m.get('groups'), rg)) - if m.get('items') != ri: errs.append('_meta.items=%r 与实际项数 %d 不符' % (m.get('items'), ri)) - return (len(errs) == 0), ('; '.join(errs) or 'groups=%d items=%d dialect=%s' % (rg, ri, m.get('dialect'))) +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) + else: + seen_group_codes.add(group_code) -def t_group_count(doc, groups): - errs = []; names = [g[0] for g in groups] - if len(groups) != EXPECT_GROUPS: errs.append('组数=%d 期望 %d' % (len(groups), EXPECT_GROUPS)) - if len(set(names)) != len(names): errs.append('组名重复：%s' % ','.join(sorted(set([n for n in names if names.count(n) > 1])))) - empty = [i for i, n in enumerate(names) if not n] - if empty: errs.append('第 %s 组组名为空' % empty) - return (len(errs) == 0), ('; '.join(errs) or '组数=%d 组名=%s' % (len(groups), ','.join([str(n) for n in names]))) + items = grp.get('items') or [] + if not isinstance(items, list): + errors.append('组 %s 的 items 不是数组' % group_code) + return 0 -def t_item_count(doc, groups): - errs = []; total = sum(len(g[2]) for g in groups) - if total != EXPECT_ITEMS: errs.append('总项数=%d 期望 %d' % (total, EXPECT_ITEMS)) - for name, _ct, items in groups: - if not items: errs.append('%s 项数为 0' % name) - declared = _meta(doc).get('group_items') or {} - if isinstance(declared, dict) and declared: - for name, _ct, items in groups: - if name in declared and declared[name] != len(items): - errs.append('%s 项数=%d 与 _meta.group_items=%r 不符' % (name, len(items), declared[name])) - detail = '总项数=%d 各组=%s' % (total, ','.join(['%s:%d' % (g[0], len(g[2])) for g in groups])) - return (len(errs) == 0), ('; '.join(errs) or detail) + 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))) -def t_code_unique(doc, groups): - errs = []; seen = {} - for name, _ct, items in groups: - for it in items: - if not isinstance(it, dict): - errs.append('%s 存在非对象项：%r' % (name, it)); continue - code = it.get('code') - if code in (None, ''): - errs.append('%s 存在空 code' % name); continue - code = str(code) - if len(code) > EXPECT_CODE_TYPE_MAXLEN: - errs.append('%s.%s code 长度 %d > %d' % (name, code, len(code), EXPECT_CODE_TYPE_MAXLEN)) - if code in seen: errs.append('code 重复：%s（%s / %s）' % (code, seen[code], name)) - else: seen[code] = name - return (len(errs) == 0), ('; '.join(errs[:8]) or '唯一 code 数=%d' % len(seen)) + 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) -def t_item_fields(doc, groups): - errs = []; n = 0 - for gname, _ct, items in groups: - for it in items: - if not isinstance(it, dict): continue - n += 1 - for fld in EXPECT_ITEM_FIELDS: - if fld not in it: errs.append('%s.%s 缺字段 %s' % (gname, it.get('code'), fld)) - if not str(it.get('name') or '').strip(): errs.append('%s.%s name 为空' % (gname, it.get('code'))) - s = it.get('sort') - if not isinstance(s, int) or isinstance(s, bool) or s <= 0: - errs.append('%s.%s sort 非正整数：%r' % (gname, it.get('code'), s)) - return (len(errs) == 0), ('; '.join(errs[:8]) or '校验项数=%d 字段=%s' % (n, '/'.join(EXPECT_ITEM_FIELDS))) + check_sort_continuous(group_code, sorts, errors) + return len(items) -def t_sort_continuous(doc, groups): - errs = [] - for gname, _ct, items in groups: - sorts = [it.get('sort') for it in items if isinstance(it, dict)] - sorts = [s for s in sorts if isinstance(s, int) and not isinstance(s, bool)] - if len(sorts) != len(items): - errs.append('%s sort 缺失或非整数' % gname); continue - if len(set(sorts)) != len(sorts): - errs.append('%s sort 重复：%s' % (gname, sorted(set([s for s in sorts if sorts.count(s) > 1])))); continue - asc = sorted(sorts) - diffs = set([asc[i + 1] - asc[i] for i in range(len(asc) - 1)]) - if len(diffs) > 1: errs.append('%s sort 不连续：%s（步长集合=%s）' % (gname, asc, sorted(diffs))) - elif diffs and list(diffs)[0] <= 0: errs.append('%s sort 步长非法：%s' % (gname, asc)) - return (len(errs) == 0), ('; '.join(errs[:6]) or '各组 sort 等差连续（8/8 组通过）') -CHECKS = (('t_meta_contract', t_meta_contract), ('t_group_count', t_group_count), ('t_item_count', t_item_count), - ('t_code_unique', t_code_unique), ('t_item_fields', t_item_fields), ('t_sort_continuous', t_sort_continuous)) +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 run(verbose=True): + +def run_self_check(path=None, verbose=False): + """执行全部核验，返回 (ok, total_items, errors)。""" + errors = [] try: - doc = _load_codes() - except Exception as e: - return (0, len(CHECKS), [('load_codes_json', False, '读取失败：%s' % e)]) - groups = _extract_groups(doc); results = [] - for name, fn in CHECKS: - try: ok, detail = fn(doc, groups) - except Exception as e: ok, detail = False, '检查异常：%s' % e - results.append((name, ok, detail)) - if verbose: print('[%s] %-20s %s' % ('PASS' if ok else 'FAIL', name, detail)) - return (len([r for r in results if r[1]]), len(results), results) + 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): - verbose = not (argv and ('-q' in argv or '--quiet' in argv)) - passed, total, results = run(verbose=verbose) - failed = [r[0] for r in results if not r[1]] - if failed: print('SELF_CHECK pbl_appcodes: FAIL %d/%d (failed: %s)' % (passed, total, ', '.join(failed))) - else: print('SELF_CHECK pbl_appcodes: PASS %d/%d' % (passed, total)) - return 0 if not failed else 1 + 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__': - sys.exit(main(sys.argv[1:])) + sys.exit(main())