app.py

import json
import logging
import os
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Any

import requests
from flask import Flask, jsonify, request, send_from_directory
from werkzeug.utils import secure_filename

from ats_client import ATSClient


VERSION = "3.2"
CONFIG_PATH = Path(os.getenv("ATS_PROXY_CONFIG", "config.json"))
SENSITIVE_LOG_FIELDS = {"authorization", "password", "random", "token"}


def load_config(path: Path) -> dict[str, Any]:
    try:
        with path.open("r", encoding="utf-8") as config_file:
            return json.load(config_file)
    except FileNotFoundError as exc:
        raise SystemExit(f"Config file not found: {path}") from exc
    except json.JSONDecodeError as exc:
        raise SystemExit(f"Invalid JSON in config file {path}: {exc}") from exc


CONFIG = load_config(CONFIG_PATH)

storage_conf = CONFIG.get("storage", {})
net_conf = CONFIG.get("network", {})
crm_conf = CONFIG.get("crm", {})
ats_conf = CONFIG.get("ats", {})
logging_conf = CONFIG.get("logging", {})

LOG_LEVEL = getattr(logging, logging_conf.get("level", "INFO").upper(), logging.INFO)
logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
logger = logging.getLogger("ats-proxy")

RECORDS_DIR = Path(storage_conf.get("path", "/app/records"))
LOCAL_IP = net_conf.get("local_ip", "0.0.0.0")
LOCAL_PORT = int(net_conf.get("local_port", 9001))
HEARTBEAT_IP = net_conf.get("heartbeat_ip") or net_conf.get("local_ip") or "127.0.0.1"
HEARTBEAT_PORT = int(net_conf.get("heartbeat_port", LOCAL_PORT))
EXTERNAL_HOST = net_conf.get("external_domain_or_ip") or HEARTBEAT_IP
EXTERNAL_PORT = int(net_conf.get("external_port", LOCAL_PORT))
PROTOCOL = net_conf.get("protocol", "http")

CRM_URL = crm_conf.get("url")
CRM_TOKEN = crm_conf.get("token")

app = Flask(__name__)
ats = ATSClient(CONFIG)
call_registry: dict[str, dict[str, Any]] = {}
registry_lock = threading.Lock()


def redact_sensitive(value: Any) -> Any:
    """Return a log-safe copy of nested data."""
    if isinstance(value, dict):
        return {
            key: "***" if str(key).lower() in SENSITIVE_LOG_FIELDS else redact_sensitive(item)
            for key, item in value.items()
        }
    if isinstance(value, list):
        return [redact_sensitive(item) for item in value]
    return value


def safe_response_excerpt(response: requests.Response, limit: int = 300) -> str:
    """Return a short response body with known credentials removed."""
    try:
        text = json.dumps(redact_sensitive(response.json()), ensure_ascii=False)
    except ValueError:
        text = response.text

    if CRM_TOKEN:
        text = text.replace(str(CRM_TOKEN), "***")
    return text[:limit]


def clean_num(value: Any) -> str:
    """Return a normalized phone number from Yeastar values like 'Name(100)'."""
    if value is None:
        return ""

    number = str(value).strip()
    if "(" in number and ")" in number:
        inside_parentheses = number.split("(", 1)[1].split(")", 1)[0].strip()
        return inside_parentheses or number

    return number


def safe_recording_filename(filename: str | None) -> str | None:
    if not filename:
        return None

    clean_name = secure_filename(Path(filename).name)
    return clean_name or None


def extract_numbers(data: dict[str, Any]) -> tuple[str | None, str | None, str | None]:
    caller = None
    callee = None
    direction = None

    if "callfrom" in data:
        caller = data.get("callfrom")
        callee = clean_num(data.get("callto"))
        direction = str(data.get("type", "")).lower() or None
        return caller, callee, direction

    ext = data.get("ext")
    if isinstance(ext, dict):
        callee = ext.get("extid") or ext.get("number") or callee

    inbound = data.get("inbound")
    if isinstance(inbound, dict):
        caller = caller or inbound.get("from")
        callee = callee or inbound.get("to")
        direction = "inbound"

    outbound = data.get("outbound")
    if isinstance(outbound, dict):
        caller = caller or outbound.get("from")
        callee = callee or outbound.get("to")
        direction = "outbound"

    return caller, callee, direction


