Add project files.

routes/helpers.py

@@ -0,0 +1,99 @@
+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()
+