llm/scripts/prepare_1c_saved_state_copy_sql.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any


SOURCE_BY_TARGET = {"ConfigSave": "Config", "ConfigCASSave": "ConfigCAS"}
COPY_COLUMNS = ("FileName", "Creation", "Modified", "Attributes", "DataSize", "BinaryData", "PartNo")


def sql_literal(value: str) -> str:
    return "N'" + value.replace("'", "''") + "'"


def bracket_name(value: str) -> str:
    return "[" + value.replace("]", "]]") + "]"


def read_json(path: Path) -> dict[str, Any]:
    data = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ValueError(f"{path} must contain a JSON object")
    return data


def unique_file_names(plan: dict[str, Any]) -> list[str]:
    rows = plan.get("source_rows") if isinstance(plan.get("source_rows"), list) else []
    names = sorted({str(row.get("file_name") or "") for row in rows if isinstance(row, dict) and row.get("file_name")})
    if not names:
        raise ValueError("Copy plan does not contain source_rows file_name values.")
    return names


def validate_plan(plan: dict[str, Any], *, expected_base_id: str | None, expected_target_table: str | None) -> tuple[list[str], dict[str, Any]]:
    failures: list[str] = []
    target = plan.get("target") if isinstance(plan.get("target"), dict) else {}
    source_family = plan.get("source_family") if isinstance(plan.get("source_family"), dict) else {}
    summary = plan.get("summary") if isinstance(plan.get("summary"), dict) else {}
    target_collisions = plan.get("target_collisions") if isinstance(plan.get("target_collisions"), dict) else {}
    target_table = str(target.get("table") or "")
    source_table = str(source_family.get("expected_source_table") or SOURCE_BY_TARGET.get(target_table) or "")
    expected_source_table = SOURCE_BY_TARGET.get(target_table)

    if plan.get("schema") != "onec_saved_state_copy_plan.v1":
        failures.append("schema must be onec_saved_state_copy_plan.v1")
    if plan.get("status") != "plan_ready" or plan.get("ready_to_copy") is not True:
        failures.append("copy plan must be plan_ready and ready_to_copy=true")
    if expected_base_id is not None and plan.get("base_id") != expected_base_id:
        failures.append(f"base_id must be {expected_base_id}")
    if target_table not in SOURCE_BY_TARGET:
        failures.append("target.table must be ConfigSave or ConfigCASSave")
    if expected_target_table is not None and target_table != expected_target_table:
        failures.append(f"target.table must be {expected_target_table}")
    if not expected_source_table or source_table != expected_source_table:
        failures.append(f"source family must match target table: {expected_source_table} -> {target_table}")
    if source_family.get("valid") is not True:
        failures.append("source_family.valid must be true")
    source_tables = source_family.get("source_tables") if isinstance(source_family.get("source_tables"), list) else []
    if expected_source_table and any(table != expected_source_table for table in source_tables):
        failures.append(f"source_family.source_tables must contain only {expected_source_table}")
    if target_collisions.get("status") != "clear":
        failures.append("target_collisions.status must be clear")
    if target_collisions.get("rows"):
        failures.append("target_collisions.rows must be empty")
    found_rows = summary.get("found_source_storage_rows")
    if not isinstance(found_rows, int) or found_rows <= 0:
        failures.append("summary.found_source_storage_rows must be a positive integer")

    return failures, {
        "base_id": plan.get("base_id"),
        "target_table": target_table,
        "source_table": source_table,
        "expected_insert_rows": found_rows,
        "source_file_names": unique_file_names(plan) if not failures else [],
    }


