From 729872faa39fd1031c4526d11ad53e9cf1907844 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Sun, 20 Sep 2026 12:34:25 +0330 Subject: [PATCH] feat: add cdr validation and normalization domain --- pyproject.toml | 4 +- src/core/domain/__init__.py | 0 src/core/domain/errors.py | 23 +++++ src/core/domain/models.py | 43 ++++++++++ src/core/domain/normalization.py | 121 +++++++++++++++++++++++++++ src/core/domain/validation.py | 139 +++++++++++++++++++++++++++++++ tests/unit/test_normalization.py | 77 +++++++++++++++++ tests/unit/test_validation.py | 71 ++++++++++++++++ 8 files changed, 477 insertions(+), 1 deletion(-) create mode 100644 src/core/domain/__init__.py create mode 100644 src/core/domain/errors.py create mode 100644 src/core/domain/models.py create mode 100644 src/core/domain/normalization.py create mode 100644 src/core/domain/validation.py create mode 100644 tests/unit/test_normalization.py create mode 100644 tests/unit/test_validation.py diff --git a/pyproject.toml b/pyproject.toml index c395c89..b910b62 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,7 +3,9 @@ name = "telecom-cdr-pipeline" version = "0.1.0" description = "High-throughput telecom CDR ingestion and analytics pipeline" requires-python = ">=3.11" -dependencies = [] +dependencies = [ + "phonenumbers>=8.13", +] [project.optional-dependencies] dev = [ diff --git a/src/core/domain/__init__.py b/src/core/domain/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/domain/errors.py b/src/core/domain/errors.py new file mode 100644 index 0000000..f270093 --- /dev/null +++ b/src/core/domain/errors.py @@ -0,0 +1,23 @@ +from enum import StrEnum + + +class FailureReason(StrEnum): + MISSING_REQUIRED_FIELD = "missing_required_field" + INVALID_RAT_VALUE = "invalid_rat_value" + INVALID_EPOCH = "invalid_epoch" + MALFORMED_JSON = "malformed_json" + NORMALIZATION_ERROR = "normalization_error" + + +class CDRProcessingError(Exception): + def __init__( + self, + reason: FailureReason, + detail: str, + failed_field: str | None = None, + ) -> None: + super().__init__(detail) + + self.reason = reason + self.detail = detail + self.failed_field = failed_field diff --git a/src/core/domain/models.py b/src/core/domain/models.py new file mode 100644 index 0000000..c0a3103 --- /dev/null +++ b/src/core/domain/models.py @@ -0,0 +1,43 @@ +from dataclasses import dataclass +from datetime import datetime +from ipaddress import IPv4Address +from uuid import UUID + + +@dataclass(frozen=True, slots=True) +class RawCDR: + raw_id: str + device_id: str + caller_num: str + svc: str + rat: int + epoch_start: int + epoch_end: int + up_bytes: int + down_bytes: int + lac: str + ci: str + src_ip: str | None = None + + +@dataclass(frozen=True, slots=True) +class NormalizedCDR: + cdr_id: UUID + 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: datetime + session_end_time: datetime + + source_ip: IPv4Address diff --git a/src/core/domain/normalization.py b/src/core/domain/normalization.py new file mode 100644 index 0000000..6e9a380 --- /dev/null +++ b/src/core/domain/normalization.py @@ -0,0 +1,121 @@ +import uuid +from datetime import datetime, timezone +from ipaddress import IPv4Address, AddressValueError + +import phonenumbers +from phonenumbers import PhoneNumberFormat + +from core.domain.errors import CDRProcessingError, FailureReason +from core.domain.models import NormalizedCDR, RawCDR + + +RAT_MAPPING = { + 1: "3G", + 6: "LTE_4G", + 9: "5G_NR", +} + +def normalize_rat(rat: int) -> str: + try: + return RAT_MAPPING[rat] + except KeyError as exc: + raise CDRProcessingError( + reason=FailureReason.INVALID_RAT_VALUE, + detail=f"Unsupported RAT value: {rat}.", + failed_field="rat", + ) from exc + +def _normalize_international_prefix(number: str) -> str: + number = number.strip() + + if number.startswith("+"): + return number + + if number.startswith("00"): + return f"+{number[2:]}" + + if number.startswith("011"): + return f"+{number[3:]}" + + return number + +def normalize_msisdn(number: str) -> str: + prepared_number = _normalize_international_prefix(number) + + try: + parsed = phonenumbers.parse( + prepared_number, + None, + ) + except phonenumbers.NumberParseException as exc: + raise CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail=f"Unable to parse caller_num: {number}.", + failed_field="caller_num", + ) from exc + + if not phonenumbers.is_possible_number(parsed): + raise CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail=f"caller_num is not a possible phone number: {number}.", + failed_field="caller_num", + ) + + return phonenumbers.format_number( + parsed, + PhoneNumberFormat.E164, + ) + +def generate_cdr_id(raw_id: str) -> uuid.UUID: + return uuid.uuid5( + uuid.NAMESPACE_DNS, + raw_id, + ) + +def epoch_to_datetime(epoch: int) -> datetime: + try: + return datetime.fromtimestamp( + epoch, + tz=timezone.utc, + ) + except (OverflowError, OSError, ValueError) as exc: + raise CDRProcessingError( + reason=FailureReason.INVALID_EPOCH, + detail=f"Invalid Unix epoch value: {epoch}.", + ) from exc + +def normalize_source_ip(source_ip: str | None) -> IPv4Address: + if source_ip is None: + return IPv4Address("0.0.0.0") + + try: + return IPv4Address(source_ip) + except AddressValueError as exc: + raise CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail=f"Invalid IPv4 address: {source_ip}.", + failed_field="src_ip", + ) from exc + +def normalize_cdr(raw: RawCDR) -> NormalizedCDR: + return NormalizedCDR( + cdr_id=generate_cdr_id(raw.raw_id), + raw_id=raw.raw_id, + + msisdn=normalize_msisdn(raw.caller_num), + imsi=raw.device_id, + + service_type=raw.svc, + radio_access_type=normalize_rat(raw.rat), + cell_id=f"{raw.lac}_{raw.ci}", + + bytes_uploaded=raw.up_bytes, + bytes_downloaded=raw.down_bytes, + total_bytes=raw.up_bytes + raw.down_bytes, + duration_seconds=raw.epoch_end - raw.epoch_start, + + session_start_time=epoch_to_datetime(raw.epoch_start), + session_end_time=epoch_to_datetime(raw.epoch_end), + + source_ip=normalize_source_ip(raw.src_ip), + ) diff --git a/src/core/domain/validation.py b/src/core/domain/validation.py new file mode 100644 index 0000000..03f2995 --- /dev/null +++ b/src/core/domain/validation.py @@ -0,0 +1,139 @@ +from collections.abc import Mapping +from typing import Any + +from core.domain.errors import CDRProcessingError, FailureReason +from core.domain.models import RawCDR + + +REQUIRED_FIELDS = ( + "raw_id", + "device_id", + "caller_num", + "svc", + "rat", + "epoch_start", + "epoch_end", + "up_bytes", + "down_bytes", + "lac", + "ci", +) + + +SUPPORTED_RAT_VALUES = { + 1, + 6, + 9, +} + +def _validate_required_fields(payload: Mapping[str, Any]) -> None: + for field in REQUIRED_FIELDS: + if field not in payload or payload[field] is None: + raise CDRProcessingError( + reason=FailureReason.MISSING_REQUIRED_FIELD, + detail=f"Required field '{field}' is missing or null.", + failed_field=field, + ) + +def _validate_type( + payload: Mapping[str, Any], + field: str, + expected_type: type, +) -> None: + value = payload[field] + + # bool is technically a subclass of int in Python. + # We do not want True/False accepted as integer CDR fields. + if expected_type is int and isinstance(value, bool): + valid = False + else: + valid = isinstance(value, expected_type) + + if not valid: + raise CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail=( + f"Field '{field}' must be of type " + f"{expected_type.__name__}." + ), + failed_field=field, + ) + +def parse_raw_cdr(payload: Mapping[str, Any]) -> RawCDR: + _validate_required_fields(payload) + + string_fields = ( + "raw_id", + "device_id", + "caller_num", + "svc", + "lac", + "ci", + ) + + integer_fields = ( + "rat", + "epoch_start", + "epoch_end", + "up_bytes", + "down_bytes", + ) + + for field in string_fields: + _validate_type(payload, field, str) + + for field in integer_fields: + _validate_type(payload, field, int) + + src_ip = payload.get("src_ip") + + if src_ip is not None: + _validate_type(payload, "src_ip", str) + + raw_cdr = RawCDR( + raw_id=payload["raw_id"], + device_id=payload["device_id"], + caller_num=payload["caller_num"], + svc=payload["svc"], + rat=payload["rat"], + epoch_start=payload["epoch_start"], + epoch_end=payload["epoch_end"], + up_bytes=payload["up_bytes"], + down_bytes=payload["down_bytes"], + lac=payload["lac"], + ci=payload["ci"], + src_ip=src_ip, + ) + + validate_raw_cdr(raw_cdr) + + return raw_cdr + +def validate_raw_cdr(raw: RawCDR) -> None: + if raw.epoch_start > raw.epoch_end: + raise CDRProcessingError( + reason=FailureReason.INVALID_EPOCH, + detail="epoch_start must be less than or equal to epoch_end.", + failed_field="epoch_start", + ) + + if raw.up_bytes < 0: + raise CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail="up_bytes must be greater than or equal to zero.", + failed_field="up_bytes", + ) + + if raw.down_bytes < 0: + raise CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail="down_bytes must be greater than or equal to zero.", + failed_field="down_bytes", + ) + + if raw.rat not in SUPPORTED_RAT_VALUES: + raise CDRProcessingError( + reason=FailureReason.INVALID_RAT_VALUE, + detail=f"Unsupported RAT value: {raw.rat}.", + failed_field="rat", + ) diff --git a/tests/unit/test_normalization.py b/tests/unit/test_normalization.py new file mode 100644 index 0000000..2e6daa0 --- /dev/null +++ b/tests/unit/test_normalization.py @@ -0,0 +1,77 @@ +from core.domain.normalization import generate_cdr_id, normalize_cdr, normalize_rat +from core.domain.validation import parse_raw_cdr + + +def test_normalize_valid_cdr() -> None: + payload = { + "raw_id": "raw-us-east-9988221", + "src_ip": "10.240.12.1", + "device_id": "310410998877665", + "caller_num": "0019875551234", + "svc": "4G_DATA", + "up_bytes": 1048576, + "down_bytes": 4194304, + "rat": 6, + "lac": "4021", + "ci": "8812", + "epoch_start": 1726070000, + "epoch_end": 1726070300, + } + + raw = parse_raw_cdr(payload) + + normalized = normalize_cdr(raw) + + assert normalized.raw_id == "raw-us-east-9988221" + assert normalized.imsi == "310410998877665" + assert normalized.msisdn == "+19875551234" + + assert normalized.service_type == "4G_DATA" + assert normalized.radio_access_type == "LTE_4G" + + assert normalized.cell_id == "4021_8812" + + assert normalized.bytes_uploaded == 1048576 + assert normalized.bytes_downloaded == 4194304 + assert normalized.total_bytes == 5242880 + + assert normalized.duration_seconds == 300 + + assert str(normalized.source_ip) == "10.240.12.1" + +def test_cdr_id_is_deterministic() -> None: + first = generate_cdr_id("raw-123") + second = generate_cdr_id("raw-123") + + assert first == second + +def test_different_raw_ids_generate_different_cdr_ids() -> None: + first = generate_cdr_id("raw-123") + second = generate_cdr_id("raw-456") + + assert first != second + +def test_rat_mapping() -> None: + assert normalize_rat(1) == "3G" + assert normalize_rat(6) == "LTE_4G" + assert normalize_rat(9) == "5G_NR" + +def test_missing_source_ip_defaults_to_zero_address() -> None: + payload = { + "raw_id": "raw-001", + "device_id": "310410998877665", + "caller_num": "0019875551234", + "svc": "4G_DATA", + "up_bytes": 100, + "down_bytes": 200, + "rat": 6, + "lac": "4021", + "ci": "8812", + "epoch_start": 1726070000, + "epoch_end": 1726070300, + } + + raw = parse_raw_cdr(payload) + normalized = normalize_cdr(raw) + + assert str(normalized.source_ip) == "0.0.0.0" diff --git a/tests/unit/test_validation.py b/tests/unit/test_validation.py new file mode 100644 index 0000000..87b030b --- /dev/null +++ b/tests/unit/test_validation.py @@ -0,0 +1,71 @@ +import pytest + +from core.domain.errors import CDRProcessingError, FailureReason +from core.domain.validation import parse_raw_cdr + + +def valid_payload() -> dict: + return { + "raw_id": "raw-001", + "src_ip": "10.240.12.1", + "device_id": "310410998877665", + "caller_num": "0019875551234", + "svc": "4G_DATA", + "up_bytes": 1048576, + "down_bytes": 4194304, + "rat": 6, + "lac": "4021", + "ci": "8812", + "epoch_start": 1726070000, + "epoch_end": 1726070300, + } + +def test_missing_required_field() -> None: + payload = valid_payload() + del payload["caller_num"] + + with pytest.raises(CDRProcessingError) as exc_info: + parse_raw_cdr(payload) + + error = exc_info.value + + assert error.reason == FailureReason.MISSING_REQUIRED_FIELD + assert error.failed_field == "caller_num" + +def test_invalid_rat() -> None: + payload = valid_payload() + payload["rat"] = 99 + + with pytest.raises(CDRProcessingError) as exc_info: + parse_raw_cdr(payload) + + error = exc_info.value + + assert error.reason == FailureReason.INVALID_RAT_VALUE + assert error.failed_field == "rat" + +def test_epoch_start_after_end_is_invalid() -> None: + payload = valid_payload() + + payload["epoch_start"] = 200 + payload["epoch_end"] = 100 + + with pytest.raises(CDRProcessingError) as exc_info: + parse_raw_cdr(payload) + + error = exc_info.value + + assert error.reason == FailureReason.INVALID_EPOCH + +def test_negative_uploaded_bytes_is_invalid() -> None: + payload = valid_payload() + + payload["up_bytes"] = -1 + + with pytest.raises(CDRProcessingError) as exc_info: + parse_raw_cdr(payload) + + error = exc_info.value + + assert error.reason == FailureReason.NORMALIZATION_ERROR + assert error.failed_field == "up_bytes"