Spaces:
Sleeping
Sleeping
File size: 4,405 Bytes
c9875bd 39be5b4 c9875bd 81a6f86 c9875bd 81a6f86 5907aab 81a6f86 c9875bd 81a6f86 c9875bd 81a6f86 c9875bd 81a6f86 c9875bd 81a6f86 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 | """
Data validation utilities for telecom data
"""
from datetime import datetime
from typing import Optional, Tuple, Any
import logging
logger = logging.getLogger(__name__)
class DataValidator:
"""Validate and clean telecom data"""
@staticmethod
def validate_msisdn(msisdn: str) -> bool:
"""Validate MSISDN format"""
if not msisdn:
return False
msisdn_str = str(msisdn).strip()
# Allow service names
if msisdn_str in ['MTN', 'MOBILEMONEY', 'MTN-DATA', 'MTNXTRATIME', 'EVD']:
return True
# Ghana format: starts with 233 and 12 digits
return msisdn_str.startswith('233') and len(msisdn_str) == 12
@staticmethod
def derive_msisdn(call_direction: str, calling_no: str, called_no: str) -> str:
"""
Derive the subscriber's MSISDN from the leg numbers when the CSV
does not provide an explicit msisdn column.
Outgoing legs -> subscriber is the caller; Incoming legs -> the callee.
Falls back to whichever side is a valid Ghana MSISDN.
"""
direction = (call_direction or '').upper()
calling = (str(calling_no).strip() if calling_no is not None else '')
called = (str(called_no).strip() if called_no is not None else '')
if 'OUT' in direction or 'MO' in direction:
primary, secondary = calling, called
elif 'IN' in direction or 'MT' in direction:
primary, secondary = called, calling
else:
primary, secondary = calling, called
if DataValidator.validate_msisdn(primary):
return primary
if DataValidator.validate_msisdn(secondary):
return secondary
return ''
@staticmethod
def validate_duration(duration: Any) -> float:
"""Validate and clean duration"""
if duration is None or duration == '':
return 0.0
try:
duration_float = float(duration)
return max(0.0, duration_float)
except (ValueError, TypeError):
return 0.0
@staticmethod
def validate_coordinates(lat: Any, lon: Any) -> Tuple[Optional[float], Optional[float]]:
"""Validate latitude and longitude"""
try:
lat_float = float(lat) if lat not in (None, '', '0.0') else None
lon_float = float(lon) if lon not in (None, '', '0.0') else None
if lat_float is not None and not (-90 <= lat_float <= 90):
lat_float = None
if lon_float is not None and not (-180 <= lon_float <= 180):
lon_float = None
return lat_float, lon_float
except (ValueError, TypeError):
return None, None
@staticmethod
def parse_datetime(date_val: Any, time_val: Any) -> Optional[datetime]:
"""Parse datetime from date and time strings or objects"""
try:
if not date_val or not time_val or date_val == 'NULL' or time_val == 'NULL':
return None
# If they are already date/time objects
from datetime import date, time
if isinstance(date_val, date) and isinstance(time_val, time):
return datetime.combine(date_val, time_val)
# If they are strings
return datetime.strptime(f"{str(date_val)} {str(time_val)}", "%Y-%m-%d %H:%M:%S")
except (ValueError, TypeError) as e:
logger.debug(f"Failed to parse datetime: {date_val} {time_val} - {e}")
return None
@staticmethod
def is_night_call(time_val: Any, night_start: int = 23, night_end: int = 5) -> bool:
"""Check if call occurred at night"""
try:
if not time_val or time_val == 'NULL':
return False
from datetime import time
if isinstance(time_val, time):
hour = time_val.hour
else:
hour = int(str(time_val).split(':')[0])
return hour >= night_start or hour < night_end
except (ValueError, IndexError, AttributeError):
return False
@staticmethod
def is_service_number(calling_no: str, service_numbers: list) -> bool:
"""Check if number is a service number"""
return calling_no in service_numbers |