"""Read-only source census; output is an audit artifact, never a runtime registry.

Run from Core with ``python -m scripts.audit_capability_modernization --out ...``.
Sibling Cloud/Dash paths can be overridden. No DB, credentials or provider calls.
Static action registration is evidence of dispatch wiring, not a provider smoke test.
"""

from __future__ import annotations

import argparse
import json
import re
from collections import Counter
from pathlib import Path

from src.domain.app_blocks.catalogs import default_known_packs, default_primitives
from src.domain.app_blocks.registry import load_default_registry
from src.domain.app_blocks.resolver import resolve_block
from src.domain.connectors.catalog import CONNECTORS, slugs_implementing
from src.domain.connectors.invoke_capabilities import (
    CAPABILITY_CONVERSION_PATTERN,
    INVOKE_CAPABILITIES,
    capability_surface,
)
from src.domain.feature_packs.catalog_v2 import CATALOG_OVERLAYS
from src.domain.feature_packs.normalize import load_pack_contract
from src.domain.starter_library.catalog import list_contracts

ROOT = Path(__file__).resolve().parents[1]
SKIP_DIRS = {"node_modules", ".git", ".nuxt", ".output", "dist", "build", "snapshots"}
SOURCE_SUFFIXES = {".ts", ".tsx", ".js", ".jsx", ".vue", ".html", ".json", ".php", ".md"}


def source_files(root: Path):
    for path in sorted(root.rglob("*")):
        if path.is_file() and path.suffix in SOURCE_SUFFIXES:
            if not SKIP_DIRS.intersection(path.relative_to(root).parts):
                yield path


def host_exists(root: Path, implementation: str | None) -> bool:
    if not implementation:
        return False
    filename, _, symbol = implementation.partition("#")
    path = root / filename
    if not path.is_file():
        return False
    if not symbol:
        return True
    content = path.read_text(encoding="utf-8")
    return bool(re.search(r"\b(?:function|const|class)\s+" + re.escape(symbol) + r"\b", content))


def registered_actions(cloud: Path) -> tuple[dict, list[str]]:
    """Inspect only modules wired into the registry barrel; don't count orphan files."""
    directory = cloud / "apps/pc-processor/src/handlers/connectors"
    barrel = (directory / "index.ts").read_text(encoding="utf-8")
    modules = re.findall(
        r'export\s*\{\s*ACTIONS\s+as\s+\w+\s*\}\s*from\s*[\'"]\./([^\'"]+)', barrel
    )
    result = {}
    for module in sorted(set(modules)):
        path = directory / f"{module}.ts"
        content = path.read_text(encoding="utf-8")
        # Both repository formats: object literals and def(slug, provider, handler).
        for match in re.finditer(
            r'^\s*[\'"]([a-z0-9_]+\.[a-z0-9_.]+)[\'"]\s*:\s*(?:\{|def\()', content, re.M
        ):
            result[match[1]] = {
                "source": f"Cloud/{path.relative_to(cloud).as_posix()}",
                "line": content.count("\n", 0, match.start()) + 1,
                "verification": "registered_source_only; provider execution not tested",
            }
    return result, sorted(k for k, v in Counter(modules).items() if v > 1)


def starter_inventory() -> list[dict]:
    patterns = {
        "sheets_path": r"google_sheets|sheets\.googleapis\.com",
        "direct_core_reference": r"beapipie\.p2hs\.com|CORE_API_BASE_URL|P2HS_BACKEND_URL",
        "cloud_reference": r"cloudgate\.p2hs\.com|CLOUD_GATEWAY|CLOUD_API_BASE",
        "provider_endpoint_or_sdk": (
            r"api\.stripe\.com|api\.openai\.com|docs\.googleapis\.com|"
            r"@aws-sdk|@googleapis|from ['\"](?:stripe|openai)['\"]"
        ),
        "legacy_dispatch_reference": r"integration\.execute",
    }
    result = []
    for directory in sorted((ROOT / "boilerplates").iterdir()):
        if not directory.is_dir() or directory.name in SKIP_DIRS:
            continue
        findings = []
        files = list(source_files(directory))
        for path in files:
            for number, line in enumerate(
                path.read_text(encoding="utf-8", errors="replace").splitlines(), 1
            ):
                for kind, pattern in patterns.items():
                    if re.search(pattern, line):
                        # Record locations only, never source lines or potential secret values.
                        findings.append(
                            {
                                "kind": kind,
                                "file": path.relative_to(ROOT).as_posix(),
                                "line": number,
                            }
                        )
        result.append(
            {
                "id": directory.name,
                "source_files_scanned": len(files),
                "findings": findings,
                "classification": "review_required" if findings else "no_scanned_markers",
                "build_verified": False,
                "scope": "local scaffold; remote DB/GitHub template catalog not enumerated",
                "migration_needed": "Review server/browser context and runtime exceptions first",
            }
        )
    return result


