Cut a long file into chunks a hosted request can outlive

An hour and a half of speech came back as "OpenRouter: HTTP 502: The
operation was aborted due to timeout", every time. The upload limit was
the only thing deciding where the file was cut, and mp3 at 48 kbps
reaches 24 MB after an hour, so a 90 minute file became two chunks and
the first one was 63 minutes of audio in a single request. Nothing
between here and the model stays on the line that long.

A chunk is capped at fifteen minutes now, whatever it weighs, and the
call is finally handed a timeout of its own: it was going out on the 300
second default sized for a dictation, which the same chunk would have
hit first anyway.

The other half is not throwing the run away when one request fails.
ApiError carries whether a second try can fix it, which is true of the
statuses a gateway raises itself and of a dropped connection, and false
of a rejected key. A chunk is asked for three times, waiting five then
ten seconds, with the Stop button still able to get through. If it does
fail in the end, what was already heard goes to the output box rather
than the bin, with the status line saying where it stops.

dikte/api.py

@@ -58,10 +58,20 @@ def timestamp_model(provider, selected=""):
     return "openai/whisper-1" if provider == "openrouter" else "whisper-1"
 
 
+# What a gateway in front of the model answers of its own accord: the request
+# never reached the model, or the model was still working when the connection
+# was given up on. Trying again is the only thing that fixes any of them, and
+# with a long file it is worth the second try rather than losing the run.
+RETRY_STATUS = frozenset({408, 429, 500, 502, 503, 504})
+
+
 class ApiError(Exception):
-    def __init__(self, message, status=None):
+    def __init__(self, message, status=None, retryable=None):
         super().__init__(message)
         self.status = status
+        # Anything not on that list is the request itself being wrong, and it
+        # will be just as wrong the second time.
+        self.retryable = status in RETRY_STATUS if retryable is None else retryable
 
 
 class Aborted(Exception):
@@ -218,7 +228,7 @@ def explain(exc, service):
     if exc.status == 429:
         return ApiError(t("{service} is rate limiting you (HTTP 429). Try again in "
                           "a moment.", service=service), exc.status)
-    return ApiError(f"{service}: {exc}", exc.status)
+    return ApiError(f"{service}: {exc}", exc.status, retryable=exc.retryable)
 
 
 def _request(url, data, headers, timeout=120, aborter=None):
@@ -234,8 +244,11 @@ def _request(url, data, headers, timeout=120, aborter=None):
         # not the network failing. URLError is an OSError, so both land here.
         if aborter is not None and aborter.aborted:
             raise Aborted from None
+        # A connection that dropped or timed out is the same bad minute as a
+        # 502, so it is worth the same second try.
         raise ApiError(t("Could not connect: {reason}",
-                         reason=getattr(exc, "reason", exc))) from exc
+                         reason=getattr(exc, "reason", exc)),
+                       retryable=True) from exc
     except json.JSONDecodeError as exc:
         raise ApiError(t("Could not parse the response: {error}", error=exc)) from exc
 
@@ -318,7 +331,7 @@ def local_failure(service, server, exc):
     """
     detail = server.error()
     return ApiError(f"{service}: {exc}" + (f" ({detail})" if detail else ""),
-                    exc.status)
+                    exc.status, retryable=exc.retryable)
 
 
 def _transcribe_request(target, audio_path, language, prompt, response_format,

dikte/filetranscribe.py

@@ -1,15 +1,19 @@
 """Transcribe an existing audio/video file with the same models.
 
 ffmpeg converts whatever comes in to 16 kHz mono WAV, and for a hosted API to
-mp3 on top of that. The upload limit is the only reason a file is ever cut up,
-and uncompressed audio reaches it after ten minutes where mp3 takes an hour.
+mp3 on top of that. Two things decide where a file is cut up: the upload limit,
+which uncompressed audio reaches after ten minutes where mp3 takes an hour, and
+the clock. An hour of audio in one request is minutes of work at the other end,
+and the gateway in front of the model hangs up long before the answer comes
+back, which arrives here as a 502 with the whole chunk lost. So a chunk is also
+capped at MAX_CHUNK_SECONDS however small it is on disk.
 
