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 -
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
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"]*>(.*?)", 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"]*>(.*?)", 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,
}