"""Excel -> runtime JSON converter (V5-M4), byte-equal via governed generation. Business edits the config Excel; this regenerates the runtime JSON. To guarantee byte-equality (so `policy_bundle_hash` and runtime behaviour never drift on a no-op regen), the converter uses the current JSON as a STRUCTURAL TEMPLATE (preserving key order) and overlays Excel cell values onto the matching base leaves, coercing each cell to the base value's type. Only paths that already exist in the base are overlaid — structurally divergent / non-value Excel columns are skipped, and non-sheet-backed JSON sections pass through unchanged. M4 adds governed runtime sheets. Rows from those sheets enter runtime JSON only when `approval_status=approved` and `runtime_enabled=true`; other rows are kept as Excel ledger/governance data and are intentionally excluded. python scripts/build_config_from_excel.py --check # byte-equal? exit 1 on drift python scripts/build_config_from_excel.py --write # regenerate JSON from Excel This handles VALUE edits (thresholds, scores, gate values, flags). Structural changes (adding/removing rows) need a base update — out of V2-M1 scope. """ from __future__ import annotations import argparse import copy import json import sys from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from openpyxl import load_workbook from content_agent.integrations import config_store from scripts.check_config_json_canonical import canonical_dumps RULE_PACK_XLSX = ROOT / "tech_documents/规则包映射/规则包映射配置表.xlsx" WALK_XLSX = ROOT / "tech_documents/游走策略/游走策略配置表.xlsx" RULE_PACK_JSON = ROOT / "product_documents/规则包/douyin_rule_packs.v1.json" WALK_JSON = ROOT / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json" SCORE_THRESHOLDS_JSON = ROOT / "tech_documents/数据接口与来源/score_thresholds.json" WALK_POLICY_JSON = ROOT / "tech_documents/数据接口与来源/walk_policy.json" PROFILE_DIR = ROOT / "tech_documents/数据接口与来源/platform_profiles" SENTINEL = "备注" # row 2 of every sheet; skipped SKIP_COLUMN = "注释" # human annotation column with no JSON home APPROVAL_STATUSES = {"draft", "waiting_for_human_reivew", "approved", "rejected", "deprecated"} RUNTIME_MODES = {"generate_runtime"} THRESHOLD_KEYS = { "pool_total", "pool_query", "review_total", "review_query", "walk_query", "walk_platform", "walk_total", } # A SheetSpec maps one Excel sheet onto a JSON section by overlay. # top-level: base[section] is a list of objects indexed by id_json (== row[id_excel]). # nested: base["rule_packs"][*][child_path...] grouped by parent_fk (rule_pack_id). class SheetSpec: def __init__( self, sheet, *, section=None, id_excel=None, id_json=None, renames=None, parent_fk=None, child_path=None, mode="overlay_existing", ): self.sheet = sheet self.section = section # top-level JSON key (for nested: "rule_packs") self.id_excel = id_excel self.id_json = id_json or id_excel self.renames = renames or {} # excel column -> dotted base path self.parent_fk = parent_fk # excel column holding rule_pack_id (nested only) self.child_path = child_path # dotted path within a rule_pack (nested only) self.mode = mode _WHEN = {"field_path": "when.field", "operator": "when.op", "expected_value": "when.value"} RULE_PACK_SPECS = [ SheetSpec("rule_pack_dispatch", section="rule_pack_dispatch", id_excel="dispatch_id"), SheetSpec("decision_reason_codes", section="decision_reason_codes", id_excel="decision_reason_code"), SheetSpec("effect_status_mapping", section="effect_status_mapping", id_excel="mapping_id"), SheetSpec("query_effect_aggregation", section="query_effect_aggregation", id_excel="aggregation_id"), SheetSpec("hard_gate_rules", section="rule_packs", child_path="hard_gates", parent_fk="rule_pack_id", id_excel="gate_id", renames={"gate_label": "label", **_WHEN}), SheetSpec("scorecard_dimensions", section="rule_packs", child_path="scorecard.dimensions", parent_fk="rule_pack_id", id_excel="dimension_key", id_json="key", renames={"dimension_label": "label"}), SheetSpec("scorecard_scoring_rules", section="rule_packs", child_path="scorecard.scoring_rules", parent_fk="rule_pack_id", id_excel="scoring_rule_id"), SheetSpec("threshold_actions", section="rule_packs", child_path="thresholds", parent_fk="rule_pack_id", id_excel="decision_reason_code"), ] # Walk sheets mirror JSON 1:1 (identity columns; only 注释 skipped). # V3 清理: 13 段收窄到 3 个仍被消费的段(其余 10 段已被 walk_graph+walk_policy 取代, # JSON 段与对应 Excel sheet 一并删除)。 _WALK = { "walk_edge_catalog": "edge_id", "walk_rule_pack_binding": "binding_id", "v4_walk_gate": "gate_id", "walk_fact_contract": "runtime_file", } WALK_SPECS = [SheetSpec(sheet, section=sheet, id_excel=idc) for sheet, idc in _WALK.items()] GOVERNED_RULE_PACK_SHEETS = { "score_weight_profiles": "generate_runtime", "score_threshold_profiles": "generate_runtime", } GOVERNED_WALK_SHEETS = { "platform_edge_matrix": "generate_runtime", "walk_policy_budget_matrix": "generate_runtime", } def _read_rows(ws) -> list[dict[str, Any]]: headers = [c.value for c in ws[1]] rows = [] for excel_row in ws.iter_rows(min_row=2, values_only=True): if excel_row and str(excel_row[0] or "").startswith(SENTINEL): continue # sentinel comment row rows.append({h: v for h, v in zip(headers, excel_row) if h is not None}) return rows def _sheet_rows(wb, sheet: str) -> list[dict[str, Any]]: if sheet not in wb.sheetnames: return [] return _read_rows(wb[sheet]) def _parse_bool(value: Any) -> bool: if isinstance(value, bool): return value return str(value or "").strip().lower() in {"true", "1", "yes", "y"} def _split_list(value: Any) -> list[str]: if value is None or value == "": return [] if isinstance(value, list): return [str(item) for item in value if str(item)] text = str(value).strip() if not text: return [] if text.startswith("["): try: parsed = json.loads(text) except json.JSONDecodeError: parsed = [] if isinstance(parsed, list): return [str(item) for item in parsed if str(item)] return [part.strip() for part in text.split(",") if part.strip()] def _require_valid_governance(row: dict[str, Any], sheet: str) -> None: status = str(row.get("approval_status") or "").strip() if status not in APPROVAL_STATUSES: raise ValueError(f"{sheet} invalid approval_status: {status!r}") if status == "approved" and _parse_bool(row.get("runtime_enabled")): if not row.get("owner"): raise ValueError(f"{sheet} approved runtime row missing owner") if not row.get("last_reviewed_at"): raise ValueError(f"{sheet} approved runtime row missing last_reviewed_at") def _runtime_row_enabled(row: dict[str, Any], sheet: str) -> bool: _require_valid_governance(row, sheet) return str(row.get("approval_status") or "").strip() == "approved" and _parse_bool(row.get("runtime_enabled")) def _load_workbook(path: Path, *, read_only: bool = True): return load_workbook(path, data_only=True, read_only=read_only) def _coerce(cell: Any, base_value: Any) -> Any: if isinstance(base_value, bool): return str(cell).strip().lower() in {"true", "1", "yes"} if not isinstance(cell, bool) else cell if isinstance(base_value, int): return base_value if cell is None or cell == "" else int(round(float(cell))) if isinstance(base_value, float): return base_value if cell is None or cell == "" else float(cell) if isinstance(base_value, (list, dict)): if isinstance(cell, str) and cell.strip(): try: return json.loads(cell) except json.JSONDecodeError: return base_value # non-JSON cell (e.g. comma-joined) -> keep base return base_value if cell is None or cell == "" else cell if base_value is None: return None if cell is None or cell == "" else cell return base_value if cell is None else str(cell) def _set_if_exists(obj: dict[str, Any], dotted: str, cell: Any) -> None: parts = dotted.split(".") cursor = obj for part in parts[:-1]: if not isinstance(cursor, dict) or part not in cursor: return cursor = cursor[part] leaf = parts[-1] if isinstance(cursor, dict) and leaf in cursor: cursor[leaf] = _coerce(cell, cursor[leaf]) def _overlay_obj(obj: dict[str, Any], row: dict[str, Any], spec: SheetSpec) -> None: for column, value in row.items(): if column == SKIP_COLUMN or column == spec.id_excel or column == spec.parent_fk: continue path = spec.renames.get(column, column) _set_if_exists(obj, path, value) def _nested_targets(base: dict[str, Any], spec: SheetSpec) -> dict[tuple, dict[str, Any]]: index: dict[tuple, dict[str, Any]] = {} for pack in base.get("rule_packs", []): cursor: Any = pack for part in spec.child_path.split("."): cursor = cursor.get(part) if isinstance(cursor, dict) else None for child in cursor or []: index[(pack.get("rule_pack_id"), child.get(spec.id_json))] = child return index def overlay_workbook(xlsx: Path, base: dict[str, Any], specs: list[SheetSpec]) -> dict[str, Any]: wb = _load_workbook(xlsx) for spec in specs: if spec.mode in RUNTIME_MODES: raise ValueError(f"{spec.sheet} is a runtime sheet and needs a dedicated builder") rows = _read_rows(wb[spec.sheet]) if spec.child_path: index = _nested_targets(base, spec) for row in rows: obj = index.get((row.get(spec.parent_fk), row.get(spec.id_excel))) if obj is not None: _overlay_obj(obj, row, spec) else: index = {obj.get(spec.id_json): obj for obj in base.get(spec.section, [])} for row in rows: obj = index.get(row.get(spec.id_excel)) if obj is not None: _overlay_obj(obj, row, spec) return base def _build_score_weight_profiles(rows: list[dict[str, Any]]) -> dict[str, Any] | None: runtime_rows = [row for row in rows if _runtime_row_enabled(row, "score_weight_profiles")] if not runtime_rows: return None profiles: dict[str, dict[str, Any]] = {} for row in runtime_rows: profile_id = str(row.get("profile_id") or "").strip() dimension_key = str(row.get("dimension_key") or "").strip() if not profile_id or not dimension_key: raise ValueError("score_weight_profiles runtime rows require profile_id and dimension_key") profile = profiles.setdefault( profile_id, { "profile_id": profile_id, "profile_label": row.get("profile_label") or profile_id, "historical_alias_ids": _split_list(row.get("historical_alias_ids")), "platform_scope": _split_list(row.get("platform_scope")) or ["*"], "content_format": row.get("content_format") or "*", "applies_when": row.get("applies_when") or "always", "weights": {}, "normalization_policy": row.get("normalization_policy") or "sum_100", "normalization_reason": row.get("normalization_reason") or "", "missing_dimension_policy": row.get("missing_dimension_policy") or "score_missing", "critical_dimensions": [], "score_source_paths": {}, "priority": int(row.get("priority") or 0), "fallback_profile_id": row.get("fallback_profile_id") or None, "runtime_enabled": True, "approval_status": "approved", "owner": row.get("owner"), "last_reviewed_at": str(row.get("last_reviewed_at")), }, ) profile["weights"][dimension_key] = _number(row.get("weight_percent")) score_source_path = row.get("score_source_path") if score_source_path: profile["score_source_paths"][dimension_key] = str(score_source_path) if _parse_bool(row.get("critical_dimension")): profile["critical_dimensions"].append(dimension_key) return { "schema_version": "score_weight_profiles.v2", "profiles": list(profiles.values()), } def _apply_score_weight_profiles(rule_pack: dict[str, Any], rows: list[dict[str, Any]]) -> None: built = _build_score_weight_profiles(rows) if built is None: return for pack in rule_pack.get("rule_packs", []): scorecard = pack.get("scorecard") if isinstance(scorecard, dict): scorecard["score_weight_profiles"] = built def _build_score_thresholds(base: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]: runtime_rows = [row for row in rows if _runtime_row_enabled(row, "score_threshold_profiles")] if not runtime_rows: return base thresholds = { key: copy.deepcopy(value) for key, value in base.items() if key.startswith("_") or key == "schema_version" } for row in runtime_rows: key = str(row.get("threshold_key") or "").strip() if key not in THRESHOLD_KEYS: raise ValueError(f"score_threshold_profiles unknown threshold_key: {key}") value = row.get("threshold_value") if value is None or value == "": raise ValueError(f"score_threshold_profiles missing value for {key}") scope_values = _split_list(row.get("platform_scope")) or ["default"] for scope in scope_values: target = "default" if scope == "*" else scope thresholds.setdefault(target, {})[key] = int(round(float(value))) return thresholds def _build_platform_profiles(rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: by_platform: dict[str, dict[str, Any]] = {} for profile_path in PROFILE_DIR.glob("*.json"): profile, _ = config_store.load_json(profile_path) by_platform[profile_path.stem] = profile for row in rows: if not _runtime_row_enabled(row, "platform_edge_matrix"): continue platform = str(row.get("platform") or "").strip() edge_id = str(row.get("edge_id") or "").strip() status = str(row.get("status") or "").strip() if platform not in by_platform: raise ValueError(f"platform_edge_matrix unknown platform: {platform}") if not edge_id or not status: raise ValueError("platform_edge_matrix runtime rows require platform, edge_id, status") spec = by_platform[platform].setdefault("edges", {}).setdefault(edge_id, {}) spec["status"] = status if row.get("status_reason"): spec["reason"] = str(row.get("status_reason")) return by_platform def _set_wrapped_value(container: dict[str, Any], key: str, value: Any) -> None: current = container.get(key) if isinstance(current, dict) and "value" in current: current["value"] = value else: container[key] = value def _build_walk_policy(base: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]: runtime_rows = [row for row in rows if _runtime_row_enabled(row, "walk_policy_budget_matrix")] if not runtime_rows: return base policy = copy.deepcopy(base) platform_rows = [] for row in runtime_rows: platform = str(row.get("platform") or "").strip() edge_id = str(row.get("edge_id") or "").strip() if not platform or not edge_id: raise ValueError("walk_policy_budget_matrix runtime rows require platform and edge_id") if platform == "default" and edge_id == "__global__": platform_rows.append( { "platform": platform, "edge_id": edge_id, "max_total_actions": _nullable_int(row.get("max_total_actions")), "max_depth": _nullable_int(row.get("max_depth")), "budget_exhausted_reason_code": row.get("budget_exhausted_reason_code") or "budget_exhausted", "operator_display_status": row.get("operator_display_status") or "budget_blocked", } ) if row.get("max_total_actions") not in {None, ""}: _set_wrapped_value(policy["global"], "max_total_actions_per_run", int(row.get("max_total_actions"))) if row.get("max_depth") not in {None, ""}: _set_wrapped_value(policy["global"], "max_depth", int(row.get("max_depth"))) continue budget_row = { "platform": platform, "edge_id": edge_id, "max_total_actions": _nullable_int(row.get("max_total_actions")), "max_per_content": _nullable_int(row.get("max_per_content")), "max_works_per_author": _nullable_int(row.get("max_works_per_author")), "max_pages": _nullable_int(row.get("max_pages")), "budget_exhausted_reason_code": row.get("budget_exhausted_reason_code") or "budget_exhausted", "operator_display_status": row.get("operator_display_status") or "budget_blocked", } platform_rows.append(budget_row) if platform == "default": legacy = next((item for item in policy.get("edge_budgets", []) if item.get("edge_id") == edge_id), None) if legacy is None: legacy = {"edge_id": edge_id} policy.setdefault("edge_budgets", []).append(legacy) for field in ["max_total_actions", "max_per_content", "max_works_per_author", "max_pages"]: if budget_row[field] is not None: legacy[field] = budget_row[field] policy["platform_edge_budgets"] = platform_rows return policy def _nullable_int(value: Any) -> int | None: if value is None or value == "": return None return int(round(float(value))) def _number(value: Any) -> int | float: parsed = float(value) return int(parsed) if parsed.is_integer() else parsed def _get_path(obj: dict[str, Any], dotted: str) -> tuple[bool, Any]: cursor: Any = obj for part in dotted.split("."): if not isinstance(cursor, dict) or part not in cursor: return False, None cursor = cursor[part] return True, cursor def _cell_value(value: Any) -> Any: if isinstance(value, bool): return value if isinstance(value, (list, dict)): return json.dumps(value, ensure_ascii=False) return value def sync_workbook(xlsx: Path, base: dict[str, Any], specs: list[SheetSpec]) -> int: """JSON -> Excel: write base values into the mapped Excel cells (round-trip safe). Used to reconcile a drifted Excel to the authoritative JSON (JSON wins). Only mapped value cells are overwritten; id / FK / 注释 / unmapped columns are left. """ wb = _load_workbook(xlsx, read_only=False) changed = 0 for spec in specs: ws = wb[spec.sheet] headers = [c.value for c in ws[1]] col_idx = {h: i + 1 for i, h in enumerate(headers) if h is not None} if spec.child_path: index = _nested_targets(base, spec) else: index = {obj.get(spec.id_json): obj for obj in base.get(spec.section, [])} for r in range(2, ws.max_row + 1): first = ws.cell(row=r, column=1).value if first is not None and str(first).startswith(SENTINEL): continue row = {h: ws.cell(row=r, column=col_idx[h]).value for h in col_idx} if spec.child_path: obj = index.get((row.get(spec.parent_fk), row.get(spec.id_excel))) else: obj = index.get(row.get(spec.id_excel)) if obj is None: continue for col, ci in col_idx.items(): if col in {SKIP_COLUMN, spec.id_excel, spec.parent_fk}: continue exists, value = _get_path(obj, spec.renames.get(col, col)) if not exists: continue new = _cell_value(value) if ws.cell(row=r, column=ci).value != new: ws.cell(row=r, column=ci, value=new) changed += 1 wb.save(xlsx) return changed def build() -> dict[str, dict[str, Any]]: """Return {json_path: regenerated_obj} for all runtime config files.""" rule_pack_base, _ = config_store.load_json(RULE_PACK_JSON) walk_base, _ = config_store.load_json(WALK_JSON) score_threshold_base, _ = config_store.load_json(SCORE_THRESHOLDS_JSON) walk_policy_base, _ = config_store.load_json(WALK_POLICY_JSON) rule_wb = _load_workbook(RULE_PACK_XLSX) walk_wb = _load_workbook(WALK_XLSX) rule_pack = overlay_workbook(RULE_PACK_XLSX, rule_pack_base, RULE_PACK_SPECS) _apply_score_weight_profiles(rule_pack, _sheet_rows(rule_wb, "score_weight_profiles")) score_thresholds = _build_score_thresholds(score_threshold_base, _sheet_rows(rule_wb, "score_threshold_profiles")) walk_strategy = overlay_workbook(WALK_XLSX, walk_base, WALK_SPECS) walk_strategy.pop("walk_edge_catalog", None) walk_policy = _build_walk_policy(walk_policy_base, _sheet_rows(walk_wb, "walk_policy_budget_matrix")) built = { str(RULE_PACK_JSON): rule_pack, str(WALK_JSON): walk_strategy, str(SCORE_THRESHOLDS_JSON): score_thresholds, str(WALK_POLICY_JSON): walk_policy, } for platform, profile in _build_platform_profiles(_sheet_rows(walk_wb, "platform_edge_matrix")).items(): built[str(PROFILE_DIR / f"{platform}.json")] = profile return built def governance_report() -> list[dict[str, Any]]: """Summarize governed sheets so no-op sheets and runtime rows are visible in gates.""" report: list[dict[str, Any]] = [] for xlsx, sheets in [ (RULE_PACK_XLSX, GOVERNED_RULE_PACK_SHEETS), (WALK_XLSX, GOVERNED_WALK_SHEETS), ]: wb = _load_workbook(xlsx) for sheet, mode in sheets.items(): rows = _sheet_rows(wb, sheet) runtime_count = sum(1 for row in rows if _runtime_row_enabled(row, sheet)) report.append( { "workbook": str(xlsx.relative_to(ROOT)), "sheet": sheet, "mode": mode, "row_count": len(rows), "runtime_row_count": runtime_count, } ) return report def _first_diff(expected: str, actual: str) -> str: e, a = expected.splitlines(), actual.splitlines() for i in range(max(len(e), len(a))): el = e[i] if i < len(e) else "" al = a[i] if i < len(a) else "" if el != al: return f"line {i + 1}:\n json: {el}\n excel: {al}" return "" def main() -> int: args = _parse_args() if args.sync_excel: rule_pack_base, _ = config_store.load_json(RULE_PACK_JSON) walk_base, _ = config_store.load_json(WALK_JSON) n1 = sync_workbook(RULE_PACK_XLSX, rule_pack_base, RULE_PACK_SPECS) n2 = sync_workbook(WALK_XLSX, walk_base, WALK_SPECS) print(json.dumps({"status": "synced", "cells_changed": {"rule_pack": n1, "walk": n2}}, ensure_ascii=False, indent=2)) return 0 built = build() findings = [] for path_str, obj in built.items(): path = Path(path_str) generated = canonical_dumps(obj) current = path.read_text(encoding="utf-8") if args.write: path.write_text(generated, encoding="utf-8") findings.append({"config_path": str(path.relative_to(ROOT)), "written": True}) else: ok = generated == current entry = {"config_path": str(path.relative_to(ROOT)), "byte_equal": ok} if not ok: entry["first_diff"] = _first_diff(current, generated) findings.append(entry) status = "written" if args.write else ("pass" if all(f.get("byte_equal") for f in findings) else "fail") print(json.dumps({"status": status, "findings": findings, "governance": governance_report()}, ensure_ascii=False, indent=2)) return 0 if status in {"written", "pass"} else 1 def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--check", action="store_true", help="byte-equal check (default)") parser.add_argument("--write", action="store_true", help="regenerate JSON from Excel instead of checking") parser.add_argument("--sync-excel", action="store_true", help="JSON wins: write JSON values back into Excel (one-time reconcile)") return parser.parse_args() if __name__ == "__main__": sys.exit(main())