Add voice support: config, meta upload, S3 utils, STT/TTS modules, wo…

app/audio_stt.py

@@ -0,0 +1,55 @@
+import io
+import asyncio
+import logging
+import subprocess
+import tempfile
+from app.config import settings
+import openai
+
+logger = logging.getLogger(__name__)
+openai.api_key = settings.OPENAI_API_KEY
+
+async def transcribe_audio_openai(audio_bytes: bytes, mime_type: str = "audio/ogg") -> str:
+    """Transcribe audio bytes using OpenAI Whisper (synchronous SDK called in thread).
+    Returns the transcript string.
+    """
+    loop = asyncio.get_event_loop()
+    def _call():
+        f = io.BytesIO(audio_bytes)
+        # Depending on SDK version: use openai.Audio.transcribe or openai.Whisper
+        try:
+            resp = openai.Audio.transcribe("whisper-1", f)
+            # newer SDKs return object with 'text'
+            if isinstance(resp, dict):
+                return resp.get('text', '')
+            return getattr(resp, 'text', '')
+        except Exception:
+            # fallback: try ChatCompletion with base64? Not implemented
+            raise
+    return await loop.run_in_executor(None, _call)
+
+
+def _convert_to_ogg_opus(input_bytes: bytes) -> bytes:
+    """Use ffmpeg (must be installed) to convert input bytes to opus-in-ogg bytes.
+    Fall back to returning original bytes on failure.
+    """
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".in", delete=True) as inf, tempfile.NamedTemporaryFile(suffix=".ogg", delete=True) as outf:
+            inf.write(input_bytes)
+            inf.flush()
+            cmd = [
+                'ffmpeg', '-y', '-i', inf.name,
+                '-c:a', 'libopus', '-b:a', '32k', '-vbr', 'on', '-application', 'voip',
+                '-f', 'ogg', outf.name
+            ]
+            subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            outf.flush()
+            outf.seek(0)
+            return outf.read()
+    except Exception:
+        logger.exception("ffmpeg conversion failed; returning input bytes")
+        return input_bytes
+
+async def convert_to_ogg_opus_async(input_bytes: bytes) -> bytes:
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(None, _convert_to_ogg_opus, input_bytes)

app/audio_tts.py

@@ -0,0 +1,18 @@
+import requests
+import logging
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+def synthesize_eleven(text: str, voice: str = "alloy") -> tuple[bytes, str]:
+    """Synthesize text to speech using ElevenLabs (blocking HTTP call). Returns (bytes, mime_type).
+    """
+    api_key = settings.ELEVENLABS_API_KEY
+    if not api_key:
+        raise RuntimeError("ELEVENLABS_API_KEY not configured")
+    url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice}"
+    headers = {"xi-api-key": api_key, "Content-Type": "application/json"}
+    data = {"text": text, "voice_settings": {"stability": 0.6, "similarity_boost": 0.75}}
+    r = requests.post(url, json=data, headers=headers, stream=True, timeout=30)
+    r.raise_for_status()
+    return r.content, r.headers.get("Content-Type", "audio/mpeg")

app/config.py

@@ -35,6 +35,20 @@ class Settings(BaseSettings):
     # Observability / Sentry
     SENTRY_DSN: str = ""
 
+    # Voice settings
+    STT_PROVIDER: str = "openai"  # openai|assemblyai|gcp|aws
+    TTS_PROVIDER: str = "elevenlabs"  # elevenlabs|gcp|aws|openai
+    ELEVENLABS_API_KEY: str = ""
+
+    # S3 (for hosting media for Twilio). If not set, Twilio uploads may fail.
+    S3_BUCKET: str = ""
+    S3_REGION: str = "us-east-1"
+    AWS_ACCESS_KEY_ID: str = ""
+    AWS_SECRET_ACCESS_KEY: str = ""
+
+    # Allowed media types
+    ALLOWED_MEDIA_TYPES: str = "audio/ogg,audio/opus,audio/mpeg,audio/mp3"
+
     class Config:
         env_file = ".env"
 

app/meta_client.py

@@ -1,5 +1,4 @@
-import hmac
-import hashlib
+import io
 import logging
 import requests
 from app.config import settings
