#!/usr/bin/env python3 """Validate and render a OneOS delivery-model Yunxiao project profile.""" from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any TARGET_LIFECYCLE = [ "待处理", "已确认", "分析中", "分析完成", "设计中", "设计完成", "待开发", "开发中", "开发完成", "待测试", "测试中", "测试完成", "发布中", "发布完成", "发布失败", "已关闭", ] REQUIRED_TASK_LABELS = {"交付", "开发", "测试", "发版"} REQUIRED_CONTROL_IDS = { *(f"Y{index:02d}" for index in range(1, 17)), "Y20", "Y21", "Y22", *(f"Y{index:02d}" for index in range(30, 36)), "Y40", "A01", "A02", "A02B", *(f"A{index:02d}" for index in range(3, 11)), *(f"TC{index:02d}" for index in range(1, 4)), *(f"DF{index:02d}" for index in range(1, 4)), *(f"CI{index:02d}" for index in range(1, 4)), "L01", "L02", } VALID_MODES = { "manual", "native_rule", "integration_bridge", "disabled_legacy", "not_supported" } VALID_PHASES = {"P0", "P1", "P2", "P3"} VALID_TASK_WORKFLOW_MODES = {"A", "B"} VALID_TASK_IDENTITY_MODES = {"work_item_type", "task_label", "title_prefix_process_control"} VALID_LEGACY_DISPOSITIONS = {"keep_until_ready", "disable", "replace", "already_disabled"} SENSITIVE_FRAGMENTS = ( "password", "passwd", "secret", "token", "cookie", "credential", "private_key", "otp", "密码", "密钥", ) def is_text(value: Any) -> bool: return isinstance(value, str) and bool(value.strip()) def require_text(container: dict[str, Any], key: str, label: str, errors: list[str]) -> None: if not is_text(container.get(key)): errors.append(f"{label}.{key} 不能为空") def walk_sensitive_keys(value: Any, prefix: str = "") -> list[str]: found: list[str] = [] if isinstance(value, dict): for key, nested in value.items(): current = f"{prefix}.{key}" if prefix else str(key) if any(fragment in str(key).lower() for fragment in SENSITIVE_FRAGMENTS): found.append(current) found.extend(walk_sensitive_keys(nested, current)) elif isinstance(value, list): for index, nested in enumerate(value): found.extend(walk_sensitive_keys(nested, f"{prefix}[{index}]")) return found def validate_controls(controls: Any, label: str, errors: list[str]) -> None: if not isinstance(controls, list): errors.append(f"{label} 必须是数组") return seen: set[str] = set() for index, control in enumerate(controls): item_label = f"{label}[{index}]" if not isinstance(control, dict): errors.append(f"{item_label} 必须是对象") continue control_id = control.get("id") if not is_text(control_id): errors.append(f"{item_label}.id 不能为空") continue if control_id in seen: errors.append(f"{label} 控制项重复：{control_id}") seen.add(control_id) if control_id not in REQUIRED_CONTROL_IDS: errors.append(f"{label} 未知控制项：{control_id}") mode = control.get("mode") if mode not in VALID_MODES: errors.append(f"{item_label}.mode 不合法") if not isinstance(control.get("enabled"), bool): errors.append(f"{item_label}.enabled 必须是布尔值") if mode in {"disabled_legacy", "not_supported"} and control.get("enabled") is not False: errors.append(f"{item_label} 的 {mode} 模式要求 enabled=false") if control_id in {"L01", "L02"}: if mode != "disabled_legacy" or control.get("enabled") is not False: errors.append(f"{item_label} 必须停用，防止发布完成绕过验收") for key in ("owner", "fallback", "evidence"): require_text(control, key, item_label, errors) missing = sorted(REQUIRED_CONTROL_IDS - seen) if missing: errors.append(f"{label} 缺少控制项：" + "、".join(missing)) def validate_project(project: Any, index: int, errors: list[str]) -> None: label = f"projects[{index}]" if not isinstance(project, dict): errors.append(f"{label} 必须是对象") return require_text(project, "name", label, errors) if project.get("migration_phase") not in VALID_PHASES: errors.append(f"{label}.migration_phase 必须是 P0/P1/P2/P3") statuses = project.get("requirement_statuses_observed") if not isinstance(statuses, list) or not statuses or not all(is_text(v) for v in statuses): errors.append(f"{label}.requirement_statuses_observed 必须是非空文本数组") if project.get("task_workflow_mode") not in VALID_TASK_WORKFLOW_MODES: errors.append(f"{label}.task_workflow_mode 必须是 A 或 B") if project.get("task_identity_mode") not in VALID_TASK_IDENTITY_MODES: errors.append(f"{label}.task_identity_mode 不合法") if not isinstance(project.get("native_rule_can_filter_related_task_identity"), bool): errors.append(f"{label}.native_rule_can_filter_related_task_identity 必须是布尔值") for key in ( "main_task_prefix", "development_task_prefix", "test_task_prefix", "release_task_prefix" ): require_text(project, key, label, errors) repositories = project.get("repositories") if not isinstance(repositories, list) or not repositories: errors.append(f"{label}.repositories 至少包含一个仓库") else: for repo_index, repo in enumerate(repositories): repo_label = f"{label}.repositories[{repo_index}]" if not isinstance(repo, dict): errors.append(f"{repo_label} 必须是对象") continue for key in ("name", "role", "integration_branch", "evidence"): require_text(repo, key, repo_label, errors) if repo.get("integration_branch") not in {"dev", "develop"}: errors.append(f"{repo_label}.integration_branch 只能是实际存在的 dev/develop") for key in ("codeup_integrated", "required_for_mr_gate"): if not isinstance(repo.get(key), bool): errors.append(f"{repo_label}.{key} 必须是布尔值") test_plan = project.get("test_plan") if not isinstance(test_plan, dict): errors.append(f"{label}.test_plan 必须是对象") else: require_text(test_plan, "name", f"{label}.test_plan", errors) require_text(test_plan, "evidence", f"{label}.test_plan", errors) for key, required in ( ("iteration_formally_related", True), ("cases_partitioned_by_requirement", True), ("fixed_defect_is_terminal", False), ("tester_retest_required", True), ): if test_plan.get(key) is not required: errors.append(f"{label}.test_plan.{key} 必须为 {str(required).lower()}") release = project.get("release") if not isinstance(release, dict): errors.append(f"{label}.release 必须是对象") else: require_text(release, "evidence", f"{label}.release", errors) for key in ( "release_task_formally_related_to_iteration", "production_evidence_separate_from_test", "acceptance_required_after_release", "production_pipeline_unchanged", ): if release.get(key) is not True: errors.append(f"{label}.release.{key} 必须为 true") legacy_rules = project.get("legacy_rules") if not isinstance(legacy_rules, list): errors.append(f"{label}.legacy_rules 必须是数组") else: for legacy_index, rule in enumerate(legacy_rules): item_label = f"{label}.legacy_rules[{legacy_index}]" if not isinstance(rule, dict): errors.append(f"{item_label} 必须是对象") continue for key in ("name", "reason", "evidence"): require_text(rule, key, item_label, errors) if rule.get("disposition") not in VALID_LEGACY_DISPOSITIONS: errors.append(f"{item_label}.disposition 不合法") if not isinstance(rule.get("enabled"), bool): errors.append(f"{item_label}.enabled 必须是布尔值") validate_controls(project.get("controls"), f"{label}.controls", errors) def validate(data: dict[str, Any]) -> list[str]: errors: list[str] = [] if data.get("schema_version") != 1: errors.append("schema_version 必须为 1") if data.get("flow_model") != "oneos_delivery": errors.append("flow_model 必须为 oneos_delivery") for key in ("organization", "work_item_type"): require_text(data, key, "profile", errors) if data.get("target_lifecycle") != TARGET_LIFECYCLE: errors.append("target_lifecycle 必须与OneOS终态16状态顺序一致") labels = data.get("task_labels") if not isinstance(labels, list) or not REQUIRED_TASK_LABELS.issubset(set(labels)): errors.append("task_labels 必须包含：交付、开发、测试、发版") policies = data.get("policies") if not isinstance(policies, dict): errors.append("policies 必须是对象") else: true_keys = ( "require_formal_relations", "require_current_state_condition", "require_nonzero_development_tasks", "require_signed_timestamped_replay_protected_callbacks", ) false_keys = ("allow_automatic_final_close_without_acceptance", "test_success_is_production_release") for key in true_keys: if policies.get(key) is not True: errors.append(f"policies.{key} 必须为 true") for key in false_keys: if policies.get(key) is not False: errors.append(f"policies.{key} 必须为 false") parts = policies.get("bridge_idempotency_key_components") if not isinstance(parts, list) or not parts or not all(is_text(v) for v in parts): errors.append("policies.bridge_idempotency_key_components 不能为空") projects = data.get("projects") if not isinstance(projects, list) or not projects: errors.append("projects 至少包含一个项目") else: for index, project in enumerate(projects): validate_project(project, index, errors) sensitive = walk_sensitive_keys(data) if sensitive: errors.append("配置中禁止出现敏感字段：" + "、".join(sensitive)) if "请填写" in json.dumps(data, ensure_ascii=False): errors.append("仍存在“请填写”占位符") return errors def render(data: dict[str, Any]) -> str: lines = [ "# OneOS云效终态交付模型执行计划", "", f"- 组织：{data['organization']}", f"- 工作项类型：{data['work_item_type']}", f"- 项目数：{len(data['projects'])}", "- 完整性口径：每项目48项Y/A/测试/缺陷/集成/旧规则控制。", "- 安全边界：不修改生产流水线；发布完成后保留产品验收门禁。", "", ] for project in data["projects"]: missing_statuses = [s for s in TARGET_LIFECYCLE if s not in project["requirement_statuses_observed"]] lines.extend([ f"## 项目：{project['name']}", "", f"- 迁移阶段：{project['migration_phase']}", f"- 任务工作流方案：{project['task_workflow_mode']}", f"- 任务身份方式：{project['task_identity_mode']}", f"- 原生规则可过滤关联任务身份：{'是' if project['native_rule_can_filter_related_task_identity'] else '否'}", f"- 缺少目标状态：{'、'.join(missing_statuses) if missing_statuses else '无'}", "", "### 旧规则处置", "", "| 规则 | 处置 | 启用 | 原因 | 证据 |", "|---|---|---|---|---|", ]) for rule in project["legacy_rules"]: lines.append( f"| {rule['name']} | {rule['disposition']} | {'是' if rule['enabled'] else '否'} | " f"{rule['reason']} | {rule['evidence']} |" ) lines.extend([ "", "### 48项控制台账", "", "| ID | 方式 | 启用 | 负责人 | 回退方案 | 证据 |", "|---|---|---|---|---|---|", ]) for control in sorted(project["controls"], key=lambda item: item["id"]): lines.append( f"| {control['id']} | {control['mode']} | {'是' if control['enabled'] else '否'} | " f"{control['owner']} | {control['fallback']} | {control['evidence']} |" ) lines.extend([ "", "### 迁移门禁", "", "- [ ] 目标状态与流转边已补齐。", "- [ ] 主任务可可靠识别或已明确交桥接。", "- [ ] 多仓库需求与开发任务均正式关联分支/MR。", "- [ ] Testhub按需求分包且缺陷复测闭环。", "- [ ] test、生产发布和产品验收证据分离。", "- [ ] L01、L02已停用，旧规则已有备份和回滚。", "", ]) return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) validate_parser = subparsers.add_parser("validate") validate_parser.add_argument("profile", type=Path) render_parser = subparsers.add_parser("render") render_parser.add_argument("profile", type=Path) render_parser.add_argument("--output", required=True, type=Path) args = parser.parse_args() try: with args.profile.open("r", encoding="utf-8-sig") as handle: data = json.load(handle) if not isinstance(data, dict): raise ValueError("配置根节点必须是JSON对象") errors = validate(data) except (OSError, json.JSONDecodeError, ValueError) as error: print(f"配置读取失败：{error}", file=sys.stderr) return 2 if errors: for error in errors: print(f"[错误] {error}", file=sys.stderr) return 1 if args.command == "validate": print(f"[通过] OneOS配置有效：{len(data['projects'])}个项目，每项目48项控制。") return 0 args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(render(data), encoding="utf-8") print(f"[完成] 已生成执行计划：{args.output}") return 0 if __name__ == "__main__": raise SystemExit(main())