Spaces:
Sleeping
Sleeping
File size: 4,926 Bytes
e4bb15d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | from typing import Any
import requests
from app.agent.orchestrator import Orchestrator
from app.voice.stt import SttClient, SttProviderError
class TelegramHandler:
def __init__(
self,
orchestrator: Orchestrator,
stt_client: SttClient,
bot_token: str,
) -> None:
self.orchestrator = orchestrator
self.stt_client = stt_client
self.bot_token = bot_token
def process_update(self, update: dict[str, Any]) -> dict[str, Any]:
message = update.get("message") or {}
chat = message.get("chat") or {}
chat_id = chat.get("id")
if chat_id is None:
return {"ok": True, "skipped": "no_chat"}
telegram_id = str(chat_id)
if "text" in message:
text = str(message.get("text", ""))
if text.strip().lower() == "/debug last_update_type":
last_type = self.orchestrator.repo.get_last_input_type_for_telegram_user(
telegram_id
)
current_type = "text"
debug_msg = (
f"debug current_update_type={current_type}, "
f"last_processed_input_type={last_type or 'none'}"
)
self._send_message(chat_id, debug_msg)
return {
"ok": True,
"input_type": "text",
"reply": debug_msg,
"debug": True,
}
reply = self.orchestrator.handle_user_text(telegram_id, text, input_type="text")
self._send_message(chat_id, reply)
return {"ok": True, "input_type": "text", "reply": reply}
if "voice" in message:
transcript = self._transcribe_media(message["voice"], "voice.ogg")
reply = self.orchestrator.handle_user_text(
telegram_id, transcript, input_type="voice"
)
self._send_message(chat_id, f"Transcribed: {transcript}\n\n{reply}")
return {"ok": True, "input_type": "voice", "reply": reply, "transcript": transcript}
if "audio" in message:
transcript = self._transcribe_media(message["audio"], "audio.m4a")
reply = self.orchestrator.handle_user_text(
telegram_id, transcript, input_type="voice"
)
self._send_message(chat_id, f"Transcribed: {transcript}\n\n{reply}")
return {"ok": True, "input_type": "voice", "reply": reply, "transcript": transcript}
return {"ok": True, "skipped": "unsupported_message_type"}
def _transcribe_media(self, media_obj: dict[str, Any], default_filename: str) -> str:
file_id = media_obj.get("file_id")
if not file_id or not self.bot_token:
return "voice transcription unavailable in local mode"
try:
file_resp = requests.get(
f"https://api.telegram.org/bot{self.bot_token}/getFile",
params={"file_id": file_id},
timeout=20,
)
file_resp.raise_for_status()
file_path = file_resp.json().get("result", {}).get("file_path")
if not file_path:
return "unable to fetch media file path from Telegram"
dl_resp = requests.get(
f"https://api.telegram.org/file/bot{self.bot_token}/{file_path}", timeout=30
)
dl_resp.raise_for_status()
filename = file_path.split("/")[-1] if "/" in file_path else default_filename
if "." not in filename:
filename = default_filename
transcript = self.stt_client.transcribe_bytes(dl_resp.content, filename=filename)
return transcript or "empty transcription"
except SttProviderError as exc:
return (
"voice transcription failed. "
f"provider_detail={exc}. "
"Tip: Telegram voice notes are usually OGG; if this keeps failing, send an audio file in m4a/mp3."
)
except requests.RequestException:
return "voice transcription failed due to Telegram network issue"
def _send_message(self, chat_id: int, text: str) -> None:
if not self.bot_token:
print("ERROR: BOT_TOKEN is totally empty or missing in environment variables!")
return
try:
print(f"Attempting to send message to {chat_id}...")
response = requests.post(
f"https://api.telegram.org/bot{self.bot_token}/sendMessage",
json={"chat_id": chat_id, "text": text, "parse_mode": "HTML"},
timeout=20,
)
print(f"Telegram API Response: HTTP {response.status_code} - {response.text}")
response.raise_for_status()
except requests.RequestException as e:
print(f"Telegram Request Failed: {e}")
|