#!/usr/bin/env python3
"""Engine/config boundary scan.

Enforces the rule in ``docs/principles/config-driven-clean-structure.md`` and
``docs/principles/engine-config-red-lines.md``: reusable engine code carries no
source or product *facts*. Onboarding the next source/product must mean editing
Git-authored config (``source.yaml``, per-product contract dirs), never editing
shared engine code.

The forbidden literals are not hand-maintained. They are *derived from the
config tree itself*: every directory name under a ``sources/`` or ``products/``
parent is an identifier that must not appear as a literal token inside engine
code. Add a source — its name automatically becomes a protected literal.

Scope
-----
Scanned ("engine") code:
  * every directory named ``engine/`` under ``ingestion/`` and ``transform/``
  * ``orchestration/airflow/engine/`` (shared orchestration compiler/operator code)
  * ``ingestion/common/`` (shared, source-agnostic modules)

Excluded within those roots: ``tests/``, ``fixtures/``, ``target/`` (dbt build
output), ``logs/``, ``__pycache__/``. Tests and fixtures are exempt by the same
convention the principle uses — they legitimately name specific sources.

A hit is an identifier matched as a whole token (word boundaries) in a ``.py``,
``.sql``, or ``.jinja`` file under a scanned root.

Escape hatches (both must carry a reason):
  * inline ``# boundary-scan: allow <reason>`` (or ``-- boundary-scan: allow``)
    at the end of the offending line.
  * an entry in ``scripts/boundary_scan_allow.txt`` — for whole-file or
    file+identifier exceptions that pre-date the gate. Each recorded exception
    is a *visible* piece of debt, not a silent exclusion.

Some source/product dir names are also common English words (``core``,
``events``, ``customers``) and cannot be enforced as bare tokens without noise.
Those are listed in ``AMBIGUOUS`` below and skipped; the list is small and
reviewed, and the principle still covers them in prose.

Usage
-----
    python scripts/boundary_scan.py            # scan; exit 1 on violations
    python scripts/boundary_scan.py --list      # print derived identifiers + roots
"""

from __future__ import annotations

import re
import sys
from dataclasses import dataclass
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]

# Trees that contain both engine code and the sources/products config tree.
SEARCH_TREES = ("ingestion", "transform")

# Extra shared roots that must obey the engine/config boundary even though they
# are outside the ingestion/transform trees scanned above.
EXTRA_ENGINE_ROOTS = ("ingestion/common", "orchestration/airflow/engine")

CODE_SUFFIXES = (".py", ".sql", ".jinja")
PRUNE_DIRS = {"tests", "fixtures", "target", "logs", "__pycache__", ".venv", ".git"}

# Identifiers that are real ``sources/`` or ``products/`` dir names but are also
# common English words; enforcing them as bare tokens would flag ordinary prose.
# Kept small and explicit. The prose principle still governs them.
AMBIGUOUS = {"core", "events", "customers"}

ALLOW_FILE = REPO / "scripts" / "boundary_scan_allow.txt"

INLINE_ALLOW = re.compile(r"(?:#|--)\s*boundary-scan:\s*allow\s+\S")


class AllowlistError(ValueError):
    """Raised when an allowlist entry is missing its required reason."""


@dataclass(frozen=True)
class Hit:
    path: Path
    line_no: int
    identifier: str
    text: str


def derive_identifiers() -> set[str]:
    """Every immediate child of a ``sources/`` or ``products/`` directory.

    Names starting with ``_`` (demo/scratch fixtures like ``_lmdemo``) are
    skipped, as are the ambiguous common-word names.
    """
    ids: set[str] = set()
    for tree in SEARCH_TREES:
        base = REPO / tree
        if not base.is_dir():
            continue
        for parent in base.rglob("*"):
            if parent.name in ("sources", "products") and parent.is_dir():
                if any(part in PRUNE_DIRS for part in parent.relative_to(REPO).parts):
                    continue
                for child in parent.iterdir():
                    if child.is_dir() and not child.name.startswith("_"):
                        ids.add(child.name)
    return {i for i in ids if i not in AMBIGUOUS}


