tcom add

telecom_anomaly/api.py

@@ -10,15 +10,22 @@ 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 (
-    extract_pdf_to_lines,
-    process_telecom_data_df,
-    strict_clean_df,
+    setup_folders,
+    extract_pdf_to_csv,
+    process_telecom_data_complete,
+    strict_clean_data,
+    RAW_CSV_DIR,
+    PROCESS_OUTPUT_DIR,
+    STRICT_CLEAN_DIR,
 )
 
 # Configure logging
 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)",
@@ -46,9 +53,10 @@ async def analyze_cdr(
     cluster_km: float = 1.0,
 ):
     """
-    Analyze a PDF or CSV file for telecom anomalies. No files are written
-    to disk; the upload is processed in a temp file that is deleted before
-    the response returns.
+    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.
     """
     filename = file.filename.lower()
     is_pdf = filename.endswith('.pdf')
@@ -57,11 +65,15 @@ async def analyze_cdr(
     if not (is_pdf or is_csv):
         raise HTTPException(status_code=400, detail="Only PDF and CSV files are supported")
 
-    # Stage the upload to a temp file (tabula needs a real path; CSV reader takes a path too)
     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]
+    raw_csv = None
+    interim_csv = None
+    final_csv = None
+
     try:
         config = TelecomConfig()
         config.service_numbers = SERVICE_NUMBERS
