diff --git a/packages/feature-import/src/jobs/transcribe.ts b/packages/feature-import/src/jobs/transcribe.ts new file mode 100644 index 0000000..7afd802 --- /dev/null +++ b/packages/feature-import/src/jobs/transcribe.ts @@ -0,0 +1,82 @@ +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import type { JobContext } from "@stc/contracts"; +import { finish, managedCourse, saveManifest, stageDir, transcriptCacheDir, type Deps } from "../context"; +import { run } from "../util/exec"; +import { optStr, withStage } from "./common"; + +export interface TranscribeResult { courseId: string; skipped?: boolean; reason?: string; transcribed: number; warnings: string[] } +interface Segment { start: number; end: number; text: string } + +const stamp = (s: number) => `${Math.floor(s / 60)}:${String(Math.floor(s % 60)).padStart(2, "0")}`; + +/** Groups timed segments into paragraphs of roughly 30 seconds, each starting with a [m:ss] marker. */ +export function segmentsToMarkdown(title: string, segments: Segment[], fallbackText: string): string { + if (segments.length === 0) return `# ${title}\n\n${fallbackText.trim()}\n`; + const paras: string[] = []; + let cur: string[] = []; + let startedAt = segments[0]!.start; + for (const s of segments) { + if (cur.length && s.start - startedAt >= 30) { + paras.push(`[${stamp(startedAt)}] ${cur.join(" ")}`); + cur = []; startedAt = s.start; + } + cur.push(s.text.trim()); + } + if (cur.length) paras.push(`[${stamp(startedAt)}] ${cur.join(" ")}`); + return `# ${title}\n\n${paras.join("\n\n")}\n`; +} + +/** + * Optional job: sends each video lesson's audio to a Whisper-compatible endpoint (STC_TRANSCRIBE_URL, OpenAI + * /v1/audio/transcriptions shape) and attaches the result as the lesson's transcript. The model never runs in this process. + */ +export async function transcribeCourse(d: Deps, payload: unknown, jc: JobContext): Promise { + const p = (payload ?? {}) as Record; + const url = d.env.STC_TRANSCRIBE_URL; + const courseId = optStr(p, "courseId") ?? ""; + if (!url) return { courseId, skipped: true, reason: "STC_TRANSCRIBE_URL is not set", transcribed: 0, warnings: [] }; + const course = await managedCourse(d, courseId); + const only = Array.isArray(p.lessonIds) ? new Set(p.lessonIds.map(String)) : null; + const lessons = (course.manifest.modules as any[]).flatMap((m) => m.lessons as any[]).filter((l) => l.type === "video" && !l.transcript && (!only || only.has(l.id))); + const warnings: string[] = []; + const stage = await stageDir(d, "transcribe"); + let done = 0; + return withStage(stage, async () => { + try { + for (const [i, l] of lessons.entries()) { + if (jc.signal.aborted) break; + await jc.reportProgress(i / Math.max(1, lessons.length), `Transcribing ${l.title}`); + try { + const audio = join(stage, `${l.id}.ogg`); + await run([d.tools.ffmpeg, "-v", "error", "-y", "-i", join(course.dir, l.file), "-vn", "-ac", "1", "-ar", "16000", "-c:a", "libopus", "-b:a", "24k", audio], { signal: jc.signal }); + const form = new FormData(); + form.set("file", Bun.file(audio), `${l.id}.ogg`); + form.set("model", d.env.STC_TRANSCRIBE_MODEL || "whisper-1"); + form.set("response_format", "verbose_json"); + const res = await fetch(url, { + method: "POST", body: form, signal: jc.signal, + headers: d.env.STC_TRANSCRIBE_KEY ? { Authorization: `Bearer ${d.env.STC_TRANSCRIBE_KEY}` } : {}, + }); + if (!res.ok) throw new Error(`HTTP ${res.status} ${(await res.text()).slice(0, 200)}`); + const body = (await res.json()) as { text?: string; segments?: Segment[] }; + const segments = (body.segments ?? []).map((s) => ({ start: Number(s.start) || 0, end: Number(s.end) || 0, text: String(s.text ?? "") })); + const rel = `transcripts/${l.id}.md`; + await mkdir(join(course.dir, "transcripts"), { recursive: true }); + await writeFile(join(course.dir, rel), segmentsToMarkdown(l.title, segments, body.text ?? "")); + await mkdir(transcriptCacheDir(d, course.id), { recursive: true }); + await writeFile(join(transcriptCacheDir(d, course.id), `${l.id}.json`), JSON.stringify({ lessonId: l.id, text: body.text ?? "", segments })); + l.transcript = rel; + done++; + } catch (e) { + warnings.push(`${l.title}: ${e instanceof Error ? e.message : String(e)}`); + } + } + } finally { + if (done > 0) await saveManifest(course.dir, course.manifest); // keep what finished even if the job was cancelled + } + if (done > 0) await finish(d, course.id, warnings); + await jc.reportProgress(1, `Transcribed ${done} of ${lessons.length} lessons`); + return { courseId: course.id, transcribed: done, warnings }; + }); +} diff --git a/packages/feature-import/test/transcribe.test.ts b/packages/feature-import/test/transcribe.test.ts new file mode 100644 index 0000000..1273e1d --- /dev/null +++ b/packages/feature-import/test/transcribe.test.ts @@ -0,0 +1,57 @@ +import { afterAll, expect, test } from "bun:test"; +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { importPacktVideo } from "../src/jobs/packt-video"; +import { segmentsToMarkdown, transcribeCourse } from "../src/jobs/transcribe"; +import { makeEnv } from "./harness"; +import { makePacktZip, tmp } from "./fixtures"; + +let calls = 0; +let auth: string | null = null; +const server = Bun.serve({ + port: 0, + async fetch(req) { + calls++; + auth = req.headers.get("authorization"); + const form = await req.formData(); + const f = form.get("file") as File; + if (!f || f.size < 100 || form.get("model") !== "whisper-1") return new Response("bad", { status: 400 }); + return Response.json({ text: "hello world", segments: [{ start: 0, end: 2, text: " hello" }, { start: 35, end: 40, text: " world" }] }); + }, +}); +afterAll(() => server.stop(true)); + +test("segments are grouped into timestamped paragraphs", () => { + expect(segmentsToMarkdown("T", [{ start: 0, end: 1, text: "a" }, { start: 31, end: 32, text: "b" }, { start: 33, end: 34, text: "c" }], "")).toBe("# T\n\n[0:00] a\n\n[0:31] b c\n"); +}); + +test("skipped when STC_TRANSCRIBE_URL is unset", async () => { + const env = await makeEnv(); + const r = await transcribeCourse(env.deps, { courseId: "x" }, env.jc); + expect(r).toMatchObject({ skipped: true, transcribed: 0 }); +}); + +test("transcribes video lessons through a Whisper-compatible endpoint", async () => { + const env = await makeEnv({ STC_TRANSCRIBE_URL: `http://127.0.0.1:${server.port}/v1/audio/transcriptions`, STC_TRANSCRIBE_KEY: "k" }); + const imp = await importPacktVideo(env.deps, { path: await makePacktZip(await tmp()), options: {} }, env.jc); + const r = await transcribeCourse(env.deps, { courseId: imp.courseId }, env.jc); + expect(r.warnings).toEqual([]); + expect(r.transcribed).toBe(3); + expect(calls).toBe(3); + expect(auth).toBe("Bearer k"); + const c = env.deps.ctx.courses.get(imp.courseId)!; + expect(c.status).toBe("ok"); + expect(c.manifest!.modules[0]!.lessons[0]!.transcript).toBe("transcripts/m1-l1.md"); + expect(await env.deps.ctx.courses.readLessonMarkdown(imp.courseId, "m1-l1")).toContain("[0:35] world"); + expect(JSON.parse(await readFile(join(env.dataDir, "cache/transcripts", imp.courseId, "m1-l1.json"), "utf8")).text).toBe("hello world"); + // a second run has nothing left to do + expect((await transcribeCourse(env.deps, { courseId: imp.courseId }, env.jc)).transcribed).toBe(0); +}); + +test("an endpoint failure becomes a warning, not a failed job", async () => { + const env = await makeEnv({ STC_TRANSCRIBE_URL: `http://127.0.0.1:1/x` }); + const imp = await importPacktVideo(env.deps, { path: await makePacktZip(await tmp()), options: {} }, env.jc); + const r = await transcribeCourse(env.deps, { courseId: imp.courseId }, env.jc); + expect(r.transcribed).toBe(0); + expect(r.warnings).toHaveLength(3); +});