diff --git a/app/audio_stt.py b/app/audio_stt.py new file mode 100644 index 0000000..bbe7d46 --- /dev/null +++ b/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) diff --git a/app/audio_tts.py b/app/audio_tts.py new file mode 100644 index 0000000..0cf78ab --- /dev/null +++ b/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") diff --git a/app/config.py b/app/config.py index e3ffda2..3861aac 100644 --- a/app/config.py +++ b/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" diff --git a/app/meta_client.py b/app/meta_client.py index 5fe1f70..2bb14b5 100644 --- a/app/meta_client.py +++ b/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= - - 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) diff --git a/app/s3_utils.py b/app/s3_utils.py new file mode 100644 index 0000000..7372d91 --- /dev/null +++ b/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 diff --git a/app/worker.py b/app/worker.py index b661cc3..b52a3f3 100644 --- a/app/worker.py +++ b/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) diff --git a/docs/voice.md b/docs/voice.md new file mode 100644 index 0000000..f7edd58 --- /dev/null +++ b/docs/voice.md @@ -0,0 +1,30 @@ +# Voice support documentation + +This document describes how voice (voice notes / audio messages) are handled by the bot. + +Overview + +- Incoming voice notes are detected by `app.parse_message.normalize_message()` and included as `media` entries in the normalized message. +- The background worker downloads the audio, converts it to opus/ogg using `ffmpeg` (if available), transcribes it via OpenAI Whisper (`app.audio_stt`), appends the transcript to the conversation store, runs the LLM reply generator (with conversation context), synthesizes the reply using ElevenLabs (`app.audio_tts`), and sends the audio reply back to the sender (Meta via media upload, Twilio via S3-hosted media URL). + +Environment variables (add to `.env`) + +- OPENAI_API_KEY - required for STT (Whisper) and LLM replies +- ELEVENLABS_API_KEY - required for ElevenLabs TTS +- S3_BUCKET, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY - required for hosting media for Twilio +- META_TOKEN, META_PHONE_NUMBER_ID - required for Meta uploads & sends + +Requirements + +- ffmpeg must be installed on the worker host for reliable audio format conversion. If ffmpeg is not present, the worker will attempt to use the original audio bytes. +- boto3 (for S3 uploads) +- requests and openai python SDK + +Notes & privacy + +- Ensure you have consent to transcribe and synthesize users' voice messages. +- Transcripts and audio replies are stored in the conversation store (Redis) by default; configure retention and access controls as needed. + +Testing + +- You can test by sending a voice note to the WhatsApp sandbox (Twilio) or via Meta; the webhook will put a job on the queue and the worker will process it. Monitor logs for STT/TTS steps.