Blame for telecom_anomaly/api.py

import asyncio
import os
import shutil
import tempfile
import logging
from pathlib import Path
from typing import Any, Dict, List

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import RedirectResponse

from telecom_anomaly.core.detector import TelecomAnomalyDetector
from telecom_anomaly.core.models import TelecomConfig
from config.settings import SERVICE_NUMBERS
from telecom_anomaly.utils.pdf_processor import (
    setup_folders,
    extract_pdf_to_csv,
    process_telecom_data_complete,
    strict_clean_data,
    RAW_CSV_DIR,
    PROCESS_OUTPUT_DIR,
    STRICT_CLEAN_DIR,
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Ensure intermediate-CSV directories exist (matches telecom_anomaly_detection)
setup_folders()

app = FastAPI(
    title="Telecom Anomaly Detection API",
    description="API for detecting anomalies in telecom Call Data Records (CDR)",
    version="1.0.0"
)

# Canonical 14-column output schema shared across CSV and PDF inputs.
# msisdn is derived internally when missing; it intentionally is not
# emitted in the response because call_direction + calling_no + called_no
# already identify the subscriber unambiguously.
REQUESTED_FIELDS: List[str] = [
    "event_date", "event_time", "call_direction",
    "calling_no", "called_no", "usage_type", "duration", "imei",
    "location_id", "region", "district", "city",
    "longitude", "latitude", "azimuth",
]


@app.get("/")
async def root():
    """Redirect to documentation"""
    return RedirectResponse(url="/docs")


@app.get("/health")
def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "version": "1.0.0"}


@app.post("/analyze")
async def analyze_cdr(
    file: UploadFile = File(...),
    filter_movements: bool = False,
    run_anomaly_detection: bool = True,
    use_model: bool = False,
    cluster_km: float = 1.0,
    max_results: int = 100,
):
    """
    Analyze a PDF or CSV file for telecom anomalies. PDFs go through the
    3-stage disk-based pipeline (extract -> repair -> strict-clean) so the
    detectors see exactly the same normalized inputs as
    telecom_anomaly_detection. All intermediate CSVs are removed before
    the response returns.

    Args:
        file: PDF or CSV file to analyze.
        filter_movements: If true, clusters coordinates within cluster_km
            and keeps only distinct locations.
        run_anomaly_detection: If true, performs anomaly detection. When
            false the endpoint just returns the cleaned/clustered rows.
        use_model: If true (and run_anomaly_detection is true), also runs
            the Isolation Forest model-based detector alongside the rule
            detectors. Default false preserves the original rule-only
            pipeline.
        cluster_km: Distance margin in kilometres for geographic
            clustering (default 1.0).
        max_results: Maximum number of records to include inline in the
            JSON response (default 100). Records are ranked by
            anomaly_count descending so the most interesting rows
            survive truncation. Set to 0 for no cap. This cap exists
            so Swagger UI / the browser does not freeze rendering huge
            payloads when use_model=true on large inputs.

    Returns:
        Detection statistics and a list of results (capped per max_results).
    """
    filename = file.filename.lower()
    is_pdf = filename.endswith('.pdf')
    is_csv = filename.endswith('.csv')

    if not (is_pdf or is_csv):
        raise HTTPException(status_code=400, detail="Only PDF and CSV files are supported")

    with tempfile.NamedTemporaryFile(delete=False, suffix=Path(file.filename).suffix) as tmp:
        shutil.copyfileobj(file.file, tmp)
        tmp_path = tmp.name

    base_name = os.path.splitext(os.path.basename(file.filename))[0]
    # The threaded helper appends intermediate-CSV paths to this list so
    # the finally block can clean them up even when an exception fires.
    paths_to_cleanup: List[str] = []

    try:
        payload = await asyncio.to_thread(
            _run_analysis,
            tmp_path=tmp_path,
            original_filename=file.filename,
            base_name=base_name,
            is_pdf=is_pdf,
            cluster_km=cluster_km,
            filter_movements=filter_movements,
            run_anomaly_detection=run_anomaly_detection,
            use_model=use_model,
            max_results=max_results,
            paths_to_cleanup=paths_to_cleanup,
        )
        return payload

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Analysis failed: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail=str(e))

    finally:
        # Remove the uploaded temp file
        if os.path.exists(tmp_path):
            try:
                os.remove(tmp_path)
            except OSError:
                pass

        # Remove intermediate CSVs from each pipeline stage so the deploy
        # stays stateless (matches the original behaviour).
        paths_to_cleanup.append(
            os.path.join(PROCESS_OUTPUT_DIR, f"{base_name}_customer_info.csv")
        )
        for path in paths_to_cleanup:
            if path and os.path.exists(path):
                try:
                    os.remove(path)
                except OSError:
                    pass


