#!/usr/bin/env python3
"""Static audit for fragile SQLAlchemy text() patterns in ClientFlow.

This is intentionally conservative: it does not prove a bug, it flags code that
should be reviewed before deploy because PostgreSQL may be unable to infer bind
parameter types in JSONB/NULL/ANY contexts.
"""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path
from typing import Dict, List

ROOT = Path(__file__).resolve().parents[1]
SCAN_DIRS = [ROOT / "app", ROOT / "scripts"]

PATTERNS = [
    ("jsonb_build_object_with_bind", re.compile(r"jsonb_build_object\s*\([^)]*:[a-zA-Z_][a-zA-Z0-9_]", re.I | re.S), "Prefer metadata || CAST(:metadata_patch AS jsonb) or cast each bind explicitly."),
    ("bind_is_null_without_cast", re.compile(r"\([^)]*:[a-zA-Z_][a-zA-Z0-9_]*\s+IS\s+NULL", re.I | re.S), "Use CAST(:param AS text/uuid/...) IS NULL when parameter can be null."),
    ("any_bind_array", re.compile(r"\bANY\s*\(\s*:[a-zA-Z_][a-zA-Z0-9_]*\s*\)", re.I), "Prefer expanding bindparams or cast array type explicitly."),
    ("metadata_jsonb_concat_without_cast", re.compile(r"metadata\s*=\s*COALESCE\([^\n]+\)\s*\|\|\s*(?!CAST\(:metadata_patch\s+AS\s+JSONB\))", re.I), "Prefer COALESCE(metadata,'{}'::jsonb) || CAST(:metadata_patch AS jsonb)."),
]

ALLOW_HINTS = [
    "CAST(:",
    "CAST(%(",
    "CAST(:metadata_patch AS JSONB)",
    "CAST(:metadata_patch AS jsonb)",
]


def iter_files():
    for base in SCAN_DIRS:
        if not base.exists():
            continue
        for path in base.rglob("*.py"):
            if "__pycache__" in path.parts:
                continue
            yield path


def line_no(text: str, pos: int) -> int:
    return text.count("\n", 0, pos) + 1


def excerpt(text: str, pos: int, size: int = 240) -> str:
    start = max(0, pos - 80)
    end = min(len(text), pos + size)
    return re.sub(r"\s+", " ", text[start:end]).strip()


def main() -> int:
    findings: List[Dict[str, str]] = []
    for path in iter_files():
        text = path.read_text(errors="ignore")
        for code, pattern, advice in PATTERNS:
            for match in pattern.finditer(text):
                sample = excerpt(text, match.start())
                # If the local expression already has explicit casts in the same
                # small sample, lower severity but still report as review.
                severity = "WARN" if any(h in sample for h in ALLOW_HINTS) else "REVIEW"
                findings.append({
                    "severity": severity,
                    "code": code,
                    "file": str(path.relative_to(ROOT)),
                    "line": str(line_no(text, match.start())),
                    "excerpt": sample,
                    "advice": advice,
                })
    print("ClientFlow SQL safety static audit")
    print("=" * 80)
    print(f"findings: {len(findings)}")
    for item in findings[:80]:
        print(f"{item['severity']} | {item['code']} | {item['file']}:{item['line']}")
        print(f"  {item['excerpt']}")
        print(f"  advice: {item['advice']}")
    if len(findings) > 80:
        print(f"... {len(findings) - 80} more findings not printed")
    out_dir = ROOT / "audit_reports"
    out_dir.mkdir(exist_ok=True)
    out = out_dir / "sql_safety_findings.json"
    out.write_text(json.dumps(findings, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"JSON: {out}")
    # Static findings are warnings by design, not deployment blockers.
    return 1 if findings else 0


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