Add YouTube -> tutorial .docx pipeline (Gradio Space)

Search top videos, rank by comment sentiment, transcribe the best with
faster-whisper, generate tutorial text with DeepSeek-V3, pick screenshots via a
weighted LLM/Whisper-timing indicator, caption them with a vision model, and
assemble a .docx. Downloaded video is auto-deleted after frame extraction.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,14 @@
+__pycache__/
+*.pyc
+.venv/
+venv/
+*.docx
+*.mp4
+*.webm
+*.wav
+*.jpg
+*.jpeg
+*.png
+runs/
+.gradio/
+.DS_Store

README.md

@@ -12,4 +12,43 @@ license: mit
 short_description: Make Tutorials from YouTube Videos
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# YouTube → Tutorial Post Generator
+
+Give a **topic** and your **Hugging Face token**, and this Space builds a downloadable
+**`.docx` tutorial** (text + AI-captioned screenshots) from the single best YouTube video
+on that topic.
+
+## Pipeline
+
+1. **Search** — top 5 videos via the [`adarshajay/youtube-search`](https://huggingface.co/spaces/adarshajay/youtube-search) Space.
+2. **Sentiment rank** — fetch top comments per video with `yt-dlp` and score them with the
+   BERT classifier [`OmarMedhat7/youtube-sentiment-analysis-model`](https://huggingface.co/OmarMedhat7/youtube-sentiment-analysis-model);
+   the video with the highest positive share wins.
+3. **Download** the winner with `yt-dlp` and extract audio with `ffmpeg`.
+4. **Transcribe** locally with `faster-whisper` (segment timestamps).
+5. **Candidate frames** are extracted densely from the video, then the **video is
+   auto-deleted** — only small JPEGs and the transcript remain.
+6. **Tutorial text** — `deepseek-ai/DeepSeek-V3` (HF Inference Providers, billed to your
+   token) turns the transcript into structured steps.
+7. **Screenshot selection** — a weighted indicator blends each step's LLM-suggested
+   timestamp with Whisper's actual speech timing.
+8. **Captions** — a vision model (default `Qwen/Qwen2.5-VL-72B-Instruct`, billed to your
+   token) captions each screenshot.
+9. **Assemble** the `.docx` for download.
+
+## Notes
+
+- **Your HF token is used only for the LLM and vision-model calls** and is billed to your
+  account. Create a fine-grained token with *"Make calls to Inference Providers"* at
+  <https://huggingface.co/settings/tokens>.
+- **Free CPU tier:** Whisper runs on CPU, so transcription is slow — keep videos short
+  (default cap ~20 min).
+- **YouTube may block the Space's IP.** If downloads/comments fail, add a Netscape-format
+  cookie file as a Space secret named `YT_COOKIES` (the file's contents).
+
+## Local run
+
+```bash
+pip install -r requirements.txt   # needs ffmpeg on PATH
+python app.py
+```

app.py

@@ -0,0 +1,213 @@
+"""Gradio Space: YouTube topic -> captioned .docx tutorial.
+
+Orchestrates the pipeline stages and streams progress/status to the UI. Heavy ML
+imports (torch/transformers/faster-whisper) are lazy inside the pipeline modules, so app
+startup stays fast.
+"""
+from __future__ import annotations
+
+import os
+import re
+import shutil
+import tempfile
+
+import gradio as gr
+
+from pipeline import (
+    captions as captions_mod,
+    docx_builder,
+    download as download_mod,
+    frames as frames_mod,
+    search as search_mod,
+    sentiment as sentiment_mod,
+    transcribe as transcribe_mod,
+    tutorial as tutorial_mod,
+)
+
+LLM_CHOICES = [
+    "deepseek-ai/DeepSeek-V3",
+    "meta-llama/Llama-3.3-70B-Instruct",
+    "openai/gpt-oss-120b",
+]
+VLM_CHOICES = [
+    "Qwen/Qwen2.5-VL-72B-Instruct",
+    "Qwen/Qwen2.5-VL-7B-Instruct",
+    "meta-llama/Llama-3.2-90B-Vision-Instruct",
+]
+
+
+def _cookiefile(workdir: str) -> str | None:
+    """Materialize the YT_COOKIES secret (Netscape cookie file contents) to disk."""
+    data = os.environ.get("YT_COOKIES")
+    if not data:
+        return None
+    path = os.path.join(workdir, "cookies.txt")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(data)
+    return path
+
+
+def _ranking_rows(scored: list[dict]) -> list[list]:
+    rows = []
+    for rank, v in enumerate(scored, start=1):
+        rows.append([
+            rank,
+            v.get("title", v["video_id"]),
+            f"{v['positive_share'] * 100:.0f}%",
+            v.get("n_comments", 0),
+            v.get("note", "") or "ok",
+            v["url"],
+        ])
+    return rows
+
+
+def _safe_name(text: str) -> str:
+    return re.sub(r"[^A-Za-z0-9._-]+", "_", text).strip("_")[:60] or "tutorial"
+
+
+def run_pipeline(topic, hf_token, llm_model, vlm_model, w_llm, w_whisper, lead,
+                 max_minutes, max_shots, progress=gr.Progress()):
+    """Generator that yields (status_md, ranking_df, transcript, docx_file)."""
+    log: list[str] = []
+
+    def status(msg: str):
+        log.append(msg)
+        return "\n\n".join(log)
+
+    topic = (topic or "").strip()
+    if not topic:
+        raise gr.Error("Please enter a topic.")
+    if not (hf_token or "").strip():
+        raise gr.Error("Please paste your Hugging Face token (used for the LLM + vision model).")
+
+    workdir = tempfile.mkdtemp(prefix="ytt_")
+    frames_dir = os.path.join(workdir, "frames")
+    video_path = None
+    try:
+        cookiefile = _cookiefile(workdir)
+
+        # 1. Search ------------------------------------------------------------------
+        progress(0.02, desc="Searching")
+        yield status(f"🔍 Searching top videos for **{topic}**…"), gr.update(), gr.update(), gr.update()
+        videos = search_mod.search_top5(topic)
+        yield status(f"Found {len(videos)} candidate videos."), gr.update(), gr.update(), gr.update()
+
+        # 2. Sentiment ranking -------------------------------------------------------
+        yield status("💬 Fetching comments and scoring sentiment…"), gr.update(), gr.update(), gr.update()
+        best, scored = sentiment_mod.rank_by_sentiment(videos, cookiefile, progress)
+        ranking = gr.update(value=_ranking_rows(scored))
+        yield (status(f"🏆 Picked **{best.get('title', best['video_id'])}** "
+                      f"({best['positive_share'] * 100:.0f}% positive)."),
+               ranking, gr.update(), gr.update())
+
+        # 3. Download + audio --------------------------------------------------------
+        progress(0.25, desc="Downloading")
+        yield status("⬇️ Downloading the chosen video…"), ranking, gr.update(), gr.update()
+        video_path, duration = download_mod.download_video(
+            best["url"], workdir, cookiefile, int(max_minutes))
+        wav = download_mod.extract_audio(video_path, workdir)
+
+        # 4. Transcribe --------------------------------------------------------------
+        progress(0.4, desc="Transcribing")
+        yield status("📝 Transcribing with Whisper (this is the slow part on CPU)…"), ranking, gr.update(), gr.update()
+        segs = transcribe_mod.transcribe(wav, progress)
+        transcript = transcribe_mod.transcript_text(segs)
+        yield (status(f"Transcript ready ({len(segs)} segments)."),
+               ranking, gr.update(value=transcript), gr.update())
+
+        # 5. Candidate frames, then DELETE the video --------------------------------
+        progress(0.6, desc="Extracting frames")
+        candidates = frames_mod.extract_candidates(video_path, frames_dir, duration)
+        frames_mod.delete_video(video_path)
+        video_path = None
+        yield (status(f"🎞️ Extracted {len(candidates)} candidate frames and "
+                      f"**deleted the downloaded video**."),
+               ranking, gr.update(value=transcript), gr.update())
+
+        # 6. Tutorial text -----------------------------------------------------------
+        progress(0.72, desc="Writing tutorial")
+        yield status(f"🤖 Generating tutorial with `{llm_model}`…"), ranking, gr.update(value=transcript), gr.update()
+        tut = tutorial_mod.generate_tutorial(transcript, hf_token.strip(), llm_model)
+
+        # 7. Weighted screenshot selection ------------------------------------------
+        selected = frames_mod.select_screenshots(
+            tut["steps"], segs, candidates,
+            w_llm=float(w_llm), w_whisper=float(w_whisper), lead=float(lead),
+            max_shots=int(max_shots),
+        )
+        yield (status(f"🖼️ Selected {len(selected)} screenshots via the weighted indicator."),
+               ranking, gr.update(value=transcript), gr.update())
+
+        # 8. Captions ----------------------------------------------------------------
+        progress(0.85, desc="Captioning")
+        yield status(f"✍️ Captioning screenshots with `{vlm_model}`…"), ranking, gr.update(value=transcript), gr.update()
+        caps = captions_mod.caption_frames(selected, tut["steps"], hf_token.strip(), vlm_model, progress)
+
+        # 9. DOCX --------------------------------------------------------------------
+        progress(0.95, desc="Building document")
+        out_path = os.path.join(workdir, f"{_safe_name(tut['title'])}.docx")
+        docx_builder.build_docx(tut, selected, caps, out_path, source_url=best["url"])
+
+        progress(1.0, desc="Done")
+        yield (status("✅ Done! Download your tutorial below."),
+               ranking, gr.update(value=transcript), gr.update(value=out_path))
+
+    except gr.Error:
+        raise
+    except (download_mod.DownloadError, RuntimeError, ValueError) as exc:
+        raise gr.Error(str(exc))
+    finally:
+        # Always remove the video if it somehow survived; keep frames/docx until the
+        # response is sent (Gradio copies the returned file out).
+        if video_path:
+            frames_mod.delete_video(video_path)
+
+
+def build_ui():
+    with gr.Blocks(title="YouTube → Tutorial Post") as demo:
+        gr.Markdown(
+            "# 📝 YouTube → Tutorial Post Generator\n"
+            "Enter a topic and your Hugging Face token. The Space picks the best video, "
+            "transcribes it, and builds a **captioned `.docx` tutorial**. Your token is "
+            "used only for the LLM + vision-model calls and **billed to your account**."
+        )
+        with gr.Row():
+            with gr.Column(scale=2):
+                topic = gr.Textbox(label="Topic", placeholder="e.g. Excel pivot tables for beginners")
+                hf_token = gr.Textbox(label="Hugging Face token", type="password",
+                                      placeholder="hf_…  (Inference Providers permission)")
+            with gr.Column(scale=1):
+                llm_model = gr.Dropdown(LLM_CHOICES, value=LLM_CHOICES[0],
+                                        label="Tutorial LLM", allow_custom_value=True)
+                vlm_model = gr.Dropdown(VLM_CHOICES, value=VLM_CHOICES[0],
+                                        label="Vision model (captions)", allow_custom_value=True)
+
+        with gr.Accordion("Advanced settings", open=False):
+            with gr.Row():
+                w_llm = gr.Slider(0.0, 1.0, value=0.4, step=0.05, label="Weight: LLM timestamp")
+                w_whisper = gr.Slider(0.0, 1.0, value=0.6, step=0.05, label="Weight: Whisper timing")
+                lead = gr.Slider(0.0, 5.0, value=1.0, step=0.5, label="Lead offset (s)")
+            with gr.Row():
+                max_minutes = gr.Slider(2, 60, value=20, step=1, label="Max video length (min)")
+                max_shots = gr.Slider(1, 15, value=8, step=1, label="Max screenshots")
+
+        run_btn = gr.Button("Generate tutorial", variant="primary")
+
+        status_md = gr.Markdown(label="Status")
+        ranking_df = gr.Dataframe(
+            headers=["#", "Title", "Positive", "Comments", "Note", "URL"],
+            label="Sentiment ranking", interactive=False, wrap=True,
+        )
+        transcript_box = gr.Textbox(label="Transcript preview", lines=10, max_lines=20, show_copy_button=True)
+        docx_file = gr.File(label="Download tutorial (.docx)")
+
+        run_btn.click(
+            run_pipeline,
+            inputs=[topic, hf_token, llm_model, vlm_model, w_llm, w_whisper, lead, max_minutes, max_shots],
+            outputs=[status_md, ranking_df, transcript_box, docx_file],
+        )
+    return demo
+
+
+if __name__ == "__main__":
+    build_ui().queue().launch()

packages.txt

@@ -0,0 +1 @@
+ffmpeg

pipeline/__init__.py

@@ -0,0 +1,6 @@
+"""Pipeline package: YouTube topic -> captioned .docx tutorial.
+
+Stages (see app.run_pipeline for orchestration):
+    search -> sentiment -> download -> transcribe -> frames (extract + delete video)
+    -> tutorial -> frames.select (weighted) -> captions -> docx_builder
+"""

pipeline/captions.py

@@ -0,0 +1,68 @@
+"""Stage 8: caption each selected screenshot with an HF vision model.
+
+Calls a vision-language chat model (default Qwen2.5-VL-72B) via HF Inference Providers,
+billed to the user's token. The image is sent inline as a base64 ``data:`` URL and the
+step heading is given as context so captions are relevant, not just literal. Failures
+degrade gracefully to the step heading so the .docx is always produced.
+"""
+from __future__ import annotations
+
+import base64
+
+from huggingface_hub import InferenceClient
+
+DEFAULT_VLM = "Qwen/Qwen2.5-VL-72B-Instruct"
+FALLBACK_VLM = "Qwen/Qwen2.5-VL-7B-Instruct"
+
+_PROMPT = (
+    "Write a single concise caption (one sentence, no preamble) describing what this "
+    "tutorial screenshot shows. Context for this step: {heading}."
+)
+
+
+def _data_url(path: str) -> str:
+    with open(path, "rb") as fh:
+        b64 = base64.b64encode(fh.read()).decode()
+    return f"data:image/jpeg;base64,{b64}"
+
+
+def _caption_one(client: InferenceClient, model: str, path: str, heading: str) -> str:
+    messages = [{
+        "role": "user",
+        "content": [
+            {"type": "text", "text": _PROMPT.format(heading=heading or "tutorial step")},
+            {"type": "image_url", "image_url": {"url": _data_url(path)}},
+        ],
+    }]
+    resp = client.chat.completions.create(model=model, messages=messages, max_tokens=120)
+    return (resp.choices[0].message.content or "").strip()
+
+
+def caption_frames(selected: dict[int, dict], steps: list[dict], hf_token: str,
+                   model: str = DEFAULT_VLM, progress=None) -> dict[int, str]:
+    """Return ``{step_index: caption}`` for each selected screenshot.
+
+    Tries ``model`` first, then ``FALLBACK_VLM``, then the step heading.
+    """
+    if not selected:
+        return {}
+    if not hf_token:
+        raise ValueError("An HF token is required for image captioning (billed to your key).")
+
+    client = InferenceClient(token=hf_token)
+    captions: dict[int, str] = {}
+    items = list(selected.items())
+    for n, (idx, sel) in enumerate(items):
+        heading = steps[idx].get("heading", "")
+        if progress:
+            progress((n + 1) / len(items), desc=f"Captioning {n + 1}/{len(items)}")
+        text = ""
+        for m in (model, FALLBACK_VLM):
+            try:
+                text = _caption_one(client, m, sel["path"], heading)
+                if text:
+                    break
+            except Exception:
+                continue
+        captions[idx] = text or heading or "Screenshot"
+    return captions

pipeline/docx_builder.py

@@ -0,0 +1,61 @@
+"""Stage 9: assemble the .docx tutorial (text + captioned screenshots)."""
+from __future__ import annotations
+
+import os
+
+from docx import Document
+from docx.enum.text import WD_ALIGN_PARAGRAPH
+from docx.shared import Inches, Pt, RGBColor
+
+
+def _fmt_ts(seconds: float) -> str:
+    m, s = divmod(int(seconds), 60)
+    h, m = divmod(m, 60)
+    return f"{h:d}:{m:02d}:{s:02d}" if h else f"{m:d}:{s:02d}"
+
+
+def build_docx(tutorial: dict, selected: dict[int, dict], captions: dict[int, str],
+               out_path: str, source_url: str = "") -> str:
+    """Write the tutorial to ``out_path`` and return it.
+
+    ``tutorial`` = {title, intro, steps}; ``selected`` = {step_idx: {time, path}};
+    ``captions`` = {step_idx: caption}.
+    """
+    doc = Document()
+    doc.add_heading(tutorial["title"], level=0)
+
+    if tutorial.get("intro"):
+        doc.add_paragraph(tutorial["intro"])
+
+    if source_url:
+        src = doc.add_paragraph()
+        run = src.add_run(f"Source video: {source_url}")
+        run.italic = True
+        run.font.size = Pt(9)
+        run.font.color.rgb = RGBColor(0x66, 0x66, 0x66)
+
+    for i, step in enumerate(tutorial["steps"], start=0):
+        doc.add_heading(f"{i + 1}. {step['heading']}", level=1)
+        if step.get("body"):
+            doc.add_paragraph(step["body"])
+
+        sel = selected.get(i)
+        if sel and os.path.exists(sel["path"]):
+            pic_par = doc.add_paragraph()
+            pic_par.alignment = WD_ALIGN_PARAGRAPH.CENTER
+            pic_par.add_run().add_picture(sel["path"], width=Inches(6))
+
+            cap_par = doc.add_paragraph()
+            cap_par.alignment = WD_ALIGN_PARAGRAPH.CENTER
+            caption = captions.get(i, "")
+            cap_run = cap_par.add_run(f"{caption}  ")
+            cap_run.italic = True
+            cap_run.font.size = Pt(9)
+            ts_run = cap_par.add_run(f"(@ {_fmt_ts(sel['time'])})")
+            ts_run.italic = True
+            ts_run.font.size = Pt(9)
+            ts_run.font.color.rgb = RGBColor(0x88, 0x88, 0x88)
+
+    os.makedirs(os.path.dirname(os.path.abspath(out_path)), exist_ok=True)
+    doc.save(out_path)
+    return out_path

pipeline/download.py

@@ -0,0 +1,98 @@
+"""Stage 3: download the chosen video and extract audio for Whisper.
+
+Downloads a <=720p mp4 with yt-dlp, then uses ffmpeg to produce a 16 kHz mono wav.
+Enforces a duration cap up front so a long video can't stall the free-CPU Space.
+"""
+from __future__ import annotations
+
+import os
+import subprocess
+
+from yt_dlp import YoutubeDL
+
+
+class DownloadError(RuntimeError):
+    """Raised for user-actionable download failures (blocked IP, too long, etc.)."""
+
+
+def probe_duration(url: str, cookiefile: str | None = None) -> float:
+    """Return the video duration in seconds without downloading."""
+    opts = {"skip_download": True, "quiet": True, "no_warnings": True}
+    if cookiefile:
+        opts["cookiefile"] = cookiefile
+    try:
+        with YoutubeDL(opts) as ydl:
+            info = ydl.extract_info(url, download=False)
+    except Exception as exc:
+        raise DownloadError(_blocked_hint(exc)) from exc
+    return float(info.get("duration") or 0.0)
+
+
+def download_video(url: str, out_dir: str, cookiefile: str | None = None,
+                   max_minutes: int = 20) -> tuple[str, float]:
+    """Download the video to ``out_dir``; return ``(mp4_path, duration_seconds)``."""
+    os.makedirs(out_dir, exist_ok=True)
+
+    duration = probe_duration(url, cookiefile)
+    if duration and duration > max_minutes * 60:
+        raise DownloadError(
+            f"Video is {duration / 60:.0f} min long, over the {max_minutes} min cap for "
+            "this free-CPU Space. Pick a shorter video or raise the cap."
+        )
+
+    out_tmpl = os.path.join(out_dir, "video.%(ext)s")
+    opts = {
+        "outtmpl": out_tmpl,
+        "format": "bestvideo[height<=720][ext=mp4]+bestaudio[ext=m4a]/best[height<=720]/best",
+        "merge_output_format": "mp4",
+        "quiet": True,
+        "no_warnings": True,
+        "noplaylist": True,
+    }
+    if cookiefile:
+        opts["cookiefile"] = cookiefile
+
+    try:
+        with YoutubeDL(opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+            path = ydl.prepare_filename(info)
+    except Exception as exc:
+        raise DownloadError(_blocked_hint(exc)) from exc
+
+    duration = float(info.get("duration") or duration or 0.0)
+
+    # merge_output_format may have rewritten the extension to .mp4
+    if not os.path.exists(path):
+        base, _ = os.path.splitext(path)
+        path = base + ".mp4"
+    if not os.path.exists(path):
+        raise DownloadError("yt-dlp reported success but no output file was found.")
+    return path, duration
+
+
+def extract_audio(video_path: str, out_dir: str) -> str:
+    """Extract 16 kHz mono wav from ``video_path`` via ffmpeg; return the wav path."""
+    wav_path = os.path.join(out_dir, "audio.wav")
+    cmd = [
+        "ffmpeg", "-y", "-i", video_path,
+        "-vn", "-ac", "1", "-ar", "16000", "-f", "wav", wav_path,
+    ]
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    if proc.returncode != 0 or not os.path.exists(wav_path):
+        raise DownloadError(
+            "ffmpeg failed to extract audio. Is ffmpeg installed (packages.txt)?\n"
+            + proc.stderr[-600:]
+        )
+    return wav_path
+
+
+def _blocked_hint(exc: Exception) -> str:
+    msg = str(exc)
+    lowered = msg.lower()
+    if any(k in lowered for k in ("sign in", "bot", "403", "429", "confirm you", "cookies")):
+        return (
+            "YouTube blocked this request (common from datacenter IPs like HF Spaces). "
+            "Add a Netscape cookie file as the Space secret YT_COOKIES and retry.\n"
+            f"Original error: {msg[:300]}"
+        )
+    return f"Download failed: {msg[:400]}"

pipeline/frames.py

@@ -0,0 +1,122 @@
+"""Stage 5 + 6: candidate frames and the weighted screenshot indicator.
+
+Flow:
+  * ``extract_candidates`` pulls a dense, downscaled grid of frames from the video in a
+    single ffmpeg pass, *before* the video is deleted.
+  * ``delete_video`` removes the source video (auto-delete once the transcript + frames
+    exist).
+  * ``select_screenshots`` runs *after* the LLM. For each step it computes a capture time
+    that blends the LLM's suggested timestamp with the Whisper segment timing of the text
+    the step quotes, then snaps that target to the nearest pre-extracted candidate frame.
+"""
+from __future__ import annotations
+
+import difflib
+import glob
+import os
+import subprocess
+
+# A 1 fps grid gives <=0.5 s snapping error. Capped so a long video can't explode disk.
+GRID_FPS = 1.0
+MAX_CANDIDATES = 1500
+
+
+def extract_candidates(video_path: str, out_dir: str, duration: float) -> list[tuple[float, str]]:
+    """Extract a downscaled candidate-frame grid; return sorted ``[(time_s, path), ...]``."""
+    os.makedirs(out_dir, exist_ok=True)
+    eff_fps = GRID_FPS
+    if duration and duration * GRID_FPS > MAX_CANDIDATES:
+        eff_fps = MAX_CANDIDATES / duration  # thin the grid for very long videos
+
+    pattern = os.path.join(out_dir, "cand_%06d.jpg")
+    cmd = [
+        "ffmpeg", "-y", "-i", video_path,
+        "-vf", f"fps={eff_fps:.6f},scale=-2:720",
+        "-q:v", "4", pattern,
+    ]
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    files = sorted(glob.glob(os.path.join(out_dir, "cand_*.jpg")))
+    if not files:
+        raise RuntimeError(
+            "ffmpeg extracted no candidate frames.\n" + proc.stderr[-600:]
+        )
+    # fps filter emits frame i (1-based) at ~ (i-1)/eff_fps seconds.
+    return [((i) / eff_fps, f) for i, f in enumerate(files)]
+
+
+def delete_video(video_path: str) -> None:
+    """Delete the downloaded video (and any sibling temp media) if present."""
+    for p in (video_path,):
+        try:
+            if p and os.path.exists(p):
+                os.remove(p)
+        except OSError:
+            pass
+
+
+def _clamp(x: float, lo: float, hi: float) -> float:
+    return max(lo, min(hi, x))
+
+
+def best_match_segment(quote: str, t_llm: float, segs: list[dict]) -> dict:
+    """Find the Whisper segment a step refers to.
+
+    Prefer the segment whose text best matches ``quote``; if there's no decent textual
+    match, fall back to the segment nearest ``t_llm``.
+    """
+    quote = (quote or "").strip().lower()
+    if quote:
+        scored = [
+            (difflib.SequenceMatcher(None, quote, s["text"].lower()).ratio(), s)
+            for s in segs
+        ]
+        ratio, best = max(scored, key=lambda t: t[0])
+        if ratio >= 0.3:
+            return best
+    return min(segs, key=lambda s: abs((s["start"] + s["end"]) / 2 - t_llm))
+
+
+def pick_time(step: dict, segs: list[dict], cand_times: list[float], *,
+              w_llm: float = 0.4, w_whisper: float = 0.6, lead: float = 1.0,
+              max_drift: float = 20.0) -> float:
+    """Weighted capture time, snapped to the nearest candidate frame.
+
+    ``t_whisper`` (segment mid + lead) is ground truth for *when* the referenced text is
+    spoken; ``t_llm`` is the model's guess of *what* matters. We blend them, but if the
+    LLM time drifts far from the matched segment we clamp it back into that segment so a
+    hallucinated timestamp can't drag the shot off-topic.
+    """
+    seg = best_match_segment(step.get("quote", ""), float(step.get("t_llm", 0.0)), segs)
+    t_whisper = (seg["start"] + seg["end"]) / 2 + lead
+    t_llm = float(step.get("t_llm", t_whisper))
+    if abs(t_llm - t_whisper) > max_drift:
+        t_llm = _clamp(t_llm, seg["start"], seg["end"])
+
+    total = w_llm + w_whisper or 1.0
+    t_target = (w_llm * t_llm + w_whisper * t_whisper) / total
+    return min(cand_times, key=lambda c: abs(c - t_target))
+
+
+def select_screenshots(steps: list[dict], segs: list[dict],
+                       candidates: list[tuple[float, str]], *,
+                       w_llm: float = 0.4, w_whisper: float = 0.6, lead: float = 1.0,
+                       max_shots: int = 8) -> dict[int, dict]:
+    """Return ``{step_index: {"time", "path"}}`` for the steps that get a screenshot.
+
+    When there are more steps than ``max_shots``, the highest-``importance`` steps win.
+    """
+    if not candidates:
+        return {}
+    cand_times = [t for t, _ in candidates]
+    time_to_path = dict(candidates)
+
+    order = sorted(range(len(steps)), key=lambda i: -float(steps[i].get("importance", 0.5)))
+    chosen = set(order[:max_shots])
+
+    out: dict[int, dict] = {}
+    for i, step in enumerate(steps):
+        if i not in chosen:
+            continue
+        t = pick_time(step, segs, cand_times, w_llm=w_llm, w_whisper=w_whisper, lead=lead)
+        out[i] = {"time": t, "path": time_to_path[t]}
+    return out

pipeline/search.py

@@ -0,0 +1,77 @@
+"""Stage 1: search the top videos for a topic via the adarshajay/youtube-search Space.
+
+The upstream Space exposes ``api_name="/youtube_search"``. It takes a single query
+string and returns a single plain-text blob listing the top 5 results (title, URL and
+video id only). We parse the ids/titles out of that text.
+"""
+from __future__ import annotations
+
+import re
+
+from gradio_client import Client
+
+SEARCH_SPACE = "adarshajay/youtube-search"
+SEARCH_API = "/youtube_search"
+
+# 11-char YouTube ids as they appear in watch?v=... or youtu.be/... URLs.
+_ID_RE = re.compile(r"(?:v=|youtu\.be/|/watch/|/embed/)([A-Za-z0-9_-]{11})")
+
+
+def _parse_titles(text: str, ids: list[str]) -> dict[str, str]:
+    """Best-effort: map each id to a nearby title line in the result blob.
+
+    The upstream format is not guaranteed, so this is intentionally forgiving: for each
+    id we take the longest non-URL line that appears on or just before the line holding
+    the id. Anything we can't resolve falls back to the id itself in ``search_top5``.
+    """
+    lines = text.splitlines()
+    titles: dict[str, str] = {}
+    for vid in ids:
+        idx = next((i for i, ln in enumerate(lines) if vid in ln), None)
+        if idx is None:
+            continue
+        window = lines[max(0, idx - 1): idx + 1]
+        cands = [ln.strip(" -*\t") for ln in window if "http" not in ln and vid not in ln]
+        cands = [c for c in cands if c]
+        if cands:
+            titles[vid] = max(cands, key=len)
+    return titles
+
+
+def search_top5(topic: str, max_results: int = 5) -> list[dict]:
+    """Return up to ``max_results`` unique videos for ``topic``.
+
+    Each item: ``{"video_id", "url", "title"}``. Raises RuntimeError if the upstream
+    Space is unreachable or returns no parseable video ids.
+    """
+    topic = (topic or "").strip()
+    if not topic:
+        raise ValueError("Please enter a topic to search for.")
+
+    try:
+        client = Client(SEARCH_SPACE)
+        raw = client.predict(topic, api_name=SEARCH_API)
+    except Exception as exc:  # network / Space down / API renamed
+        raise RuntimeError(
+            f"Could not reach the search Space '{SEARCH_SPACE}'. It may be sleeping or "
+            f"its API changed. Details: {exc}"
+        ) from exc
+
+    text = raw if isinstance(raw, str) else str(raw)
+    ids = list(dict.fromkeys(_ID_RE.findall(text)))  # preserve order, dedupe
+    if not ids:
+        raise RuntimeError(
+            "The search Space returned no recognizable YouTube video ids. Raw output:\n"
+            + text[:500]
+        )
+
+    ids = ids[:max_results]
+    titles = _parse_titles(text, ids)
+    return [
+        {
+            "video_id": vid,
+            "url": f"https://www.youtube.com/watch?v={vid}",
+            "title": titles.get(vid, vid),
+        }
+        for vid in ids
+    ]

pipeline/sentiment.py

@@ -0,0 +1,109 @@
+"""Stage 2: rank candidate videos by the sentiment of their top comments.
+
+We fetch comments with yt-dlp (no video download) and score them with the BERT text
+classifier ``OmarMedhat7/youtube-sentiment-analysis-model`` running inside the Space.
+The video with the highest positive share wins. Comment scraping is flaky, so any
+single video that fails simply scores 0 and we never abort the whole ranking.
+"""
+from __future__ import annotations
+
+import os
+from functools import lru_cache
+
+from yt_dlp import YoutubeDL
+
+SENTIMENT_MODEL = "OmarMedhat7/youtube-sentiment-analysis-model"
+MAX_COMMENTS = 30
+
+# Label strings vary by model; treat anything that looks positive/negative accordingly.
+_POSITIVE = {"positive", "pos", "label_2", "label_1"}
+_NEGATIVE = {"negative", "neg", "label_0"}
+
+
+@lru_cache(maxsize=1)
+def _classifier():
+    """Lazily build and cache the text-classification pipeline (heavy import)."""
+    from transformers import pipeline
+
+    return pipeline(
+        "text-classification",
+        model=SENTIMENT_MODEL,
+        truncation=True,
+        max_length=256,
+        top_k=None,  # return all label scores so we can read the dominant one
+    )
+
+
+def _polarity(scores) -> float:
+    """Map one classifier output to a polarity in [-1, 1].
+
+    ``scores`` is a list of ``{"label", "score"}`` (top_k=None). Positive labels add,
+    negative labels subtract; neutral/unknown contribute nothing.
+    """
+    val = 0.0
+    for s in scores:
+        label = str(s["label"]).strip().lower()
+        if label in _POSITIVE:
+            val += s["score"]
+        elif label in _NEGATIVE:
+            val -= s["score"]
+    return val
+
+
+def _fetch_comments(url: str, cookiefile: str | None) -> list[str]:
+    opts = {
+        "skip_download": True,
+        "getcomments": True,
+        "quiet": True,
+        "no_warnings": True,
+        "extractor_args": {"youtube": {"max_comments": [str(MAX_COMMENTS)], "comment_sort": ["top"]}},
+    }
+    if cookiefile:
+        opts["cookiefile"] = cookiefile
+    with YoutubeDL(opts) as ydl:
+        info = ydl.extract_info(url, download=False)
+    comments = info.get("comments") or []
+    texts = [c.get("text", "").strip() for c in comments if c.get("text")]
+    return texts[:MAX_COMMENTS]
+
+
+def rank_by_sentiment(videos: list[dict], cookiefile: str | None = None,
+                      progress=None) -> tuple[dict, list[dict]]:
+    """Score each video and return ``(best_video, scored)``.
+
+    ``scored`` mirrors ``videos`` with added keys: ``positive_share`` (0..1),
+    ``n_comments``, ``note``. Ordering of ``scored`` is by positive_share desc, with the
+    original search order as the tie-break / fallback.
+    """
+    if not videos:
+        raise ValueError("No videos to rank.")
+
+    clf = _classifier()
+    scored: list[dict] = []
+    for i, v in enumerate(videos):
+        if progress:
+            progress((i + 1) / len(videos), desc=f"Sentiment {i + 1}/{len(videos)}")
+        item = dict(v)
+        try:
+            comments = _fetch_comments(v["url"], cookiefile)
+            if comments:
+                results = clf(comments)  # list aligned with comments (each = list of labels)
+                pols = [_polarity(r) for r in results]
+                positives = sum(1 for p in pols if p > 0)
+                item["positive_share"] = positives / len(pols)
+                item["n_comments"] = len(comments)
+                item["note"] = ""
+            else:
+                item["positive_share"] = 0.0
+                item["n_comments"] = 0
+                item["note"] = "no comments fetched"
+        except Exception as exc:
+            item["positive_share"] = 0.0
+            item["n_comments"] = 0
+            item["note"] = f"comment fetch failed: {type(exc).__name__}"
+        item["search_rank"] = i
+        scored.append(item)
+
+    # Best = highest positive share; ties and all-zero fall back to original search order.
+    scored.sort(key=lambda d: (-d["positive_share"], d["search_rank"]))
+    return scored[0], scored

pipeline/transcribe.py

@@ -0,0 +1,48 @@
+"""Stage 4: timestamped transcription with faster-whisper (CPU/int8).
+
+Returns a list of segments ``{"start", "end", "text"}`` in seconds. The model is cached
+at module level so repeated runs in the same Space process don't reload weights.
+"""
+from __future__ import annotations
+
+import os
+from functools import lru_cache
+
+# Small model keeps CPU transcription tractable on the free tier. Override with
+# WHISPER_MODEL (e.g. "small", "tiny") via env if needed.
+WHISPER_MODEL = os.environ.get("WHISPER_MODEL", "base")
+
+
+@lru_cache(maxsize=1)
+def _model():
+    from faster_whisper import WhisperModel
+
+    return WhisperModel(WHISPER_MODEL, device="cpu", compute_type="int8")
+
+
+def transcribe(wav_path: str, progress=None) -> list[dict]:
+    """Transcribe ``wav_path`` and return timestamped segments."""
+    model = _model()
+    segments, info = model.transcribe(wav_path, vad_filter=True)
+
+    total = float(getattr(info, "duration", 0.0)) or None
+    segs: list[dict] = []
+    for s in segments:  # generator -> streamed, so we can report progress
+        text = s.text.strip()
+        if text:
+            segs.append({"start": float(s.start), "end": float(s.end), "text": text})
+        if progress and total:
+            progress(min(s.end / total, 1.0), desc="Transcribing")
+
+    if not segs:
+        raise RuntimeError("Whisper produced an empty transcript (no speech detected).")
+    return segs
+
+
+def transcript_text(segs: list[dict]) -> str:
+    """Render segments as ``[mm:ss] text`` lines for the LLM and the UI preview."""
+    lines = []
+    for s in segs:
+        m, sec = divmod(int(s["start"]), 60)
+        lines.append(f"[{m:02d}:{sec:02d}] {s['text']}")
+    return "\n".join(lines)

pipeline/tutorial.py

@@ -0,0 +1,129 @@
+"""Stage 7: transform the timestamped transcript into a structured tutorial.
+
+Uses an HF Inference Providers chat model (default DeepSeek-V3) billed to the user's
+token. The model is asked for strict JSON so the downstream weighted indicator and the
+.docx builder have stable fields to work with.
+"""
+from __future__ import annotations
+
+import json
+import re
+
+from huggingface_hub import InferenceClient
+
+DEFAULT_LLM = "deepseek-ai/DeepSeek-V3"
+
+# Rough char budget to stay clear of context limits on the free path. Long transcripts
+# are truncated (with a marker); good enough for a tutorial summary.
+MAX_TRANSCRIPT_CHARS = 24000
+
+_SYSTEM = (
+    "You are a technical writer. You convert a timestamped video transcript into a clear, "
+    "step-by-step written tutorial. You ALWAYS respond with a single JSON object and no "
+    "prose outside it."
+)
+
+_INSTRUCTIONS = """\
+Turn the transcript below into a tutorial. Return ONLY a JSON object with this schema:
+
+{
+  "title": "string - concise tutorial title",
+  "intro": "string - 2-4 sentence overview",
+  "steps": [
+    {
+      "heading": "string - short step title",
+      "body": "string - 1-3 paragraphs explaining this step in your own words",
+      "quote": "string - a short, near-verbatim snippet (<=120 chars) copied from the "
+               "transcript line this step is based on, used to locate the moment",
+      "t_llm": number,        // best timestamp IN SECONDS for an illustrative screenshot
+      "importance": number    // 0..1, how worth screenshotting this step is
+    }
+  ]
+}
+
+Rules:
+- 4 to 10 steps. Keep quotes copied from the transcript so they can be matched back.
+- t_llm must be within the transcript's time range.
+- Output valid JSON only. No markdown, no comments in the actual output.
+
+Transcript (each line is "[mm:ss] text"):
+---
+{transcript}
+---
+"""
+
+
+def _extract_json(text: str) -> dict:
+    """Parse the model output into a dict, tolerating code fences / stray prose."""
+    text = text.strip()
+    if text.startswith("```"):
+        text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.DOTALL).strip()
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        start, end = text.find("{"), text.rfind("}")
+        if start != -1 and end != -1 and end > start:
+            return json.loads(text[start:end + 1])
+        raise
+
+
+def _normalize(data: dict) -> dict:
+    """Coerce/clean fields so downstream stages never crash on missing keys."""
+    steps = []
+    for s in data.get("steps", []) or []:
+        try:
+            t = float(s.get("t_llm", 0) or 0)
+        except (TypeError, ValueError):
+            t = 0.0
+        try:
+            imp = float(s.get("importance", 0.5) or 0.5)
+        except (TypeError, ValueError):
+            imp = 0.5
+        steps.append({
+            "heading": str(s.get("heading", "Step")).strip() or "Step",
+            "body": str(s.get("body", "")).strip(),
+            "quote": str(s.get("quote", "")).strip(),
+            "t_llm": max(0.0, t),
+            "importance": min(1.0, max(0.0, imp)),
+        })
+    if not steps:
+        raise RuntimeError("The LLM returned no usable steps.")
+    return {
+        "title": str(data.get("title", "Tutorial")).strip() or "Tutorial",
+        "intro": str(data.get("intro", "")).strip(),
+        "steps": steps,
+    }
+
+
+def generate_tutorial(transcript: str, hf_token: str, model: str = DEFAULT_LLM) -> dict:
+    """Call the chat model and return a normalized ``{title, intro, steps}`` dict."""
+    if not hf_token:
+        raise ValueError("An HF token is required for the tutorial LLM (billed to your key).")
+
+    truncated = transcript[:MAX_TRANSCRIPT_CHARS]
+    if len(transcript) > MAX_TRANSCRIPT_CHARS:
+        truncated += "\n[... transcript truncated for length ...]"
+    prompt = _INSTRUCTIONS.replace("{transcript}", truncated)
+
+    client = InferenceClient(token=hf_token)
+    try:
+        resp = client.chat.completions.create(
+            model=model,
+            messages=[
+                {"role": "system", "content": _SYSTEM},
+                {"role": "user", "content": prompt},
+            ],
+            temperature=0.3,
+            max_tokens=4000,
+        )
+    except Exception as exc:
+        raise RuntimeError(f"Tutorial LLM call failed ({model}): {exc}") from exc
+
+    content = resp.choices[0].message.content or ""
+    try:
+        data = _extract_json(content)
+    except Exception as exc:
+        raise RuntimeError(
+            f"Could not parse JSON from the LLM. First 400 chars:\n{content[:400]}"
+        ) from exc
+    return _normalize(data)

requirements.txt

@@ -0,0 +1,9 @@
+gradio>=4.44
+gradio_client>=1.3
+huggingface_hub>=0.27
+yt-dlp>=2024.12.13
+faster-whisper>=1.1.0
+transformers>=4.46
+torch>=2.2
+python-docx>=1.1.2
+Pillow>=10.4