parser.py

"""
parser.py — Parse call log CSV files and Telnyx CDR API responses.
"""

import csv
from datetime import datetime


REQUIRED_FIELDS = {"timestamp", "phone_number", "status"}


def parse_csv(filepath: str) -> list[dict]:
    """
    Parse a call log CSV file into a list of call record dicts.

    Args:
        filepath: Path to the CSV file

    Returns:
        list of call record dicts
    """
    records = []
    with open(filepath, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        fields = set(reader.fieldnames or [])
        missing = REQUIRED_FIELDS - fields
        if missing:
            raise ValueError(f"CSV is missing required columns: {missing}")

        for row in reader:
            record = dict(row)
            # Normalize timestamp
            try:
                record["timestamp"] = datetime.fromisoformat(record["timestamp"])
            except (ValueError, KeyError):
                record["timestamp"] = None

            # Normalize duration to seconds
            if "duration" in record:
                record["duration_seconds"] = _parse_duration(record["duration"])

            records.append(record)

    return records


def parse_telnyx_cdr(api_response: list[dict]) -> list[dict]:
    """
    Parse a list of Telnyx CDR API records into a normalized format.

    Args:
        api_response: List of CDR record dicts from the Telnyx API

    Returns:
        list of normalized call record dicts
    """
    records = []
    for item in api_response:
        record = {
            "timestamp": _parse_iso(item.get("start_time")),
            "phone_number": item.get("to", ""),
            "from_number": item.get("from", ""),
            "status": item.get("status", "unknown"),
            "duration_seconds": item.get("duration_secs", 0),
            "call_id": item.get("call_leg_id", ""),
            "direction": item.get("direction", "outbound"),
        }
        records.append(record)
    return records


def _parse_duration(value: str) -> int:
    """Convert a duration string like '2m 14s' or '134' to integer seconds."""
    if not value:
        return 0
    value = str(value).strip()
    if value.isdigit():
        return int(value)
    total = 0
    if "m" in value:
        parts = value.split("m")
        total += int(parts[0].strip()) * 60
        value = parts[1]
    if "s" in value:
        total += int(value.replace("s", "").strip())
    return total


def _parse_iso(value: str) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None