@@ -15,6 +14,7 @@ def __init__(self):
         self.phone_number_id = getattr(settings, "META_PHONE_NUMBER_ID", "")
         self.base_url = f"https://graph.facebook.com/v15.0/{self.phone_number_id}/messages"
         self.app_secret = getattr(settings, "META_APP_SECRET", "")
+        self.media_url = f"https://graph.facebook.com/v15.0/{self.phone_number_id}/media"
 
     def send(self, to: str, body: str):
         if not (self.token and self.phone_number_id):
@@ -30,19 +30,24 @@ def send(self, to: str, body: str):
         r.raise_for_status()
         return r.json()
 
-    def validate_request(self, full_url: str, params, headers: dict, raw_body: bytes = None) -> bool:
+    def upload_media(self, content_bytes: bytes, mime_type: str = "audio/ogg") -> str:
         """
-        Validate Meta (Facebook) webhook signature. Must compute HMAC-SHA256 over raw request body bytes
-        and compare to X-Hub-Signature-256 header which is of the form: sha256=<hex>
-
-        Args:
-            full_url: not used for Meta but kept for interface compatibility
-            params: parsed params/payload (not used for signature check)
-            headers: request headers mapping
-            raw_body: raw request body bytes (required)
-        Returns:
-            bool indicating whether signature is valid
+        Upload media to Meta and return media_id. Uses the /{phone_number_id}/media endpoint.
         """
+        if not (self.token and self.phone_number_id):
+            raise RuntimeError("META_TOKEN or META_PHONE_NUMBER_ID not configured")
+        files = {
+            'file': ('voice.ogg', io.BytesIO(content_bytes), mime_type)
+        }
+        params = {"messaging_product": "whatsapp", "type": "audio"}
+        headers = {"Authorization": f"Bearer {self.token}"}
+        r = requests.post(self.media_url, headers=headers, files=files, data=params, timeout=30)
+        r.raise_for_status()
+        data = r.json()
+        # response contains 'id' key for the uploaded media
+        return data.get('id')
+
+    def validate_request(self, full_url: str, params, headers: dict, raw_body: bytes = None) -> bool:
         sig_header = headers.get("X-Hub-Signature-256") or headers.get("x-hub-signature-256")
         if not sig_header:
             logger.warning("Missing X-Hub-Signature-256 header")
