From 44214b6084909b62b404616e3016f5f7baf3eee7 Mon Sep 17 00:00:00 2001 From: BadStorm Date: Fri, 11 Sep 2026 21:54:50 +0200 Subject: [PATCH] feat(talkbot): reprocess cited messages and raise pipeline timeouts --- containers/talkbot/server.py | 143 ++++++++++++++++++++++++++- containers/talkbot/talkbot.container | 14 ++- 2 files changed, 153 insertions(+), 4 deletions(-) diff --git a/containers/talkbot/server.py b/containers/talkbot/server.py index fa2ac14..26261ec 100644 --- a/containers/talkbot/server.py +++ b/containers/talkbot/server.py @@ -24,6 +24,7 @@ NC_USER = os.environ["NC_USER"] NC_PASSWORD = os.environ["NC_PASSWORD"] WHISPER_URL = os.environ.get("WHISPER_URL", "http://whisper:8080").rstrip("/") +WHISPER_TIMEOUT = int(os.environ.get("WHISPER_TIMEOUT") or "300") LLAMACPP_URL = os.environ.get("LLAMACPP_URL", "http://llamacpp:7000").rstrip("/") LLAMACPP_MODEL = os.environ.get("LLAMACPP_MODEL", "").strip() LLAMACPP_API_KEY = os.environ.get("LLAMACPP_API_KEY", "").strip() @@ -328,6 +329,28 @@ def extract_file_param(parameters) -> Optional[dict]: return None +def is_bot_mentioned(parameters) -> bool: + if isinstance(parameters, dict): + values = list(parameters.values()) + elif isinstance(parameters, list): + values = parameters + else: + values = [] + + for value in values: + if not isinstance(value, dict) or value.get("type") != "user": + continue + + mentioned_id = str(value.get("id", "") or "") + if "/" in mentioned_id: + mentioned_id = mentioned_id.split("/", 1)[-1] + + if mentioned_id == MANAGER_USER: + return True + + return False + + def is_audio_file(file_param: dict) -> bool: mimetype = str(file_param.get("mimetype", "") or "").lower() if mimetype.startswith("audio/"): @@ -370,6 +393,80 @@ def extract_reply_to_message_id(obj: dict) -> Optional[int]: return None +def fetch_chat_message(token: str, message_id: int) -> Optional[dict]: + response = httpx.get( + f"{NC_URL}/ocs/v2.php/apps/spreed/api/v1/chat/{token}", + headers={ + "OCS-APIRequest": "true", + "Accept": "application/json", + }, + auth=(NC_USER, NC_PASSWORD), + params={ + "lookIntoFuture": 0, + "limit": 1, + "lastKnownMessageId": message_id, + "includeLastKnown": 1, + }, + timeout=30, + ) + + if response.status_code == 304: + return None + + response.raise_for_status() + + messages = response.json().get("ocs", {}).get("data", []) + + for message in messages: + if isinstance(message, dict) and int(message.get("id", -1)) == message_id: + return message + + return None + + +def build_reprocess_job(token: str, message_id: int) -> Optional[dict]: + """Build a job that reprocesses an already-posted message. + + Used when someone replies to a message mentioning the bot instead of + sending a new message: we fetch the cited message from Nextcloud Talk + and re-run the pipeline on it exactly as if it had just arrived, be it + a voice message or a text message. + """ + message = fetch_chat_message(token, message_id) + + if not message: + return None + + parameters = message.get("messageParameters", {}) + + file_param = extract_file_param(parameters) + if file_param and not is_audio_file(file_param): + file_param = None + + message_text = "" if file_param else str(message.get("message") or "").strip() + + if not file_param and not message_text: + return None + + sender_user_id = actor_user_id({"id": message.get("actorId", "")}) + sender_display_name = ( + message.get("actorDisplayName") + or sender_user_id + or "User" + ) + + return { + "token": token, + "message_id": int(message_id), + "sender_user_id": sender_user_id, + "sender_display_name": sender_display_name, + "file_param": file_param, + "message_text": message_text, + "preserve_original": False, + "reply_to_message_id": None, + } + + def get_conversation_participants(token: str) -> list[dict]: response = httpx.get( f"{NC_URL}/ocs/v2.php/apps/spreed/api/v4/room/{token}/participants", @@ -610,7 +707,7 @@ def transcribe(audio_bytes: bytes) -> tuple[str, str]: data={ "response_format": "json", }, - timeout=300, + timeout=WHISPER_TIMEOUT, ) if response.is_error: @@ -1435,6 +1532,7 @@ def health(): return { "status": "ok", "manager": MANAGER_USER, + "whisper_timeout": WHISPER_TIMEOUT, "llamacpp_timeout": LLAMACPP_TIMEOUT, "llamacpp_thinking": LLAMACPP_ENABLE_THINKING, "qwen_tts_timeout": QWEN_TTS_TIMEOUT, @@ -1551,6 +1649,49 @@ async def webhook( parameters, ) + if ( + not file_param + and reply_to_message_id + and is_bot_mentioned(parameters) + ): + reprocess_job = build_reprocess_job( + token, + reply_to_message_id, + ) + + if reprocess_job: + enqueue_event(reprocess_job) + + log.info( + "Queued reprocessing of message %s (requested by %s citing " + "the bot in message %s)", + reply_to_message_id, + sender_user_id, + message_id, + ) + + return JSONResponse( + { + "status": "queued", + "type": "voice-reprocess" if reprocess_job["file_param"] else "text-reprocess", + "reprocessed_message_id": reply_to_message_id, + } + ) + + log.info( + "Message %s mentions the bot but cited message %s has no " + "content to reprocess", + message_id, + reply_to_message_id, + ) + + return JSONResponse( + { + "status": "ignored", + "reason": "cited message has no content to reprocess", + } + ) + preserve_original = False if file_param and not is_audio_file(file_param): diff --git a/containers/talkbot/talkbot.container b/containers/talkbot/talkbot.container index 01280f1..6d65d1a 100644 --- a/containers/talkbot/talkbot.container +++ b/containers/talkbot/talkbot.container @@ -31,19 +31,27 @@ Environment=NC_BOT_SECRET=changeme Environment=NC_USER=changeme Environment=NC_PASSWORD=changeme -# --- Backend services (adjust host:port to match your setup/network) --- +# Whisper Environment=WHISPER_URL=http://whisper:8080 +Environment=WHISPER_TIMEOUT=1800 + +# LLAMAcpp Environment=LLAMACPP_URL=http://llamacpp:8090 #Environment=LLAMACPP_MODEL= #Environment=LLAMACPP_API_KEY= Environment=LLAMACPP_MAX_TOKENS=8192 Environment=LLAMACPP_ENABLE_THINKING=false -#Environment=LOG_WEBHOOK_PAYLOAD=false +Environment=LLAMACPP_TIMEOUT=1800 + +# QWEN Environment=QWEN_TTS_URL=http://qwen-tts:8000 +Environment=QWEN_TTS_TIMEOUT=5400 +Environment=QWEN_TTS_SPEAKER=Ryan + # --- Behaviour --- +#Environment=LOG_WEBHOOK_PAYLOAD=false Environment=DEFAULT_TARGET_LANGUAGE=it -Environment=QWEN_TTS_SPEAKER=Ryan [Service] Restart=on-failure