twilio_service.py

import logging
import os
from typing import Any, Dict, List, Optional, cast

from twilio.rest import Client

TwilioRestException: type[BaseException]
try:
    from twilio.base.exceptions import TwilioRestException as _TwilioRestException

    TwilioRestException = _TwilioRestException
except ImportError:

    class _FallbackTwilioRestException(Exception):
        status: int | None
        code: int | None

        def __init__(self, *args: object, status: int | None = None, code: int | None = None) -> None:
            super().__init__(*args)
            self.status = status
            self.code = code

    TwilioRestException = _FallbackTwilioRestException


from twilio.jwt.access_token import AccessToken
from twilio.jwt.access_token.grants import VoiceGrant
from twilio.request_validator import RequestValidator

from database import phone_calls as phone_calls_db

logger = logging.getLogger(__name__)

account_sid = os.getenv('TWILIO_ACCOUNT_SID')
auth_token = os.getenv('TWILIO_AUTH_TOKEN')
api_key_sid = os.getenv('TWILIO_API_KEY_SID')
api_key_secret = os.getenv('TWILIO_API_KEY_SECRET')
twiml_app_sid = os.getenv('TWILIO_TWIML_APP_SID')

_client = None
_TWILIO_NOT_FOUND_CODES = {20404}


def _get_client() -> Client:
    global _client
    if _client is None:
        if not account_sid or not auth_token:
            raise ValueError("TWILIO_ACCOUNT_SID and TWILIO_AUTH_TOKEN must be set")
        _client = Client(account_sid, auth_token)
    return _client


def generate_access_token(uid: str, ttl: int = 3600) -> Dict[str, Any]:
    """
    Generate a Twilio Access Token with Voice grant for the given user.

    Args:
        uid: User ID used as the token identity
        ttl: Token time-to-live in seconds (default 1 hour)

    Returns:
        dict with access_token, ttl, and identity
    """
    if not api_key_sid or not api_key_secret:
        raise ValueError("TWILIO_API_KEY_SID and TWILIO_API_KEY_SECRET must be set")
    if not twiml_app_sid:
        raise ValueError("TWILIO_TWIML_APP_SID must be set")

    token = AccessToken(
        account_sid,
        api_key_sid,
        api_key_secret,
        identity=uid,
        ttl=ttl,
    )

    voice_grant = VoiceGrant(
        outgoing_application_sid=twiml_app_sid,
        incoming_allow=False,
    )
    token.add_grant(voice_grant)  # type: ignore[reportUnknownMemberType]  # twilio add_grant untyped

    return {
        'access_token': token.to_jwt(),  # type: ignore[reportUnknownMemberType]  # twilio to_jwt untyped
        'ttl': ttl,
        'identity': uid,
    }


def start_caller_id_verification(phone_number: str) -> Dict[str, Any]:
    """
    Start the caller ID verification process via Twilio.
    Twilio will call the user's phone with a verification code.

    Args:
        phone_number: Phone number in E.164 format

    Returns:
        dict with verification_sid and status
    """
    client = _get_client()
    validation_request = client.validation_requests.create(
        friendly_name=phone_number,
        phone_number=phone_number,
    )
    return {
        'verification_sid': validation_request.call_sid,
        'phone_number': phone_number,
        'validation_code': validation_request.validation_code,
        'status': 'pending',
    }


def check_caller_id_verified(phone_number: str) -> bool:
    """
    Check if a phone number has been verified as a caller ID.

    Args:
        phone_number: Phone number in E.164 format

    Returns:
        True if the number is verified
    """
    client = _get_client()
    outgoing_caller_ids = client.outgoing_caller_ids.list(phone_number=phone_number)
    return len(outgoing_caller_ids) > 0


def get_caller_id(phone_number: str) -> Optional[Dict[str, Any]]:
    """
    Get the caller ID record for a verified phone number.

    Args:
        phone_number: Phone number in E.164 format

    Returns:
        dict with sid, phone_number, friendly_name or None
    """
    client = _get_client()
    outgoing_caller_ids = client.outgoing_caller_ids.list(phone_number=phone_number)
    if not outgoing_caller_ids:
        return None
    cid = outgoing_caller_ids[0]
    return {
        'sid': cid.sid,
        'phone_number': cid.phone_number,
        'friendly_name': cid.friendly_name,
    }


def _delete_caller_id_status(sid: str) -> str:
    try:
        client = _get_client()
        client.outgoing_caller_ids(sid).delete()
        return 'deleted'
    except TwilioRestException as e:
        twilio_error = cast(Any, e)
        status = getattr(twilio_error, 'status', None)
        code = getattr(twilio_error, 'code', None)
        if status == 404 or code in _TWILIO_NOT_FOUND_CODES:
            return 'already_deleted'
        logger.warning(f'delete_caller_id: twilio error sid={sid} status={status} code={code}')
        return 'failed'
    except Exception as e:
        logger.warning(f'delete_caller_id: unexpected error sid={sid}: {e}')
        return 'failed'


def delete_caller_id(sid: str) -> bool:
    """
    Delete a verified caller ID.

    Args:
        sid: The Twilio SID of the outgoing caller ID

    Returns:
        True if deleted successfully
    """
    return _delete_caller_id_status(sid) in ('deleted', 'already_deleted')


def _delete_user_caller_ids(uid: str, *, strict: bool) -> int:
    try:
        numbers = phone_calls_db.get_phone_numbers(uid)
    except Exception as e:
        logger.error(f'delete_user_caller_ids: list phone_numbers failed: {e}')
        if strict:
            raise
        return 0

    cleaned = 0
    failures = []
    for number in numbers:
        sid = number.get('twilio_sid')
        if not sid:
            continue
        status = _delete_caller_id_status(sid)
        if status in ('deleted', 'already_deleted'):
            cleaned += 1
            continue
        failures.append(sid)
        logger.warning(f'delete_user_caller_ids: twilio delete failed for sid={sid}')

    if failures and strict:
        raise RuntimeError(f'twilio caller-id delete failed for {len(failures)} caller id(s)')
    return cleaned


def delete_user_caller_ids(uid: str) -> int:
    # Best-effort caller-ID cleanup for non-compliance paths.
    return _delete_user_caller_ids(uid, strict=False)


def delete_user_caller_ids_strict(uid: str) -> int:
    """Delete every verified Twilio caller ID owned by ``uid``.

    Unlike ``delete_user_caller_ids``, this account-deletion variant raises if
    listing or deleting any caller ID fails. The account deletion worker must
    keep Firestore metadata until Twilio cleanup has either succeeded or can be
    retried with the phone-number documents still present.
    """
    return _delete_user_caller_ids(uid, strict=True)


def list_caller_ids() -> List[Dict[str, Any]]:
    """
    List all verified outgoing caller IDs for the account.

    Returns:
        List of caller ID records
    """
    client = _get_client()
    caller_ids = client.outgoing_caller_ids.list()
    return [
        {
            'sid': cid.sid,
            'phone_number': cid.phone_number,
            'friendly_name': cid.friendly_name,
        }
        for cid in caller_ids
    ]


def validate_twilio_signature(url: str, params: Dict[str, Any], signature: str) -> bool:
    """
    Validate that a request originated from Twilio using the X-Twilio-Signature header.

    Args:
        url: The full URL of the request
        params: The POST parameters
        signature: The X-Twilio-Signature header value

    Returns:
        True if the signature is valid
    """
    if not auth_token:
        return False
    validator = RequestValidator(auth_token)
    return validator.validate(url, params, signature)  # type: ignore[reportUnknownMemberType, reportUnknownVariableType]  # twilio RequestValidator.validate untyped