def update_call_registry(callid: str, data: dict[str, Any]) -> dict[str, Any]:
    caller, callee, direction = extract_numbers(data)

    with registry_lock:
        call_info = call_registry.setdefault(
            callid,
            {
                "events_sent": set(),
                "caller": None,
                "callee": None,
                "direction": None,
                "start_time": time.time(),
                "timestart": None,
            },
        )

        if caller:
            call_info["caller"] = clean_num(caller)
        if callee:
            if "ext" in data or not call_info["callee"]:
                call_info["callee"] = clean_num(callee)
        if direction:
            call_info["direction"] = direction
        if data.get("timestart"):
            call_info["timestart"] = data["timestart"]

        return call_info


def parse_yeastar_int(value: Any, default: int = 0) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def extract_timestamp_from_callid(callid: str) -> int:
    try:
        return int(str(callid).split(".", 1)[0])
    except (TypeError, ValueError):
        return int(time.time())


def build_record_link(filename: str | None) -> str:
    safe_filename = safe_recording_filename(filename)
    if not safe_filename:
        return ""

    port_part = "" if EXTERNAL_PORT in (80, 443) else f":{EXTERNAL_PORT}"
    return f"{PROTOCOL}://{EXTERNAL_HOST}{port_part}/download/{safe_filename}"


def build_crm_payload(
    callid: str,
    event_type: str,
    filename: str | None = None,
    cdr_data: dict[str, Any] | None = None,
) -> dict[str, Any]:
    with registry_lock:
        call_info = dict(call_registry.get(callid, {}))

    if cdr_data and event_type == "cdr":
        caller = clean_num(cdr_data.get("callfrom")) or call_info.get("caller") or "unknown"
        callee = clean_num(cdr_data.get("callto")) or call_info.get("callee") or "unknown"
        direction = str(cdr_data.get("type") or call_info.get("direction") or "inbound").lower()
        total_duration = parse_yeastar_int(cdr_data.get("callduraction"))
        talk_duration = parse_yeastar_int(cdr_data.get("talkduraction"))
        wait_duration = max(total_duration - talk_duration, 0)
        state = str(cdr_data.get("status", "ANSWERED")).lower()
        start_timestamp = extract_timestamp_from_callid(callid)
        date_start = cdr_data.get("timestart") or datetime.fromtimestamp(start_timestamp).strftime(
            "%Y-%m-%d %H:%M:%S"
        )
    else:
        caller = call_info.get("caller") or "unknown"
        callee = call_info.get("callee") or "unknown"
        direction = str(call_info.get("direction") or "inbound").lower()
        total_duration = 0
        talk_duration = 0
        wait_duration = 0
        state = ""
        start_timestamp = extract_timestamp_from_callid(callid)
        date_start = datetime.fromtimestamp(start_timestamp).strftime("%Y-%m-%d %H:%M:%S")

    payload = {
        "token": CRM_TOKEN,
        "caller": clean_num(caller) or "unknown",
        "callee": clean_num(callee) or "unknown",
        "direction": "in" if direction == "inbound" else "out",
        "event": event_type,
        "start_timestamp": start_timestamp,
        "state": state,
        "duration": total_duration,
        "talk_diuration": talk_duration,
        "record_link": build_record_link(filename),
    }

    if event_type == "cdr":
        payload.update(
            {
                "call_uuid": callid,
                "date_start": date_start,
                "wait_duration": wait_duration,
            }
        )

    return payload