def _run_analysis(
    tmp_path: str,
    original_filename: str,
    base_name: str,
    is_pdf: bool,
    cluster_km: float,
    filter_movements: bool,
    run_anomaly_detection: bool,
    use_model: bool,
    max_results: int,
    paths_to_cleanup: List[str],
) -> Dict[str, Any]:
    """
    Synchronous heavy work for /analyze.

    Runs in a worker thread (via asyncio.to_thread) so the FastAPI event
    loop stays responsive — without this, the browser sees the page as
    frozen during long Isolation Forest fits and the user is prompted to
    abandon the tab.

    Mutates `paths_to_cleanup` with each intermediate CSV created so the
    caller's finally block can delete them stateless-style.
    """
    config = TelecomConfig()
    config.service_numbers = SERVICE_NUMBERS
    config.output_dir = None
    config.save_pdf_dir = None
    config.location_cluster_km = cluster_km

    detector = TelecomAnomalyDetector(config)

    if is_pdf:
        logger.info(f"Processing PDF: {original_filename}")

        raw_csv = extract_pdf_to_csv(tmp_path, base_name)
        if not raw_csv:
            raise HTTPException(status_code=500, detail="Failed to extract tables from PDF")
        paths_to_cleanup.append(raw_csv)

        interim_csv = process_telecom_data_complete(raw_csv, base_name)
        if not interim_csv:
            raise HTTPException(status_code=500, detail="Failed to repair fragmented rows in CSV")
        paths_to_cleanup.append(interim_csv)

        final_csv = strict_clean_data(interim_csv, base_name)
        if not final_csv or not os.path.exists(final_csv):
            raise HTTPException(status_code=500, detail="Strict-clean stage produced no output")
        paths_to_cleanup.append(final_csv)

        detector.load_data([final_csv], filter_movements=filter_movements)
    else:
        logger.info(f"Processing CSV: {original_filename}")
        detector.load_data([tmp_path], filter_movements=filter_movements)

    # Detection runs when EITHER the rule-based pipeline ("Anomaly
    # Detection") OR the model pipeline ("Use AI Model") was requested.
    # `run_rules=run_anomaly_detection` makes the rule detectors honour
    # the rule toggle independently, so the four combinations work as
    # expected: rules-only, model-only, both, or neither (clustering /
    # preview only).
    any_detection = run_anomaly_detection or use_model
    if any_detection:
        detector.detect_anomalies(
            use_model=use_model,
            run_rules=run_anomaly_detection,
        )

    stats = detector.get_statistics()
    kept_original_indices = set(detector.data_indices.values())

    full_records: List[Dict[str, Any]] = []
    for idx, row in enumerate(detector.original_data):
        if idx not in kept_original_indices:
            continue
        methods = detector.row_anomaly_map.get(idx, [])
        # When any detection ran, only flagged rows make it into the
        # response. When no detection ran (clustering-only), every kept
        # row goes through with has_anomaly=NO so the caller still sees
        # the clustered dataset.
        if any_detection and not methods:
            continue
        reasons = detector.row_anomaly_reasons.get(idx, [])
        attributions = detector.row_anomaly_attribution.get(idx, [])
        record = {field: row.get(field, "") for field in REQUESTED_FIELDS}
        record.update({
            "has_anomaly": "YES" if methods else "NO",
            "anomaly_count": len(methods),
            "anomaly_methods": "; ".join(methods),
            # Index-aligned with anomaly_methods. Currently only the iforest
            # detector populates a non-empty reason; rule detectors leave
            # their slot blank. Empty entries are dropped so the joined
            # string stays readable.
            "anomaly_reasons": "; ".join(r for r in reasons if r),
            # Structured per-method attribution. Each entry is the top-k
            # {feature, z, value, direction} dicts the detector produced.
            # Rule detectors contribute empty lists; iforest contributes
            # 1-3 entries per flag. The frontend can render these as
            # per-row bar charts ("why was this flagged"), or aggregate
            # them client-side for a global feature-importance view.
            "anomaly_attribution": attributions,
        })
        full_records.append(record)

    # Inline response cap. Rank by anomaly_count desc so the most
    # interesting rows survive truncation. max_results == 0 disables.
    if max_results and max_results > 0 and len(full_records) > max_results:
        response_records = sorted(
            full_records,
            key=lambda r: r.get("anomaly_count", 0),
            reverse=True,
        )[:max_results]
        truncated = True
    else:
        response_records = full_records
        truncated = False

    return {
        "filename": original_filename,
        # Full stats whenever ANY detection ran (rules or model). The
        # stripped `{"total_rows": …}` shape is reserved for the
        # clustering-only / preview pass, where there are no anomalies
        # to summarise. Without this, AI-only mode silently dropped
        # rows_with_anomalies / anomaly_rate / anomaly_types from the
        # response.
        "statistics": (
            stats.to_dict()
            if any_detection
            else {"total_rows": len(detector.data)}
        ),
        "results": response_records,
        "total_results": len(full_records),
        "results_truncated": truncated,
        "results_cap": max_results,
    }


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)