97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
|
import base64
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import re
|
||
|
|
from typing import Dict, Any
|
||
|
|
from urllib.error import HTTPError, URLError
|
||
|
|
from urllib.request import Request as UrlRequest, urlopen
|
||
|
|
|
||
|
|
from app.core.config import settings
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
class SmsService:
|
||
|
|
API_URL = "https://api.cpsms.dk/v2/send"
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def normalize_recipient(number: str) -> str:
|
||
|
|
cleaned = re.sub(r"[^0-9+]", "", (number or "").strip())
|
||
|
|
if not cleaned:
|
||
|
|
raise ValueError("Mobilnummer mangler")
|
||
|
|
|
||
|
|
if cleaned.startswith("+"):
|
||
|
|
cleaned = cleaned[1:]
|
||
|
|
if cleaned.startswith("00"):
|
||
|
|
cleaned = cleaned[2:]
|
||
|
|
|
||
|
|
if not cleaned.isdigit():
|
||
|
|
raise ValueError("Ugyldigt mobilnummer")
|
||
|
|
|
||
|
|
if len(cleaned) == 8:
|
||
|
|
cleaned = "45" + cleaned
|
||
|
|
|
||
|
|
if len(cleaned) < 8:
|
||
|
|
raise ValueError("Ugyldigt mobilnummer")
|
||
|
|
|
||
|
|
return cleaned
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _authorization_header() -> str:
|
||
|
|
username = (settings.SMS_USERNAME or "").strip()
|
||
|
|
api_key = (settings.SMS_API_KEY or "").strip()
|
||
|
|
|
||
|
|
if not username or not api_key:
|
||
|
|
raise ValueError("SMS er ikke konfigureret (SMS_USERNAME/SMS_API_KEY mangler)")
|
||
|
|
|
||
|
|
raw = f"{username}:{api_key}".encode("utf-8")
|
||
|
|
token = base64.b64encode(raw).decode("ascii")
|
||
|
|
return f"Basic {token}"
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def send_sms(cls, to: str, message: str, sender: str | None = None) -> Dict[str, Any]:
|
||
|
|
sms_message = (message or "").strip()
|
||
|
|
if not sms_message:
|
||
|
|
raise ValueError("SMS-besked må ikke være tom")
|
||
|
|
if len(sms_message) > 1530:
|
||
|
|
raise ValueError("SMS-besked er for lang (max 1530 tegn)")
|
||
|
|
|
||
|
|
recipient = cls.normalize_recipient(to)
|
||
|
|
sms_sender = (sender or settings.SMS_SENDER or "").strip()
|
||
|
|
if not sms_sender:
|
||
|
|
raise ValueError("SMS afsender mangler (SMS_SENDER)")
|
||
|
|
|
||
|
|
payload = {
|
||
|
|
"to": recipient,
|
||
|
|
"message": sms_message,
|
||
|
|
"from": sms_sender,
|
||
|
|
}
|
||
|
|
|
||
|
|
body = json.dumps(payload).encode("utf-8")
|
||
|
|
request = UrlRequest(cls.API_URL, data=body, method="POST")
|
||
|
|
request.add_header("Content-Type", "application/json")
|
||
|
|
request.add_header("Authorization", cls._authorization_header())
|
||
|
|
|
||
|
|
try:
|
||
|
|
with urlopen(request, timeout=15) as response:
|
||
|
|
response_body = response.read().decode("utf-8")
|
||
|
|
response_data = json.loads(response_body) if response_body else {}
|
||
|
|
return {
|
||
|
|
"http_status": getattr(response, "status", 200),
|
||
|
|
"provider": "cpsms",
|
||
|
|
"recipient": recipient,
|
||
|
|
"result": response_data,
|
||
|
|
}
|
||
|
|
except HTTPError as e:
|
||
|
|
error_body = ""
|
||
|
|
try:
|
||
|
|
error_body = e.read().decode("utf-8")
|
||
|
|
except Exception:
|
||
|
|
error_body = ""
|
||
|
|
|
||
|
|
logger.error("❌ CPSMS HTTP error: status=%s body=%s", e.code, error_body)
|
||
|
|
raise RuntimeError(f"CPSMS fejl ({e.code})")
|
||
|
|
except URLError as e:
|
||
|
|
logger.error("❌ CPSMS connection error: %s", e)
|
||
|
|
raise RuntimeError("Kunne ikke kontakte CPSMS")
|