def send_to_crm(
    callid: str,
    event_type: str,
    filename: str | None = None,
    cdr_data: dict[str, Any] | None = None,
) -> None:
    if not CRM_URL or not CRM_TOKEN:
        logger.warning("CRM URL or token is not configured")
        return

    payload = build_crm_payload(callid, event_type, filename, cdr_data)
    logger.info(
        "CRM event %s: %s -> %s",
        event_type,
        payload["caller"],
        payload["callee"],
    )
    logger.debug(
        "CRM payload: %s",
        json.dumps(redact_sensitive(payload), ensure_ascii=False),
    )

    def send_task() -> None:
        attempts = int(crm_conf.get("retry_attempts", 3))
        retry_delay = int(crm_conf.get("retry_delay", 5))

        for attempt in range(1, attempts + 1):
            try:
                response = requests.post(
                    CRM_URL,
                    json=payload,
                    timeout=crm_conf.get("timeout", 10),
                    headers={"Content-Type": "application/json"},
                )

                if 200 <= response.status_code < 300:
                    logger.info("CRM accepted %s for call %s", event_type, callid)
                    return

                logger.warning(
                    "CRM returned HTTP %s on attempt %s/%s: %s",
                    response.status_code,
                    attempt,
                    attempts,
                    safe_response_excerpt(response),
                )
            except requests.RequestException as exc:
                logger.warning("CRM request failed on attempt %s/%s: %s", attempt, attempts, exc)

            if attempt < attempts:
                time.sleep(retry_delay)

        logger.error("CRM delivery failed for %s after %s attempts", callid, attempts)

    threading.Thread(target=send_task, daemon=True).start()


def heartbeat_worker() -> None:
    interval = int(ats_conf.get("heartbeat_interval", 30))
    logger.info("Heartbeat worker started, interval=%ss", interval)

    while True:
        try:
            token = ats.token or ats.login()
            if token:
                heartbeat_url = f"{ats_conf['url']}/api/{ats_conf['api_version']}/heartbeat"
                payload = {
                    "ipaddr": HEARTBEAT_IP,
                    "port": str(HEARTBEAT_PORT),
                    "url": "/webhook",
                }
                response = requests.post(
                    heartbeat_url,
                    params={"token": token},
                    json=payload,
                    verify=ats.verify_tls,
                    timeout=5,
                )
                response.raise_for_status()
                result = response.json()
                if result.get("status") != "Success":
                    logger.warning(
                        "ATS heartbeat rejected: errno=%s",
                        result.get("errno", "unknown"),
                    )
                    ats.token = None
        except Exception as exc:
            ats.token = None
            logger.error("Heartbeat failed: %s", type(exc).__name__)

        time.sleep(interval)


def cleanup_worker() -> None:
    interval = int(storage_conf.get("cleanup_interval", 3600))
    days = int(storage_conf.get("cleanup_days", 90))
    logger.info("Cleanup worker started, days=%s, interval=%ss", days, interval)

    while True:
        try:
            RECORDS_DIR.mkdir(parents=True, exist_ok=True)
            cutoff = time.time() - (days * 86400)
            deleted = 0

            for filepath in RECORDS_DIR.iterdir():
                if filepath.is_file() and filepath.stat().st_mtime < cutoff:
                    filepath.unlink()
                    deleted += 1

            if deleted:
                logger.info("Deleted old recording files: %s", deleted)

            registry_cutoff = time.time() - 3600
            with registry_lock:
                old_callids = [
                    callid
                    for callid, info in call_registry.items()
                    if info.get("start_time", 0) < registry_cutoff
                ]
                for callid in old_callids:
                    call_registry.pop(callid, None)
        except Exception:
            logger.exception("Cleanup failed")

        time.sleep(interval)


