#!/usr/bin/env python3 """Gradio UI for Chorus with durable jobs, automatic caching, polling, and card UI. Behavior: - One visible action button. - Completed results are cached automatically. - In-progress work is recorded in cache/jobs/.json. - Duplicate work is prevented with an atomic directory lock. - The UI polls every 5 seconds after a job is started/attached. - Results are rendered as full-page cluster cards using normal page scroll. """ from __future__ import annotations import html import json import os import shutil import threading import time import traceback from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Iterator import gradio as gr from dotenv import load_dotenv from chorus.api.youtube import YouTubeClient from chorus.filters import apply_filters from chorus.logging_config import configure_logging from filter_comments import build_llm_client APP_TITLE = "Chorus" APP_SUBTITLE = "Comment filter for signal and insight" CACHE_STALE_AFTER = timedelta(hours=6) RUNNING_JOB_STALE_AFTER = timedelta(hours=12) LOCK_STALE_AFTER = timedelta(minutes=10) MAX_BACKGROUND_JOBS = int(os.getenv("CHORUS_MAX_BACKGROUND_JOBS", "1")) POLL_SECONDS = 5 BACKGROUND_SEMAPHORE = threading.Semaphore(MAX_BACKGROUND_JOBS) def utc_now() -> datetime: return datetime.now(timezone.utc) def iso_now() -> str: return utc_now().isoformat() def parse_dt(value: Any) -> datetime | None: if not value: return None try: dt = datetime.fromisoformat(str(value)) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt except Exception: return None def default_cache_dir() -> Path: if os.getenv("CHORUS_CACHE_DIR"): return Path(os.environ["CHORUS_CACHE_DIR"]).expanduser() if Path("/data").exists(): return Path("/data/chorus-cache") return Path(".cache/chorus") CACHE_DIR = default_cache_dir() RAW_DIR = CACHE_DIR / "raw" PROCESSED_DIR = CACHE_DIR / "processed" JOBS_DIR = CACHE_DIR / "jobs" LOCKS_DIR = CACHE_DIR / "locks" load_dotenv() log = configure_logging() def ensure_dirs() -> None: for d in (RAW_DIR, PROCESSED_DIR, JOBS_DIR, LOCKS_DIR): d.mkdir(parents=True, exist_ok=True) def safe_video_id_from_url(url: str) -> str: url = url.strip() if not url: raise gr.Error("Please enter a YouTube video URL.") return YouTubeClient._video_id_from_uri(url) def raw_path(video_id: str) -> Path: return RAW_DIR / f"{video_id}.json" def processed_path(video_id: str) -> Path: return PROCESSED_DIR / f"{video_id}.json" def job_path(video_id: str) -> Path: return JOBS_DIR / f"{video_id}.json" def lock_path(video_id: str) -> Path: return LOCKS_DIR / f"{video_id}.lock" def atomic_write_json(path: Path, data: dict[str, Any] | list[Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp") with tmp.open("w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) tmp.replace(path) def read_json(path: Path) -> Any | None: try: with path.open("r", encoding="utf-8") as f: return json.load(f) except FileNotFoundError: return None except json.JSONDecodeError: log.warning("Ignoring corrupt JSON file: %s", path) return None def remove_stale_lock(path: Path) -> None: try: created = datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc) except FileNotFoundError: return if utc_now() - created > LOCK_STALE_AFTER: shutil.rmtree(path, ignore_errors=True) @contextmanager def directory_lock(video_id: str) -> Iterator[bool]: ensure_dirs() path = lock_path(video_id) remove_stale_lock(path) acquired = False try: os.mkdir(path) acquired = True except FileExistsError: acquired = False try: yield acquired finally: if acquired: shutil.rmtree(path, ignore_errors=True) def update_job(video_id: str, **updates: Any) -> dict[str, Any]: data = read_json(job_path(video_id)) or {"video_id": video_id} data.update(updates) data["updated_at"] = iso_now() atomic_write_json(job_path(video_id), data) return data def is_running_job_fresh(job: dict[str, Any]) -> bool: if job.get("state") not in {"queued", "running"}: return False updated = parse_dt(job.get("updated_at")) or parse_dt(job.get("started_at")) return bool(updated and utc_now() - updated < RUNNING_JOB_STALE_AFTER) def cache_age(payload: dict[str, Any]) -> timedelta | None: created = parse_dt(payload.get("created_at")) return None if created is None else utc_now() - created def is_stale_processed(payload: dict[str, Any]) -> bool: age = cache_age(payload) return age is None or age > CACHE_STALE_AFTER def format_age(age: timedelta | None) -> str: if age is None: return "unknown age" total_minutes = max(0, int(age.total_seconds() // 60)) days, rem_minutes = divmod(total_minutes, 24 * 60) hours, minutes = divmod(rem_minutes, 60) if days: return f"{days}d {hours}h ago" if hours: return f"{hours}h {minutes}m ago" return f"{minutes}m ago" def format_timestamp(value: Any) -> str: if value is None: return "" try: return datetime.fromtimestamp(float(value), tz=timezone.utc).strftime("%b %-d %Y %-I:%M %p") except Exception: return str(value) def esc(value: Any) -> str: return html.escape("" if value is None else str(value), quote=True) def first_string_by_keys(obj: Any, keys: tuple[str, ...]) -> str | None: if isinstance(obj, dict): for key in keys: value = obj.get(key) if isinstance(value, str) and value.strip(): return value.strip() for value in obj.values(): found = first_string_by_keys(value, keys) if found: return found elif isinstance(obj, list): for value in obj: found = first_string_by_keys(value, keys) if found: return found return None def first_number_by_keys(obj: Any, keys: tuple[str, ...]) -> int | float | None: if isinstance(obj, dict): for key in keys: value = obj.get(key) if isinstance(value, (int, float)) and not isinstance(value, bool): return value for value in obj.values(): found = first_number_by_keys(value, keys) if found is not None: return found return None def looks_like_cluster(item: dict[str, Any]) -> bool: if item.get("kind") in {"comment_cluster", "rescued_comment"}: return True cluster_keys = { "comments", "top_comments", "representative_comments", "key_comments", "summary", "cluster_summary", "description", "title", "cluster_title", "label", "topic", "enrichment", "metadata", "highlighted", "is_highlighted", } return bool(cluster_keys.intersection(item.keys())) def get_result_list(payload: dict[str, Any]) -> list[Any]: for key in ("clusters", "clustered_comments", "result", "results", "items"): value = payload.get(key) if isinstance(value, list): return value if isinstance(payload.get("data"), dict): return get_result_list(payload["data"]) return [] def get_clusters(result: list[Any]) -> list[dict[str, Any]]: if result and all(isinstance(x, dict) and looks_like_cluster(x) for x in result): return [x for x in result if isinstance(x, dict)] comments = [x for x in result if isinstance(x, dict)] if comments: return [{"title": "Filtered comments", "summary": "Comments that survived the filter pipeline.", "comments": comments, "highlighted": True}] return [] def get_cluster_comments(cluster: dict[str, Any]) -> list[Any]: for key in ("top_comments", "representative_comments", "key_comments", "comments", "examples", "sample_comments"): value = cluster.get(key) if isinstance(value, list): return value return [] def get_cluster_title(cluster: dict[str, Any], index: int) -> str: return first_string_by_keys(cluster, ("title", "cluster_title", "label", "topic", "name", "headline")) or f"Cluster {index + 1}" def get_cluster_summary(cluster: dict[str, Any]) -> str: return first_string_by_keys( cluster, ( "summary", "cluster_summary", "llm_summary", "description", "synopsis", "rationale", "explanation", "why_it_matters", "takeaway", "abstract", "short_summary", "long_summary", ), ) or "No summary available." def cluster_comment_count(cluster: dict[str, Any]) -> int: explicit = first_number_by_keys(cluster, ("comment_count", "n_comments", "size", "count", "cluster_size")) if explicit is not None: try: return int(explicit) except Exception: pass return len(get_cluster_comments(cluster)) def is_highlighted(cluster: dict[str, Any]) -> bool: if "highlighted" in cluster: highlighted = bool(cluster["highlighted"]) if highlighted and cluster.get("cluster_scoring_skipped"): import logging logging.getLogger("chorus.app").warning( "Cluster %s marked as highlighted but LLM scoring was skipped. Forcing to False.", cluster.get("cluster_id") ) return False return highlighted if cluster.get("cluster_label") == "highlight": if cluster.get("cluster_scoring_skipped"): return False return True return False def sort_clusters(clusters: list[dict[str, Any]]) -> list[tuple[int, dict[str, Any]]]: indexed = list(enumerate(clusters)) return sorted(indexed, key=lambda t: (not is_highlighted(t[1]), -cluster_comment_count(t[1]), t[0])) def comment_text(comment: Any) -> str: if isinstance(comment, str): return comment if isinstance(comment, dict): return str(comment.get("text") or comment.get("body") or comment.get("comment") or comment.get("content") or "") return str(comment) def comment_author(comment: Any) -> str: if isinstance(comment, dict): return str(comment.get("author") or comment.get("username") or comment.get("user") or "") return "" def comment_date(comment: Any) -> str: if isinstance(comment, dict): return format_timestamp(comment.get("timestamp") or comment.get("created_utc") or comment.get("created_at") or comment.get("date")) return "" def render_comments_html(comments: list[Any]) -> str: if not comments: return '
No representative comments found for this cluster.
' parts = [] for c in comments: author = esc(comment_author(c)) date = esc(comment_date(c)) text = esc(comment_text(c)).replace("\n", "
") parts.append( f'
' f'
{author}{date}
' f'
{text}
' f'
' ) return "".join(parts) def render_video_header(payload: dict[str, Any]) -> str: title = payload.get("video_title") or payload.get("title") or "" url = payload.get("video_url") or "" return ( f'
' f'

