Spaces:
Sleeping
Sleeping
| import tabula | |
| import pandas as pd | |
| import io | |
| import re | |
| import os | |
| 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 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_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 | |
| header_idx = -1 | |
| for i, line in enumerate(lines): | |
| if "event_date_time" in line: | |
| header_idx = i | |
| break | |
| 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): | |
| if current_row is not None: | |
| fixed_rows.append(current_row) | |
| current_row = row.copy() | |
| elif current_row is not None: | |
| current_row[main_col] = (str(current_row[main_col]) + " " + val).replace('\r', '').replace('\n', '') | |
| for col in df_raw.columns: | |
| if col != main_col and pd.api.types.is_string_dtype(df_raw[col]): | |
| fragment = str(row[col]).strip() | |
| if fragment.lower() != 'nan': | |
| current_row[col] = (str(current_row[col]) + " " + fragment).replace('\r', '').replace('\n', ' ') | |
| if current_row is not None: | |
| 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 | |
| df_calls['event_time'] = dt_series.dt.time | |
| df_calls.drop(columns=[main_col], inplace=True) | |
| uppercase_cols = ['district', 'city', 'usage_type', 'usage_sub_type', 'calling_no', 'called_no', 'imei', 'location_id', 'msisdn'] | |
| for col in df_calls.columns: | |
| c_low = col.lower() | |
| if df_calls[col].dtype == 'object': | |
| df_calls[col] = df_calls[col].astype(str).str.replace(r'[\r\n\s]+', ' ', regex=True).str.strip() | |
| if c_low in uppercase_cols: | |
| df_calls[col] = df_calls[col].str.upper() | |
| elif c_low == 'region': | |
| df_calls[col] = df_calls[col].str.title() | |
| elif c_low == 'call_direction': | |
| df_calls[col] = df_calls[col].str.title().replace({'Incomingmt': 'Incoming (MT)', '\(\(': '(', '\)\)': ')'}, regex=True) | |
| 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] | |
| 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_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'], format='%H:%M:%S', errors='coerce') | |
| df_cleaned = df.dropna().copy() | |
| df_cleaned = df_cleaned.drop(columns=['event_time_check']) | |
| 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 | |