def build_sql(metadata: dict[str, Any]) -> str:
    source_table = str(metadata["source_table"])
    target_table = str(metadata["target_table"])
    expected_rows = int(metadata["expected_insert_rows"])
    file_names = [str(name) for name in metadata["source_file_names"]]
    column_list = ", ".join(bracket_name(column) for column in COPY_COLUMNS)
    source_columns = ", ".join(f"s.{bracket_name(column)}" for column in COPY_COLUMNS)
    values = ",\n    ".join(f"({sql_literal(name)})" for name in file_names)
    return "\n".join([
        "-- Generated by scripts/prepare_1c_saved_state_copy_sql.py.",
        "-- Review before execution. This script prepares a saved-state working copy only.",
        "SET XACT_ABORT ON;",
        "BEGIN TRANSACTION;",
        "",
        "DECLARE @planned TABLE ([FileName] nvarchar(260) NOT NULL PRIMARY KEY);",
        "INSERT INTO @planned ([FileName]) VALUES",
        f"    {values};",
        "",
        "IF EXISTS (",
        f"    SELECT 1 FROM dbo.{bracket_name(target_table)} AS t",
        "    INNER JOIN @planned AS p ON p.[FileName] = t.[FileName]",
        ")",
        "BEGIN",
        f"    SELECT t.[FileName], t.[PartNo], t.[DataSize] FROM dbo.{bracket_name(target_table)} AS t",
        "    INNER JOIN @planned AS p ON p.[FileName] = t.[FileName]",
        "    ORDER BY t.[FileName], t.[PartNo];",
        "    ROLLBACK TRANSACTION;",
        "    THROW 51001, 'Target save layer already contains one or more planned FileName values.', 1;",
        "END;",
        "",
        f"INSERT INTO dbo.{bracket_name(target_table)} ({column_list})",
        f"SELECT {source_columns}",
        f"FROM dbo.{bracket_name(source_table)} AS s",
        "INNER JOIN @planned AS p ON p.[FileName] = s.[FileName];",
        "",
        f"IF @@ROWCOUNT <> {expected_rows}",
        "BEGIN",
        "    ROLLBACK TRANSACTION;",
        "    THROW 51002, 'Copied row count did not match the reviewed copy plan.', 1;",
        "END;",
        "",
        "COMMIT TRANSACTION;",
        "",
    ])


def main() -> int:
    parser = argparse.ArgumentParser(description="Generate a reviewed SQL script for preparing a 1C saved-state copy from a copy-plan report. Does not execute SQL.")
    parser.add_argument("--plan", type=Path, default=Path("reports/1c-sql/upo_test/saved-state-copy-plan.json"))
    parser.add_argument("--sql-out", type=Path, required=True)
    parser.add_argument("--report", type=Path)
    parser.add_argument("--expected-base-id")
    parser.add_argument("--expected-target-table", choices=sorted(SOURCE_BY_TARGET))
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()

    report: dict[str, Any] = {
        "schema": "onec_saved_state_copy_sql_plan.v1",
        "plan_path": str(args.plan),
        "sql_path": str(args.sql_out),
        "read_only": True,
        "sql_write_performed": False,
        "status": "error",
        "failures": [],
    }
    try:
        plan = read_json(args.plan)
        failures, metadata = validate_plan(plan, expected_base_id=args.expected_base_id, expected_target_table=args.expected_target_table)
        report.update({
            "base_id": metadata.get("base_id"),
            "target_table": metadata.get("target_table"),
            "source_table": metadata.get("source_table"),
            "expected_insert_rows": metadata.get("expected_insert_rows"),
            "source_file_names": metadata.get("source_file_names"),
            "failures": failures,
        })
        if failures:
            report["status"] = "blocked"
        else:
            sql_text = build_sql(metadata)
            args.sql_out.parent.mkdir(parents=True, exist_ok=True)
            args.sql_out.write_text(sql_text, encoding="utf-8")
            report["status"] = "ready"
            report["sql_sha1_hint"] = "review_file_contents"
    except Exception as exc:
        report["failures"] = [str(exc)]

    if args.report:
        args.report.parent.mkdir(parents=True, exist_ok=True)
        args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    if args.json or report["status"] != "ready":
        print(json.dumps(report, ensure_ascii=False, indent=2))
    else:
        print(f"OK: saved-state copy SQL script written to {args.sql_out}")
    return 0 if report["status"] == "ready" else 1


if __name__ == "__main__":
    raise SystemExit(main())