fix(3-6): accept Integer default for non-Boolean cells

scripts/check_semantic.py

@@ -1,116 +1,200 @@
-"""Check DPM-XL expression semantics for all rules in a validations xlsx."""
+"""Check DPM-XL expression semantics for all OperationVersion rules in the DB."""
 
 import argparse
+import csv
 import sys
+import time
+from datetime import datetime
 from pathlib import Path
-from typing import Any
 
-import openpyxl
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, text
 from sqlalchemy.orm import sessionmaker
 
 from dpmcore.services.semantic import SemanticService
 
-_DEFAULT_XLSX = (
-    Path(__file__).resolve().parents[1]
-    / "tests"
-    / "fixtures"
-    / "validations_export.xlsx"
-)
 _DEFAULT_DB = (
     Path(__file__).resolve().parents[1] / "tests" / "fixtures" / "test_data.db"
 )
 
-
-def _check_rows(
-    rows: list[dict[str, Any]],
-    svc: SemanticService,
-) -> tuple[list[tuple[str, str, str, str, str]], int]:
-    items = [
-        (
-            str(row.get("Code") or ""),
-            col,
-            str(row.get("StartRelease") or "").strip(),
-            str(row.get(col)).strip(),
-        )
-        for row in rows
-        for col in ("Expression", "Precondition")
-        if row.get(col) is not None and str(row.get(col)).strip()
-    ]
-    total = len(items)
-    width = len(str(total))
-    failures: list[tuple[str, str, str, str, str]] = []
-    for i, (code, col, release, expr) in enumerate(items, start=1):
-        result = svc.validate(expr, release_code=release or None)
-        prefix = f"[{i:{width}}/{total}] {code} | {col} | release {release}"
-        if result.is_valid:
-            print(f"{prefix} | PASS")
-        else:
-            error = result.error_message or ""
-            print(f"{prefix} | FAIL: {error}")
-            print(f"  Expression: {expr}")
-            failures.append((code, col, release, expr, error))
-    return failures, total
+WIDTH = 80
+BAR = "=" * WIDTH
+
+_TTY = sys.stdout.isatty()
+_GREEN = "\033[32m" if _TTY else ""
+_RED = "\033[31m" if _TTY else ""
+_DIM = "\033[2m" if _TTY else ""
+_RESET = "\033[0m" if _TTY else ""
+
+
+def _fmt_size(path: Path) -> str:
+    mb = path.stat().st_size / (1024 * 1024)
+    return f"{mb:.1f} MB"
+
+
+def _fmt_duration(seconds: float) -> str:
+    m, s = divmod(int(seconds), 60)
+    h, m = divmod(m, 60)
+    if h:
+        return f"{h}h {m}m {s}s"
+    if m:
+        return f"{m}m {s}s"
+    return f"{s}s"
+
+
+
+def _load_rows(db_path: Path):
+    engine = create_engine(f"sqlite:///{db_path}")
+    with engine.connect() as conn:
+        rows = conn.execute(
+            text(
+                """
+                SELECT
+                    ov.OperationVID,
+                    o.Code,
+                    r.Code  AS ReleaseCode,
+                    ov.Expression
+                FROM OperationVersion ov
+                JOIN   Operation o ON o.OperationID = ov.OperationID
+                LEFT JOIN Release r ON r.ReleaseID  = ov.StartReleaseID
+                WHERE  ov.Expression IS NOT NULL
+                  AND  trim(ov.Expression) != ''
+                ORDER BY o.Code, ov.OperationVID
+                """
+            )
+        ).fetchall()
+    engine.dispose()
+    return rows
 
 
 def main() -> int:
-    """Run semantic checks and print results to stdout."""
     parser = argparse.ArgumentParser(
-        description="Check DPM-XL expression semantics in a validations xlsx.",
-    )
-    parser.add_argument(
-        "--xlsx",
-        default=str(_DEFAULT_XLSX),
-        help=f"Path to the validations xlsx (default: {_DEFAULT_XLSX})",
+        description="Check DPM-XL expression semantics for all rules in the DB.",
     )
     parser.add_argument(
         "--db",
-        default=str(_DEFAULT_DB),
+        type=Path,
+        default=_DEFAULT_DB,
         help=f"Path to the SQLite DPM database (default: {_DEFAULT_DB})",
     )
+    parser.add_argument(
+        "--csv",
+        type=Path,
+        default=None,
+        help="Path for the failures CSV output (default: <db_stem>_failures.csv next to the DB)",
+    )
     args = parser.parse_args()
 
