Spaces:
Paused
Paused
| from __future__ import annotations | |
| import html | |
| import ipaddress | |
| import re | |
| import socket | |
| import time | |
| import xml.etree.ElementTree as ET | |
| from typing import Any | |
| from urllib.parse import urlparse | |
| import httpx | |
| _FEED_CACHE: dict[str, Any] = {"timestamp": 0.0, "items": []} | |
| CACHE_TTL_SECONDS = 600 | |
| RSS_SOURCES = [ | |
| { | |
| "name": "Ars Technica", | |
| "url": "https://feeds.arstechnica.com/arstechnica/technology-lab", | |
| "default_category": "Systems & Cloud", | |
| }, | |
| { | |
| "name": "Hacker News", | |
| "url": "https://hnrss.org/frontpage?points=120&count=12", | |
| "default_category": "DevTools & Engineering", | |
| }, | |
| { | |
| "name": "The Verge Tech", | |
| "url": "https://www.theverge.com/rss/tech/index.xml", | |
| "default_category": "Product & Startups", | |
| }, | |
| ] | |
| CURATED_TECH_STORIES: list[dict[str, Any]] = [ | |
| { | |
| "id": "curated-ai-agent-evals", | |
| "headline": "Why Engineering Teams Are Shifting From Prompt Tweaking to Deterministic Agent Evals", | |
| "summary": "As coding and workflow agents move into production pipelines, teams are discovering that reliability comes from sandboxes, tool guardrails, and automated regression suites rather than longer system prompts.", | |
| "source": "Tech Presenter Briefing", | |
| "category": "AI & Models", | |
| "published_label": "Featured Briefing", | |
| "url": "", | |
| "talking_points": [ | |
| "Single-turn chat demos hide compounding multi-step tool errors.", | |
| "Production teams treat agent trajectories like CI pipelines with pass/fail assertions.", | |
| "Fast feedback loops and sandboxing beat brittle prompt engineering.", | |
| ], | |
| }, | |
| { | |
| "id": "curated-sqlite-edge-renaissance", | |
| "headline": "The Modern SQLite & Single-Node Renaissance in High-Throughput Web Apps", | |
| "summary": "With NVMe storage, WAL concurrency, and read-replica streaming tools, single-node and embedded databases now handle workloads that teams previously assumed required distributed clusters.", | |
| "source": "Systems Architecture Desk", | |
| "category": "Systems & Cloud", | |
| "published_label": "Featured Briefing", | |
| "url": "", | |
| "talking_points": [ | |
| "Eliminating network hops between app and data slashes tail latency to microseconds.", | |
| "Operational simplicity means fewer 3 a.m. distributed consensus failures.", | |
| "Know the exact write-contention threshold where horizontal Postgres sharding still wins.", | |
| ], | |
| }, | |
| { | |
| "id": "curated-on-device-speech-vision", | |
| "headline": "Browser-Native ML: Running Vision & Audio Analytics Locally With WebAssembly and WebGPU", | |
| "summary": "Modern browsers now run real-time face landmarking, audio signal analysis, and compact speech models directly on user devices—preserving privacy while cutting cloud inference bills to zero.", | |
| "source": "Web Platform Digest", | |
| "category": "DevTools & Engineering", | |
| "published_label": "Featured Briefing", | |
| "url": "", | |
| "talking_points": [ | |
| "Zero network round-trip enables 30–60 FPS interactive coaching feedback.", | |
| "Sensitive camera and microphone streams stay private by default.", | |
| "Hybrid architectures combine instant browser cues with optional deeper server models.", | |
| ], | |
| }, | |
| { | |
| "id": "curated-passkeys-security", | |
| "headline": "Passkeys & Hardware-Bound Credentials Are Finally Replacing Phishable Passwords at Scale", | |
| "summary": "Major consumer and developer platforms report dramatic drops in account takeovers and sign-in friction after adopting WebAuthn passkeys synced across desktop and mobile devices.", | |
| "source": "Security Engineering Brief", | |
| "category": "Security & Privacy", | |
| "published_label": "Featured Briefing", | |
| "url": "", | |
| "talking_points": [ | |
| "Cryptographic origin binding makes lookalike phishing domains useless.", | |
| "Biometric sign-in cuts login time by more than half on mobile browsers.", | |
| "Cross-ecosystem recovery UX remains the critical detail product teams must nail.", | |
| ], | |
| }, | |
| { | |
| "id": "curated-small-teams-leverage", | |
| "headline": "How Tiny Product-Minded Engineering Teams Ship Like 50-Person Organizations", | |
| "summary": "Armed with typed monorepos, automated preview environments, and AI-assisted testing, lean 4-to-8 engineer teams are shipping polished full-stack products at unprecedented velocity.", | |
| "source": "Builder & Startup Desk", | |
| "category": "Product & Startups", | |
| "published_label": "Featured Briefing", | |
| "url": "", | |
| "talking_points": [ | |
| "Communication overhead scales quadratically; small senior teams keep context unified.", | |
| "Shared TypeScript/Python contracts eliminate integration drift.", | |
| "Clear product taste and storytelling have become the true competitive moat.", | |
| ], | |
| }, | |
| { | |
| "id": "curated-inference-economics", | |
| "headline": "The New Unit Economics of AI Features: Latency Budgets, Caching, and Model Routing", | |
| "summary": "Smart AI products no longer route every user click to the largest frontier model; instead, they combine semantic caching, compact specialist models, and frontier escalation only when needed.", | |
| "source": "AI Infrastructure Brief", | |
| "category": "AI & Models", | |
| "published_label": "Featured Briefing", | |
| "url": "", | |
| "talking_points": [ | |
| "Users abandon interactive UI features when latency exceeds 2 seconds.", | |
| "Routing 80% of routine tasks to compact models cuts cost by 5–10x.", | |
| "Designing transparent progressive UI states turns wait time into perceived intelligence.", | |
| ], | |
| }, | |
| ] | |
| STORY_FRAMEWORKS: dict[str, dict[str, Any]] = { | |
| "briefing_60s": { | |
| "id": "briefing_60s", | |
| "label": "60s Breaking Tech Briefing", | |
| "duration_sec": 60, | |
| "mode": "Storytelling", | |
| "goal": "Improve pace", | |
| "description": "Punchy creator/anchor update: Headline Hook → What Changed → Why Builders Care → Sharp Closing Take.", | |
| }, | |
| "explainer_2m": { | |
| "id": "explainer_2m", | |
| "label": "2-Min Deep-Dive Explainer", | |
| "duration_sec": 120, | |
| "mode": "Storytelling", | |
| "goal": "Stronger eye contact", | |
| "description": "Educational breakdown: Relatable Problem → Under-the-Hood Mechanism → Visual Analogy → Practical Impact.", | |
| }, | |
| "contrarian_take": { | |
| "id": "contrarian_take", | |
| "label": "90s Contrarian Tech Take", | |
| "duration_sec": 90, | |
| "mode": "Debate", | |
| "goal": "Reduce fillers", | |
| "description": "Opinionated commentary: Conventional Hype → Where Reality Differs → Concrete Evidence → Nuanced Prediction.", | |
| }, | |
| "war_story": { | |
| "id": "war_story", | |
| "label": "2-Min Builder Lesson & Story", | |
| "duration_sec": 120, | |
| "mode": "Pitch", | |
| "goal": "Stronger eye contact", | |
| "description": "Narrative arc connecting news to real engineering: The Stakes → The Tradeoff → How to Apply It Monday Morning.", | |
| }, | |
| } | |
| def _strip_html(raw: str) -> str: | |
| no_tags = re.sub(r"<[^>]+>", " ", raw or "") | |
| unescaped = html.unescape(no_tags) | |
| return re.sub(r"\s+", " ", unescaped).strip() | |
| def _infer_category(title: str, summary: str, fallback: str) -> str: | |
| blob = f"{title} {summary}".lower() | |
| if any(k in blob for k in ("ai", "llm", "model", "gpt", "claude", "gemini", "neural", "inference", "agent")): | |
| return "AI & Models" | |
| if any(k in blob for k in ("security", "vulnerability", "cve", "passkey", "encryption", "breach", "privacy", "zero-day")): | |
| return "Security & Privacy" | |
| if any(k in blob for k in ("cloud", "linux", "kernel", "postgres", "sqlite", "kubernetes", "server", "cpu", "gpu", "chip")): | |
| return "Systems & Cloud" | |
| if any(k in blob for k in ("rust", "typescript", "python", "compiler", "git", "open source", "api", "browser", "ide")): | |
| return "DevTools & Engineering" | |
| if any(k in blob for k in ("startup", "launch", "funding", "apple", "product", "design", "market")): | |
| return "Product & Startups" | |
| return fallback | |
| def _parse_feed_xml(xml_text: str, source_name: str, default_category: str) -> list[dict[str, Any]]: | |
| items: list[dict[str, Any]] = [] | |
| try: | |
| root = ET.fromstring(xml_text) | |
| except ET.ParseError: | |
| return items | |
| # RSS 2.0 <item> | |
| for idx, item in enumerate(root.findall(".//item")[:8]): | |
| title = _strip_html(item.findtext("title") or "") | |
| desc = _strip_html(item.findtext("description") or "") | |
| link = (item.findtext("link") or "").strip() | |
| if not title or len(title) < 12: | |
| continue | |
| if "Comments URL:" in desc: | |
| desc = "Trending technical discussion on Hacker News — ideal for practicing concise synthesis and your own engineering point of view." | |
| summary = desc[:340] + ("..." if len(desc) > 340 else "") | |
| category = _infer_category(title, summary, default_category) | |
| items.append({ | |
| "id": f"live-{source_name.lower().replace(' ', '-')}-{idx}", | |
| "headline": title, | |
| "summary": summary or f"Fresh technology story from {source_name}. Frame what happened, why it matters, and your practical builder takeaway.", | |
| "source": source_name, | |
| "category": category, | |
| "published_label": "Live Feed", | |
| "url": link, | |
| "talking_points": _extract_talking_points(title, summary), | |
| }) | |
| # Atom <entry> | |
| ns = {"atom": "http://www.w3.org/2005/Atom"} | |
| for idx, entry in enumerate(root.findall(".//atom:entry", ns)[:8]): | |
| title = _strip_html(entry.findtext("atom:title", default="", namespaces=ns)) | |
| summary_raw = ( | |
| entry.findtext("atom:summary", default="", namespaces=ns) | |
| or entry.findtext("atom:content", default="", namespaces=ns) | |
| ) | |
| desc = _strip_html(summary_raw) | |
| link_el = entry.find("atom:link", ns) | |
| link = (link_el.attrib.get("href", "") if link_el is not None else "").strip() | |
| if not title or len(title) < 12: | |
| continue | |
| summary = desc[:340] + ("..." if len(desc) > 340 else "") | |
| category = _infer_category(title, summary, default_category) | |
| items.append({ | |
| "id": f"live-atom-{source_name.lower().replace(' ', '-')}-{idx}", | |
| "headline": title, | |
| "summary": summary or f"Latest technology report from {source_name}.", | |
| "source": source_name, | |
| "category": category, | |
| "published_label": "Live Feed", | |
| "url": link, | |
| "talking_points": _extract_talking_points(title, summary), | |
| }) | |
| return items | |
| def _extract_talking_points(headline: str, summary: str) -> list[str]: | |
| clean_summary = summary.strip() | |
| sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", clean_summary) if len(s.strip()) > 25] | |
| points = [f"Core shift: {headline}"] | |
| if sentences: | |
| points.append(sentences[0][:160]) | |
| points.append("Builder lens: How does this change architecture, latency, cost, or user trust?") | |
| return points[:3] | |
| def fetch_tech_news_feed(category: str | None = None, force_refresh: bool = False) -> dict[str, Any]: | |
| now = time.time() | |
| live_items: list[dict[str, Any]] = [] | |
| if not force_refresh and _FEED_CACHE["items"] and (now - _FEED_CACHE["timestamp"] < CACHE_TTL_SECONDS): | |
| live_items = list(_FEED_CACHE["items"]) | |
| else: | |
| fetched: list[dict[str, Any]] = [] | |
| try: | |
| with httpx.Client(timeout=3.0, follow_redirects=True, headers={"User-Agent": "SpeechMachine-TechCoach/1.0"}) as client: | |
| for src in RSS_SOURCES: | |
| try: | |
| resp = client.get(src["url"]) | |
| if resp.status_code == 200 and resp.text: | |
| fetched.extend(_parse_feed_xml(resp.text, src["name"], src["default_category"])) | |
| except Exception: | |
| continue | |
| except Exception: | |
| fetched = [] | |
| if fetched: | |
| # Deduplicate by normalized headline | |
| seen: set[str] = set() | |
| deduped: list[dict[str, Any]] = [] | |
| for entry in fetched: | |
| norm = re.sub(r"\W+", "", entry["headline"].lower())[:60] | |
| if norm not in seen: | |
| seen.add(norm) | |
| deduped.append(entry) | |
| live_items = deduped[:15] | |
| _FEED_CACHE["timestamp"] = now | |
| _FEED_CACHE["items"] = live_items | |
| combined = live_items + CURATED_TECH_STORIES | |
| if category and category.lower() not in {"all", ""}: | |
| filtered = [item for item in combined if item["category"].lower() == category.lower()] | |
| else: | |
| filtered = combined | |
| categories = [ | |
| "All", | |
| "AI & Models", | |
| "Systems & Cloud", | |
| "DevTools & Engineering", | |
| "Security & Privacy", | |
| "Product & Startups", | |
| ] | |
| frameworks_out = [ | |
| { | |
| **fw, | |
| "target_seconds": fw["duration_sec"], | |
| "target_words": round((fw["duration_sec"] / 60) * 140), | |
| } | |
| for fw in STORY_FRAMEWORKS.values() | |
| ] | |
| stories_out = [ | |
| { | |
| **st, | |
| "published_at": st.get("published_label", "Today"), | |
| "is_live": str(st.get("id", "")).startswith("live-"), | |
| } | |
| for st in filtered[:24] | |
| ] | |
| return { | |
| "live_connected": len(live_items) > 0, | |
| "live_count": len(live_items), | |
| "categories": categories, | |
| "frameworks": frameworks_out, | |
| "stories": stories_out, | |
| } | |
| def _is_safe_public_url(url: str) -> bool: | |
| try: | |
| parsed = urlparse(url) | |
| if parsed.scheme not in {"http", "https"} or not parsed.hostname: | |
| return False | |
| hostname = parsed.hostname.lower() | |
| if hostname in {"localhost", "127.0.0.1", "0.0.0.0", "::1"} or hostname.endswith(".local"): | |
| return False | |
| infos = socket.getaddrinfo(hostname, None) | |
| for info in infos: | |
| ip_obj = ipaddress.ip_address(info[4][0]) | |
| if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local or ip_obj.is_reserved: | |
| return False | |
| return True | |
| except Exception: | |
| return False | |
| def _try_fetch_article_excerpt(url: str) -> tuple[str | None, str | None]: | |
| if not url or not _is_safe_public_url(url): | |
| return None, None | |
| try: | |
| with httpx.Client(timeout=3.5, follow_redirects=True, headers={"User-Agent": "SpeechMachine-TechCoach/1.0"}) as client: | |
| resp = client.get(url) | |
| if resp.status_code != 200: | |
| return None, None | |
| raw_html = resp.text[:120000] | |
| title_match = re.search(r"<title[^>]*>(.*?)</title>", raw_html, flags=re.IGNORECASE | re.DOTALL) | |
| page_title = _strip_html(title_match.group(1)) if title_match else None | |
| paragraphs = [ | |
| _strip_html(m) | |
| for m in re.findall(r"<p[^>]*>(.*?)</p>", raw_html, flags=re.IGNORECASE | re.DOTALL) | |
| ] | |
| substantive = [p for p in paragraphs if len(p) > 60][:4] | |
| excerpt = " ".join(substantive)[:700] if substantive else None | |
| return page_title, excerpt | |
| except Exception: | |
| return None, None | |
| def _extract_keywords(headline: str, summary: str) -> list[str]: | |
| stop = { | |
| "the", "and", "for", "with", "that", "this", "from", "into", "your", "are", "why", | |
| "how", "what", "when", "where", "over", "more", "than", "their", "they", "have", | |
| "has", "been", "will", "just", "new", "modern", "teams", "using", "about", | |
| } | |
| words = re.findall(r"[a-zA-Z][a-zA-Z0-9\-]{3,}", f"{headline} {summary}".lower()) | |
| picked: list[str] = [] | |
| for w in words: | |
| clean = w.strip("-") | |
| if clean not in stop and clean not in picked and len(clean) >= 4: | |
| picked.append(clean) | |
| if len(picked) >= 4: | |
| break | |
| return picked + ["because", "however"] | |
| def build_storyboard( | |
| headline: str, | |
| summary: str = "", | |
| source: str | None = None, | |
| url: str | None = None, | |
| category: str | None = "AI & Models", | |
| framework: str = "briefing_60s", | |
| personal_angle: str | None = None, | |
| ) -> dict[str, Any]: | |
| effective_headline = headline.strip() | |
| effective_summary = summary.strip() | |
| if url and not effective_summary: | |
| fetched_title, fetched_excerpt = _try_fetch_article_excerpt(url) | |
| if fetched_title and len(effective_headline) < 10: | |
| effective_headline = fetched_title | |
| if fetched_excerpt: | |
| effective_summary = fetched_excerpt | |
| if not effective_summary: | |
| effective_summary = ( | |
| f"This story explores {effective_headline} and how it reshapes engineering tradeoffs, " | |
| "product velocity, and real-world user outcomes." | |
| ) | |
| fw = STORY_FRAMEWORKS.get(framework, STORY_FRAMEWORKS["briefing_60s"]) | |
| angle_line = ( | |
| personal_angle.strip() | |
| if personal_angle and personal_angle.strip() | |
| else "For builders, the real signal is focusing on measurable reliability, latency, and user value over surface hype." | |
| ) | |
| keywords = _extract_keywords(effective_headline, effective_summary) | |
| if framework == "briefing_60s": | |
| sections = [ | |
| { | |
| "label": "1. Cold-Open Hook (0:00–0:10)", | |
| "text": f"Big shift in {category or 'tech'} today: {effective_headline}.", | |
| }, | |
| { | |
| "label": "2. What Just Happened (0:10–0:28)", | |
| "text": f"First, here is the core development: {effective_summary}", | |
| }, | |
| { | |
| "label": "3. Why Builders Should Care (0:28–0:46)", | |
| "text": "However, look past the headline—this directly changes how teams weigh speed, cost, and architecture decisions.", | |
| }, | |
| { | |
| "label": "4. My Sharp Takeaway (0:46–1:00)", | |
| "text": f"My take: {angle_line}", | |
| }, | |
| ] | |
| elif framework == "explainer_2m": | |
| sections = [ | |
| { | |
| "label": "1. Relatable Problem Hook (0:00–0:20)", | |
| "text": f"You may have seen the headline '{effective_headline}'—let us unpack what is actually happening under the hood.", | |
| }, | |
| { | |
| "label": "2. Under-the-Hood Mechanism (0:20–0:55)", | |
| "text": f"First, the technical context: {effective_summary}", | |
| }, | |
| { | |
| "label": "3. Plain-English Mental Model (0:55–1:30)", | |
| "text": "Second, think of this like upgrading from manual city streets to an automated express lane—throughput rises, however edge-case guardrails matter even more.", | |
| }, | |
| { | |
| "label": "4. Practical Builder Takeaway (1:30–2:00)", | |
| "text": f"Therefore, here is how I am thinking about applying this: {angle_line}", | |
| }, | |
| ] | |
| elif framework == "contrarian_take": | |
| sections = [ | |
| { | |
| "label": "1. Conventional Consensus (0:00–0:18)", | |
| "text": f"Everyone is reacting to '{effective_headline}' as if it changes everything overnight.", | |
| }, | |
| { | |
| "label": "2. Where Reality Differs (0:18–0:45)", | |
| "text": f"However, when you inspect the details—{effective_summary}—the real bottleneck is still integration and human workflow.", | |
| }, | |
| { | |
| "label": "3. Concrete Evidence (0:45–1:10)", | |
| "text": "First, tools evolve fast, yet production reliability and debugging discipline still determine who ships sustainably.", | |
| }, | |
| { | |
| "label": "4. Nuanced Prediction (1:10–1:30)", | |
| "text": f"So my prediction is simple: {angle_line}", | |
| }, | |
| ] | |
| else: # war_story | |
| sections = [ | |
| { | |
| "label": "1. The Stakes (0:00–0:20)", | |
| "text": f"Reading '{effective_headline}' reminded me of a critical engineering lesson every builder runs into eventually.", | |
| }, | |
| { | |
| "label": "2. The Core Tension (0:20–0:55)", | |
| "text": f"First, look at what happened here: {effective_summary}", | |
| }, | |
| { | |
| "label": "3. The Tradeoff Under Pressure (0:55–1:30)", | |
| "text": "However, in real production systems, every optimization introduces a tradeoff between simplicity, latency, and observability.", | |
| }, | |
| { | |
| "label": "4. Monday-Morning Principle (1:30–2:00)", | |
| "text": f"Therefore, the principle worth carrying into your own stack is this: {angle_line}", | |
| }, | |
| ] | |
| teleprompter_script = "\n\n".join(sec["text"] for sec in sections) | |
| bullet_cues = [f"{sec['label']}: {sec['text'][:115]}..." if len(sec["text"]) > 118 else f"{sec['label']}: {sec['text']}" for sec in sections] | |
| delivery_tips = [ | |
| "Lock your eyes onto the camera lens for the entire 1st section hook.", | |
| "Pause silently for 1 full beat right before 'However' and 'My take / Therefore'.", | |
| "Keep speaking cadence between 130–155 WPM so technical nouns land crisply.", | |
| ] | |
| sections_normalized = [ | |
| { | |
| "label": sec["label"], | |
| "text": sec["text"], | |
| "beat": sec["label"].split("(")[0].strip(), | |
| "timing": "(" + sec["label"].split("(")[1] if "(" in sec["label"] else "", | |
| "guidance": fw["description"], | |
| "script": sec["text"], | |
| } | |
| for sec in sections | |
| ] | |
| return { | |
| "headline": effective_headline, | |
| "summary": effective_summary, | |
| "source": source or "Custom Tech Story", | |
| "url": url or "", | |
| "category": category or "AI & Models", | |
| "framework": fw["id"], | |
| "framework_info": fw, | |
| "framework_label": fw["label"], | |
| "target_seconds": fw["duration_sec"], | |
| "target_words": round((fw["duration_sec"] / 60) * 140), | |
| "sections": sections_normalized, | |
| "teleprompter_script": teleprompter_script, | |
| "bullet_cues": bullet_cues, | |
| "target_keywords": keywords[:6], | |
| "delivery_tips": delivery_tips, | |
| "camera_tips": delivery_tips, | |
| } | |