def build_audit(cloud: Path, dash: Path) -> dict:
    cloud_actions, duplicate_exports = registered_actions(cloud)
    capabilities = {
        key: capability_surface(key, list(slugs_implementing(key))) for key in INVOKE_CAPABILITIES
    }
    registry = load_default_registry()
    packs = {
        key: load_pack_contract(key, {}).model_dump(mode="json") for key in sorted(CATALOG_OVERLAYS)
    }
    blocks = []
    for block in registry.all():
        required_packs, primitives, nested, gaps, configs, _optionals = resolve_block(
            block,
            registry,
            primitives=default_primitives(),
            known_packs=default_known_packs(),
        )
        hosts = []
        for framework in ("html", "nuxt_app", "react_app"):
            root = ROOT / "boilerplates" / framework
            host_file = root / "p2hs/app-block-hosts.json"
            entries = (
                json.loads(host_file.read_text(encoding="utf-8")).get("hosts", [])
                if host_file.exists()
                else []
            )
            by_slot = {entry["slot"]: entry for entry in entries if entry["block"] == block.id}
            for slot in block.slots:
                if slot.type != "ui_slot":
                    continue
                implementation = by_slot.get(slot.id, {}).get("implementation")
                hosts.append(
                    {
                        "framework": framework,
                        "slot": slot.id,
                        "implementation": implementation,
                        "exists": host_exists(root, implementation),
                    }
                )
        provider_bindings = [binding.model_dump(mode="json") for binding in block.provider_bindings]
        pack_bindings = [
            binding
            for key in required_packs
            for binding in packs.get(key, {}).get("provider_bindings", [])
        ]
        classification = "READY"
        if any(g.severity == "build_blocking" for g in gaps):
            classification = "CAPABILITY_GAP"
        elif any(not host["exists"] for host in hosts):
            classification = "BROKEN"
        elif provider_bindings or pack_bindings:
            classification = "PROVIDER_BOUND"
        blocks.append(
            {
                "id": block.id,
                "current_status": block.status,
                "classification": classification,
                "classification_scope": "static composition only; not end-to-end readiness",
                "required_feature_packs": required_packs,
                "required_primitives": primitives,
                "required_app_blocks": nested,
                "optional_requires": [d.model_dump(mode="json") for d in block.optional_requires],
                "required_bindings": [b.model_dump(mode="json") for b in block.bindings],
                "provider_bindings": provider_bindings,
                "pack_provider_bindings": pack_bindings,
                "configuration_requirements": [c.model_dump(mode="json") for c in configs],
                "gaps": [g.model_dump(mode="json") for g in gaps],
                "hosts": hosts,
                "tests": ["tests/test_app_blocks.py", "tests/test_starter_library.py"],
                "migration_needed": "No Sheets dependency; build/pack verification pending",
            }
        )
    connectors = []
    for key, connector in sorted(CONNECTORS.items()):
        actions = []
        for action in connector.actions:
            wired = cloud_actions.get(action.slug)
            if not wired:
                category, reason = "F", "Catalog action has no observed registered Cloud action"
            elif action.implements:
                category, reason = (
                    "A",
                    "Catalog implements existing capability; verify family runtime tests",
                )
            elif key in {"cloudflare_r2", "hubspot", "salesforce", "twilio"}:
                category, reason = (
                    "C",
                    "Family queued for semantic audit; no new verb or normalization claimed",
                )
            else:
                category, reason = (
                    "B",
                    "Retain provider surface pending semantic review; classification provisional",
                )
            actions.append(
                {
                    "slug": action.slug,
                    "classification": category,
                    "reason": reason,
                    "cloud_registration": wired,
                    "implements": list(action.implements),
                    "required_scopes": list(action.required_scopes),
                    "required_entitlements": list(action.required_capabilities),
                }
            )
        connectors.append(
            {
                "provider": key,
                "customer_connector_exists": True,
                "execution_domain": "C",
                "domain_scope": "Catalog path only; vendor P/B observations in audit report",
                "catalog_status": connector.status,
                "client_side": connector.client_side,
                "connection_config_fields": list(connector.config_fields),
                "actions": actions,
                "resolver_capabilities": sorted(
                    {cap for a in connector.actions for cap in a.implements}
                ),
                "bff_status": "shared connector proxy; capabilities use public allowlist",
                "codegen_status": "derived capability defaults; provider surface otherwise",
                "feature_pack_dependencies": [
                    slug
                    for slug, pack in packs.items()
                    if any(key in b.get("providers", []) for b in pack.get("provider_bindings", []))
                ],
                "app_block_dependencies": [
                    b["id"]
                    for b in blocks
                    if any(
                        key in pb.get("providers", [])
                        for pb in b["provider_bindings"] + b["pack_provider_bindings"]
                    )
                ],
                "starter_dependencies": "see starter findings; semantic review pending",
            }
        )
    catalog_slugs = {a.slug for c in CONNECTORS.values() for a in c.actions}
    proxy = dash / "server/api/projects/[projectId]/capabilities/[capability].post.ts"
    proxy_text = proxy.read_text(encoding="utf-8")
    allowlist = re.search(r"const PUBLIC_BFF_PROXY = new Set\(\[(.*?)\]\)", proxy_text, re.S)
    if not allowlist:
        raise ValueError(
            "BFF allowlist shape changed; update audit parser instead of assuming no routes"
        )
    public_verbs = re.findall(r"'([^']+)'", allowlist[1])
    return {
        "schema_version": 1,
        "phase": CAPABILITY_CONVERSION_PATTERN["current_program_phase"],
        "authority": "audit snapshot only; never consumed by runtime",
        "verification": "local sources only; DB/provider/deployment/generated-build checks pending",
        "sheets": {
            "customer_catalog_exists": "google_sheets" in CONNECTORS,
            "registered_cloud_actions": sorted(
                s for s in cloud_actions if s.startswith("google_sheets.")
            ),
            "declared_capabilities": sorted(
                c for c in capabilities if c.startswith("spreadsheet.")
            ),
            "public_bff_capabilities": sorted(
                c for c in public_verbs if c.startswith("spreadsheet.")
            ),
            "verbs_frozen": any(c.startswith("spreadsheet.") for c in capabilities),
        },
        "capabilities": capabilities,
        "public_bff_capabilities": public_verbs,
        "connectors": connectors,
        "cloud_actions_without_catalog": [
            {"slug": s, "classification": "G", **cloud_actions[s]}
            for s in sorted(cloud_actions.keys() - catalog_slugs)
        ],
        "duplicate_cloud_module_exports": duplicate_exports,
        "app_blocks": blocks,
        "feature_pack_overlays": packs,
        "feature_pack_scope": "Code overlays only; DB compositions and installs not read",
        "starter_contracts": [c.model_dump(mode="json") for c in list_contracts()],
        "starters": starter_inventory(),
        "remote_template_catalog": {
            "status": "not_audited",
            "source": "src/domain/templates/repository.py:TemplateRepository.list_all",
            "reason": "DB catalog and referenced GitHub repos are separate from local scaffolds",
        },
    }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--cloud", type=Path, default=ROOT.parent / "P2HSCloud")
    parser.add_argument("--dash", type=Path, default=ROOT.parent / "P2HS-Dash")
    parser.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    audit = build_audit(args.cloud, args.dash)
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(audit, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    print(
        json.dumps(
            {
                "connectors": len(audit["connectors"]),
                "actions": sum(len(c["actions"]) for c in audit["connectors"]),
                "app_blocks": len(audit["app_blocks"]),
                "feature_pack_overlays": len(audit["feature_pack_overlays"]),
                "starters": len(audit["starters"]),
                "sheets": audit["sheets"],
            }
        )
    )


if __name__ == "__main__":
    main()