-That is worth the encoder, because a cut is not free. Whisper hears in thirty
-second windows and decides for itself where one cue ends and the next begins; a
-chunk that starts in the middle of a sentence can come back as one cue per
-window, twenty seconds of text at a time, for the whole rest of the chunk. So
-the file is cut as rarely as the limit allows, what is cut overlaps, and
-stitch() drops the half that was heard twice.
+A cut is not free, which is what the encoder buys and why nothing is cut more
+finely than that. Whisper hears in thirty second windows and decides for itself
+where one cue ends and the next begins; a chunk that starts in the middle of a
+sentence can come back as one cue per window, twenty seconds of text at a time,
+for the whole rest of the chunk. So what is cut overlaps, and stitch() drops
+the half that was heard twice.
 """
 
 import contextlib
@@ -19,6 +23,7 @@ import shutil
 import subprocess
 import tempfile
 import threading
+import time
 import wave
 
 from PyQt6.QtCore import QObject, pyqtSignal
@@ -29,10 +34,14 @@ from . import ggml
 from .i18n import t
 
 UPLOAD_LIMIT = 24 * 1024 * 1024  # the APIs take 25 MB; leave the form its room
+MAX_CHUNK_SECONDS = 900      # as much audio as a hosted request can outlive
 MP3_BITRATE = "48k"          # mono speech at 16 kHz: whisper hears nothing less
 OVERLAP_SECONDS = 30         # a whisper window: how far back a chunk starts
 WAV_CHUNK_SECONDS = 600      # 19 MB, for the caller that uploads the WAV itself
 CLEANUP_CHUNK_CHARS = 12000  # keep each cleanup call comfortably small
+HOSTED_TIMEOUT = 600         # a quarter hour of audio, with room for the upload
+RETRIES = 3                  # how many times one chunk is asked for in all
+RETRY_WAIT = 5               # seconds before the second try, doubled after that
 RATE = 16000
 MIN_SUBTITLE_SECONDS = 1.5   # how long a cue with no end time of its own stays up
 
@@ -86,9 +95,40 @@ class FileTranscriber(QObject):
     def _check(self):
         self._abort.check()
 
+    def _wait(self, seconds):
+        """Sleep on it, with the Stop button still able to get through."""
+        deadline = time.monotonic() + seconds
+        while time.monotonic() < deadline:
+            self._check()
+            time.sleep(0.25)
+        self._check()
+
+    def _attempt(self, call, stage):
+        """`call`, asked again when what failed was the network rather than us.
+
+        One chunk is a quarter hour of audio that took a minute to encode and a
+        minute to upload, so a gateway having a bad moment is worth waiting out
+        rather than throwing the run away over. `stage` is what the status line
+        said before the failure, put back once the wait is over.
+        """
+        for attempt in range(1, RETRIES + 1):
+            self._check()
+            try:
+                return call()
+            except api.ApiError as exc:
+                if attempt == RETRIES or not exc.retryable:
+                    raise
+                self.progress.emit(t(
+                    "{error} Trying again ({attempt}/{total})…",
+                    error=exc, attempt=attempt + 1, total=RETRIES))
+                self._wait(RETRY_WAIT * 2 ** (attempt - 1))
+                self.progress.emit(stage)
+
     def _work(self, path, timestamps, do_cleanup):
         conf = self.conf
         workdir = None
+        pieces = []
+        segments = []
         try:
             if not shutil.which("ffmpeg"):
                 raise api.ApiError(t("ffmpeg not found. Install it to transcribe files."))
@@ -104,39 +144,36 @@ class FileTranscriber(QObject):
             if len(chunks) > 1:
                 self.progress.emit(t("Splitting into {count} chunks…", count=len(chunks)))
 
-            pieces = []
-            segments = []
             for index, (chunk_path, offset) in enumerate(chunks, start=1):
                 self._check()
-                self.progress.emit(
-                    t("Transcribing chunk {index}/{count}…",
-                      index=index, count=len(chunks))
-                    if len(chunks) > 1 else t("Transcribing…")
-                )
+                stage = (t("Transcribing chunk {index}/{count}…",
+                           index=index, count=len(chunks))
+                         if len(chunks) > 1 else t("Transcribing…"))
+                self.progress.emit(stage)
                 if timestamps:
-                    segments = stitch(segments, [
-                        (start + offset, end + offset, line)
-                        for start, end, line in api.transcribe_segments(
-                            target,
-                            chunk_path,
-                            language=conf["language"],
-                            prompt=conf["transcribe_prompt"],
-                            aborter=self._abort,
-                        )
-                    ])
-                else:
-                    pieces.append(api.transcribe(
+                    heard = self._attempt(lambda: api.transcribe_segments(
                         target,
                         chunk_path,
                         language=conf["language"],
                         prompt=conf["transcribe_prompt"],
+                        timeout=HOSTED_TIMEOUT,
                         aborter=self._abort,
-                    ))
+                    ), stage)
+                    segments = stitch(segments, [
+                        (start + offset, end + offset, line)
+                        for start, end, line in heard
+                    ])
+                else:
+                    pieces.append(self._attempt(lambda: api.transcribe(
+                        target,
+                        chunk_path,
+                        language=conf["language"],
+                        prompt=conf["transcribe_prompt"],
+                        timeout=HOSTED_TIMEOUT,
+                        aborter=self._abort,
+                    ), stage))
 
-            if timestamps:
-                pieces = [f"[{format_timestamp(start)}] {line}"
-                          for start, _, line in segments]
-            text = "\n".join(pieces) if timestamps else " ".join(pieces)
+            text = _joined(pieces, segments, timestamps)
 
             if do_cleanup and text:
                 self._check()
@@ -148,7 +185,16 @@ class FileTranscriber(QObject):
         except Cancelled:
             self.progress.emit(t("Stopped."))
         except (api.ApiError, OSError, subprocess.SubprocessError, wave.Error) as exc:
-            self.failed.emit(str(exc))
+            # An hour of a long file already heard is not worth throwing away
+            # because the chunk after it failed, or because cleanup did. Hand
+            # over what there is, and say in the same breath where it stops.
+            partial = _joined(pieces, segments, timestamps)
+            if partial:
+                self.finished.emit(partial, segments)
+                self.failed.emit(t("{error} The transcript up to there is below.",
+                                   error=exc))
+            else:
+                self.failed.emit(str(exc))
         finally:
             self._local = None
             if workdir:
@@ -181,12 +227,21 @@ class FileTranscriber(QObject):
         self._local = ggml.llm if cleanup.provider(conf) == "local" else None
         prompt = conf.cleanup_prompt(with_timestamps=timestamps, subtitles=True)
         out = []
+        stage = t("Cleaning up…")
         for block in split_text(text, timestamps):
             self._check()
-            out.append(cleanup.run(block, conf, prompt, aborter=self._abort))
+            out.append(self._attempt(
+                lambda: cleanup.run(block, conf, prompt, aborter=self._abort), stage))
         return ("\n" if timestamps else "\n\n").join(out)
 
 
+def _joined(pieces, segments, timestamps):
+    """The transcript as one string, out of whichever of the two is holding it."""
+    if timestamps:
+        pieces = [f"[{format_timestamp(start)}] {line}" for start, _, line in segments]
+    return "\n".join(pieces) if timestamps else " ".join(pieces)
+
+
 def format_timestamp(seconds):
     seconds = int(seconds)
     hours, rest = divmod(seconds, 3600)
@@ -309,13 +364,20 @@ def wav_seconds(wav_path):
 def chunk_seconds(path, duration):
     """How many seconds of this audio fit in one request, or 0 when all of it does.
 
