62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
"""Pluggable SMS / notification providers.
|
|
|
|
Providers are selected at runtime from the dynamic settings store, so an SMS
|
|
gateway can be configured from the admin panel without touching code. Business
|
|
logic only ever talks to the provider interface.
|
|
"""
|
|
import json
|
|
import re
|
|
import urllib.request
|
|
from typing import Protocol
|
|
|
|
from apps.core.settings_service import SettingsService
|
|
|
|
|
|
class SmsProviderInterface(Protocol):
|
|
def send(self, recipient: str, message: str) -> str:
|
|
"""Return the provider message id."""
|
|
...
|
|
|
|
|
|
class MockSmsProvider:
|
|
def send(self, recipient: str, message: str) -> str:
|
|
return f"MOCK-{recipient}-{abs(hash(message)) % 10_000_000}"
|
|
|
|
|
|
class HttpSmsProvider:
|
|
"""Generic HTTP gateway: POSTs a JSON payload to a configurable endpoint."""
|
|
|
|
def __init__(self, endpoint: str, api_key: str = "", sender: str = ""):
|
|
self.endpoint = endpoint
|
|
self.api_key = api_key
|
|
self.sender = sender
|
|
|
|
def send(self, recipient: str, message: str) -> str:
|
|
payload = {"to": recipient, "text": message, "from": self.sender, "api_key": self.api_key}
|
|
data = json.dumps(payload).encode("utf-8")
|
|
request = urllib.request.Request(
|
|
self.endpoint, data=data, headers={"Content-Type": "application/json"}, method="POST"
|
|
)
|
|
with urllib.request.urlopen(request, timeout=15) as response:
|
|
return response.read().decode("utf-8", "replace")[:120]
|
|
|
|
|
|
def get_sms_provider() -> SmsProviderInterface:
|
|
service = SettingsService()
|
|
provider = service.get("sms.provider", "mock")
|
|
if provider == "http":
|
|
return HttpSmsProvider(
|
|
endpoint=service.get("sms.endpoint", ""),
|
|
api_key=service.get("sms.api_key", ""),
|
|
sender=service.get("sms.sender", ""),
|
|
)
|
|
return MockSmsProvider()
|
|
|
|
|
|
def render_template(template_text: str, context: dict) -> str:
|
|
def _replace(match):
|
|
key = match.group(1)
|
|
return str(context.get(key, ""))
|
|
|
|
return re.sub(r"\{\{\s*(\w+)\s*\}\}", _replace, template_text)
|