utils/normalization.py

from __future__ import annotations

import ipaddress
import re
from typing import Any, Dict


_WHITESPACE_RE = re.compile(r"\s+")


def normalize_signal(name: str, value: Any) -> Any:
    if value is None:
        return None
    if not isinstance(value, str):
        return value

    key = name.lower()
    text = _WHITESPACE_RE.sub(" ", value.strip())
    if not text:
        return None

    if key in {"email", "username", "device", "device_fingerprint"}:
        return text.lower()

    if key == "phone":
        digits = re.sub(r"\D+", "", text)
        if not digits:
            return None
        if len(digits) == 11 and digits.startswith("1"):
            return f"+{digits}"
        if len(digits) == 10:
            return f"+1{digits}"
        return f"+{digits}"

    if key == "ip":
        try:
            return ipaddress.ip_address(text).compressed.lower()
        except ValueError:
            return text.lower()

    if key in {"content", "signup_time"}:
        return text

    return text


def normalize_signals(signals: Dict[str, Any]) -> Dict[str, Any]:
    normalized = {key: normalize_signal(key, value) for key, value in signals.items()}
    metadata = normalized.get("metadata")
    if not isinstance(metadata, dict):
        normalized["metadata"] = None
    return normalized


def entity_values(signals: Dict[str, Any]) -> Dict[str, str]:
    entity_keys = {
        "email": "email",
        "phone": "phone",
        "username": "username",
        "ip": "ip",
        "device_fingerprint": "device",
    }
    entities: Dict[str, str] = {}
    for signal_key, entity_type in entity_keys.items():
        value = signals.get(signal_key)
        if isinstance(value, str) and value:
            entities[entity_type] = value
    return entities