Spaces:
Sleeping
Sleeping
| """ | |
| 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""" | |
| 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 | |
| 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 '' | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| def is_service_number(calling_no: str, service_numbers: list) -> bool: | |
| """Check if number is a service number""" | |
| return calling_no in service_numbers |