import json from ipaddress import IPv4Address from app.common.validation import PayloadError, parse_payload, validate_structure import uuid from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Dict, Optional, Tuple import phonenumbers RAT_MAP = { 1: "3G", 6: "LTE_4G", 9: "5G_NR", } REQUIRED_FIELDS = [ "raw_id", "device_id", "caller_num", "svc", "rat", "epoch_start", "epoch_end", "up_bytes", "down_bytes", "lac", "ci", ] @dataclass(frozen=True) class NormalizedRecord: cdr_id: str raw_id: str msisdn: str imsi: str service_type: str radio_access_type: str cell_id: str bytes_uploaded: int bytes_downloaded: int total_bytes: int duration_seconds: int session_start_time: str session_end_time: str source_ip: str def to_row(self) -> Tuple[Any, ...]: """Returns ordered tuple matching ClickHouse cdr_records schema columns.""" return ( uuid.UUID(self.cdr_id), self.raw_id, self.msisdn, self.imsi, self.service_type, self.radio_access_type, self.cell_id, self.bytes_uploaded, self.bytes_downloaded, self.total_bytes, self.duration_seconds, datetime.strptime(self.session_start_time, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc), datetime.strptime(self.session_end_time, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc), IPv4Address(self.source_ip), ) def normalize_phone_number(raw_num: str) -> Optional[str]: """ Normalizes telephone number into E.164 format: - Replaces leading '00' or '011' with '+' - Preserves existing '+' - Validates with phonenumbers library """ if not isinstance(raw_num, str): return None cleaned = raw_num.strip() if cleaned.startswith("00"): cleaned = "+" + cleaned[2:] elif cleaned.startswith("011"): cleaned = "+" + cleaned[3:] elif not cleaned.startswith("+"): cleaned = "+" + cleaned try: parsed = phonenumbers.parse(cleaned, None) if not ( phonenumbers.is_valid_number(parsed) or phonenumbers.is_possible_number(parsed) ): return None return phonenumbers.format_number( parsed, phonenumbers.PhoneNumberFormat.E164 ) except Exception: return None def normalize_cdr_payload( raw_payload_bytes: bytes, ) -> Tuple[Optional[NormalizedRecord], Optional[str], Optional[str], Optional[str]]: """ Validates and normalizes raw CDR payload. Returns: (record, failure_reason, failed_field, raw_id) If record is None, failure_reason describes why it should be routed to DLQ. """ try: payload = parse_payload(raw_payload_bytes) except PayloadError: return None, "malformed_json", None, None raw_id = payload.get("raw_id") raw_id_str = raw_id if isinstance(raw_id, str) else None try: validate_structure(payload) except PayloadError as exc: reason = "missing_required_field" if exc.reason == "missing_required_field" else ( "invalid_epoch" if exc.field in ("epoch_start", "epoch_end") else "invalid_rat_value" if exc.field == "rat" else "normalization_error") return None, reason, exc.field, raw_id_str for field in ("epoch_start", "epoch_end"): if not 0 <= payload[field] <= 2**32 - 1: return None, "invalid_epoch", field, raw_id_str epoch_start, epoch_end = payload["epoch_start"], payload["epoch_end"] if epoch_start > epoch_end: return None, "invalid_epoch", "epoch_start", raw_id_str for field in ("up_bytes", "down_bytes"): if not 0 <= payload[field] <= 2**64 - 1: return None, "normalization_error", field, raw_id_str up_bytes, down_bytes = payload["up_bytes"], payload["down_bytes"] if up_bytes + down_bytes > 2**64 - 1: return None, "normalization_error", "total_bytes", raw_id_str if payload["rat"] not in RAT_MAP: return None, "invalid_rat_value", "rat", raw_id_str msisdn = normalize_phone_number(payload["caller_num"]) if msisdn is None: return None, "normalization_error", "caller_num", raw_id_str try: source_ip = str(IPv4Address(payload["src_ip"] if payload.get("src_ip") is not None else "0.0.0.0")) except (ValueError, TypeError): return None, "normalization_error", "src_ip", raw_id_str try: start_dt = datetime.fromtimestamp(epoch_start, timezone.utc).strftime("%Y-%m-%d %H:%M:%S") end_dt = datetime.fromtimestamp(epoch_end, timezone.utc).strftime("%Y-%m-%d %H:%M:%S") except (ValueError, OverflowError, OSError): return None, "invalid_epoch", "epoch_start", raw_id_str record = NormalizedRecord( cdr_id=str(uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)), raw_id=raw_id, msisdn=msisdn, imsi=payload["device_id"], service_type=payload["svc"], radio_access_type=RAT_MAP[payload["rat"]], cell_id=f"{payload['lac']}_{payload['ci']}", bytes_uploaded=up_bytes, bytes_downloaded=down_bytes, total_bytes=up_bytes+down_bytes, duration_seconds=epoch_end-epoch_start, session_start_time=start_dt, session_end_time=end_dt, source_ip=source_ip, ) return record, None, None, raw_id