-    Measured rather than worked out: what an encoder makes of an hour of speech
-    depends on the speech, and the file on disk is the only honest answer.
+    Whichever of the two limits bites first. How much fits under the upload
+    limit is measured rather than worked out: what an encoder makes of an hour
+    of speech depends on the speech, and the file on disk is the only honest
+    answer. The other limit is MAX_CHUNK_SECONDS, and it is the one that catches
+    a long file at this bitrate: an hour and a half of mp3 is two chunks by size
+    and one of them is an hour of audio in a single request, which no hosted
+    gateway stays on the line for.
     """
-    size = os.path.getsize(path)
-    if size <= UPLOAD_LIMIT or duration <= 0:
+    if duration <= 0:
         return 0.0
-    return max(60.0, duration * UPLOAD_LIMIT / size * 0.95)
+    size = os.path.getsize(path)
+    fits = duration * UPLOAD_LIMIT / size * 0.95 if size > UPLOAD_LIMIT else duration
+    seconds = max(60.0, min(fits, MAX_CHUNK_SECONDS))
+    return 0.0 if seconds >= duration else seconds
 
 
 def split_wav(wav_path, workdir, seconds=WAV_CHUNK_SECONDS, overlap=OVERLAP_SECONDS):

dikte/i18n.py

@@ -297,6 +297,10 @@ TR = {
     "Converting audio…": "Ses dönüştürülüyor…",
     "Splitting into {count} chunks…": "{count} parçaya bölünüyor…",
     "Transcribing chunk {index}/{count}…": "{index}/{count} parça yazıya çevriliyor…",
+    "{error} Trying again ({attempt}/{total})…":
+        "{error} Yeniden deneniyor ({attempt}/{total})…",
+    "{error} The transcript up to there is below.":
+        "{error} Oraya kadar çevrilen metin aşağıda.",
     "Done: {chars} characters.": "Bitti: {chars} karakter.",
     "Stopped.": "Durduruldu.",
     "Failed: {error}": "Başarısız: {error}",

tests/test_api.py

@@ -79,6 +79,33 @@ class Explain(DikteTest):
     def test_the_status_is_carried_through(self):
         self.assertEqual(self.error(429).status, 429)
 
+    def test_so_is_whether_it_is_worth_asking_again(self):
+        self.assertTrue(self.error(502).retryable)
+        self.assertFalse(self.error(401).retryable)
+
+
+class Retryable(unittest.TestCase):
+    """Which failures a second try can fix, and which will fail the same way."""
+
+    def test_a_gateway_that_gave_up_waiting(self):
+        for status in (408, 429, 500, 502, 503, 504):
+            with self.subTest(status=status):
+                self.assertTrue(api.ApiError("x", status).retryable)
+
+    def test_a_request_that_was_wrong(self):
+        for status in (400, 401, 402, 403, 404, 413, 422):
+            with self.subTest(status=status):
+                self.assertFalse(api.ApiError("x", status).retryable)
+
+    def test_an_error_of_our_own_is_not_the_network(self):
+        self.assertFalse(api.ApiError("Transcript came back empty.").retryable)
+
+    def test_a_connection_that_dropped_is_worth_a_second_try(self):
+        with fake_urlopen(url_error("connection reset")):
+            with self.assertRaises(api.ApiError) as caught:
+                api._request("https://example.test", b"{}", {})
+        self.assertTrue(caught.exception.retryable)
+
 
 class ExtractError(unittest.TestCase):
     def test_the_usual_shape(self):

tests/test_filetranscribe.py

@@ -214,10 +214,22 @@ class ChunkSeconds(DikteTest):
         self.assertEqual(ft.chunk_seconds(self.file(1024), 600), 0.0)
 
     def test_a_file_over_the_limit_is_cut_by_what_it_measured(self):
-        # Twice the limit over an hour, so a little under half an hour fits.
-        seconds = ft.chunk_seconds(self.file(ft.UPLOAD_LIMIT * 2), 3600)
-        self.assertGreater(seconds, 1500)
-        self.assertLess(seconds, 1800)
+        # Twice the limit over twenty minutes, so a little under ten fits.
+        seconds = ft.chunk_seconds(self.file(ft.UPLOAD_LIMIT * 2), 1200)
+        self.assertGreater(seconds, 500)
+        self.assertLess(seconds, 600)
+
+    def test_a_chunk_is_never_more_audio_than_a_request_can_outlive(self):
+        """An hour in one request is a 502 from the gateway, whatever it weighs."""
+        self.assertEqual(ft.chunk_seconds(self.file(ft.UPLOAD_LIMIT * 2), 3600),
+                         ft.MAX_CHUNK_SECONDS)
+
+    def test_a_small_file_that_is_still_hours_long_is_cut_on_the_clock(self):
+        self.assertEqual(ft.chunk_seconds(self.file(1024), 7200),
+                         ft.MAX_CHUNK_SECONDS)
+
+    def test_a_file_short_enough_on_both_counts_is_not_cut(self):
+        self.assertEqual(ft.chunk_seconds(self.file(1024), ft.MAX_CHUNK_SECONDS), 0.0)
 
     def test_a_file_with_no_length_is_left_whole(self):
         self.assertEqual(ft.chunk_seconds(self.file(ft.UPLOAD_LIMIT * 2), 0), 0.0)
@@ -378,6 +390,58 @@ class Transcriber(DikteTest):
         worker.stop()
         self.assertTrue(worker._abort.aborted)
 
+    def test_a_chunk_is_given_longer_to_answer_than_a_dictation(self):
+        """A quarter hour of audio is not a sentence: the default would cut it off."""
+        worker = ft.FileTranscriber(self.conf)
+        with mock.patch.object(ft, "_to_wav", side_effect=lambda *a: self.source), \
+                mock.patch.object(ft, "_to_mp3",
+                                  side_effect=lambda path, *a, **k: path), \
+                mock.patch.object(ft.shutil, "which", return_value="/usr/bin/ffmpeg"), \
+                mock.patch.object(api, "transcribe", return_value="text") as call:
+            worker._work(self.source, False, False)
+        self.assertEqual(call.call_args.kwargs["timeout"], ft.HOSTED_TIMEOUT)
+
+    def test_a_gateway_having_a_bad_moment_is_asked_again(self):
+        with mock.patch.object(ft.FileTranscriber, "_wait"):
+            done, failures, progress, _ = self.run_chain(
+                fail=[api.ApiError("HTTP 502: timeout", 502), "raw text"])
+        self.assertEqual(failures, [])
+        self.assertEqual(done[0][0], "raw text")
+        self.assertTrue(any("Trying again" in message for message in progress))
+
+    def test_a_rejected_key_is_not_asked_again(self):
+        """Trying again with the same key is only a slower way to fail."""
+        call = mock.Mock(side_effect=api.ApiError("rejected the API key", 401))
+        with mock.patch.object(ft.FileTranscriber, "_wait"):
+            _, failures, _, _ = self.run_chain(fail=call)
+        self.assertEqual(call.call_count, 1)
+        self.assertIn("rejected", failures[0])
+
+    def test_a_chunk_is_given_up_on_after_the_last_try(self):
+        call = mock.Mock(side_effect=api.ApiError("HTTP 502: timeout", 502))
+        with mock.patch.object(ft.FileTranscriber, "_wait"):
+            _, failures, _, _ = self.run_chain(fail=call)
+        self.assertEqual(call.call_count, ft.RETRIES)
+        self.assertIn("502", failures[0])
+
+    def test_what_was_heard_before_the_failure_is_still_handed_over(self):
+        """An hour already transcribed is not thrown away over the chunk after it."""
+        boom = api.ApiError("HTTP 502: timeout", 502)
+        with mock.patch.object(ft.FileTranscriber, "_wait"), \
+                mock.patch.object(ft.FileTranscriber, "_chunks",
+                                  side_effect=lambda wav, *a: [(wav, 0.0), (wav, 10.0)]):
+            done, failures, _, _ = self.run_chain(
+                fail=["first half"] + [boom] * ft.RETRIES)
+        self.assertEqual(done[0][0], "first half")
+        self.assertIn("502", failures[0])
+
+    def test_nothing_heard_at_all_is_a_plain_failure(self):
+        call = mock.Mock(side_effect=api.ApiError("rejected the API key", 401))
+        with mock.patch.object(ft.FileTranscriber, "_wait"):
+            done, failures, _, _ = self.run_chain(fail=call)
+        self.assertEqual(done, [])
+        self.assertEqual(failures[0], "rejected the API key")
+
     def test_a_second_start_while_one_is_running_is_ignored(self):
         worker = ft.FileTranscriber(self.conf)
         worker._thread = mock.Mock(is_alive=lambda: True)