| 1 | #!/usr/bin/env python3 |
| 2 | """Render the press context prompt section with TechCrunch and correlation data. |
| 3 | |
| 4 | Reads crawl data and correlation data, then renders the press context |
| 5 | prompt template with real values. Output can be piped into the analyzer. |
| 6 | |
| 7 | Usage: |
| 8 | python scripts/render_press_context.py [--topic ai-ml] [--week 2026-W21] |
| 9 | """ |
| 10 | |
| 11 | import argparse |
| 12 | import json |
| 13 | import re |
| 14 | import sys |
| 15 | import urllib.request |
| 16 | from datetime import datetime |
| 17 | from pathlib import Path |
| 18 | from urllib.parse import urlparse |
| 19 | |
| 20 | # Allow imports when run from repo root or scripts/ |
| 21 | _REPO_ROOT = Path(__file__).resolve().parent.parent |
| 22 | sys.path.insert(0, str(_REPO_ROOT / "scripts")) |
| 23 | |
| 24 | from topic_paths import analyzed_dir, raw_dir # noqa: E402 |
| 25 | |
| 26 | PRESS_CONTEXT_TOKEN_BUDGET = 8000 |
| 27 | PRESS_CONTEXT_CHAR_BUDGET = PRESS_CONTEXT_TOKEN_BUDGET * 4 |
| 28 | MAX_RENDERED_ARTICLES = 40 |
| 29 | MAX_RENDERED_CORRELATIONS = 20 |
| 30 | GITHUB_REPO_FULL_NAME_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") |
| 31 | |
| 32 | # Single source of truth for the "no press this week" sentinel. This is a |
| 33 | # NON-EMPTY string, so downstream code must use NO_PRESS_SENTINEL_MARKER to |
| 34 | # detect it rather than treating any non-empty press_content as real press. |
| 35 | NO_PRESS_SENTINEL = ( |
| 36 | "No press data available for this week. Analyze repos based on GitHub signals only." |
| 37 | ) |
| 38 | NO_PRESS_SENTINEL_MARKER = re.compile(r"no press data available for this week", re.IGNORECASE) |
| 39 | |
| 40 | |
| 41 | def validate_https_url(url: str, *, label: str) -> None: |
| 42 | parsed = urlparse(url) |
| 43 | if parsed.scheme.lower() != "https": |
| 44 | raise ValueError(f"{label} must use HTTPS: {url}") |
| 45 | if parsed.username or parsed.password: |
| 46 | raise ValueError(f"{label} must not include credentials: {url}") |
| 47 | host = (parsed.hostname or "").rstrip(".").lower() |
| 48 | if not host: |
| 49 | raise ValueError(f"{label} must include a hostname: {url}") |
| 50 | try: |
| 51 | port = parsed.port |
| 52 | except ValueError as exc: |
| 53 | raise ValueError(f"{label} has an invalid port: {url}") from exc |
| 54 | if port not in (None, 443): |
| 55 | raise ValueError(f"{label} must not use unexpected ports: {url}") |
| 56 | |
| 57 | |
| 58 | def _escape_markdown_url(url: str) -> str: |
| 59 | """Escape parentheses in URLs used inside markdown link syntax [text](url). |
| 60 | |
| 61 | A bare ')' in the URL would prematurely close the markdown link, potentially |
| 62 | allowing content injection in the rendered prompt. |
| 63 | """ |
| 64 | return url.replace("(", "%28").replace(")", "%29") |
| 65 | |
| 66 | |
| 67 | def current_week() -> str: |
| 68 | """Return the current ISO week as YYYY-WNN.""" |
| 69 | now = datetime.now() |
| 70 | iso = now.isocalendar() |
| 71 | return f"{iso[0]}-W{iso[1]:02d}" |
| 72 | |
| 73 | |
| 74 | def load_json(path: Path) -> dict | None: |
| 75 | """Load a JSON file, returning None if it doesn't exist.""" |
| 76 | if not path.exists(): |
| 77 | return None |
| 78 | with open(path, "r", encoding="utf-8") as f: |
| 79 | return json.load(f) |
| 80 | |
| 81 | |
| 82 | def format_articles_list(articles: list[dict]) -> str: |
| 83 | """Format articles into a markdown list.""" |
| 84 | from sanitize_repo_content import sanitize_text |
| 85 | |
| 86 | if not articles: |
| 87 | return "- (none)" |
| 88 | lines = [] |
| 89 | for article in articles[:MAX_RENDERED_ARTICLES]: |
| 90 | title = sanitize_text( |
| 91 | article.get("title", "Untitled"), |
| 92 | max_length=200, |
| 93 | label="article_title", |
| 94 | ) |
| 95 | url = sanitize_text( |
| 96 | article.get("url", ""), |
| 97 | max_length=300, |
| 98 | label="article_url", |
| 99 | ) |
| 100 | categories = [ |
| 101 | sanitize_text(c, max_length=50, label="article_category") |
| 102 | for c in article.get("categories", []) |
| 103 | if isinstance(c, str) |
| 104 | ] |
| 105 | source = sanitize_text( |
| 106 | article.get("source", "unknown"), |
| 107 | max_length=100, |
| 108 | label="article_source", |
| 109 | ) |
| 110 | published_at = sanitize_text( |
| 111 | article.get("published_at", ""), |
| 112 | max_length=20, |
| 113 | label="article_published_at", |
| 114 | ) |
| 115 | cat_str = f" [{', '.join(categories)}]" if categories else "" |
| 116 | source_str = f" — {source}" |
| 117 | if published_at: |
| 118 | source_str += f", {published_at[:10]}" |
| 119 | if url: |
| 120 | lines.append(f"- [{title}]({_escape_markdown_url(url)}){cat_str}{source_str}") |
| 121 | else: |
| 122 | lines.append(f"- {title}{cat_str}{source_str}") |
| 123 | omitted = len(articles) - MAX_RENDERED_ARTICLES |
| 124 | if omitted > 0: |
| 125 | lines.append(f"…and {omitted} more relevant articles within budget") |
| 126 | return "\n".join(lines) |
| 127 | |
| 128 | |
| 129 | _HYPE_RISK_SEVERITY: dict[str, int] = {"high": 3, "medium": 2, "low": 1, "none": 0} |
| 130 | |
| 131 | |
| 132 | def _fetch_readme_snippet(full_name: str, max_chars: int = 500) -> str: |
| 133 | """Fetch the first max_chars of a repo README from raw.githubusercontent.com. |
| 134 | |
| 135 | Returns an empty string on any failure (network error, 404, timeout). |
| 136 | Should only be called in reader_mode=True paths. |
| 137 | """ |
| 138 | if not GITHUB_REPO_FULL_NAME_RE.fullmatch(full_name): |
| 139 | return "" |
| 140 | url = f"https://raw.githubusercontent.com/{full_name}/HEAD/README.md" |
| 141 | try: |
| 142 | validate_https_url(url, label="README URL") |
| 143 | req = urllib.request.Request(url, headers={"User-Agent": "SquadScope/1.0"}) |
| 144 | with urllib.request.urlopen(req, timeout=5) as resp: # nosec B310 |
| 145 | raw = resp.read(max_chars * 3) |
| 146 | return raw.decode("utf-8", errors="replace")[:max_chars] |
| 147 | except Exception: |
| 148 | return "" |
| 149 | |
| 150 | |
| 151 | def _extract_readme_description(snippet: str) -> str: |
| 152 | """Return the first readable descriptive line from a README snippet. |
| 153 | |
| 154 | Skips headings, badge lines, image tags, and blank lines. |
| 155 | Trims each candidate line to the last complete sentence boundary |
| 156 | (. ! ?) so truncated snippets never show a broken mid-sentence tail. |
| 157 | If no sentence boundary is found within a line, that line is skipped. |
| 158 | Returns an empty string if nothing usable is found. |
| 159 | """ |
| 160 | for line in snippet.splitlines(): |
| 161 | line = line.strip() |
| 162 | if not line: |
| 163 | continue |
| 164 | if line.startswith(("#", "!", "<", "|", "[")): |
| 165 | continue |
| 166 | # Strip markdown formatting: links → text, remove bold/italic/code, strip HTML |
| 167 | line = re.sub(r"!\[.*?\]\(.*?\)", "", line) |
| 168 | line = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", line) |
| 169 | line = re.sub(r"<[^>]+>", "", line) |
| 170 | line = re.sub(r"[*_`>]", "", line) |
| 171 | line = line.strip() |
| 172 | if not line: |
| 173 | continue |
| 174 | # Trim to the last complete sentence boundary (. ! ? followed by space or end) |
| 175 | last_boundary = -1 |
| 176 | for i in range(len(line) - 1, -1, -1): |
| 177 | if line[i] in ".!?" and (i + 1 >= len(line) or line[i + 1] == " "): |
| 178 | last_boundary = i |
| 179 | break |
| 180 | if last_boundary < 0: |
| 181 | # No sentence boundary — drop this line to avoid broken sentences |
| 182 | continue |
| 183 | line = line[: last_boundary + 1].strip(" .,;:") |
| 184 | if len(line) >= 20: |
| 185 | return line |
| 186 | return "" |
| 187 | |
| 188 | |
| 189 | def _format_correlations_narrative(correlations: list[dict], articles: list[dict]) -> str: |
| 190 | """Generate narrative paragraphs explaining press-to-code correlations. |
| 191 | |
| 192 | Groups correlations by GitHub org, fetches README snippets for top repos, |
| 193 | and produces 1–3 prose paragraphs with inline links to repos and articles. |
| 194 | Only called in reader_mode=True — README network fetches happen here. |
| 195 | """ |
| 196 | if not correlations: |
| 197 | return "(No significant press correlations this week.)" |
| 198 | |
| 199 | # URL → title lookup for inline article links |
| 200 | url_to_title: dict[str, str] = { |
| 201 | a["url"]: a["title"] for a in articles if a.get("url") and a.get("title") |
| 202 | } |
| 203 | |
| 204 | # Sort correlations by confidence desc, hype_risk severity desc |
| 205 | sorted_corrs = sorted( |
| 206 | correlations, |
| 207 | key=lambda c: ( |
| 208 | -c.get("correlation_confidence", 0.0), |
| 209 | -_HYPE_RISK_SEVERITY.get(c.get("hype_risk", "none"), 0), |
| 210 | ), |
| 211 | ) |
| 212 | |
| 213 | # Group by org (first segment of "owner/repo") |
| 214 | org_groups: dict[str, list[dict]] = {} |
| 215 | for corr in sorted_corrs: |
| 216 | repo = corr.get("repo", "") |
| 217 | if not repo: |
| 218 | continue |
| 219 | org = repo.split("/")[0] |
| 220 | org_groups.setdefault(org, []).append(corr) |
| 221 | |
| 222 | def _group_score(corrs: list[dict]) -> float: |
| 223 | return sum(c.get("correlation_confidence", 0.0) for c in corrs) |
| 224 | |
| 225 | top_groups = sorted( |
| 226 | org_groups.items(), |
| 227 | key=lambda kv: _group_score(kv[1]), |
| 228 | reverse=True, |
| 229 | )[:4] |
| 230 | |
| 231 | # Fetch README snippets for the top repos across groups (max 6 total) |
| 232 | repos_to_fetch: list[str] = [] |
| 233 | for _, group_corrs in top_groups: |
| 234 | for corr in group_corrs[:2]: |
| 235 | repo = corr.get("repo", "") |
| 236 | if repo and repo not in repos_to_fetch and len(repos_to_fetch) < 6: |
| 237 | repos_to_fetch.append(repo) |
| 238 | |
| 239 | readme_snippets: dict[str, str] = {} |
| 240 | for repo in repos_to_fetch: |
| 241 | snippet = _fetch_readme_snippet(repo) |
| 242 | if snippet: |
| 243 | readme_snippets[repo] = snippet |
| 244 | |
| 245 | total = len(correlations) |
| 246 | paragraphs: list[str] = [] |
| 247 | |
| 248 | for idx, (org, group_corrs) in enumerate(top_groups[:3]): |
| 249 | # Collect up to 2 article links for this group |
| 250 | article_links: list[str] = [] |
| 251 | seen_article_urls: set[str] = set() |
| 252 | for corr in group_corrs: |
| 253 | for url in corr.get("matched_articles", []): |
| 254 | if url not in seen_article_urls and len(article_links) < 2: |
| 255 | seen_article_urls.add(url) |
| 256 | title = url_to_title.get(url, "") |
| 257 | if title: |
| 258 | article_links.append(f"[{title}]({_escape_markdown_url(url)})") |
| 259 | |
| 260 | # Collect up to 3 repo links with optional README description |
| 261 | repo_parts: list[str] = [] |
| 262 | for corr in group_corrs[:3]: |
| 263 | repo = corr.get("repo", "") |
| 264 | if not repo: |
| 265 | continue |
| 266 | link = _repo_link(repo) |
| 267 | desc = _extract_readme_description(readme_snippets.get(repo, "")) |
| 268 | repo_parts.append(f"{link} — {desc}" if desc else link) |
| 269 | |
| 270 | if not repo_parts: |
| 271 | continue |
| 272 | |
| 273 | repos_str = _join_links(repo_parts) |
| 274 | |
| 275 | if article_links: |
| 276 | arts_str = _join_links(article_links) |
| 277 | if idx == 0: |
| 278 | para = ( |
| 279 | f"This week's external press coverage closely tracks developer activity " |
| 280 | f"across {total} repos. {org.capitalize()} featured prominently: " |
| 281 | f"coverage of {arts_str} aligns with activity in {repos_str}." |
| 282 | ) |
| 283 | else: |
| 284 | para = ( |
| 285 | f"{org.capitalize()}'s press footprint also intersects with GitHub: " |
| 286 | f"coverage of {arts_str} tracks activity in {repos_str}." |
| 287 | ) |
| 288 | else: |
| 289 | if idx == 0: |
| 290 | para = ( |
| 291 | f"This week's external press coverage closely tracks developer activity " |
| 292 | f"across {total} repos. {org.capitalize()} shows the strongest signal, " |
| 293 | f"with {repos_str} seeing notable GitHub traction." |
| 294 | ) |
| 295 | else: |
| 296 | para = ( |
| 297 | f"{org.capitalize()} also shows strong press-to-code correlation, " |
| 298 | f"with activity in {repos_str}." |
| 299 | ) |
| 300 | |
| 301 | paragraphs.append(para) |
| 302 | |
| 303 | return ( |
| 304 | "\n\n".join(paragraphs) if paragraphs else "(No significant press correlations this week.)" |
| 305 | ) |
| 306 | |
| 307 | |
| 308 | def format_correlations_list( |
| 309 | correlations: list[dict], |
| 310 | *, |
| 311 | top_n: int | None = None, |
| 312 | reader_mode: bool = False, |
| 313 | articles: list[dict] | None = None, |
| 314 | ) -> str: |
| 315 | """Format correlations into a markdown list or narrative prose. |
| 316 | |
| 317 | Args: |
| 318 | correlations: List of correlation dicts. |
| 319 | top_n: When set (and reader_mode=False), show only the top N entries |
| 320 | (sorted by confidence desc, then hype_risk severity desc) and |
| 321 | append a "…and N more" summary line. |
| 322 | reader_mode: When True, delegate to _format_correlations_narrative() |
| 323 | which produces prose paragraphs with inline links. top_n |
| 324 | is ignored in this mode. |
| 325 | articles: Article list used by the narrative formatter for URL→title |
| 326 | lookup. Ignored when reader_mode=False. |
| 327 | """ |
| 328 | if not correlations: |
| 329 | return "- (none)" |
| 330 | |
| 331 | if reader_mode: |
| 332 | return _format_correlations_narrative(correlations, articles or []) |
| 333 | |
| 334 | if top_n is not None: |
| 335 | sorted_corrs = sorted( |
| 336 | correlations, |
| 337 | key=lambda c: ( |
| 338 | -c.get("correlation_confidence", 0.0), |
| 339 | -_HYPE_RISK_SEVERITY.get(c.get("hype_risk", "none"), 0), |
| 340 | ), |
| 341 | ) |
| 342 | omitted = max(0, len(sorted_corrs) - top_n) |
| 343 | display = sorted_corrs[:top_n] |
| 344 | else: |
| 345 | display = correlations |
| 346 | omitted = 0 |
| 347 | |
| 348 | from sanitize_repo_content import sanitize_text as _sanitize |
| 349 | |
| 350 | lines = [] |
| 351 | for corr in display: |
| 352 | repo = corr.get("repo", "unknown") |
| 353 | match_type = corr.get("match_type", "unknown") |
| 354 | confidence = corr.get("correlation_confidence", 0.0) |
| 355 | strength = corr.get("correlation_strength", corr.get("confidence_label", "unknown")) |
| 356 | hype_risk = corr.get("hype_risk", "none") |
| 357 | details = corr.get("matched_article_details", []) |
| 358 | sources = sorted( |
| 359 | { |
| 360 | source |
| 361 | for detail in details |
| 362 | for source in detail.get("sources", [detail.get("source", "unknown")]) |
| 363 | } |
| 364 | ) |
| 365 | citation = "" |
| 366 | if details: |
| 367 | first = details[0] |
| 368 | title = _sanitize( |
| 369 | first.get("title", "article"), |
| 370 | max_length=200, |
| 371 | label="correlation_article_title", |
| 372 | ) |
| 373 | url = _sanitize( |
| 374 | first.get("url", ""), |
| 375 | max_length=300, |
| 376 | label="correlation_article_url", |
| 377 | ) |
| 378 | citation = ( |
| 379 | f", cited: [{title}]({_escape_markdown_url(url)})" if url else f", cited: {title}" |
| 380 | ) |
| 381 | lines.append( |
| 382 | f"- {repo} — match: {match_type}, " |
| 383 | f"strength: {strength}, confidence: {confidence:.1f}, " |
| 384 | f"sources: {', '.join(sources) if sources else 'unknown'}, " |
| 385 | f"hype_risk: {hype_risk}{citation}" |
| 386 | ) |
| 387 | |
| 388 | if omitted > 0: |
| 389 | lines.append(f"…and {omitted} more repos with press correlation") |
| 390 | |
| 391 | return "\n".join(lines) |
| 392 | |
| 393 | |
| 394 | def _repo_link(full_name: str) -> str: |
| 395 | """Format a repo as a markdown link using only the repo name (after the slash).""" |
| 396 | repo_name = full_name.split("/")[-1] |
| 397 | return f"[{repo_name}](https://github.com/{full_name})" |
| 398 | |
| 399 | |
| 400 | def _join_links(links: list[str]) -> str: |
| 401 | """Join a list of markdown links into a readable phrase.""" |
| 402 | if len(links) == 1: |
| 403 | return links[0] |
| 404 | if len(links) == 2: |
| 405 | return f"{links[0]} and {links[1]}" |
| 406 | return f"{', '.join(links[:-1])}, and {links[-1]}" |
| 407 | |
| 408 | |
| 409 | def _format_unpublicized_narrative(items: list[dict]) -> str: |
| 410 | """Generate narrative paragraph(s) for dev activity without press coverage.""" |
| 411 | if not items: |
| 412 | return "" |
| 413 | |
| 414 | # Sort topics by total stars, cap at 6 |
| 415 | sorted_items = sorted( |
| 416 | items, |
| 417 | key=lambda x: sum(r.get("stars", 0) for r in x.get("github_repos", [])), |
| 418 | reverse=True, |
| 419 | )[:6] |
| 420 | |
| 421 | topic_parts: list[tuple[str, list[str]]] = [] |
| 422 | for item in sorted_items: |
| 423 | topic = item.get("topic", "unknown") |
| 424 | repos = sorted( |
| 425 | item.get("github_repos", []), |
| 426 | key=lambda r: r.get("stars", 0), |
| 427 | reverse=True, |
| 428 | ) |
| 429 | links = [_repo_link(r["full_name"]) for r in repos[:3] if r.get("full_name")] |
| 430 | if links: |
| 431 | topic_parts.append((topic, links)) |
| 432 | |
| 433 | if not topic_parts: |
| 434 | return "" |
| 435 | |
| 436 | # First paragraph: intro + first three topics |
| 437 | first_batch = topic_parts[:3] |
| 438 | fragments = [f"{topic} saw activity with {_join_links(links)}" for topic, links in first_batch] |
| 439 | para1 = ( |
| 440 | "Developer activity this week shows momentum in areas the tech press isn't covering. " |
| 441 | + "; ".join(fragments) |
| 442 | + "." |
| 443 | ) |
| 444 | |
| 445 | paragraphs = [para1] |
| 446 | |
| 447 | # Second paragraph for remaining topics |
| 448 | if len(topic_parts) > 3: |
| 449 | second_batch = topic_parts[3:] |
| 450 | fragments2 = [f"{topic} with {_join_links(links)}" for topic, links in second_batch] |
| 451 | paragraphs.append("Additional activity surfaced in " + ", ".join(fragments2) + ".") |
| 452 | |
| 453 | paragraphs.append( |
| 454 | "These gaps suggest that foundational developer tooling — the infrastructure " |
| 455 | "that powers daily workflows — grows through community word-of-mouth rather than press cycles." |
| 456 | ) |
| 457 | |
| 458 | return "\n\n".join(paragraphs) |
| 459 | |
| 460 | |
| 461 | def _format_uncovered_narrative(items: list[dict]) -> str: |
| 462 | """Generate a narrative paragraph for tech trends without dev activity.""" |
| 463 | if not items: |
| 464 | return "" |
| 465 | |
| 466 | display = items[:5] |
| 467 | |
| 468 | topic_names = [item.get("topic", "unknown") for item in display] |
| 469 | |
| 470 | # Collect up to two article links across all topics |
| 471 | article_links: list[str] = [] |
| 472 | for item in display: |
| 473 | for a in item.get("news_articles", item.get("techcrunch_articles", []))[:1]: |
| 474 | title = a.get("title", "article") |
| 475 | url = a.get("url", "") |
| 476 | if url: |
| 477 | article_links.append(f"[{title}]({_escape_markdown_url(url)})") |
| 478 | if len(article_links) >= 2: |
| 479 | break |
| 480 | |
| 481 | if len(topic_names) == 1: |
| 482 | topics_str = topic_names[0] |
| 483 | elif len(topic_names) == 2: |
| 484 | topics_str = f"{topic_names[0]} and {topic_names[1]}" |
| 485 | else: |
| 486 | topics_str = f"{', '.join(topic_names[:-1])}, and {topic_names[-1]}" |
| 487 | |
| 488 | if article_links: |
| 489 | if len(article_links) == 1: |
| 490 | article_str = f"Articles like {article_links[0]} generated buzz" |
| 491 | else: |
| 492 | article_str = f"Articles like {article_links[0]} and {article_links[1]} generated buzz" |
| 493 | else: |
| 494 | article_str = "Press articles generated buzz" |
| 495 | |
| 496 | return ( |
| 497 | f"External press heavily covered {topics_str} this week, but GitHub shows minimal " |
| 498 | f"matching developer activity. {article_str}, yet no significant new repositories " |
| 499 | f"emerged in these spaces — suggesting these are still in the narrative or " |
| 500 | f"announcement phase rather than implementation." |
| 501 | ) |
| 502 | |
| 503 | |
| 504 | def format_divergences(divergences: dict, *, reader_mode: bool = False) -> str: |
| 505 | """Format divergences section into markdown. |
| 506 | |
| 507 | Args: |
| 508 | divergences: Divergence data dict. |
| 509 | reader_mode: When True, renders narrative paragraphs with inline repo/article |
| 510 | links instead of raw bullet lists. When False (AI prompt mode), |
| 511 | the original bullet-list format is preserved unchanged. |
| 512 | """ |
| 513 | if not divergences: |
| 514 | return "" |
| 515 | |
| 516 | uncovered = divergences.get("uncovered_tech_trends", []) |
| 517 | unpublicized = divergences.get("unpublicized_dev_activity", []) |
| 518 | |
| 519 | if not uncovered and not unpublicized: |
| 520 | return "" |
| 521 | |
| 522 | lines = ["\n### Divergence Analysis\n"] |
| 523 | |
| 524 | if reader_mode: |
| 525 | # Narrative mode: flowing prose with inline links, no raw data dumps |
| 526 | if uncovered: |
| 527 | lines.append("#### 🔍 Tech Trends Without Dev Activity\n") |
| 528 | lines.append(_format_uncovered_narrative(uncovered)) |
| 529 | lines.append("") |
| 530 | |
| 531 | if unpublicized: |
| 532 | lines.append("#### 🚀 Dev Activity Without Press Coverage\n") |
| 533 | lines.append(_format_unpublicized_narrative(unpublicized)) |
| 534 | lines.append("") |
| 535 | else: |
| 536 | # AI prompt mode: full raw data for model consumption — keep unchanged |
| 537 | if uncovered: |
| 538 | lines.append("#### 🔍 Tech Trends Without Dev Activity") |
| 539 | lines.append( |
| 540 | "Topics heavily covered by external press with no matching GitHub repos:\n" |
| 541 | ) |
| 542 | for item in uncovered: |
| 543 | topic = item.get("topic", "unknown") |
| 544 | articles = item.get("news_articles", item.get("techcrunch_articles", [])) |
| 545 | article_refs = ", ".join( |
| 546 | f"[{a.get('title', 'article')}]({_escape_markdown_url(a.get('url', ''))})" |
| 547 | for a in articles[:3] |
| 548 | ) |
| 549 | lines.append(f"- **{topic}**: {article_refs}") |
| 550 | lines.append("") |
| 551 | |
| 552 | if unpublicized: |
| 553 | lines.append("#### 🚀 Dev Activity Without Press Coverage") |
| 554 | lines.append("GitHub repos/trends with no matching external press coverage:\n") |
| 555 | for item in unpublicized: |
| 556 | topic = item.get("topic", "unknown") |
| 557 | repos = item.get("github_repos", []) |
| 558 | repo_refs = ", ".join( |
| 559 | f"{r.get('full_name', '?')} (⭐{r.get('stars', 0)})" for r in repos[:3] |
| 560 | ) |
| 561 | lines.append(f"- **{topic}**: {repo_refs}") |
| 562 | lines.append("") |
| 563 | |
| 564 | lines.append("#### Divergence Instructions") |
| 565 | lines.append("Use divergences to identify:") |
| 566 | lines.append("- 🔮 Where industry is moving but devs haven't caught up") |
| 567 | lines.append("- 💡 Where devs are innovating ahead of media attention") |
| 568 | lines.append("- 📊 Opportunity gaps between narrative and reality") |
| 569 | |
| 570 | return "\n".join(lines) |
| 571 | |
| 572 | |
| 573 | def _source_caveats(techcrunch_data: dict | None, correlation_data: dict | None) -> str: |
| 574 | """Render concise partial-failure caveats from crawl/correlation metadata.""" |
| 575 | metadata: dict = {} |
| 576 | if techcrunch_data: |
| 577 | metadata = techcrunch_data.get("metadata", {}) |
| 578 | corr_sources = {} |
| 579 | if correlation_data: |
| 580 | corr_sources = correlation_data.get("metadata", {}).get("news_sources", {}) |
| 581 | |
| 582 | requested = metadata.get("sources_requested") or corr_sources.get("sources_requested") or [] |
| 583 | succeeded = metadata.get("sources_succeeded") or corr_sources.get("sources_succeeded") or [] |
| 584 | failed = metadata.get("sources_failed") or corr_sources.get("sources_failed") or [] |
| 585 | errors = metadata.get("errors") or corr_sources.get("errors") or [] |
| 586 | if not requested and not failed: |
| 587 | return "" |
| 588 | lines = [ |
| 589 | "### Source Coverage", |
| 590 | f"- Sources requested: {', '.join(requested) if requested else 'unknown'}", |
| 591 | f"- Sources succeeded: {', '.join(succeeded) if succeeded else 'none'}", |
| 592 | ] |
| 593 | if failed: |
| 594 | lines.append(f"- Partial crawl caveat: failed sources: {', '.join(failed)}") |
| 595 | for error in errors[:3]: |
| 596 | lines.append( |
| 597 | f" - {error.get('source', 'unknown')}: " |
| 598 | f"{error.get('error_class', 'error')} {error.get('error', '')}".strip() |
| 599 | ) |
| 600 | return "\n".join(lines) |
| 601 | |
| 602 | |
| 603 | def _source_coverage( |
| 604 | techcrunch_data: dict | None, correlation_data: dict | None |
| 605 | ) -> dict[str, list[str]]: |
| 606 | metadata = techcrunch_data.get("metadata", {}) if techcrunch_data else {} |
| 607 | corr_sources = ( |
| 608 | correlation_data.get("metadata", {}).get("news_sources", {}) if correlation_data else {} |
| 609 | ) |
| 610 | requested = metadata.get("sources_requested") or corr_sources.get("sources_requested") or [] |
| 611 | succeeded = metadata.get("sources_succeeded") or corr_sources.get("sources_succeeded") or [] |
| 612 | failed = metadata.get("sources_failed") or corr_sources.get("sources_failed") or [] |
| 613 | return { |
| 614 | "requested": [str(item) for item in requested], |
| 615 | "succeeded": [str(item) for item in succeeded], |
| 616 | "failed": [str(item) for item in failed], |
| 617 | } |
| 618 | |
| 619 | |
| 620 | def estimate_tokens(markdown: str) -> int: |
| 621 | """Return a rough token estimate used for telemetry and hard budget checks.""" |
| 622 | return max(1, (len(markdown) + 3) // 4) |
| 623 | |
| 624 | |
| 625 | def press_token_estimate(content: str) -> int: |
| 626 | """Return 0 for empty content or no-press sentinel so gate fallback matches path logic.""" |
| 627 | stripped = content.strip() |
| 628 | if not stripped or NO_PRESS_SENTINEL_MARKER.search(stripped): |
| 629 | return 0 |
| 630 | return estimate_tokens(stripped) |
| 631 | |
| 632 | |
| 633 | def enforce_press_context_budget(markdown: str) -> str: |
| 634 | """Keep press context below the documented token budget.""" |
| 635 | if estimate_tokens(markdown) <= PRESS_CONTEXT_TOKEN_BUDGET: |
| 636 | return markdown |
| 637 | budget_note = ( |
| 638 | "\n\n### Budget Notice\n" |
| 639 | f"Press context truncated to ~{PRESS_CONTEXT_TOKEN_BUDGET} tokens; " |
| 640 | "citations and source caveats above are prioritized.\n" |
| 641 | ) |
| 642 | keep_chars = max(0, PRESS_CONTEXT_CHAR_BUDGET - len(budget_note)) |
| 643 | truncated = markdown[:keep_chars].rsplit("\n", 1)[0] |
| 644 | return truncated + budget_note |
| 645 | |
| 646 | |
| 647 | def render_press_context( |
| 648 | techcrunch_data: dict | None, |
| 649 | correlation_data: dict | None, |
| 650 | week: str, |
| 651 | *, |
| 652 | reader_mode: bool = False, |
| 653 | ) -> str: |
| 654 | """Render the press context prompt section. |
| 655 | |
| 656 | Args: |
| 657 | techcrunch_data: Parsed TechCrunch crawl JSON or None. |
| 658 | correlation_data: Parsed correlation JSON or None. |
| 659 | week: The week string (YYYY-WNN). |
| 660 | reader_mode: When True, produces reader-facing output: top-10 correlations |
| 661 | only, no AI instruction blocks, and a narrative divergence |
| 662 | conclusion instead of model directives. |
| 663 | |
| 664 | Returns: |
| 665 | Rendered markdown prompt section. |
| 666 | """ |
| 667 | if techcrunch_data is None and correlation_data is None: |
| 668 | return NO_PRESS_SENTINEL |
| 669 | |
| 670 | template_path = _REPO_ROOT / "prompts" / "analyze-press-context.md" |
| 671 | template = template_path.read_text(encoding="utf-8") |
| 672 | |
| 673 | # Extract articles (filter to relevant ones) |
| 674 | articles = [] |
| 675 | if techcrunch_data: |
| 676 | all_articles = techcrunch_data.get("articles", []) |
| 677 | articles = [a for a in all_articles if a.get("relevance_score", 0) >= 0.4] |
| 678 | |
| 679 | # Extract correlations |
| 680 | correlations = [] |
| 681 | if correlation_data: |
| 682 | correlations = correlation_data.get("correlations", []) |
| 683 | |
| 684 | article_count = len(articles) |
| 685 | correlation_count = len(correlations) |
| 686 | |
| 687 | # Extract divergences |
| 688 | divergences = {} |
| 689 | if correlation_data: |
| 690 | divergences = correlation_data.get("divergences", {}) |
| 691 | |
| 692 | # In reader mode, sort correlations by confidence desc then hype_risk severity |
| 693 | if reader_mode and correlations: |
| 694 | correlations = sorted( |
| 695 | correlations, |
| 696 | key=lambda c: ( |
| 697 | -c.get("correlation_confidence", 0.0), |
| 698 | -_HYPE_RISK_SEVERITY.get(c.get("hype_risk", "none"), 0), |
| 699 | ), |
| 700 | ) |
| 701 | |
| 702 | top_n = MAX_RENDERED_CORRELATIONS if reader_mode else None |
| 703 | |
| 704 | # Render template |
| 705 | source_label = "External news" |
| 706 | rendered = template.replace("TechCrunch", source_label) |
| 707 | rendered = rendered.replace("{date}", week) |
| 708 | rendered = rendered.replace("{article_count}", str(article_count)) |
| 709 | rendered = rendered.replace("{articles_list}", format_articles_list(articles)) |
| 710 | rendered = rendered.replace("{correlation_count}", str(correlation_count)) |
| 711 | rendered = rendered.replace( |
| 712 | "{correlations_list}", |
| 713 | format_correlations_list( |
| 714 | correlations, |
| 715 | top_n=top_n, |
| 716 | reader_mode=reader_mode, |
| 717 | articles=articles, |
| 718 | ), |
| 719 | ) |
| 720 | |
| 721 | # Strip the AI-only ### Instructions block in reader mode |
| 722 | if reader_mode: |
| 723 | instructions_marker = "\n### Instructions\n" |
| 724 | if instructions_marker in rendered: |
| 725 | rendered = rendered[: rendered.index(instructions_marker)] |
| 726 | # Remove the count header "N repos have press correlation:" — narrative |
| 727 | # paragraphs are self-contained; the raw count line is noise in reader mode. |
| 728 | rendered = re.sub(r"\d+ repos have press correlation:\n", "", rendered) |
| 729 | |
| 730 | # Append divergences section |
| 731 | divergence_section = format_divergences(divergences, reader_mode=reader_mode) |
| 732 | if divergence_section: |
| 733 | rendered += "\n" + divergence_section |
| 734 | |
| 735 | caveats = _source_caveats(techcrunch_data, correlation_data) |
| 736 | if caveats: |
| 737 | rendered += "\n\n" + caveats |
| 738 | |
| 739 | rendered += ( |
| 740 | "\n\n### Press Context Telemetry\n" |
| 741 | f"- token_estimate: {estimate_tokens(rendered)}\n" |
| 742 | f"- token_budget: {PRESS_CONTEXT_TOKEN_BUDGET}\n" |
| 743 | f"- article_limit: {MAX_RENDERED_ARTICLES}\n" |
| 744 | f"- articles_retained: {min(article_count, MAX_RENDERED_ARTICLES)}\n" |
| 745 | f"- articles_dropped: {max(0, article_count - MAX_RENDERED_ARTICLES)}\n" |
| 746 | f"- correlation_limit: {MAX_RENDERED_CORRELATIONS if reader_mode else 'unbounded-input'}\n" |
| 747 | f"- correlations_retained: {min(correlation_count, MAX_RENDERED_CORRELATIONS) if reader_mode else correlation_count}\n" |
| 748 | f"- correlations_dropped: {max(0, correlation_count - MAX_RENDERED_CORRELATIONS) if reader_mode else 0}\n" |
| 749 | ) |
| 750 | coverage = _source_coverage(techcrunch_data, correlation_data) |
| 751 | rendered += ( |
| 752 | f"- sources_requested: {', '.join(coverage['requested']) if coverage['requested'] else 'unknown'}\n" |
| 753 | f"- sources_succeeded: {', '.join(coverage['succeeded']) if coverage['succeeded'] else 'unknown'}\n" |
| 754 | f"- sources_failed: {', '.join(coverage['failed']) if coverage['failed'] else 'none'}\n" |
| 755 | ) |
| 756 | |
| 757 | return enforce_press_context_budget(rendered) |
| 758 | |
| 759 | |
| 760 | def resolve_paths(topic: str | None, week: str) -> tuple[Path, Path]: |
| 761 | """Resolve file paths for external news and correlation data.""" |
| 762 | raw_path = raw_dir(topic) |
| 763 | external_path = raw_path / f"{week}-external-news.json" |
| 764 | legacy_path = raw_path / f"{week}-techcrunch.json" |
| 765 | tc_path = legacy_path if legacy_path.exists() and not external_path.exists() else external_path |
| 766 | corr_path = analyzed_dir(topic) / f"{week}-correlations.json" |
| 767 | return tc_path, corr_path |
| 768 | |
| 769 | |
| 770 | def main() -> None: |
| 771 | parser = argparse.ArgumentParser(description="Render press context prompt section") |
| 772 | parser.add_argument("--topic", default=None, help="Topic ID (e.g., ai-ml)") |
| 773 | parser.add_argument("--week", default=None, help="Week in YYYY-WNN format (default: current)") |
| 774 | args = parser.parse_args() |
| 775 | |
| 776 | week = args.week or current_week() |
| 777 | topic = args.topic |
| 778 | |
| 779 | tc_path, corr_path = resolve_paths(topic, week) |
| 780 | techcrunch_data = load_json(tc_path) |
| 781 | correlation_data = load_json(corr_path) |
| 782 | |
| 783 | output = render_press_context(techcrunch_data, correlation_data, week) |
| 784 | print(output) |
| 785 | |
| 786 | |
| 787 | if __name__ == "__main__": |
| 788 | main() |