Spaces:
Sleeping
Sleeping
File size: 7,386 Bytes
c9875bd 39be5b4 c9875bd 81a6f86 c9875bd 81a6f86 c9875bd 39be5b4 c9875bd 39be5b4 c9875bd 39be5b4 c9875bd 39be5b4 c9875bd 39be5b4 c9875bd 39be5b4 c9875bd 39be5b4 c9875bd 39be5b4 c9875bd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 | """
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 |