From 33f2d8278ecaf549f7f658a611229257ea23b8da Mon Sep 17 00:00:00 2001 From: BadStorm Date: Fri, 11 Sep 2026 22:15:39 +0200 Subject: [PATCH] feat(talkbot): split voice/text queues and trim verbose log output --- containers/talkbot/server.py | 73 ++++++++++++++++++++++++++++-------- 1 file changed, 58 insertions(+), 15 deletions(-) diff --git a/containers/talkbot/server.py b/containers/talkbot/server.py index 26261ec..13ee874 100644 --- a/containers/talkbot/server.py +++ b/containers/talkbot/server.py @@ -38,6 +38,7 @@ LOG_WEBHOOK_PAYLOAD = ( os.environ.get("LOG_WEBHOOK_PAYLOAD", "false").lower() in {"1", "true", "yes", "on"} ) +LOG_TEXT_PREVIEW_CHARS = int(os.environ.get("LOG_TEXT_PREVIEW_CHARS") or "120") QWEN_TTS_URL = os.environ.get("QWEN_TTS_URL", "http://qwen-tts:8000").rstrip("/") QWEN_TTS_TIMEOUT = int(os.environ.get("QWEN_TTS_TIMEOUT") or "1800") @@ -85,9 +86,14 @@ MANAGER_USER = NC_USER app = FastAPI(title="Nextcloud Talk voice translator") -# Events are processed one at a time by a single background worker: the -# pipeline (whisper -> llama.cpp -> qwen-tts) is heavy and the host has -# limited resources, so we serialize instead of processing concurrently. +# Events are processed by two background workers, one per job type: +# - "voice" jobs run the heavy whisper -> llama.cpp -> qwen-tts pipeline +# and are processed one at a time, since whisper/qwen-tts are the +# resource-heavy steps and the host has limited GPU/CPU for them. +# - "text" jobs only call llama.cpp for translation, which is fast and +# uses its own service, so they are processed on a separate worker and +# never wait behind a long-running voice job. +# Each worker still processes its own queue serially. # The webhook handler only validates/parses the payload (no blocking I/O) # and enqueues the job, so Nextcloud always gets an immediate response # and never times out waiting for a translation to finish. @@ -99,6 +105,10 @@ app = FastAPI(title="Nextcloud Talk voice translator") new_job_signal = threading.Event() +def job_type_of(job: dict) -> str: + return "voice" if job.get("file_param") else "text" + + def init_queue_db() -> None: os.makedirs(os.path.dirname(QUEUE_DB_PATH), exist_ok=True) @@ -127,16 +137,31 @@ def init_queue_db() -> None: if "next_attempt_at" not in existing_columns: conn.execute("ALTER TABLE queue ADD COLUMN next_attempt_at TEXT") + if "job_type" not in existing_columns: + conn.execute("ALTER TABLE queue ADD COLUMN job_type TEXT") + + # Backfill rows left over from before job_type existed so they + # are picked up by the right worker instead of neither. + for row_id, payload in conn.execute( + "SELECT id, payload FROM queue WHERE job_type IS NULL" + ): + job_type = job_type_of(json.loads(payload)) + conn.execute( + "UPDATE queue SET job_type = ? WHERE id = ?", + (job_type, row_id), + ) + conn.commit() def enqueue_event(job: dict) -> None: with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: conn.execute( - "INSERT INTO queue (payload, created_at) VALUES (?, ?)", + "INSERT INTO queue (payload, created_at, job_type) VALUES (?, ?, ?)", ( json.dumps(job), datetime.now(timezone.utc).isoformat(), + job_type_of(job), ), ) conn.commit() @@ -144,17 +169,18 @@ def enqueue_event(job: dict) -> None: new_job_signal.set() -def dequeue_next_event() -> Optional[tuple[int, dict, int]]: +def dequeue_next_event(job_type: str) -> Optional[tuple[int, dict, int]]: now = datetime.now(timezone.utc).isoformat() with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: row = conn.execute( """ SELECT id, payload, attempts FROM queue - WHERE next_attempt_at IS NULL OR next_attempt_at <= ? + WHERE job_type = ? + AND (next_attempt_at IS NULL OR next_attempt_at <= ?) ORDER BY id LIMIT 1 """, - (now,), + (job_type, now), ).fetchone() if not row: @@ -245,6 +271,13 @@ def language_short_code(language: str) -> str: return LANGUAGE_CODES.get(code, code.upper()[:8]) +def truncate_for_log(text: str) -> str: + text = text or "" + if len(text) <= LOG_TEXT_PREVIEW_CHARS: + return text + return text[:LOG_TEXT_PREVIEW_CHARS] + "…" + + def sanitize_filename_part(value: str) -> str: value = (value or "").strip() value = re.sub(r"[^\w.-]+", "_", value, flags=re.UNICODE) @@ -1170,7 +1203,7 @@ def process_voice_message( log.info( "Transcribed (%s): %s", detected_language, - text, + truncate_for_log(text), ) if is_blank_transcription(text): @@ -1232,7 +1265,7 @@ def process_voice_message( log.info( "Translated (%s): %s", recipient_language_name, - translated_text, + truncate_for_log(translated_text), ) translations.append( @@ -1416,7 +1449,7 @@ def process_event(job: dict) -> None: "Translated text for %s (%s): %s", recipient, language_name(target_language), - translated, + truncate_for_log(translated), ) if translated.strip().lower() == message_text.lower(): @@ -1472,9 +1505,9 @@ def notify_processing_failure(job: dict) -> None: ) -def event_worker() -> None: +def event_worker(job_type: str) -> None: while True: - job_row = dequeue_next_event() + job_row = dequeue_next_event(job_type) if job_row is None: new_job_signal.wait(timeout=5) @@ -1488,7 +1521,8 @@ def event_worker() -> None: process_event(job) except Exception as exc: log.exception( - "Error processing message %s (attempt %s/%s): %s", + "Error processing %s message %s (attempt %s/%s): %s", + job_type, job.get("message_id"), attempts, MAX_ATTEMPTS, @@ -1500,7 +1534,8 @@ def event_worker() -> None: continue log.error( - "Giving up on message %s after %s attempts", + "Giving up on %s message %s after %s attempts", + job_type, job.get("message_id"), attempts, ) @@ -1522,7 +1557,15 @@ def start_event_worker() -> None: threading.Thread( target=event_worker, - name="talkbot-event-worker", + args=("voice",), + name="talkbot-voice-worker", + daemon=True, + ).start() + + threading.Thread( + target=event_worker, + args=("text",), + name="talkbot-text-worker", daemon=True, ).start()