-    print("Semantic check")
-    print(f"  xlsx: {args.xlsx}")
-    print(f"  db:   {args.db}")
+    db_path: Path = args.db.resolve()
+    if not db_path.exists():
+        print(f"ERROR: database not found: {db_path}", file=sys.stderr)
+        return 1
+
+    csv_path: Path = (
+        args.csv.resolve()
+        if args.csv
+        else db_path.parent / f"{db_path.stem}_failures.csv"
+    )
+
+    # ------------------------------------------------------------------
+    # Header
+    # ------------------------------------------------------------------
+    print(f"\n{BAR}")
+    print("  DPM Semantic Validation Runner")
+    print(BAR)
+
+    rows = _load_rows(db_path)
+    total = len(rows)
+    total_str = f"{total:,}"
+    counter_width = len(str(total))
+
+    print(f"  Database   : {db_path}")
+    print(f"  Size       : {_fmt_size(db_path)}")
+    print(f"  Validations: {total:,}")
+    print(f"  CSV output : {csv_path}")
+    print(f"  Started    : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    print(BAR)
     print()
 
-    engine = create_engine(f"sqlite:///{args.db}")
-    Session = sessionmaker(bind=engine)
-    session = Session()
-
-    try:
-        wb = openpyxl.load_workbook(args.xlsx)
-        ws = wb["Validations"]
-        headers = [cell.value for cell in ws[1]]
-        rows: list[dict[str, Any]] = [
-            dict(zip(headers, row, strict=False))
-            for row in ws.iter_rows(min_row=2, values_only=True)
-        ]
-
-        svc = SemanticService(session)
-        failures, total = _check_rows(rows, svc)
-
-        passed = total - len(failures)
-        failed = len(failures)
-        print(
-            f"\n{total} expressions checked — {passed} passed, {failed} failed"
-        )
-
-        if failures:
-            width_f = len(str(len(failures)))
-            print("\nFailed expressions:")
-            for j, (code, col, release, expr, error) in enumerate(failures, 1):
-                print(
-                    f"  [{j:{width_f}}/{len(failures)}]"
-                    f" {code} | {col} | release {release} | FAIL: {error}"
-                )
-                print(f"    Expression: {expr}")
-
-        return 1 if failures else 0
-    finally:
-        session.close()
-        engine.dispose()
+    # ------------------------------------------------------------------
+    # Run
+    # ------------------------------------------------------------------
+    engine = create_engine(f"sqlite:///{db_path}")
+    session = sessionmaker(bind=engine)()
+    svc = SemanticService(session)
+
+    passed = 0
+    failures = []
+
+    t_start = time.monotonic()
+
+    for i, (operation_vid, op_code, release_code, expression) in enumerate(rows, 1):
+        label = op_code or str(operation_vid)
+        release = release_code or ""
+
+        result = svc.validate(expression.strip(), release_code=release or None)
+
+        idx = f"[{i:{counter_width},}/{total_str}]"
+        preview = expression.replace("\n", " ").strip()
+        line = f"{idx} {label:<12} {_DIM}{preview}{_RESET}"
+
+        if result.is_valid:
+            passed += 1
+            print(f"{line}  {_GREEN}PASS{_RESET}")
+        else:
+            err = result.error_message or "(no message)"
+            err_code = result.error_code or ""
+            failures.append((i, operation_vid, label, release, expression.strip(), err_code, err))
+            print(f"{line}  {_RED}FAIL{_RESET}")
+            print(f"  {_DIM}└─{_RESET} {err}")
+
+    session.close()
+    engine.dispose()
+
+    elapsed = time.monotonic() - t_start
+    failed = len(failures)
+    pct_pass = passed / total * 100 if total else 0.0
+    pct_fail = failed / total * 100 if total else 0.0
+
+    # ------------------------------------------------------------------
+    # Summary
+    # ------------------------------------------------------------------
+    print()
+    print(BAR)
+    print("  SUMMARY")
+    print(BAR)
+    print(f"  Database   : {db_path}")
+    print(f"  Size       : {_fmt_size(db_path)}")
+    print(f"  Total      : {total:,}")
+    print(f"  {_GREEN}Passed{_RESET}     : {passed:,}  ({pct_pass:.1f}%)")
+    print(f"  {_RED}Failed{_RESET}     : {failed:,}  ({pct_fail:.1f}%)")
+    print(f"  Duration   : {_fmt_duration(elapsed)}")
+    print(BAR)
+
+    # ------------------------------------------------------------------
+    # Failures list
+    # ------------------------------------------------------------------
+    if failures:
+        print()
+        print(BAR)
+        print("  FAILURES")
+        print(BAR)
+        for idx, operation_vid, label, release, expression, err_code, error in failures:
+            idx_str = f"[{idx:{counter_width},}/{total_str}]"
+            release_info = f" | release: {release}" if release else ""
+            print(f"\n{idx_str} {label}{release_info}")
+            print(f"  Expression : {expression}")
+            print(f"  Error      : {error}")
+        print()
+        print(BAR)
+
+        with csv_path.open("w", newline="", encoding="utf-8") as fh:
+            writer = csv.writer(fh)
+            writer.writerow(["index", "operation_vid", "code", "release", "error_code", "error", "expression"])
+            for idx, operation_vid, label, release, expression, err_code, error in failures:
+                writer.writerow([idx, operation_vid, label, release, err_code, error, expression])
+        print(f"\n  CSV written: {csv_path}")
+
+    return 1 if failures else 0
 
 
 if __name__ == "__main__":