{esc(title)}

' f'{esc(url)}' f'
' ) def render_clusters_html(payload: dict[str, Any]) -> str: items = get_clusters(get_result_list(payload)) if not items: return '
No clusters found.
' highlighted_clusters = [] other_clusters = [] rescued_comments = [] for i, item in enumerate(items): if item.get("kind") == "rescued_comment": rescued_comments.append(item) else: if is_highlighted(item): highlighted_clusters.append((i, item)) else: other_clusters.append((i, item)) # Sort clusters: largest first highlighted_clusters.sort(key=lambda t: cluster_comment_count(t[1]), reverse=True) other_clusters.sort(key=lambda t: cluster_comment_count(t[1]), reverse=True) # Sort rescued comments by score rescued_comments.sort(key=lambda c: c.get("cheap_information_score") or 0, reverse=True) sections = [] # 1. Highlighted clusters for original_index, cluster in highlighted_clusters: sections.append(render_cluster_card(cluster, original_index, is_highlighted=True)) # 2. Rescued comments if rescued_comments: sections.append('

Notable individual comments

') for comment in rescued_comments: sections.append(render_rescued_comment_card(comment)) # 3. Other clusters if other_clusters: if highlighted_clusters or rescued_comments: sections.append('

Other kept clusters

') for original_index, cluster in other_clusters: sections.append(render_cluster_card(cluster, original_index, is_highlighted=False)) return "\n".join(sections) def render_cluster_card(cluster: dict[str, Any], index: int, is_highlighted: bool) -> str: css_class = "chorus-card chorus-highlight" if is_highlighted else "chorus-card chorus-muted" title = esc(get_cluster_title(cluster, index)) summary = esc(get_cluster_summary(cluster)).replace("\n", "
") count = cluster_comment_count(cluster) label = "" if is_highlighted else '
Non-highlighted cluster
' comments = render_comments_html(get_cluster_comments(cluster)) return ( f'
' f'' f'
' f'{label}' f'
{title}
' f'
{summary}
' f'' f'
' f'
{comments}
' f'
' f'
' ) def render_rescued_comment_card(comment: dict[str, Any]) -> str: author = esc(comment.get("author") or "Unknown") text = esc(comment.get("text") or "").replace("\n", "
") likes = comment.get("likeCount", 0) replies = comment.get("totalReplyCount", 0) score = comment.get("cheap_information_score", 0) source = esc(comment.get("source_cluster_label") or f"Cluster {comment.get('source_cluster_id')}") return ( f'
' f'
' f'
Rescued from: {source}
' f'
{author}{likes} likes, {replies} repliesInfo score: {score}
' f'
{text}
' f'
' f'
' ) def analyze_button() -> Any: return gr.update(value="Analyze comments", interactive=True) def refresh_button() -> Any: return gr.update(value="Refresh cached data", interactive=True) def render_processed(payload: dict[str, Any]) -> tuple[Any, ...]: clusters = get_clusters(get_result_list(payload)) age_text = format_age(cache_age(payload)) stale = is_stale_processed(payload) status = ( f"✅ Loaded **{len(clusters)} cluster(s)**. " f"Cache age: **{age_text}**. " f"Raw comments: **{payload.get('raw_comment_count', '?')}**." ) if stale: status += "\n\nCached result is older than 6 hours. Click **Refresh cached data** to fetch and process it again." state = {"video_id": payload.get("video_id"), "video_url": payload.get("video_url"), "mode": "processed_stale" if stale else "processed_fresh", "payload": payload} return status, render_video_header(payload), render_clusters_html(payload), state, gr.update(active=False), (refresh_button() if stale else analyze_button()) def render_job(job: dict[str, Any]) -> tuple[Any, ...]: state = job.get("state", "unknown") stage = job.get("stage", "starting") details = [] if job.get("raw_comment_count") is not None: details.append(f"downloaded {job['raw_comment_count']} comments") if job.get("result_count") is not None: details.append(f"produced {job['result_count']} result items") detail_text = f" ({'; '.join(details)})" if details else "" if state == "failed": status = f"❌ Previous run failed during **{stage}**. Click **Analyze comments** to retry." if job.get("error"): status += f"\n\n```text\n{job['error']}\n```" active = False elif state == "done": status = "✅ Job finished. Loading cached result..." active = True else: started = parse_dt(job.get("started_at")) age_text = format_age(utc_now() - started) if started else "unknown age" status = f"⏳ This video is being processed. Stage: **{stage}**{detail_text}. Started **{age_text}**." active = True return status, "", "", {"video_id": job.get("video_id"), "video_url": job.get("video_url"), "mode": "job", "job": job}, gr.update(active=active), analyze_button() def fetch_video_title(client: YouTubeClient, video_id: str) -> str | None: try: response = client.youtube.videos().list(part="snippet", id=video_id, maxResults=1).execute() items = response.get("items") or [] if items: return items[0].get("snippet", {}).get("title") except Exception: log.exception("Could not fetch YouTube title for %s", video_id) return None def run_pipeline_background(video_id: str, video_url: str, force_refresh: bool = False) -> None: with BACKGROUND_SEMAPHORE: try: update_job(video_id, state="running", stage="fetching", started_processing_at=iso_now()) video_title = None llm_client = build_llm_client() # Start the ready check early if force_refresh or not raw_path(video_id).exists(): api_key = os.getenv("YOUTUBE_API_KEY") if not api_key: raise RuntimeError("YOUTUBE_API_KEY is not set. Add it to .env locally or as a Hugging Face Space secret.") client = YouTubeClient(api_key=api_key) video_title = fetch_video_title(client, video_id) comments = client.get_comments_by_uri(video_url) atomic_write_json(raw_path(video_id), {"video_url": video_url, "video_id": video_id, "video_title": video_title, "created_at": iso_now(), "comments": comments}) else: raw = read_json(raw_path(video_id)) or [] if isinstance(raw, dict): video_title = raw.get("video_title") comments = raw.get("comments") or [] else: comments = raw update_job(video_id, stage="filtering", raw_comment_count=len(comments), result_count=0) if not comments: raise RuntimeError("No comments were returned for this video.") filtered: list[dict[str, Any]] = [] for i, item in enumerate(apply_filters(comments, llm_client=llm_client), start=1): filtered.append(item) if i == 1 or i % 5 == 0: update_job(video_id, stage="filtering", raw_comment_count=len(comments), result_count=i) payload = { "video_url": video_url, "video_id": video_id, "video_title": video_title, "created_at": iso_now(), "raw_comment_count": len(comments), "result_count": len(filtered), "result": filtered, } atomic_write_json(processed_path(video_id), payload) update_job(video_id, state="done", stage="done", raw_comment_count=len(comments), result_count=len(filtered), finished_at=iso_now()) except Exception as exc: log.exception("Pipeline failed for video %s", video_id) update_job(video_id, state="failed", stage="failed", error=str(exc), traceback=traceback.format_exc(limit=8), finished_at=iso_now()) def start_background_pipeline(video_id: str, video_url: str, force_refresh: bool = False) -> None: t = threading.Thread(target=run_pipeline_background, args=(video_id, video_url, force_refresh), name=f"chorus-{video_id}", daemon=True) t.start() def submit_or_attach(video_url: str, state: dict[str, Any] | None = None) -> tuple[Any, ...]: ensure_dirs() video_id = safe_video_id_from_url(video_url) video_url = video_url.strip() state = state or {} force_refresh = state.get("mode") == "processed_stale" and state.get("video_id") == video_id if not force_refresh: processed = read_json(processed_path(video_id)) if isinstance(processed, dict): return render_processed(processed) existing_job = read_json(job_path(video_id)) if isinstance(existing_job, dict) and is_running_job_fresh(existing_job): return render_job(existing_job) with directory_lock(video_id) as acquired: if not acquired: time.sleep(0.1) job = read_json(job_path(video_id)) or {"video_id": video_id, "video_url": video_url, "state": "running", "stage": "starting", "started_at": iso_now()} return render_job(job) if not force_refresh: processed = read_json(processed_path(video_id)) if isinstance(processed, dict): return render_processed(processed) existing_job = read_json(job_path(video_id)) if isinstance(existing_job, dict) and is_running_job_fresh(existing_job): return render_job(existing_job) job = {"video_id": video_id, "video_url": video_url, "state": "queued", "stage": "queued", "started_at": iso_now(), "updated_at": iso_now(), "force_refresh": force_refresh} atomic_write_json(job_path(video_id), job) start_background_pipeline(video_id, video_url, force_refresh=force_refresh) return render_job(job) def analyze(video_url: str, state: dict[str, Any] | None): return submit_or_attach(video_url, state) def poll_status(video_url: str, state: dict[str, Any] | None): ensure_dirs() state = state or {} video_id = None if video_url and video_url.strip(): try: video_id = safe_video_id_from_url(video_url) except Exception: video_id = None if not video_id: video_id = state.get("video_id") if not video_id: return "Waiting for a video URL.", "", "", state, gr.update(active=False), analyze_button() processed = read_json(processed_path(video_id)) if isinstance(processed, dict): return render_processed(processed) job = read_json(job_path(video_id)) if isinstance(job, dict): return render_job(job) return "No job or cached result found for this video.", "", "", state, gr.update(active=False), analyze_button() CSS = """ .gradio-container { max-width: 980px !important; margin: 0 auto !important; } #chorus-brand { display: flex; align-items: center; gap: 18px; margin: 8px 0 18px 0; } #chorus-logo { width: 112px; height: 90px; position: relative; flex: 0 0 auto; } #chorus-logo .bubble { position: absolute; left: 4px; top: 8px; width: 78px; height: 62px; border-radius: 32px; background: linear-gradient(135deg, #6672f0, #4f67d8); } #chorus-logo .tail { position: absolute; left: 58px; top: 52px; width: 30px; height: 34px; background: #079f8f; clip-path: polygon(0 0, 100% 0, 100% 100%); border-radius: 0 0 6px 0; } #chorus-logo .funnel { position: absolute; left: 24px; top: 15px; width: 47px; height: 52px; background: #f8f7ff; clip-path: polygon(0 0, 100% 0, 67% 40%, 67% 88%, 33% 100%, 33% 40%); } #chorus-logo .check { position: absolute; left: 37px; top: 30px; width: 34px; height: 17px; border-left: 5px solid #0f172a; border-bottom: 5px solid #0f172a; transform: rotate(-45deg); } #chorus-brand h1 { margin: 0; font-size: 58px; line-height: .92; font-weight: 900; letter-spacing: -3px; color: #0f172a; } #chorus-brand p { margin: 10px 0 0 4px; font-size: 22px; font-weight: 700; color: #475569; } #chorus-form-row { align-items: end; gap: 14px; } .chorus-video { margin: 20px 0 14px; } .chorus-video h2 { margin: 0 0 4px; font-size: 24px; font-weight: 900; color: #111; } .chorus-video a { color: #2563eb; font-weight: 700; text-decoration: none; } .chorus-card { border: 1.5px solid #d6c800; border-radius: 28px; margin: 16px 0; overflow: hidden; cursor: pointer; } .chorus-card summary { list-style: none; } .chorus-card summary::-webkit-details-marker { display: none; } .chorus-card:hover { filter: brightness(.99); box-shadow: 0 2px 10px rgba(15, 23, 42, .09); } .chorus-card-head { padding: 14px 18px 12px; } .chorus-card-title { font-size: 21px; font-weight: 900; margin-bottom: 4px; color: #111; } .chorus-card-summary { font-size: 18px; line-height: 1.25; color: #222; } .chorus-card-footer { display: flex; justify-content: space-between; gap: 18px; margin-top: 5px; font-size: 16px; color: #333; } .chorus-card[open] .chorus-click-hint { display: none; } .chorus-card .chorus-comments { display: none; } .chorus-card[open] .chorus-comments { display: block; } .chorus-highlight { background: #fffde8; border-color: #e7d600; } .chorus-muted { background: #f0f7ff; border-color: #bed0f7; } .chorus-rescued { background: #f9f9f9; border-color: #e0e0e0; cursor: default; } .chorus-rescued:hover { filter: none; box-shadow: none; } .chorus-badge { font-size: 16px; font-weight: 900; margin-bottom: 2px; color: #475569; } .chorus-section-heading { margin: 32px 0 16px 4px; font-size: 24px; font-weight: 900; color: #475569; } .chorus-comments { padding: 0 34px 22px 34px; } .chorus-comment { margin: 22px 0; } .chorus-comment-meta { display: flex; gap: 32px; font-size: 16px; color: #4b5563; margin-bottom: 6px; } .chorus-comment-text { font-size: 17px; line-height: 1.28; color: #222; white-space: normal; } .chorus-empty, .chorus-no-comments { padding: 18px; color: #555; } """ with gr.Blocks() as demo: state = gr.State({}) timer = gr.Timer(POLL_SECONDS, active=False) gr.HTML( '
' '' f'

{APP_TITLE}

{APP_SUBTITLE}

' '
' ) with gr.Row(elem_id="chorus-form-row"): url = gr.Textbox(label="Paste a YouTube URL", placeholder="https://www.youtube.com/watch?v=…", scale=5) run = gr.Button("Analyze comments", variant="primary", scale=2) status = gr.Markdown("Waiting for a video URL.") video_header = gr.HTML("") cluster_cards = gr.HTML("") outputs = [status, video_header, cluster_cards, state, timer, run] run.click(analyze, inputs=[url, state], outputs=outputs) timer.tick(poll_status, inputs=[url, state], outputs=outputs, show_progress="hidden") if __name__ == "__main__": demo.queue(default_concurrency_limit=8).launch(css=CSS)