Spaces:
Sleeping
Sleeping
| """ | |
| Data reader for telecom CSV files | |
| """ | |
| import csv | |
| import os | |
| import logging | |
| import pandas as pd | |
| from typing import List, Dict, Any, Optional, Union | |
| from datetime import datetime | |
| from telecom_anomaly.validation.validator import DataValidator | |
| logger = logging.getLogger(__name__) | |
| class TelecomDataReader: | |
| """Read and validate telecom data from CSV files""" | |
| def __init__(self, config, batch_size: int = 10000): | |
| self.config = config | |
| self.batch_size = batch_size | |
| self.headers: List[str] = [] | |
| self.validator = DataValidator() | |
| def read_files(self, file_paths: Union[str, List[str]], | |
| encoding: str = 'utf-8-sig') -> List[Dict[str, Any]]: | |
| """ | |
| Read and validate data from CSV files | |
| Returns list of cleaned records with original indices | |
| """ | |
| if isinstance(file_paths, str): | |
| file_paths = [file_paths] | |
| all_records = [] | |
| original_data = [] | |
| for file_path in file_paths: | |
| if not os.path.exists(file_path): | |
| logger.warning(f"File not found: {file_path}") | |
| continue | |
| logger.info(f"Reading data from {file_path}") | |
| try: | |
| file_records, file_original = self._read_single_file( | |
| file_path, encoding, len(original_data) | |
| ) | |
| all_records.extend(file_records) | |
| original_data.extend(file_original) | |
| except Exception as e: | |
| logger.error(f"Failed to read {file_path}: {e}") | |
| logger.info(f"Read {len(all_records)} valid records from {len(original_data)} total rows") | |
| return all_records, original_data | |
| def read_dataframe(self, df: 'pd.DataFrame') -> List[Dict[str, Any]]: | |
| """ | |
| Read and validate data from a pandas DataFrame | |
| Returns list of cleaned records with original indices | |
| """ | |
| records = [] | |
| original_data = [] | |
| # Always extend the known-headers set with this DataFrame's | |
| # columns so multi-call usage doesn't silently misalign columns. | |
| for col in df.columns: | |
| name = str(col).strip() | |
| if name not in self.headers: | |
| self.headers.append(name) | |
| # Convert DataFrame to records (list of dicts) | |
| from datetime import date, time | |
| for idx, row in df.iterrows(): | |
| # Create row dictionary | |
| row_dict = {} | |
| for k, v in row.to_dict().items(): | |
| if pd.isna(v): | |
| row_dict[k] = "" | |
| elif isinstance(v, (date, time, datetime)): | |
| row_dict[k] = v | |
| else: | |
| row_dict[k] = str(v) | |
| # Store original | |
| original_copy = row_dict.copy() | |
| original_copy['_file_source'] = "dataframe_input" | |
| original_copy['_row_number'] = idx + 1 | |
| original_data.append(original_copy) | |
| # Clean record | |
| cleaned = self._clean_record(row_dict, len(original_data) - 1) | |
| if cleaned: | |
| records.append(cleaned) | |
| logger.info(f"Processed {len(records)} valid records from {len(original_data)} DataFrame rows") | |
| return records, original_data | |
| def _read_single_file(self, file_path: str, encoding: str, | |
| base_idx: int) -> tuple: | |
| """Read and validate a single CSV file""" | |
| records = [] | |
| original_data = [] | |
| with open(file_path, 'r', encoding=encoding) as f: | |
| reader = csv.reader(f) | |
| # Headers must be read fresh for every file: different sources | |
| # ship slightly different column sets (e.g. some have msisdn, | |
| # others have first_site_name/tbl_dt instead). The reader keeps | |
| # the union of headers seen so downstream code can still iterate. | |
| file_headers = [h.strip() for h in next(reader)] | |
| for h in file_headers: | |
| if h not in self.headers: | |
| self.headers.append(h) | |
| for row_num, row in enumerate(reader, start=1): | |
| if not row or all(not cell for cell in row): | |
| continue | |
| # Ensure row length matches headers | |
| if len(row) < len(file_headers): | |
| row.extend([''] * (len(file_headers) - len(row))) | |
| elif len(row) > len(file_headers): | |
| row = row[:len(file_headers)] | |
| # Create row dictionary | |
| row_dict = dict(zip(file_headers, row)) | |
| # Store original | |
| original_copy = row_dict.copy() | |
| original_copy['_file_source'] = os.path.basename(file_path) | |
| original_copy['_row_number'] = row_num | |
| original_data.append(original_copy) | |
| # Clean record | |
| cleaned = self._clean_record(row_dict, base_idx + len(original_data) - 1) | |
| if cleaned: | |
| records.append(cleaned) | |
| return records, original_data | |
| def _clean_record(self, row: Dict[str, Any], original_idx: int) -> Optional[Dict[str, Any]]: | |
| """Clean and validate a single record""" | |
| cleaned = {} | |
| # Strip string values | |
| for key, value in row.items(): | |
| cleaned[key] = value.strip() if isinstance(value, str) else value | |
| # Add metadata | |
| cleaned['_original_idx'] = original_idx | |
| # MSISDN: prefer an explicit column, otherwise derive from the leg | |
| # numbers using call_direction. Files (or DataFrames) where msisdn | |
| # is absent must still produce a subscriber id so the per-subscriber | |
| # detectors and exporters keep working. | |
| msisdn = cleaned.get('msisdn', '') | |
| if not self.validator.validate_msisdn(msisdn): | |
| msisdn = self.validator.derive_msisdn( | |
| cleaned.get('call_direction', ''), | |
| cleaned.get('calling_no', ''), | |
| cleaned.get('called_no', ''), | |
| ) | |
| if not msisdn: | |
| return None | |
| cleaned['msisdn'] = msisdn | |
| # Validate duration | |
| cleaned['duration'] = self.validator.validate_duration(cleaned.get('duration')) | |
| # Validate coordinates | |
| lat, lon = self.validator.validate_coordinates( | |
| cleaned.get('latitude'), cleaned.get('longitude') | |
| ) | |
| cleaned['latitude'] = lat | |
| cleaned['longitude'] = lon | |
| # Parse datetime | |
| cleaned['datetime'] = self.validator.parse_datetime( | |
| cleaned.get('event_date', ''), | |
| cleaned.get('event_time', '') | |
| ) | |
| # Check night call | |
| cleaned['is_night'] = self.validator.is_night_call( | |
| cleaned.get('event_time', ''), | |
| self.config.night_hours_start, | |
| self.config.night_hours_end | |
| ) | |
| # Check service number | |
| calling = cleaned.get('calling_no', '') | |
| cleaned['is_service'] = self.validator.is_service_number( | |
| calling, self.config.service_numbers | |
| ) | |
| return cleaned |