@@ -61,6 +66,8 @@ def validate_request(self, full_url: str, params, headers: dict, raw_body: bytes
         if prefix.lower() != "sha256":
             logger.warning("Unsupported signature algorithm: %s", prefix)
             return False
+        import hmac, hashlib
         mac = hmac.new(self.app_secret.encode(), msg=raw_body, digestmod=hashlib.sha256)
         expected = mac.hexdigest()
-        return hmac.compare_digest(expected, signature)
+        import hmac as _h
+        return _h.compare_digest(expected, signature)

app/s3_utils.py

@@ -0,0 +1,24 @@
+import boto3
+import logging
+import uuid
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+def upload_bytes_to_s3(content: bytes, content_type: str, key_prefix: str = "media/") -> str:
+    """Upload bytes to S3 and return a presigned GET URL (expires in 1 hour).
+    Requires AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY and S3_BUCKET in env.
+    """
+    bucket = settings.S3_BUCKET
+    if not bucket:
+        raise RuntimeError("S3_BUCKET not configured")
+    session = boto3.session.Session(
+        aws_access_key_id=settings.AWS_ACCESS_KEY_ID or None,
+        aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY or None,
+        region_name=settings.S3_REGION or None,
+    )
+    s3 = session.client('s3')
+    key = f"{key_prefix}{uuid.uuid4().hex}.ogg"
+    s3.put_object(Bucket=bucket, Key=key, Body=content, ContentType=content_type)
+    url = s3.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': key}, ExpiresIn=3600)
+    return url

app/worker.py

@@ -1,11 +1,19 @@
 import asyncio
 import logging
+import tempfile
 from typing import Any, Dict
 from app.reply_generator import generate_reply
 from app.messaging import client as messaging_client
 from app.metrics_extra import set_worker_queue_depth, inc_message_retry, inc_llm_calls
+from app.parse_message import normalize_message
+from app.audio_stt import transcribe_audio_openai, convert_to_ogg_opus_async
+from app.audio_tts import synthesize_eleven
+from app.s3_utils import upload_bytes_to_s3
+from app.config import settings
+from app.convo_store import ConvoStore
 
 logger = logging.getLogger(__name__)
+store = ConvoStore(redis_url=settings.REDIS_URL)
 
 class Worker:
     def __init__(self):
@@ -52,31 +60,165 @@ async def _run(self):
                     pass
 
     async def _process_item(self, item: Dict[str, Any]):
-        """Process a queued message: generate reply and send it."""
+        """Process a queued message: detect audio, transcribe, generate reply and send audio/text."""
         from_number = item.get("from")
         body = item.get("body", "")
         history = item.get("history", [])
+        provider = item.get("provider", getattr(settings, "WHATSAPP_PROVIDER", "twilio"))
+        media = item.get("media") or []
 
-        # Generate reply (may internally call blocking code via run_in_executor)
+        # If there's audio media, handle STT -> LLM -> TTS flow
+        audio_media = None
+        for m in media:
+            mtype = (m.get("type") or "").lower()
+            if mtype.startswith("audio") or mtype in ("voice", "voice_note", "ogg", "opus"):
+                audio_media = m
+                break
+
+        transcript = None
+        if audio_media:
+            try:
+                # Determine how to obtain bytes: normalized 'raw' media may include url or id
+                # Prefer raw.url or raw['url'] or media.get('url') else media.get('id') for Meta
+                media_url = audio_media.get('url') or audio_media.get('media_url')
+                media_id = audio_media.get('id')
+                media_bytes = None
+
+                # If provider is meta and we have id, fetch via Meta media endpoint
+                if provider == 'meta' and media_id:
+                    # Use Meta client to fetch media URL
+                    import requests
+                    meta_token = settings.META_TOKEN
+                    # GET /{media_id}?access_token={token}
+                    meta_media_url = f"https://graph.facebook.com/v15.0/{media_id}?access_token={meta_token}"
+                    r = requests.get(meta_media_url, timeout=10)
+                    r.raise_for_status()
+                    data = r.json()
+                    fetch_url = data.get('url')
+                    r2 = requests.get(fetch_url, timeout=20)
+                    r2.raise_for_status()
+                    media_bytes = r2.content
+                elif media_url:
+                    import requests
+                    # Twilio media URLs may require basic auth
+                    try:
+                        auth = None
+                        if settings.TWILIO_SID and settings.TWILIO_AUTH:
+                            auth = (settings.TWILIO_SID, settings.TWILIO_AUTH)
+                        r = requests.get(media_url, timeout=20, auth=auth)
+                        r.raise_for_status()
+                        media_bytes = r.content
+                    except Exception:
+                        logger.exception("Failed to fetch media url %s", media_url)
+                else:
+                    logger.warning("No media URL or id found for audio media")
+
+                if not media_bytes:
+                    raise RuntimeError("Could not fetch audio bytes")
+
+                # Convert to opus-in-ogg for better compatibility
+                converted = await convert_to_ogg_opus_async(media_bytes)
+
+                # Transcribe
+                transcript = await transcribe_audio_openai(converted, mime_type='audio/ogg')
+                # append transcript to convo store
+                await store.append_message(from_number, {"text": transcript, "type": "voice"})
+            except Exception:
+                logger.exception("Audio STT failed")
+                transcript = None
+
+        # Build prompt text: use transcript if present else body
+        prompt_text = (transcript or body or "").strip()
+        # include recent history
+        if history:
+            # history is an array of dicts; include last N texts
+            recent = [h.get('text') for h in history if h.get('text')]
+            # include as context
+            context = "\n".join(recent[-settings.LLM_MAX_HISTORY_MESSAGES:])
+        else:
+            context = ""
+
+        combined = (context + "\n" + prompt_text).strip() if context else prompt_text
+
+        # Generate reply text (LLM)
         try:
-            reply = await generate_reply(body, history)
-            # mark llm call success
+            reply_text = await generate_reply(combined, history)
             inc_llm_calls(True)
         except Exception:
-            logger.exception("Failed to generate reply; using fallback")
+            logger.exception("LLM reply generation failed")
             inc_llm_calls(False)
-            reply = "Thanks — I received your message. Can you tell me more?"
+            reply_text = "Thanks — I received your message. Can you tell me more?"
 
-        # Send message (messaging_client.send may be blocking -> run in thread)
-        try:
-            await asyncio.to_thread(messaging_client.send, to=from_number, body=reply)
-        except Exception:
-            logger.exception("Failed to send message to %s", from_number)
-            # increment retry counter (best-effort; sending retries should be implemented elsewhere)
+        # If we have TTS configured and we received audio, synthesize audio reply and send audio
+        if audio_media:
+            try:
+                # Synthesize using ElevenLabs (blocking) in thread
+                loop = asyncio.get_event_loop()
+                tts_bytes, mime = await loop.run_in_executor(None, synthesize_eleven, reply_text)
+                # Convert to opus if needed
+                converted_tts = await convert_to_ogg_opus_async(tts_bytes)
+
+                if provider == 'meta':
+                    # Upload to Meta and send audio message
+                    from app.meta_client import MetaProvider
+                    meta = MetaProvider()
+                    media_id = meta.upload_media(converted_tts, mime_type='audio/ogg')
+                    # send audio message
+                    payload = {
+                        "messaging_product": "whatsapp",
+                        "to": from_number.replace('whatsapp:', ''),
+                        "type": "audio",
+                        "audio": {"id": media_id}
+                    }
+                    # POST to messages endpoint
+                    import requests
+                    headers = {"Authorization": f"Bearer {settings.META_TOKEN}", "Content-Type": "application/json"}
+                    r = requests.post(f"https://graph.facebook.com/v15.0/{settings.META_PHONE_NUMBER_ID}/messages", headers=headers, json=payload, timeout=10)
+                    r.raise_for_status()
+                else:
+                    # Upload to S3 and send via Twilio with media_url
+                    url = upload_bytes_to_s3(converted_tts, 'audio/ogg')
+                    await asyncio.to_thread(messaging_client.send, to=from_number, body=reply_text)  # send text + media
+                    # Twilio's send wrapper may not accept media url; use underlying client if available
+                    try:
+                        # prefer low-level twilio client
+                        tw = messaging_client._client
+                        tw.send(to=from_number, body=reply_text)
+                    except Exception:
+                        # fallback to TwilioProvider via messaging_client
+                        try:
+                            await asyncio.to_thread(messaging_client.send, to=from_number, body=reply_text)
+                        except Exception:
+                            logger.exception("Failed to send text via messaging client")
+                    # If provider supports media_url param, attempt to call Twilio directly
+                    try:
+                        from app.twilio_client import TwilioProvider
+                        tprov = TwilioProvider()
+                        # Twilio SDK: messages.create(body=..., from_=..., to=..., media_url=[url])
+                        await asyncio.to_thread(tprov.client.messages.create, body=reply_text, from_=settings.TWILIO_WHATSAPP_NUMBER, to=from_number, media_url=[url])
+                    except Exception:
+                        logger.exception("Failed to send Twilio media message")
+
+                # Persist reply to convo store
+                await store.append_message(from_number, {"text": reply_text, "type": "voice_reply"})
+            except Exception:
+                logger.exception("Failed to synthesize/send audio reply; falling back to text")
+                try:
+                    await asyncio.to_thread(messaging_client.send, to=from_number, body=reply_text)
+                except Exception:
+                    logger.exception("Failed to send fallback text reply")
+        else:
+            # No audio: send text reply as before
             try:
-                inc_message_retry(getattr(messaging_client, "_client", type(messaging_client)).__class__.__name__)
+                await asyncio.to_thread(messaging_client.send, to=from_number, body=reply_text)
+                # persist
+                await store.append_message(from_number, {"text": reply_text, "type": "bot_reply"})
             except Exception:
-                inc_message_retry("unknown")
+                logger.exception("Failed to send message to %s", from_number)
+                try:
+                    inc_message_retry(getattr(messaging_client, "_client", type(messaging_client)).__class__.__name__)
+                except Exception:
+                    inc_message_retry("unknown")
 
     async def enqueue(self, item: Dict[str, Any]):
         await self.queue.put(item)