yeastar-voipms-webhook-proxy/routes/voipms.py

# routes/voipms.py

import json
import hmac, hashlib
import phonenumbers
from datetime import datetime
from flask import Blueprint, request, jsonify
from config import YEASTAR_WEBHOOK_URL, YEASTAR_SECRET, ENDPOINT_OBSCURITY
import requests
import logging
import uuid

bp = Blueprint("voipms", __name__)

@bp.route(f"/{ENDPOINT_OBSCURITY}/voipms-inbound/verbose", methods=["POST"])
@bp.route(f"/{ENDPOINT_OBSCURITY}/voipms-inbound", methods=["POST"])
def voipms_inbound():
    request_id = uuid.uuid4().hex[:8]
    verbose = request.path.endswith("/verbose")

    def get_client_ip():
        for header in ["X-Forwarded-For", "X-Real-IP"]:
            ip = request.headers.get(header)
            if ip:
                return ip.split(",")[0].strip()
        return request.remote_addr or "unknown"

    client_ip = get_client_ip()

    if verbose: print(f"[{request_id}] voipms_inbound called{' [VERBOSE]' if verbose else ''}")

    try:
        if verbose:
            print(f"[{request_id}] Headers: {dict(request.headers)}")

        raw_data = request.get_data(as_text=True)
        if verbose:
            print(f"[{request_id}] Raw Body: {raw_data}")

        data = request.get_json(force=True)

        payload = data.get("data", {}).get("payload", {})
        from_number = payload.get("from", {}).get("phone_number")
        to_list = payload.get("to", [])
        text = payload.get("text", "")
        media_urls = [m["url"] for m in payload.get("media", [])] if payload.get("media") else []

        from_number = to_e164(from_number) if from_number else ""
        to_numbers = [to_e164(entry.get("phone_number")) for entry in to_list if entry.get("phone_number")]

        if verbose:
            print(f"[{request_id}] Normalized From: {from_number}")
            print(f"[{request_id}] Normalized To: {to_numbers}")
            print(f"[{request_id}] Text: {text}")
            print(f"[{request_id}] Media URLs: {media_urls}")
        else:
            msg_type = "MMS" if media_urls else "SMS"
            now_str = datetime.now().isoformat()
            print(f"[{now_str}] IP:{client_ip} From Voip.ms | {msg_type} | From:{from_number} ? To:{', '.join(to_numbers)} | Images:{len(media_urls)}")

        ys_payload = create_yeastar_payload({
            "from": from_number,
            "to": to_numbers,
            "text": text,
            "media_urls": media_urls
        })

        if verbose:
            print(f"[{request_id}] Outbound Payload: {json.dumps(ys_payload, indent=2)}")

        body_str = json.dumps(ys_payload)
        signature = generate_signature(body_str)
        headers = {
            "Content-Type": "application/json",
            "X-Signature-256": signature
        }

        response = requests.post(YEASTAR_WEBHOOK_URL, data=body_str, headers=headers)
        if verbose: print(f"[{request_id}] Yeastar Response Code: {response.status_code}")
        if verbose:
            print(f"[{request_id}] Yeastar Response Body: {response.text}")

        return jsonify({"status": "ok"}), 200

    except Exception as e:
        logging.error(f"[{request_id}] Error in voipms_inbound: {e}", exc_info=True)
        return jsonify({"error": "internal server error"}), 500

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