call-center/scripts/track9_check.py

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import sys
import time

import httpx
from sqlalchemy import create_engine, text

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from services.shared.security import issue_app_token


def _load_env_file(path: str) -> None:
    env_path = Path(path).expanduser().resolve()
    if not env_path.exists() or not env_path.is_file():
        return
    for line in env_path.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith("#") or "=" not in stripped:
            continue
        key, value = stripped.split("=", 1)
        key = key.strip()
        if key and key not in os.environ:
            os.environ[key] = value.strip()


def _bool_env(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def _ops_headers() -> dict[str, str]:
    if _bool_env("ALLOW_LEGACY_HEADER_AUTH", True):
        return {"X-User": "admin", "X-Role": "admin"}
    token = issue_app_token(
        subject="ops:track9-check",
        username="track9-check",
        role="admin",
        auth_source="service",
        provider="track9-script",
        ttl_seconds=300,
    )
    return {"Authorization": f"Bearer {token}"}


def _get_json(url: str) -> dict:
    with httpx.Client(timeout=10, trust_env=False) as client:
        response = client.get(url, headers=_ops_headers())
        response.raise_for_status()
        return response.json()


def _wait_for_ami_connected(base_url: str, retries: int = 8, delay_seconds: float = 1.0) -> dict:
    last_status: dict = {}
    for _ in range(retries):
        status = _get_json(f"{base_url.rstrip('/')}/proxy/asterisk-bridge/asterisk/status")
        last_status = status
        if status.get("ami_connected"):
            return status
        time.sleep(delay_seconds)
    return last_status


def _table_exists(conn, name: str) -> bool:
    try:
        conn.execute(text(f"SELECT 1 FROM {name} LIMIT 1"))
        return True
    except Exception:  # noqa: BLE001
        return False


def _scalar(conn, sql: str) -> int:
    value = conn.execute(text(sql)).scalar()
    return int(value or 0)


def main() -> int:
    parser = argparse.ArgumentParser(description="Track 9 Asterisk integration acceptance validator")
    parser.add_argument("--base-url", default="http://127.0.0.1:8080")
    parser.add_argument("--env-file", default=".env.production")
    parser.add_argument("--database-url", required=True)
    parser.add_argument("--require-recording", action="store_true")
    args = parser.parse_args()

    _load_env_file(args.env_file)

    failures: list[str] = []

    health = _get_json(f"{args.base_url.rstrip('/')}/proxy/asterisk-bridge/health")
    status = _wait_for_ami_connected(args.base_url)
    if health.get("status") != "ok":
        failures.append("Bridge health is not ok")
    if not status.get("ami_connected"):
        failures.append("AMI is not connected")

    engine = create_engine(args.database_url, future=True)
    with engine.begin() as conn:
        for table in ("asterisk_event_log", "asterisk_call_links", "voice_events"):
            if not _table_exists(conn, table):
                failures.append(f"Missing table: {table}")

        started = _scalar(
            conn,
            "SELECT COUNT(*) FROM asterisk_event_log WHERE ami_event_name = 'MVPCCCallStarted' AND forward_status = 'forwarded'",
        )
        call_started = _scalar(
            conn,
            "SELECT COUNT(*) FROM voice_events WHERE event_type = 'call.started' AND payload_json LIKE '%\"source\": \"asterisk\"%'",
        )
        links = _scalar(conn, "SELECT COUNT(*) FROM asterisk_call_links")
        stale_failures = _scalar(
            conn,
            "SELECT COUNT(*) FROM asterisk_event_log WHERE forward_status = 'failed'",
        )
        recordings = _scalar(
            conn,
            "SELECT COUNT(*) FROM asterisk_event_log WHERE ami_event_name = 'MVPCCRecordingReady' AND recording_id IS NOT NULL",
        )

    if started < 1:
        failures.append("No forwarded MVPCCCallStarted events")
    if call_started < 1:
        failures.append("No platform call.started events from Asterisk source")
    if links < 1:
        failures.append("No linked Asterisk calls")
    if stale_failures > 0:
        failures.append(f"Failed bridge events present: {stale_failures}")
    if args.require_recording and recordings < 1:
        failures.append("No uploaded recordings linked to Asterisk events")

    if failures:
        print("[FAIL] Track 9 validation failed")
        for item in failures:
            print(f"- {item}")
        return 1

    print("[PASS] Track 9 validation checks passed")
    print(f"- started_events: {started}")
    print(f"- platform_call_started: {call_started}")
    print(f"- call_links: {links}")
    print(f"- recording_events_with_upload: {recordings}")
    return 0


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