52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
import re
|
|
from uuid import UUID
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models import PhoneNumber, UserAccount
|
|
|
|
|
|
def normalize_phone_digits(value: str) -> str:
|
|
digits = re.sub(r"\D", "", value.strip())
|
|
if len(digits) > 10 and digits.startswith("91"):
|
|
digits = digits[2:]
|
|
if len(digits) > 10:
|
|
digits = digits[-10:]
|
|
return digits
|
|
|
|
|
|
def format_bonvoice_destination(value: str) -> str:
|
|
digits = re.sub(r"\D", "", value.strip())
|
|
if len(digits) == 10:
|
|
return f"91{digits}"
|
|
return digits
|
|
|
|
|
|
def require_owned_did(db: Session, user: UserAccount, from_value: str) -> PhoneNumber:
|
|
normalized = normalize_phone_digits(from_value)
|
|
if len(normalized) != 10:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="from must be a valid 10-digit purchased number",
|
|
)
|
|
number = db.scalar(
|
|
select(PhoneNumber).where(
|
|
PhoneNumber.user_id == user.id,
|
|
PhoneNumber.status == "assigned",
|
|
PhoneNumber.phone_number == normalized,
|
|
)
|
|
)
|
|
if not number:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="from must be a purchased number assigned to your account",
|
|
)
|
|
return number
|
|
|
|
def parse_user_id(value: str) -> UUID:
|
|
try:
|
|
return UUID(value)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid user id") from exc
|