100 lines
3 KiB
Python
100 lines
3 KiB
Python
|
import random, time
|
||
|
|
import phonenumbers
|
||
|
|
import hashlib
|
||
|
|
import hmac
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
from datetime import datetime
|
||
|
|
from config import YEASTAR_WEBHOOK_URL, YEASTAR_SECRET
|
||
|
|
|
||
|
|
def generate_signature(body: str) -> str:
|
||
|
|
signature = hmac.new(YEASTAR_SECRET.encode(), body.encode(), hashlib.sha256).hexdigest()
|
||
|
|
return f"sha256={signature}"
|
||
|
|
|
||
|
|
def to_e164(number):
|
||
|
|
try:
|
||
|
|
parsed = phonenumbers.parse(number, "US")
|
||
|
|
return phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164)
|
||
|
|
except Exception as e:
|
||
|
|
logging.warning(f"Failed to normalize phone number {number}: {e}")
|
||
|
|
return number
|
||
|
|
|
||
|
|
def create_yeastar_payload(data):
|
||
|
|
now_iso = datetime.now().astimezone().isoformat()
|
||
|
|
message_id = str(int(datetime.utcnow().timestamp()))
|
||
|
|
payload = {
|
||
|
|
"data": {
|
||
|
|
"event_type": "message.received",
|
||
|
|
"payload": {
|
||
|
|
"id": message_id,
|
||
|
|
"from": {
|
||
|
|
"phone_number": data["from"]
|
||
|
|
},
|
||
|
|
"to": [
|
||
|
|
{"phone_number": num} for num in data["to"]
|
||
|
|
],
|
||
|
|
"text": data["text"],
|
||
|
|
"record_type": "message",
|
||
|
|
"received_at": now_iso
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
media_urls = data.get("media_urls", [])
|
||
|
|
if media_urls:
|
||
|
|
payload["data"]["payload"].pop("text", None)
|
||
|
|
payload["data"]["payload"]["media"] = [{"url": url} for url in media_urls]
|
||
|
|
|
||
|
|
return payload
|
||
|
|
|
||
|
|
def generate_message_id():
|
||
|
|
return f"{random.randint(100000, 999999)}{int(time.time())}"
|
||
|
|
|
||
|
|
def parse_bearer_token(auth_header):
|
||
|
|
"""
|
||
|
|
Expect: Authorization: Bearer username|||password
|
||
|
|
"""
|
||
|
|
if not auth_header.startswith("Bearer "):
|
||
|
|
return None, None
|
||
|
|
token = auth_header.split("Bearer ")[1].strip()
|
||
|
|
parts = token.split("|||")
|
||
|
|
if len(parts) == 2:
|
||
|
|
return parts[0], parts[1]
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
def build_soap_envelope(
|
||
|
|
username, password, from_number, to_number,
|
||
|
|
message_text, media_urls=None
|
||
|
|
):
|
||
|
|
media_urls = media_urls or []
|
||
|
|
is_mms = len(media_urls) > 0
|
||
|
|
soap_method = "sendMMS" if is_mms else "sendSMS"
|
||
|
|
|
||
|
|
envelope = f"""
|
||
|
|
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
|
||
|
|
xmlns:wsd="https://voip.ms/api/wsdl"
|
||
|
|
xmlns:xsd1="https://voip.ms/api/schema">
|
||
|
|
<soapenv:Header/>
|
||
|
|
<soapenv:Body>
|
||
|
|
<wsd:{soap_method}>
|
||
|
|
<wsd:params>
|
||
|
|
<xsd1:api_username>{username}</xsd1:api_username>
|
||
|
|
<xsd1:api_password>{password}</xsd1:api_password>
|
||
|
|
<xsd1:did>{from_number}</xsd1:did>
|
||
|
|
<xsd1:dst>{to_number}</xsd1:dst>
|
||
|
|
<xsd1:message>{message_text}</xsd1:message>
|
||
|
|
"""
|
||
|
|
|
||
|
|
# Add media tags if MMS, max 3 allowed
|
||
|
|
if is_mms:
|
||
|
|
for i, url in enumerate(media_urls[:3]):
|
||
|
|
envelope += f" <xsd1:media{i+1}>{url}</xsd1:media{i+1}>\n"
|
||
|
|
|
||
|
|
envelope += f""" </wsd:params>
|
||
|
|
</wsd:{soap_method}>
|
||
|
|
</soapenv:Body>
|
||
|
|
</soapenv:Envelope>"""
|
||
|
|
|
||
|
|
return envelope.strip()
|
||
|
|
|