deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
a28264bd72
commit
970a6256fb
@ -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": "产出物种类，运行时自动采集"}
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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 []
|
||||
|
||||
@ -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())
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user