content-find-agent-new/scripts/build_config_from_excel.py

"""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 "<missing>"
        al = a[i] if i < len(a) else "<missing>"
        if el != al:
            return f"line {i + 1}:\n  json:  {el}\n  excel: {al}"
    return "<no line diff>"


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())