"""Gradio Space: YouTube topic -> captioned .docx tutorial (API-based acquisition). Pipeline: search top videos -> rank by YouTube Data API comment sentiment -> transcript via youtube-transcript-api -> DeepSeek-V3 tutorial -> real screenshots via yt-dlp stream URL + ffmpeg -ss (weighted timestamps) -> VLM captions -> .docx. No video download and no cookies/proxy/PO-token UI. A thin fallback remains via the optional Space secrets YT_COOKIES / YT_PROXY (used only to resolve the stream URL and the transcript when the Space's datacenter IP is blocked). """ from __future__ import annotations import base64 import binascii import os import re import shutil import tempfile import time def _install_proxy_ca() -> None: """Trust a self-signed proxy CA (secret ``YT_PROXY_CA``) alongside the system roots, so an HTTPS (TLS-wrapped) ``YT_PROXY`` validates. TLS-wrapping hides the target host (``CONNECT www.youtube.com``) from the Space's egress DPI, which otherwise resets YouTube-bound connections. Covers requests / youtube-transcript-api (``REQUESTS_CA_BUNDLE``) and stdlib ssl / yt-dlp (``SSL_CERT_FILE``).""" ca = os.environ.get("YT_PROXY_CA", "").strip() if not ca: return try: import certifi bundle = os.path.join(tempfile.gettempdir(), "yt_proxy_ca_bundle.pem") with open(certifi.where(), encoding="utf-8") as fh: roots = fh.read() with open(bundle, "w", encoding="utf-8") as fh: fh.write(roots.rstrip() + "\n" + ca + "\n") os.environ["REQUESTS_CA_BUNDLE"] = bundle # requests / youtube-transcript-api os.environ["SSL_CERT_FILE"] = bundle # stdlib ssl + yt-dlp (see frames.py) except Exception: pass _install_proxy_ca() import gradio as gr from pipeline import ( asr as asr_mod, captions as captions_mod, docx_builder, downloader as downloader_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", ] # --------------------------------------------------------------- thin access fallback def _looks_like_netscape(text: str) -> bool: head = text.lstrip() return head.startswith("#") or "\tTRUE\t" in text or "\tFALSE\t" in text def _maybe_b64_decode(text: str) -> str | None: compact = "".join(text.split()) if len(compact) < 16 or re.search(r"[^A-Za-z0-9+/=]", compact): return None try: decoded = base64.b64decode(compact, validate=True).decode("utf-8", "replace") except (binascii.Error, ValueError): return None return decoded if _looks_like_netscape(decoded) else None def _cookiefile(workdir: str) -> str | None: """Materialize the optional YT_COOKIES secret to a Netscape file; return path or None.""" data = os.environ.get("YT_COOKIES") if not data or not data.strip(): return None if not _looks_like_netscape(data): decoded = _maybe_b64_decode(data) if decoded: data = decoded if not data.lstrip().startswith(("# Netscape", "# HTTP")): data = "# Netscape HTTP Cookie File\n" + data.lstrip("\n") if not data.endswith("\n"): data += "\n" path = os.path.join(workdir, "cookies.txt") with open(path, "w", encoding="utf-8", newline="\n") as fh: fh.write(data) return path def _resolve_proxy() -> str | None: proxy = os.environ.get("YT_PROXY", "").strip() return proxy or None def _resolve_media_proxy() -> str | None: """Plain HTTP proxy for large media (googlevideo.com) downloads. The TLS YT_PROXY can't sustain multi-MB transfers, so screenshots use this instead; falls back to YT_PROXY when unset.""" mp = os.environ.get("YT_MEDIA_PROXY", "").strip() return mp or None def _resolve_api_key(ui_key: str | None) -> str | None: key = (ui_key or "").strip() or os.environ.get("YOUTUBE_API_KEY", "").strip() return key or None def check_access(): """Health check: are the required secrets set and is the video-download API live?""" import requests rk = os.environ.get("RAPIDAPI_KEY", "").strip() lines = [ f"- **Video download** (`RAPIDAPI_KEY`): {'set ✅' if rk else 'not set ❌ — required'}", f"- **Comment sentiment** (`YOUTUBE_API_KEY`): " f"{'set ✅' if os.environ.get('YOUTUBE_API_KEY') else 'not set ⚪ (sentiment skipped)'}", "- **Transcript**: faster-whisper, local (no key/proxy needed) ✅", "- **Search**: Piped API (no key needed) ✅", ] api_ok = None if rk: host = "youtube-video-fast-downloader-24-7.p.rapidapi.com" try: r = requests.get(f"https://{host}/get-video-info/dQw4w9WgXcQ", headers={"X-RapidAPI-Key": rk, "X-RapidAPI-Host": host}, timeout=25) if r.status_code == 200: api_ok = True lines.append("- ✅ Download API reachable and key valid (HTTP 200)") downloader_mod._update_quota(r.headers) # prime the monthly-cap tracker q = downloader_mod.quota_status() if q.get("limit") is not None and q.get("remaining") is not None: days = (q["reset"] // 86400) if q.get("reset") else "?" cap = f" · self-cap {q['self_cap']}" if q.get("self_cap") else "" lines.append(f"- 📊 Monthly requests: **{q['remaining']} / {q['limit']} left** " f"({q.get('used', 0)} used, resets in ~{days} days){cap}") elif r.status_code in (401, 403): api_ok = False lines.append(f"- ❌ Download API rejected the key (HTTP {r.status_code}) — " "check RAPIDAPI_KEY and that you're subscribed") else: api_ok = False lines.append(f"- ⚠️ Download API HTTP {r.status_code}: {r.text[:120]}") except Exception as exc: api_ok = False lines.append(f"- ❌ Download API unreachable: {type(exc).__name__}") if not rk: verdict = "### 🔴 Set `RAPIDAPI_KEY` — it's required to download the source video." elif api_ok: verdict = ("### 🟢 Ready. Video downloads via the API; the transcript is produced " "locally by faster-whisper (first run downloads the model, ~1 min).") else: verdict = "### 🔴 The video-download API isn't working — see the details below." return verdict + "\n\n" + "\n".join(lines) # ----------------------------------------------------------------------------- helpers 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.get('positive_share', 0) * 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 _collect_keywords(primary_kw, secondary_kw) -> dict: primary = (primary_kw or "").strip() secondary, seen = [], {primary.lower()} for part in (secondary_kw or "").split(","): kw = part.strip() if kw and kw.lower() not in seen: seen.add(kw.lower()) secondary.append(kw) return {"primary": primary, "secondary": secondary} def run_pipeline(topic, hf_token, yt_api_key, llm_model, vlm_model, w_llm, w_whisper, lead, max_shots, primary_kw, secondary_kw, content_brief="", progress=gr.Progress()): """Generator yielding (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 = os.path.join(workdir, "source.mp4") try: api_key = _resolve_api_key(yt_api_key) # 1. Search ------------------------------------------------------------------ progress(0.03, desc="Searching") yield status(f"🔍 Searching top videos for **{topic}** (engagement-ranked)…"), gr.update(), gr.update(), gr.update() videos = search_mod.search_top5(topic, api_key=api_key) eng_note = "" if videos and "engagement" in videos[0]: eng_note = " • ranked by likes+comments+subscribers−dislikes (normalized)" yield status(f"Found {len(videos)} candidate videos{eng_note}."), gr.update(), gr.update(), gr.update() # 2. Sentiment ranking (YouTube Data API comments) --------------------------- if api_key: yield status("💬 Fetching comments (YouTube Data API) and scoring sentiment…"), gr.update(), gr.update(), gr.update() best, scored = sentiment_mod.rank_by_sentiment(videos, api_key, progress) picked_msg = f"({best['positive_share'] * 100:.0f}% positive comments)" else: best = videos[0] scored = [{**v, "positive_share": 0.0, "n_comments": 0, "note": "sentiment skipped (no API key)", "search_rank": i} for i, v in enumerate(videos)] picked_msg = "(no YouTube Data API key → used top search result)" ranking = gr.update(value=_ranking_rows(scored)) yield (status(f"🏆 Picked **{best.get('title', best['video_id'])}** {picked_msg}."), ranking, gr.update(), gr.update()) # 3. Download the source video once (RapidAPI, own-CDN — no proxy) ----------- progress(0.25, desc="Downloading video") yield status("⬇️ Downloading the video (server-side prep can take up to ~5 min)…"), ranking, gr.update(), gr.update() downloader_mod.download_video(best["video_id"], video_path, progress=progress) size_mb = os.path.getsize(video_path) / 1024 / 1024 yield (status(f"Video downloaded ({size_mb:.0f} MB)."), ranking, gr.update(), gr.update()) # 4. Transcript via faster-whisper on the local video ------------------------ progress(0.4, desc="Transcribing (faster-whisper)") yield status("📝 Transcribing the audio with faster-whisper (can take a while on CPU)…"), ranking, gr.update(), gr.update() segs = asr_mod.transcribe_file(video_path, progress=progress) transcript = transcribe_mod.transcript_text(segs) yield (status(f"Transcript ready ({len(segs)} segments)."), ranking, gr.update(value=transcript), gr.update()) # 5. Tutorial text ----------------------------------------------------------- progress(0.6, desc="Writing tutorial") keywords = _collect_keywords(primary_kw, secondary_kw) kw_note = f" • primary: '{keywords['primary']}'" if keywords["primary"] else "" if keywords["secondary"]: kw_note += f" • secondary: {', '.join(keywords['secondary'])}" if (content_brief or "").strip(): kw_note += " • honoring your content brief" yield status(f"🤖 Generating tutorial with `{llm_model}`{kw_note}…"), ranking, gr.update(value=transcript), gr.update() tut = tutorial_mod.generate_tutorial(transcript, hf_token.strip(), llm_model, keywords, brief=content_brief) if keywords["primary"]: n = tutorial_mod.count_keyword(tut, keywords["primary"]) yield (status(f"🔑 Primary keyword '{keywords['primary']}' appears {n}× in the post."), ranking, gr.update(value=transcript), gr.update()) # 6. Screenshots from the local video (ffmpeg, no network) ------------------- selected, caps = {}, {} times = frames_mod.compute_shot_times( tut["steps"], segs, w_llm=float(w_llm), w_whisper=float(w_whisper), lead=float(lead), max_shots=int(max_shots)) if times: progress(0.8, desc="Screenshots") yield status(f"🎞️ Capturing {len(times)} screenshots from the video…"), ranking, gr.update(value=transcript), gr.update() try: selected = frames_mod.capture_from_file(times, video_path, frames_dir, progress) except Exception as exc: yield (status(f"⚠️ Couldn't extract screenshots — text-only tutorial. " f"`{type(exc).__name__}: {str(exc)[:300]}`"), ranking, gr.update(value=transcript), gr.update()) if selected: progress(0.9, desc="Captioning") yield status(f"✍️ Captioning {len(selected)} 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) # Done with the video — delete it (keep only the .docx for download). try: os.remove(video_path) except OSError: pass # 7. DOCX -------------------------------------------------------------------- progress(0.96, 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") shots_msg = f"{len(selected)} screenshots" if selected else "text-only" yield (status(f"✅ Done ({shots_msg}). Download your tutorial below."), ranking, gr.update(value=transcript), gr.update(value=out_path)) except gr.Error: raise except (transcribe_mod.TranscriptError, sentiment_mod.SentimentError, RuntimeError, ValueError) as exc: raise gr.Error(str(exc)) finally: # Always delete the downloaded video (keep only the .docx for download). try: os.remove(video_path) except OSError: pass def build_ui(): with gr.Blocks(title="YouTube → Tutorial Post") as demo: gr.Markdown( "# 📝 YouTube → Tutorial Post Generator\n" "Enter a topic, your **Hugging Face token** (LLM + vision model, billed to you) " "and a **YouTube Data API key** (for comments). The Space picks the best video by " "comment sentiment, pulls its transcript, writes an AEO-friendly tutorial, grabs " "real screenshots at the right moments, and builds a **.docx**." ) 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)") yt_api_key = gr.Textbox(label="YouTube Data API key", type="password", placeholder="for comments (or set the YOUTUBE_API_KEY secret)") 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) content_brief = gr.Textbox( label="What the content must cover (optional)", lines=4, placeholder=("List the points, steps, or questions the tutorial must address — " "one per line or comma-separated.\n" "e.g. How to create a pivot table\nHow to refresh data\n" "Common pivot table errors"), info=("Required coverage for the tutorial writer. Leave blank to just follow " "the video. Screenshots stay optional — if none are suitable the post is " "produced without them."), ) with gr.Accordion("SEO / AEO keywords (optional)", open=False): gr.Markdown( "The **primary keyword** is used naturally ~3× in the body and placed in " "the title, URL slug, meta description, the first 100 words, and one or two " "H2 headings. Each **secondary keyword** is used once. The post also follows " "answer-engine best practices (direct answer up top, FAQ, last-updated date, " "source citation)." ) primary_kw = gr.Textbox(label="Primary keyword", placeholder="e.g. excel pivot tables") secondary_kw = gr.Textbox(label="Secondary keywords (comma-separated)", placeholder="e.g. pivot chart, data summary") 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: transcript timing") lead = gr.Slider(0.0, 5.0, value=1.0, step=0.5, label="Lead offset (s)") max_shots = gr.Slider(1, 15, value=8, step=1, label="Max screenshots") with gr.Row(): health_btn = gr.Button("🩺 Check YouTube access (proxy + reachability)") health_md = gr.Markdown() 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) docx_file = gr.File(label="Download tutorial (.docx)") health_btn.click(check_access, outputs=health_md) run_btn.click( run_pipeline, inputs=[topic, hf_token, yt_api_key, llm_model, vlm_model, w_llm, w_whisper, lead, max_shots, primary_kw, secondary_kw, content_brief], outputs=[status_md, ranking_df, transcript_box, docx_file], ) return demo # Expose a module-level `demo` so HF Spaces' SSR launcher finds it (avoids the # "Launching demo not found in __main__" fallback warning). demo = build_ui() demo.queue() if __name__ == "__main__": demo.launch()