feat: add cdr validation and normalization domain

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 = [

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

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

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),
+    )

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",
+        )

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"

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"