feat: use official iCourse transcript when available, skip ASR

Adds get_transcript_segments() to ICourseClient returning timed
segments.  In LectureRunner, probes the official transcript API before
firing up ASR — if the transcript exists and has no silence gaps >20
minutes, uses it directly.  Saves ~5 min of ASR CPU per lecture.

Segments use the same {start_ms, end_ms, text} format as our ASR,
so the bucketer works unchanged.

Author: LeafCreeper

src/api/icourse.py

@@ -362,11 +362,25 @@ def get_lecture_detail(self, course_id: str, sub_id: str) -> dict:
         )
 
     def get_transcript(self, sub_id: str) -> str | None:
-        """Get the transcript text for a lecture.
+        """Get the transcript text for a lecture (flat string).
 
         Returns the full transcript text, empty string if no transcript,
         or None on error.
         """
+        segments = self.get_transcript_segments(sub_id)
+        if segments is None:
+            return None
+        if not segments:
+            return ""
+        return " ".join(s["text"] for s in segments if s["text"])
+
+    def get_transcript_segments(self, sub_id: str) -> list[dict] | None:
+        """Get transcript as timed segments.  Returns None on API error,
+        empty list if no transcript exists.
+
+        Each segment: {"start_ms": int, "end_ms": int, "text": str}
+        Sorted by start_ms ascending.
+        """
         url = f"{self.base_url}/courseapi/v3/web-socket/search-trans-result"
         resp = self.vpn.get(
             url, params={"sub_id": sub_id, "format": "json"}
@@ -379,15 +393,23 @@ def get_transcript(self, sub_id: str) -> str | None:
 
         result_list = data.get("list", [])
         if not result_list:
-            return ""
+            return []
 
         all_content = result_list[0].get("all_content", [])
         if not all_content:
-            return ""
-
-        all_content.sort(key=lambda x: x.get("BeginSec", 0))
-        return " ".join(
-            seg.get("Text", "") for seg in all_content if seg.get("Text")
+            return []
+
+        return sorted(
+            (
+                {
+                    "start_ms": int(seg.get("BeginSec", 0)) * 1000,
+                    "end_ms": int(seg.get("EndSec", seg.get("BeginSec", 0))) * 1000,
+                    "text": seg.get("Text", ""),
+                }
+                for seg in all_content
+                if seg.get("Text", "").strip()
+            ),
+            key=lambda s: s["start_ms"],
         )
 
     def get_sub_detail(self, course_id: str, sub_id: str) -> dict:

src/pipeline/lecture_runner.py

@@ -173,15 +173,26 @@ def _schedule_next(self, next_info: Optional[tuple[str, str]]):
         # semaphore until a download slot frees.
         self._scheduler.prefetch_lecture(self._client, next_course, next_sub)
 
+    @staticmethod
+    def _official_transcript_usable(segments: list[dict] | None,
+                                    max_gap_minutes: int = 20) -> bool:
+        """True if the official transcript is complete enough to use."""
+        if not segments:
+            return False
+        max_gap_s = 0
+        for i in range(1, len(segments)):
+            gap = segments[i]["start_ms"] - segments[i - 1]["end_ms"]
+            if gap > max_gap_s:
+                max_gap_s = gap
+        return max_gap_s <= max_gap_minutes * 60_000
+
     def _get_transcript(self, existing: dict | None, course_id: str,
                         sub_id: str) -> tuple[Optional[str], Optional[list]]:
         """Return (transcript, segments) or (None, None) on skip.
 
-        Reuses an existing transcript if present (segments==None — bucketer
-        falls back to flat mode).  Otherwise pulls the prefetched audio
-        handle, runs ``transcribe_tail``, and writes the transcript.  On
-        ``NoAudioStreamError`` returns (None, None) after marking the
-        lecture as a deliberate skip.
+        Tries the official iCourse transcript first — when available and
+        complete-enough (no >20 min silence gaps) it replaces the ASR
+        step entirely, saving ~5 min of CPU time per lecture.
         """
         if existing and existing.get("transcript"):
             self._reporter.info(
@@ -191,6 +202,20 @@ def _get_transcript(self, existing: dict | None, course_id: str,
             )
             return existing["transcript"], None
 
+        # Try official transcript before firing up ASR.
+        try:
+            official = self._client.get_transcript_segments(sub_id)
+            if self._official_transcript_usable(official):
+                text = " ".join(s["text"] for s in official)
+                self._reporter.info(
+                    f"    Using official transcript "
+                    f"({len(text)} chars, {len(official)} segments)"
+                )
+                self._db.update_transcript(sub_id, text)
+                return text, official
+        except Exception:
+            pass  # fall through to ASR
+
         # Pull the audio handle.  ``schedule`` is idempotent — usually the
         # previous lecture already kicked it off (Phase C), but for the
         # first lecture in the batch we still need to fire it ourselves.