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)