@@ -73,21 +85,20 @@ async def analyze_cdr(
 
         if is_pdf:
             logger.info(f"Processing PDF: {file.filename}")
-            base_name = os.path.splitext(file.filename)[0]
 
-            raw_lines = extract_pdf_to_lines(tmp_path)
-            if not raw_lines:
+            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")
 
-            df_processed = process_telecom_data_df(raw_lines, base_name)
-            if df_processed is None:
-                raise HTTPException(status_code=500, detail="Failed to process extracted telecom data")
+            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")
 
-            df_final = strict_clean_df(df_processed)
-            if df_final is None or df_final.empty:
-                raise HTTPException(status_code=500, detail="No valid records found after cleaning")
+            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")
 
-            detector.load_dataframe(df_final, filter_movements=filter_movements)
+            detector.load_data([final_csv], filter_movements=filter_movements)
         else:
             logger.info(f"Processing CSV: {file.filename}")
             detector.load_data([tmp_path], filter_movements=filter_movements)
@@ -137,12 +148,26 @@ async def analyze_cdr(
         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
+        for path in (
+            raw_csv,
+            interim_csv,
+            final_csv,
+            os.path.join(PROCESS_OUTPUT_DIR, f"{base_name}_customer_info.csv"),
+        ):
+            if path and os.path.exists(path):
+                try:
+                    os.remove(path)
+                except OSError:
+                    pass
+
 
 if __name__ == "__main__":
     import uvicorn

telecom_anomaly/utils/pdf_processor.py

@@ -3,28 +3,45 @@ import pandas as pd
 import io
 import re
 import os
-import tempfile
+from pathlib import Path
 
+# --- CONFIGURATION ---
+# Match telecom_anomaly_detection's pipeline exactly: every stage writes a CSV
+# and the next stage reads it back. The CSV roundtrip normalizes dtypes and
+# whitespace, which is what produces the anomaly counts the user expects.
+DATA_BASE = Path("data")
+RAW_CSV_DIR = DATA_BASE / "raw_csv_temp"
+PROCESS_OUTPUT_DIR = DATA_BASE / "output_process_data"
+STRICT_CLEAN_DIR = DATA_BASE / "strictly_cleaned_data"
 
-def extract_pdf_to_lines(pdf_file_path):
-    """Stage 1 (In-Memory): Run tabula on the PDF and return raw CSV lines."""
-    with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as tmp:
-        tmp_path = tmp.name
+
+def setup_folders():
+    """Ensure all necessary output directories exist."""
+    for folder in [RAW_CSV_DIR, PROCESS_OUTPUT_DIR, STRICT_CLEAN_DIR]:
+        if not os.path.exists(folder):
+            os.makedirs(folder)
+
+
+def extract_pdf_to_csv(pdf_path, base_name):
+    """Stage 1: Convert PDF tables to a raw CSV file."""
+    csv_path = os.path.join(RAW_CSV_DIR, f"{base_name}_raw.csv")
     try:
-        tabula.convert_into(pdf_file_path, tmp_path, output_format="csv", pages="all", lattice=True)
-        with open(tmp_path, 'r', encoding='utf-8') as f:
-            lines = f.readlines()
-        return lines
-    finally:
-        if os.path.exists(tmp_path):
-            try:
-                os.remove(tmp_path)
-            except OSError:
-                pass
+        tabula.convert_into(pdf_path, csv_path, output_format="csv", pages="all", lattice=True)
+        return csv_path
+    except Exception as e:
+        print(f"Error extracting {pdf_path}: {e}")
+        return None
+
 
+def process_telecom_data_complete(input_path, base_name):
+    """Stage 2: Parse raw CSV and repair fragmented rows."""
+    try:
+        with open(input_path, 'r', encoding='utf-8') as f:
+            lines = f.readlines()
+    except Exception as e:
+        print(f"Error reading {input_path}: {e}")
+        return None
 
-def process_telecom_data_df(lines, base_name):
-    """Stage 2 (In-Memory): Parse raw CSV lines and repair fragmented rows. Returns a DataFrame."""
     header_idx = -1
     for i, line in enumerate(lines):
         if "event_date_time" in line:
@@ -34,14 +51,24 @@ def process_telecom_data_df(lines, base_name):
     if header_idx == -1:
         return None
 
+    # Customer info section (before the call-records header)
+    customer_text = "".join(lines[:header_idx])
+    df_customer = pd.read_csv(io.StringIO(customer_text))
+    df_customer.columns = [re.sub(r'[\r\n\s]+', ' ', c).strip() for c in df_customer.columns]
+    cust_path = os.path.join(PROCESS_OUTPUT_DIR, f'{base_name}_customer_info.csv')
+    df_customer.to_csv(cust_path, index=False)
+
+    # Call records
     call_text = "".join(lines[header_idx:])
     df_raw = pd.read_csv(io.StringIO(call_text))
     df_raw.columns = [c.strip() for c in df_raw.columns]
     main_col = df_raw.columns[0]
+
     df_raw = df_raw[df_raw[main_col].astype(str).str.contains("event_date_time") == False]
 
     fixed_rows = []
     current_row = None
+
     for _, row in df_raw.iterrows():
         val = str(row[main_col]).strip()
         if re.match(r'^\d{4}-\d{2}-\d{2}', val):
@@ -60,6 +87,7 @@ def process_telecom_data_df(lines, base_name):
         fixed_rows.append(current_row)
 
     df_calls = pd.DataFrame(fixed_rows)
+
     df_calls[main_col] = df_calls[main_col].str.replace(r'[\r\n]+', '', regex=True).str.replace(r'\s+', ' ', regex=True).str.strip()
     dt_series = pd.to_datetime(df_calls[main_col], errors='coerce')
     df_calls['event_date'] = dt_series.dt.date
@@ -80,29 +108,25 @@ def process_telecom_data_df(lines, base_name):
 
     ordered_cols = ['event_date', 'event_time'] + [c for c in df_calls.columns if c not in ['event_date', 'event_time']]
     df_calls = df_calls[ordered_cols]
-    return df_calls
+
+    interim_path = os.path.join(PROCESS_OUTPUT_DIR, f'{base_name}_call_records_final.csv')
+    df_calls.to_csv(interim_path, index=False)
+    return interim_path
 
 
-def strict_clean_df(df):
-    """
-    Stage 3 (In-Memory): Final strict cleaning. Returns a DataFrame whose
-    event_date and event_time are plain strings ('YYYY-MM-DD' and 'HH:MM:SS'),
-    matching exactly what the reference project produces after a CSV roundtrip.
-    Keeping these as strings ensures the downstream reader / validator behave
-    identically to the CSV-only path used by telecom_anomaly_detection.
-    """
+def strict_clean_data(file_path, base_name):
+    """Stage 3: Final strict cleaning."""
+    df = pd.read_csv(file_path)
+
     if 'duration' in df.columns and df['duration'].dtype == 'object':
         df['duration'] = df['duration'].str.replace(',', '.').astype(float)
 
     df['event_date'] = pd.to_datetime(df['event_date'], errors='coerce')
-    df['event_time_check'] = pd.to_datetime(df['event_time'].astype(str), format='%H:%M:%S', errors='coerce')
+    df['event_time_check'] = pd.to_datetime(df['event_time'], format='%H:%M:%S', errors='coerce')
 
     df_cleaned = df.dropna().copy()
     df_cleaned = df_cleaned.drop(columns=['event_time_check'])
 
-    # Serialize date/time as strings so the in-memory path matches the
-    # reference project's CSV-roundtrip path byte-for-byte.
-    df_cleaned['event_date'] = df_cleaned['event_date'].dt.strftime('%Y-%m-%d')
-    df_cleaned['event_time'] = df_cleaned['event_time'].astype(str)
-
-    return df_cleaned
+    final_path = os.path.join(STRICT_CLEAN_DIR, f'{base_name}_strictly_cleaned.csv')
+    df_cleaned.to_csv(final_path, index=False)
+    return final_path