def engine_roots() -> list[Path]:
    roots: list[Path] = []
    for tree in SEARCH_TREES:
        base = REPO / tree
        if not base.is_dir():
            continue
        for d in base.rglob("engine"):
            if d.is_dir() and not any(p in PRUNE_DIRS for p in d.relative_to(REPO).parts):
                roots.append(d)
    for extra in EXTRA_ENGINE_ROOTS:
        p = REPO / extra
        if p.is_dir():
            roots.append(p)
    return sorted(set(roots))


def iter_code_files(root: Path):
    for path in root.rglob("*"):
        if path.suffix not in CODE_SUFFIXES:
            continue
        if any(part in PRUNE_DIRS for part in path.relative_to(root).parts):
            continue
        yield path


def load_allowlist() -> list[tuple[str, str | None]]:
    """Parse the allowlist into ``(path_substr, identifier_or_None)`` rules."""
    rules: list[tuple[str, str | None]] = []
    if not ALLOW_FILE.exists():
        return rules
    for n, raw in enumerate(ALLOW_FILE.read_text().splitlines(), 1):
        line, sep, reason = raw.partition("#")
        line = line.strip()
        if not line:
            continue
        if not sep or not reason.strip():
            raise AllowlistError(f"{ALLOW_FILE.relative_to(REPO)}:{n}: missing reason")
        if ":" in line:
            path_part, ident = line.split(":", 1)
            rules.append((path_part.strip(), ident.strip() or None))
        else:
            rules.append((line, None))
    return rules


def is_allowlisted(rel_path: str, identifier: str, rules) -> bool:
    for path_substr, ident in rules:
        if path_substr in rel_path and (ident is None or ident == identifier):
            return True
    return False


def scan() -> list[Hit]:
    identifiers = derive_identifiers()
    patterns = {i: re.compile(rf"\b{re.escape(i)}\b") for i in identifiers}
    rules = load_allowlist()
    hits: list[Hit] = []
    for root in engine_roots():
        for path in iter_code_files(root):
            rel = str(path.relative_to(REPO))
            try:
                lines = path.read_text(encoding="utf-8").splitlines()
            except (UnicodeDecodeError, OSError):
                continue
            for n, text in enumerate(lines, 1):
                if INLINE_ALLOW.search(text):
                    continue
                for ident, pat in patterns.items():
                    if pat.search(text) and not is_allowlisted(rel, ident, rules):
                        hits.append(Hit(path, n, ident, text.strip()))
    return hits


def main(argv: list[str]) -> int:
    if "--list" in argv:
        print("Derived protected identifiers:")
        for i in sorted(derive_identifiers()):
            print(f"  {i}")
        print(f"\nAmbiguous (skipped, common words): {sorted(AMBIGUOUS)}")
        print("\nEngine roots scanned:")
        for r in engine_roots():
            print(f"  {r.relative_to(REPO)}")
        return 0

    try:
        hits = scan()
    except AllowlistError as e:
        print(f"boundary-scan: invalid allowlist entry — {e}")
        return 1
    if not hits:
        print("boundary-scan: OK — no source/product literals in engine code.")
        return 0

    print(f"boundary-scan: {len(hits)} violation(s) — source/product literal in engine code:\n")
    for h in sorted(hits, key=lambda x: (str(x.path), x.line_no)):
        rel = h.path.relative_to(REPO)
        print(f"  {rel}:{h.line_no}: '{h.identifier}'")
        print(f"      {h.text}")
    print(
        "\nMove the fact to config (source.yaml / per-product contract dir) and keep the\n"
        "engine generic, or record a justified exception in scripts/boundary_scan_allow.txt\n"
        "or with an inline '# boundary-scan: allow <reason>'. See\n"
        "docs/principles/engine-config-red-lines.md."
    )
    return 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))