@app.route("/webhook", methods=["POST"])
def webhook():
    data = request.get_json(silent=True) or {}

    if not data:
        return jsonify({"status": "no_data"}), 200

    if logging_conf.get("show_full_events", False):
        logger.debug(
            "Webhook payload: %s",
            json.dumps(redact_sensitive(data), ensure_ascii=False),
        )

    action = data.get("action")
    callid = data.get("callid")

    if not callid:
        return jsonify({"status": "Success"}), 200

    callid = str(callid)
    call_info = update_call_registry(callid, data)
    events_sent = call_info["events_sent"]

    logger.info(
        "Webhook %s callid=%s %s -> %s",
        action,
        callid,
        call_info.get("caller"),
        call_info.get("callee"),
    )

    if action in {"Invite", "RING", "ALERT", "Incoming"} and "invite" not in events_sent:
        send_to_crm(callid, "invite")
        events_sent.add("invite")
    elif action in {"ANSWER", "ANSWERED"} and "answer" not in events_sent:
        send_to_crm(callid, "answer")
        events_sent.add("answer")
    elif action == "BYE" and "hangup" not in events_sent:
        send_to_crm(callid, "hangup")
        events_sent.add("hangup")
    elif action == "NewCdr":
        filename = safe_recording_filename(data.get("recording"))
        logger.info("CDR received for callid=%s, recording=%s", callid, filename or "none")

        if filename:
            start_recording_download(callid, filename, data)
        else:
            send_to_crm(callid, "cdr", None, cdr_data=data)

        threading.Thread(target=cleanup_call_later, args=(callid,), daemon=True).start()

    return jsonify({"status": "Success"}), 200


def start_recording_download(callid: str, filename: str, cdr_data: dict[str, Any]) -> None:
    def download_task() -> None:
        save_path = RECORDS_DIR / filename
        attempts = int(ats_conf.get("download_attempts", 6))
        delay = int(ats_conf.get("download_retry_delay", 10))

        for attempt in range(1, attempts + 1):
            time.sleep(delay)

            if ats.download_file(filename, save_path):
                size_mb = save_path.stat().st_size / (1024 * 1024)
                logger.info("Recording downloaded: %s (%.2f MB)", filename, size_mb)
                send_to_crm(callid, "cdr", filename, cdr_data=cdr_data)
                return

            logger.info("Recording download attempt %s/%s failed: %s", attempt, attempts, filename)

        logger.error("Recording download failed after %s attempts: %s", attempts, filename)
        send_to_crm(callid, "cdr", None, cdr_data=cdr_data)

    threading.Thread(target=download_task, daemon=True).start()


def cleanup_call_later(callid: str) -> None:
    time.sleep(120)
    with registry_lock:
        call_registry.pop(callid, None)


@app.route("/download/<path:filename>")
def download(filename: str):
    safe_filename = safe_recording_filename(filename)
    if not safe_filename:
        return jsonify({"error": "Invalid filename"}), 400

    return send_from_directory(RECORDS_DIR, safe_filename, as_attachment=True)


@app.route("/health")
def health():
    return jsonify(
        {
            "status": "ok",
            "version": VERSION,
            "ats_connected": bool(ats.token),
            "active_calls": len(call_registry),
            "network": {
                "listen_ip": LOCAL_IP,
                "heartbeat_ip": HEARTBEAT_IP,
                "heartbeat_port": HEARTBEAT_PORT,
                "external_host": EXTERNAL_HOST,
                "external_port": EXTERNAL_PORT,
            },
        }
    )


@app.route("/calls")
def calls():
    with registry_lock:
        calls_data = [
            {
                "callid": callid,
                "caller": info.get("caller"),
                "callee": info.get("callee"),
                "direction": info.get("direction"),
                "events_sent": sorted(info.get("events_sent", [])),
            }
            for callid, info in call_registry.items()
        ]

    return jsonify({"active": len(calls_data), "calls": calls_data})


def start_background_workers() -> None:
    RECORDS_DIR.mkdir(parents=True, exist_ok=True)
    threading.Thread(target=heartbeat_worker, daemon=True).start()
    threading.Thread(target=cleanup_worker, daemon=True).start()


if __name__ == "__main__":
    start_background_workers()

    logger.info("Yeastar Proxy v%s starting", VERSION)
    if ats.login():
        logger.info("Connected to ATS")
    else:
        logger.warning("Could not connect to ATS on startup")

    logger.info("Listen: %s:%s", LOCAL_IP, LOCAL_PORT)
    logger.info("External: %s://%s:%s", PROTOCOL, EXTERNAL_HOST, EXTERNAL_PORT)
    logger.info("Storage: %s", RECORDS_DIR)
    logger.info("CRM: %s", CRM_URL)

    app.run(host="0.0.0.0", port=LOCAL_PORT, debug=False)