| 1 | #!/usr/bin/env python3 |
| 2 | """External news RSS crawler with entity extraction for SquadScope. |
| 3 | |
| 4 | Fetches configured RSS feeds in parallel, extracts structured metadata, |
| 5 | GitHub URLs, and entities (company/project names). |
| 6 | |
| 7 | Usage: |
| 8 | python scripts/techcrunch_crawler.py [--topic ai-ml] \ |
| 9 | [--output data/raw/ai-ml/2026-W21-external-news.json] [--since 2026-05-11] |
| 10 | """ |
| 11 | |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import argparse |
| 15 | import hashlib |
| 16 | import ipaddress |
| 17 | import json |
| 18 | import math |
| 19 | import os |
| 20 | import re |
| 21 | import secrets |
| 22 | import sys |
| 23 | import time |
| 24 | from collections import Counter |
| 25 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 26 | from dataclasses import dataclass |
| 27 | from datetime import UTC, date, datetime, timedelta, timezone |
| 28 | from email.utils import parsedate_to_datetime |
| 29 | from pathlib import Path |
| 30 | from typing import Any |
| 31 | from urllib.error import HTTPError, URLError |
| 32 | from urllib.parse import urlparse |
| 33 | from urllib.request import Request, urlopen |
| 34 | |
| 35 | import feedparser |
| 36 | |
| 37 | from scripts.observability_metrics import ( |
| 38 | DEFAULT_OBSERVABILITY_DIR, |
| 39 | METRICS_SCHEMA_VERSION, |
| 40 | CrawlMetrics, |
| 41 | ObservabilityLedger, |
| 42 | duration_p95, |
| 43 | emit_ledger, |
| 44 | ) |
| 45 | from scripts.topic_paths import raw_dir |
| 46 | |
| 47 | FEED_URL = "https://techcrunch.com/feed/" |
| 48 | DEFAULT_SOURCES_PATH = Path("config/external_news_sources.json") |
| 49 | DEFAULT_FETCH_TIMEOUT_SECONDS = 15 |
| 50 | DEFAULT_FETCH_RETRIES = 3 |
| 51 | DEFAULT_MAX_WORKERS = 8 |
| 52 | # HTTP statuses worth retrying on a transient failure; everything else (e.g. |
| 53 | # 400/401/404) is treated as permanent and fails the single source fast. |
| 54 | RETRYABLE_STATUSES = frozenset({403, 408, 429, 500, 502, 503, 504}) |
| 55 | # Exponential backoff bounds for per-source retries (mirrors scripts/crawl.py). |
| 56 | RETRY_BASE_DELAY_SECONDS = 2.0 |
| 57 | RETRY_MAX_DELAY_SECONDS = 30.0 |
| 58 | # Bound hostile/absurd Retry-After values while honoring reasonable server |
| 59 | # requests well above the crawler's computed 30s backoff cap. |
| 60 | RETRY_AFTER_MAX_SECONDS = 120.0 |
| 61 | _JITTER_RANDOM = secrets.SystemRandom() |
| 62 | CANONICAL_SCHEMA_VERSION = 2 |
| 63 | APPROVED_FEED_HOSTS = frozenset( |
| 64 | { |
| 65 | "techcrunch.com", |
| 66 | "blogs.nvidia.com", |
| 67 | "huggingface.co", |
| 68 | "www.technologyreview.com", |
| 69 | "github.blog", |
| 70 | } |
| 71 | ) |
| 72 | |
| 73 | GITHUB_URL_RE = re.compile(r"https?://github\.com/[a-zA-Z0-9_.\-]+/[a-zA-Z0-9_.\-]+") |
| 74 | |
| 75 | TECH_KEYWORDS = { |
| 76 | "ai", |
| 77 | "ml", |
| 78 | "machine learning", |
| 79 | "deep learning", |
| 80 | "open-source", |
| 81 | "open source", |
| 82 | "github", |
| 83 | "developer", |
| 84 | "api", |
| 85 | "framework", |
| 86 | "sdk", |
| 87 | "llm", |
| 88 | "gpt", |
| 89 | "model", |
| 90 | "neural", |
| 91 | "transformer", |
| 92 | "cloud", |
| 93 | "devops", |
| 94 | "kubernetes", |
| 95 | "docker", |
| 96 | "rust", |
| 97 | "python", |
| 98 | "javascript", |
| 99 | "typescript", |
| 100 | "golang", |
| 101 | "database", |
| 102 | "vector", |
| 103 | "embedding", |
| 104 | "agent", |
| 105 | "rag", |
| 106 | "fine-tuning", |
| 107 | "inference", |
| 108 | "startup", |
| 109 | "oss", |
| 110 | } |
| 111 | |
| 112 | # Common lowercase words that should not be treated as entities |
| 113 | STOP_WORDS = { |
| 114 | "a", |
| 115 | "an", |
| 116 | "the", |
| 117 | "and", |
| 118 | "or", |
| 119 | "but", |
| 120 | "in", |
| 121 | "on", |
| 122 | "at", |
| 123 | "to", |
| 124 | "for", |
| 125 | "of", |
| 126 | "with", |
| 127 | "by", |
| 128 | "from", |
| 129 | "is", |
| 130 | "are", |
| 131 | "was", |
| 132 | "were", |
| 133 | "be", |
| 134 | "been", |
| 135 | "has", |
| 136 | "have", |
| 137 | "had", |
| 138 | "do", |
| 139 | "does", |
| 140 | "did", |
| 141 | "will", |
| 142 | "would", |
| 143 | "could", |
| 144 | "should", |
| 145 | "may", |
| 146 | "might", |
| 147 | "can", |
| 148 | "this", |
| 149 | "that", |
| 150 | "these", |
| 151 | "those", |
| 152 | "it", |
| 153 | "its", |
| 154 | "new", |
| 155 | "how", |
| 156 | "why", |
| 157 | "what", |
| 158 | "when", |
| 159 | "where", |
| 160 | "who", |
| 161 | "all", |
| 162 | "just", |
| 163 | "more", |
| 164 | "most", |
| 165 | "some", |
| 166 | "any", |
| 167 | "no", |
| 168 | "not", |
| 169 | "than", |
| 170 | "too", |
| 171 | "very", |
| 172 | "also", |
| 173 | "about", |
| 174 | "up", |
| 175 | "out", |
| 176 | "into", |
| 177 | "over", |
| 178 | "after", |
| 179 | "before", |
| 180 | "between", |
| 181 | "under", |
| 182 | "again", |
| 183 | "here", |
| 184 | "there", |
| 185 | "now", |
| 186 | "then", |
| 187 | "once", |
| 188 | "well", |
| 189 | "back", |
| 190 | "still", |
| 191 | "even", |
| 192 | "big", |
| 193 | "first", |
| 194 | "last", |
| 195 | "next", |
| 196 | "says", |
| 197 | "said", |
| 198 | "gets", |
| 199 | "got", |
| 200 | "makes", |
| 201 | "made", |
| 202 | "takes", |
| 203 | "took", |
| 204 | "goes", |
| 205 | "went", |
| 206 | "comes", |
| 207 | "came", |
| 208 | "wants", |
| 209 | "launches", |
| 210 | "raises", |
| 211 | "builds", |
| 212 | "looks", |
| 213 | "like", |
| 214 | "use", |
| 215 | "using", |
| 216 | "used", |
| 217 | } |
| 218 | |
| 219 | |
| 220 | @dataclass(frozen=True, slots=True) |
| 221 | class NewsSourceConfig: |
| 222 | """Configuration for one external RSS source.""" |
| 223 | |
| 224 | name: str |
| 225 | feed_url: str |
| 226 | requests_per_minute: int = 10 |
| 227 | |
| 228 | def __post_init__(self) -> None: |
| 229 | validate_feed_url(self.feed_url) |
| 230 | |
| 231 | @property |
| 232 | def host(self) -> str: |
| 233 | """Return the normalized feed host.""" |
| 234 | return (urlparse(self.feed_url).hostname or "").rstrip(".").lower() |
| 235 | |
| 236 | |
| 237 | def validate_feed_url(url: str) -> None: |
| 238 | """Validate an external RSS URL against the approved egress allowlist.""" |
| 239 | parsed = urlparse(url) |
| 240 | if parsed.scheme.lower() != "https": |
| 241 | raise ValueError(f"External RSS feed URL must use HTTPS: {url}") |
| 242 | if parsed.username or parsed.password: |
| 243 | raise ValueError(f"External RSS feed URL must not include credentials: {url}") |
| 244 | host = (parsed.hostname or "").rstrip(".").lower() |
| 245 | if not host: |
| 246 | raise ValueError(f"External RSS feed URL must include a hostname: {url}") |
| 247 | try: |
| 248 | port = parsed.port |
| 249 | except ValueError as exc: |
| 250 | raise ValueError(f"External RSS feed URL has an invalid port: {url}") from exc |
| 251 | if port not in (None, 443): |
| 252 | raise ValueError(f"External RSS feed URL must not use unexpected ports: {url}") |
| 253 | if host in {"localhost", "localhost.localdomain"} or host.endswith(".local"): |
| 254 | raise ValueError(f"External RSS feed URL must not target local hosts: {url}") |
| 255 | try: |
| 256 | ip_addr = ipaddress.ip_address(host) |
| 257 | except ValueError: |
| 258 | pass |
| 259 | else: |
| 260 | if ip_addr.is_private or ip_addr.is_loopback or ip_addr.is_link_local: |
| 261 | raise ValueError(f"External RSS feed URL must not target private/local IPs: {url}") |
| 262 | if host not in APPROVED_FEED_HOSTS: |
| 263 | approved = ", ".join(sorted(APPROVED_FEED_HOSTS)) |
| 264 | raise ValueError(f"External RSS feed host is not approved: {host} (approved: {approved})") |
| 265 | |
| 266 | |
| 267 | def load_source_configs(path: Path = DEFAULT_SOURCES_PATH) -> list[NewsSourceConfig]: |
| 268 | """Load external RSS source config from JSON.""" |
| 269 | payload = json.loads(path.read_text(encoding="utf-8")) |
| 270 | if not isinstance(payload, list): |
| 271 | raise ValueError(f"Expected a list of source configs in {path}") |
| 272 | |
| 273 | sources: list[NewsSourceConfig] = [] |
| 274 | seen_names: set[str] = set() |
| 275 | for raw in payload: |
| 276 | if not isinstance(raw, dict): |
| 277 | raise ValueError(f"Invalid source config in {path}: {raw!r}") |
| 278 | name = str(raw.get("name", "")).strip() |
| 279 | feed_url = str(raw.get("feed_url", "")).strip() |
| 280 | if not name or not feed_url: |
| 281 | raise ValueError(f"Source configs require name and feed_url: {raw!r}") |
| 282 | if name in seen_names: |
| 283 | raise ValueError(f"Duplicate external news source name: {name}") |
| 284 | seen_names.add(name) |
| 285 | sources.append( |
| 286 | NewsSourceConfig( |
| 287 | name=name, |
| 288 | feed_url=feed_url, |
| 289 | requests_per_minute=int(raw.get("requests_per_minute", 10)), |
| 290 | ) |
| 291 | ) |
| 292 | return sources |
| 293 | |
| 294 | |
| 295 | def source_config_checksum(sources: list[NewsSourceConfig]) -> str: |
| 296 | """Return a stable checksum for the effective source config.""" |
| 297 | canonical = [ |
| 298 | { |
| 299 | "feed_url": source.feed_url, |
| 300 | "name": source.name, |
| 301 | "requests_per_minute": source.requests_per_minute, |
| 302 | } |
| 303 | for source in sorted(sources, key=lambda item: item.name) |
| 304 | ] |
| 305 | payload = json.dumps(canonical, sort_keys=True, separators=(",", ":")) |
| 306 | return hashlib.sha256(payload.encode("utf-8")).hexdigest() |
| 307 | |
| 308 | |
| 309 | def schema_checksum() -> str: |
| 310 | """Return a stable checksum for the external-news artifact contract.""" |
| 311 | schema_contract = { |
| 312 | "schema_version": CANONICAL_SCHEMA_VERSION, |
| 313 | "top_level": [ |
| 314 | "schema_version", |
| 315 | "week", |
| 316 | "source", |
| 317 | "crawled_at", |
| 318 | "crawl_window", |
| 319 | "articles", |
| 320 | "metadata", |
| 321 | ], |
| 322 | "metadata": [ |
| 323 | "source_config_checksum", |
| 324 | "schema_checksum", |
| 325 | "sources_requested", |
| 326 | "sources_succeeded", |
| 327 | "sources_failed", |
| 328 | "source_status", |
| 329 | "source_reuse_summary", |
| 330 | "source_artifact_provenance", |
| 331 | "total_articles", |
| 332 | "relevant_articles", |
| 333 | "dedupe_count", |
| 334 | "errors", |
| 335 | "artifact_checksum", |
| 336 | ], |
| 337 | } |
| 338 | payload = json.dumps(schema_contract, sort_keys=True, separators=(",", ":")) |
| 339 | return hashlib.sha256(payload.encode("utf-8")).hexdigest() |
| 340 | |
| 341 | |
| 342 | def source_content_checksum(source_id: str, articles: list[dict[str, Any]]) -> str: |
| 343 | """Return a stable checksum for one source's article payload.""" |
| 344 | source_articles = [ |
| 345 | article |
| 346 | for article in articles |
| 347 | if article.get("source") == source_id or source_id in article.get("sources", []) |
| 348 | ] |
| 349 | payload = json.dumps(source_articles, sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
| 350 | return hashlib.sha256(payload.encode("utf-8")).hexdigest() |
| 351 | |
| 352 | |
| 353 | def parse_iso_datetime(value: Any) -> datetime | None: |
| 354 | if not isinstance(value, str) or not value.strip(): |
| 355 | return None |
| 356 | candidate = value.strip() |
| 357 | if candidate.endswith("Z"): |
| 358 | candidate = f"{candidate[:-1]}+00:00" |
| 359 | try: |
| 360 | parsed = datetime.fromisoformat(candidate) |
| 361 | except ValueError: |
| 362 | return None |
| 363 | return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) |
| 364 | |
| 365 | |
| 366 | def _load_json_object(path: Path) -> dict[str, Any] | None: |
| 367 | try: |
| 368 | payload = json.loads(path.read_text(encoding="utf-8")) |
| 369 | except (OSError, json.JSONDecodeError): |
| 370 | return None |
| 371 | return payload if isinstance(payload, dict) else None |
| 372 | |
| 373 | |
| 374 | def _same_window(payload: dict[str, Any], since: datetime, until: datetime) -> bool: |
| 375 | window = payload.get("crawl_window") |
| 376 | return ( |
| 377 | isinstance(window, dict) |
| 378 | and window.get("since") == iso_timestamp(since) |
| 379 | and window.get("until") == iso_timestamp(until) |
| 380 | ) |
| 381 | |
| 382 | |
| 383 | def same_utc_day(value: str | None, expected: date) -> bool: |
| 384 | parsed = parse_iso_datetime(value) |
| 385 | return parsed is not None and parsed.astimezone(UTC).date() == expected |
| 386 | |
| 387 | |
| 388 | def source_reuse_decisions( |
| 389 | payload: dict[str, Any] | None, |
| 390 | sources: list[NewsSourceConfig], |
| 391 | *, |
| 392 | week: str, |
| 393 | run_date: date, |
| 394 | since: datetime, |
| 395 | until: datetime, |
| 396 | policy: str, |
| 397 | current_config_checksum: str, |
| 398 | current_code_sha: str | None, |
| 399 | ) -> tuple[ |
| 400 | list[dict[str, Any]], list[NewsSourceConfig], list[dict[str, Any]], list[dict[str, Any]] |
| 401 | ]: |
| 402 | """Compatibility planner for callers that pass a loaded artifact.""" |
| 403 | decisions: list[dict[str, Any]] = [] |
| 404 | reused_articles: list[dict[str, Any]] = [] |
| 405 | reused_statuses: list[dict[str, Any]] = [] |
| 406 | to_crawl: list[NewsSourceConfig] = [] |
| 407 | metadata = payload.get("metadata", {}) if isinstance(payload, dict) else {} |
| 408 | statuses = metadata.get("source_status", []) if isinstance(metadata, dict) else [] |
| 409 | if not isinstance(statuses, list): |
| 410 | statuses = [] |
| 411 | status_by_source = { |
| 412 | str(status.get("source")): status for status in statuses if isinstance(status, dict) |
| 413 | } |
| 414 | raw_articles = payload.get("articles", []) if isinstance(payload, dict) else [] |
| 415 | articles: list[dict[str, Any]] = [] |
| 416 | articles_malformed = False |
| 417 | if isinstance(raw_articles, list): |
| 418 | for article in raw_articles: |
| 419 | article_sources = article.get("sources", []) if isinstance(article, dict) else None |
| 420 | if not isinstance(article, dict) or not isinstance(article_sources, list): |
| 421 | articles_malformed = True |
| 422 | break |
| 423 | articles.append(article) |
| 424 | else: |
| 425 | articles_malformed = True |
| 426 | global_reasons: list[str] = [] |
| 427 | if policy == "force-refresh": |
| 428 | global_reasons.append("source_refresh_policy=force-refresh") |
| 429 | if payload is None: |
| 430 | global_reasons.append("artifact missing or malformed") |
| 431 | else: |
| 432 | if payload.get("week") != week: |
| 433 | global_reasons.append(f"week mismatch: expected {week}, found {payload.get('week')!r}") |
| 434 | if not same_utc_day(payload.get("crawled_at"), run_date): |
| 435 | global_reasons.append("artifact is not from the current UTC run date") |
| 436 | window = ( |
| 437 | payload.get("crawl_window") if isinstance(payload.get("crawl_window"), dict) else {} |
| 438 | ) |
| 439 | if window.get("since") != iso_timestamp(since) or window.get("until") != iso_timestamp( |
| 440 | until |
| 441 | ): |
| 442 | global_reasons.append("crawl window mismatch") |
| 443 | if isinstance(metadata, dict): |
| 444 | if metadata.get("source_config_checksum") != current_config_checksum: |
| 445 | global_reasons.append("source config checksum mismatch") |
| 446 | artifact_code_sha = metadata.get("crawler_code_sha") |
| 447 | if current_code_sha and artifact_code_sha != current_code_sha: |
| 448 | global_reasons.append("crawler/config fingerprint mismatch") |
| 449 | if articles_malformed: |
| 450 | global_reasons.append("artifact articles malformed") |
| 451 | for source in sources: |
| 452 | source_reasons = list(global_reasons) |
| 453 | status = status_by_source.get(source.name) |
| 454 | if not status or status.get("success") is not True: |
| 455 | source_reasons.append("source missing or previously failed") |
| 456 | if source_reasons: |
| 457 | decisions.append( |
| 458 | {"source": source.name, "decision": "refresh", "reasons": source_reasons} |
| 459 | ) |
| 460 | to_crawl.append(source) |
| 461 | continue |
| 462 | source_articles = [ |
| 463 | article |
| 464 | for article in articles |
| 465 | if source.name |
| 466 | in {str(article.get("source", "")), *[str(item) for item in article.get("sources", [])]} |
| 467 | ] |
| 468 | reused_articles.extend(source_articles) |
| 469 | reused_status = dict(status) |
| 470 | reused_status["reused_same_day"] = True |
| 471 | reused_status["success"] = True |
| 472 | reused_statuses.append(reused_status) |
| 473 | decisions.append({"source": source.name, "decision": "reuse", "reasons": []}) |
| 474 | return reused_articles, to_crawl, reused_statuses, decisions |
| 475 | |
| 476 | |
| 477 | def plan_source_reuse( |
| 478 | previous_path: Path, |
| 479 | sources: list[NewsSourceConfig], |
| 480 | *, |
| 481 | now: datetime, |
| 482 | since: datetime, |
| 483 | until: datetime, |
| 484 | config_checksum: str, |
| 485 | forced_sources: set[str] | None = None, |
| 486 | source_refresh_policy: str = "reuse-same-day", |
| 487 | run_started_at: datetime | None = None, |
| 488 | current_code_sha: str | None = None, |
| 489 | ) -> tuple[ |
| 490 | list[dict[str, Any]], |
| 491 | list[NewsSourceConfig], |
| 492 | list[dict[str, Any]], |
| 493 | list[dict[str, Any]], |
| 494 | list[dict[str, str]], |
| 495 | ]: |
| 496 | """Load eligible same-day source artifacts and return reused articles plus sources to crawl.""" |
| 497 | forced = forced_sources or set() |
| 498 | run_time = run_started_at or now |
| 499 | requested = {source.name for source in sources} |
| 500 | pending: list[NewsSourceConfig] = [] |
| 501 | reused_articles: list[dict[str, Any]] = [] |
| 502 | summary: list[dict[str, Any]] = [] |
| 503 | provenance: list[dict[str, Any]] = [] |
| 504 | stale_reasons: list[str] = [] |
| 505 | previous = _load_json_object(previous_path) if previous_path.exists() else None |
| 506 | expected_schema_checksum = schema_checksum() |
| 507 | |
| 508 | if source_refresh_policy == "force-refresh": |
| 509 | stale_reasons = ["source_refresh_policy=force-refresh"] |
| 510 | elif previous is None: |
| 511 | stale_reasons = [ |
| 512 | "missing previous artifact" |
| 513 | if not previous_path.exists() |
| 514 | else "previous artifact is not valid JSON" |
| 515 | ] |
| 516 | else: |
| 517 | crawled_at = parse_iso_datetime(previous.get("crawled_at")) |
| 518 | metadata = previous.get("metadata") if isinstance(previous.get("metadata"), dict) else {} |
| 519 | try: |
| 520 | validate_canonical_output(previous) |
| 521 | except ValueError as exc: |
| 522 | stale_reasons.append(str(exc)) |
| 523 | if previous.get("week") != week_slug(now): |
| 524 | stale_reasons.append( |
| 525 | f"week mismatch: expected {week_slug(now)}, found {previous.get('week')!r}" |
| 526 | ) |
| 527 | if ( |
| 528 | crawled_at is None |
| 529 | or crawled_at.astimezone(UTC).date() != run_time.astimezone(UTC).date() |
| 530 | ): |
| 531 | stale_reasons.append("crawled_at is not from the current UTC day") |
| 532 | if not _same_window(previous, since, until): |
| 533 | stale_reasons.append("crawl_window mismatch") |
| 534 | if metadata.get("source_config_checksum") != config_checksum: |
| 535 | stale_reasons.append("source_config_checksum mismatch") |
| 536 | if metadata.get("schema_checksum") != expected_schema_checksum: |
| 537 | stale_reasons.append("schema_checksum mismatch") |
| 538 | artifact_code_sha = metadata.get("crawler_code_sha") |
| 539 | if current_code_sha and artifact_code_sha != current_code_sha: |
| 540 | stale_reasons.append("crawler/config fingerprint mismatch") |
| 541 | |
| 542 | previous_metadata = ( |
| 543 | previous.get("metadata", {}) |
| 544 | if isinstance(previous, dict) and isinstance(previous.get("metadata"), dict) |
| 545 | else {} |
| 546 | ) |
| 547 | previous_statuses = { |
| 548 | str(status.get("source")): status |
| 549 | for status in previous_metadata.get("source_status", []) |
| 550 | if isinstance(status, dict) and status.get("source") in requested |
| 551 | } |
| 552 | previous_articles = [ |
| 553 | article |
| 554 | for article in (previous.get("articles", []) if isinstance(previous, dict) else []) |
| 555 | if isinstance(article, dict) |
| 556 | ] |
| 557 | previous_run_id = str(previous_metadata.get("run_id") or "") |
| 558 | previous_checksum = previous_metadata.get("artifact_checksum") |
| 559 | |
| 560 | for source in sources: |
| 561 | source_id = source.name |
| 562 | status = previous_statuses.get(source_id) |
| 563 | reasons = list(stale_reasons) |
| 564 | action = "missing" |
| 565 | if source_id in forced: |
| 566 | action = "forced" |
| 567 | reasons.append("source explicitly refreshed") |
| 568 | elif stale_reasons: |
| 569 | action = "missing" if stale_reasons == ["missing previous artifact"] else "stale" |
| 570 | elif status is None: |
| 571 | action = "missing" |
| 572 | reasons.append("source missing from previous artifact") |
| 573 | elif not status.get("success"): |
| 574 | action = "failed" |
| 575 | reasons.append("previous source crawl failed") |
| 576 | else: |
| 577 | action = "reused" |
| 578 | |
| 579 | matching_articles = [ |
| 580 | article for article in previous_articles if article.get("source") == source_id |
| 581 | ] |
| 582 | summary.append( |
| 583 | { |
| 584 | "source": source_id, |
| 585 | "action": action, |
| 586 | "reused": action == "reused", |
| 587 | "refreshed": action != "reused", |
| 588 | "reasons": reasons, |
| 589 | } |
| 590 | ) |
| 591 | provenance.append( |
| 592 | { |
| 593 | "source_id": source_id, |
| 594 | "action": action, |
| 595 | "artifact_path": previous_path.as_posix(), |
| 596 | "original_run_id": previous_run_id, |
| 597 | "original_crawled_at": previous.get("crawled_at") |
| 598 | if isinstance(previous, dict) |
| 599 | else None, |
| 600 | "evaluated_at": iso_timestamp(now), |
| 601 | "date": now.astimezone(UTC).date().isoformat(), |
| 602 | "week": week_slug(now), |
| 603 | "crawl_window": {"since": iso_timestamp(since), "until": iso_timestamp(until)}, |
| 604 | "source_config_checksum": config_checksum, |
| 605 | "schema_checksum": expected_schema_checksum, |
| 606 | "artifact_checksum": previous_checksum, |
| 607 | "content_checksum": source_content_checksum(source_id, matching_articles), |
| 608 | "reasons": reasons, |
| 609 | } |
| 610 | ) |
| 611 | if action == "reused": |
| 612 | reused_articles.extend(matching_articles) |
| 613 | else: |
| 614 | pending.append(source) |
| 615 | |
| 616 | return reused_articles, pending, summary, provenance, [] |
| 617 | |
| 618 | |
| 619 | def merge_reuse_results( |
| 620 | initial_summary: list[dict[str, Any]], |
| 621 | initial_provenance: list[dict[str, Any]], |
| 622 | refreshed_statuses: list[dict[str, Any]], |
| 623 | refreshed_articles: list[dict[str, Any]], |
| 624 | *, |
| 625 | now: datetime, |
| 626 | since: datetime, |
| 627 | until: datetime, |
| 628 | config_checksum: str, |
| 629 | output_path: Path, |
| 630 | run_id: str, |
| 631 | ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: |
| 632 | """Update reuse plan entries with the result of sources crawled in this run.""" |
| 633 | summary_by_source = {entry["source"]: dict(entry) for entry in initial_summary} |
| 634 | provenance_by_source = {entry["source_id"]: dict(entry) for entry in initial_provenance} |
| 635 | for status in refreshed_statuses: |
| 636 | source_id = str(status.get("source")) |
| 637 | if not source_id: |
| 638 | continue |
| 639 | action = "refreshed" if status.get("success") else "failed" |
| 640 | reasons = ( |
| 641 | [] if status.get("success") else [status.get("error_message") or "source crawl failed"] |
| 642 | ) |
| 643 | summary_by_source[source_id] = { |
| 644 | "source": source_id, |
| 645 | "action": action, |
| 646 | "reused": False, |
| 647 | "refreshed": True, |
| 648 | "reasons": reasons, |
| 649 | } |
| 650 | provenance_by_source[source_id] = { |
| 651 | "source_id": source_id, |
| 652 | "action": action, |
| 653 | "artifact_path": output_path.as_posix(), |
| 654 | "original_run_id": run_id, |
| 655 | "original_crawled_at": iso_timestamp(now), |
| 656 | "evaluated_at": iso_timestamp(now), |
| 657 | "date": now.astimezone(UTC).date().isoformat(), |
| 658 | "week": week_slug(now), |
| 659 | "crawl_window": {"since": iso_timestamp(since), "until": iso_timestamp(until)}, |
| 660 | "source_config_checksum": config_checksum, |
| 661 | "schema_checksum": schema_checksum(), |
| 662 | "artifact_checksum": None, |
| 663 | "content_checksum": source_content_checksum(source_id, refreshed_articles), |
| 664 | "reasons": reasons, |
| 665 | } |
| 666 | return ( |
| 667 | [summary_by_source[source] for source in sorted(summary_by_source)], |
| 668 | [provenance_by_source[source] for source in sorted(provenance_by_source)], |
| 669 | ) |
| 670 | |
| 671 | |
| 672 | def iso_timestamp(value: datetime) -> str: |
| 673 | """Format datetime as ISO 8601 UTC string.""" |
| 674 | return value.astimezone(UTC).isoformat().replace("+00:00", "Z") |
| 675 | |
| 676 | |
| 677 | def week_slug(value: datetime) -> str: |
| 678 | """Return ISO week string like '2026-W21'.""" |
| 679 | year, week, _ = value.isocalendar() |
| 680 | return f"{year}-W{week:02d}" |
| 681 | |
| 682 | |
| 683 | def extract_github_urls(text: str) -> list[str]: |
| 684 | """Extract unique GitHub repository URLs from text.""" |
| 685 | if not text: |
| 686 | return [] |
| 687 | urls = GITHUB_URL_RE.findall(text) |
| 688 | # Deduplicate while preserving order |
| 689 | seen: set[str] = set() |
| 690 | result: list[str] = [] |
| 691 | for url in urls: |
| 692 | # Strip trailing periods/commas that may be captured |
| 693 | url = url.rstrip(".,;)") |
| 694 | if url not in seen: |
| 695 | seen.add(url) |
| 696 | result.append(url) |
| 697 | return result |
| 698 | |
| 699 | |
| 700 | def extract_entities(title: str) -> list[str]: |
| 701 | """Extract likely entity names (companies, projects) from a title. |
| 702 | |
| 703 | Heuristic: capitalized words that aren't common English words. |
| 704 | """ |
| 705 | if not title: |
| 706 | return [] |
| 707 | words = re.findall(r"\b[A-Z][a-zA-Z0-9]*(?:\.[a-zA-Z]+)*\b", title) |
| 708 | entities: list[str] = [] |
| 709 | seen: set[str] = set() |
| 710 | for word in words: |
| 711 | lower = word.lower() |
| 712 | if lower in STOP_WORDS: |
| 713 | continue |
| 714 | if len(word) < 2: |
| 715 | continue |
| 716 | if word not in seen: |
| 717 | seen.add(word) |
| 718 | entities.append(word) |
| 719 | return entities |
| 720 | |
| 721 | |
| 722 | def compute_relevance_score(article: dict[str, Any]) -> float: |
| 723 | """Compute a 0-1 relevance score based on tech/OSS keyword density.""" |
| 724 | text = " ".join( |
| 725 | [ |
| 726 | article.get("title", ""), |
| 727 | article.get("summary", ""), |
| 728 | " ".join(article.get("categories", [])), |
| 729 | ] |
| 730 | ).lower() |
| 731 | |
| 732 | if not text.strip(): |
| 733 | return 0.0 |
| 734 | |
| 735 | matches = sum(1 for kw in TECH_KEYWORDS if kw in text) |
| 736 | # Normalize: cap at 1.0, scale so 5+ keywords = 1.0 |
| 737 | score = min(matches / 5.0, 1.0) |
| 738 | # Boost if GitHub links found |
| 739 | if article.get("github_links"): |
| 740 | score = min(score + 0.2, 1.0) |
| 741 | return round(score, 2) |
| 742 | |
| 743 | |
| 744 | def parse_published_date(entry: Any) -> datetime | None: |
| 745 | """Parse the published date from a feedparser entry.""" |
| 746 | published_parsed = getattr(entry, "published_parsed", None) |
| 747 | if published_parsed: |
| 748 | return datetime(*published_parsed[:6], tzinfo=UTC) |
| 749 | # Fallback: try updated_parsed |
| 750 | updated_parsed = getattr(entry, "updated_parsed", None) |
| 751 | if updated_parsed: |
| 752 | return datetime(*updated_parsed[:6], tzinfo=UTC) |
| 753 | return None |
| 754 | |
| 755 | |
| 756 | def _sleep_before_retry(attempt: int, retry_after: float | None = None) -> float: |
| 757 | """Sleep with exponential backoff + jitter before the next fetch attempt. |
| 758 | |
| 759 | Server-supplied Retry-After is honored up to RETRY_AFTER_MAX_SECONDS, while |
| 760 | computed backoff is bounded by RETRY_MAX_DELAY_SECONDS. Returns the delay |
| 761 | slept so callers/tests can reason about it. |
| 762 | """ |
| 763 | if retry_after and retry_after > 0: |
| 764 | delay = min(retry_after, RETRY_AFTER_MAX_SECONDS) |
| 765 | else: |
| 766 | base_delay = min( |
| 767 | RETRY_BASE_DELAY_SECONDS * (2**attempt), |
| 768 | RETRY_MAX_DELAY_SECONDS, |
| 769 | ) |
| 770 | delay = min( |
| 771 | base_delay + _JITTER_RANDOM.uniform(0.3, 1.7), |
| 772 | RETRY_MAX_DELAY_SECONDS, |
| 773 | ) |
| 774 | time.sleep(delay) |
| 775 | return delay |
| 776 | |
| 777 | |
| 778 | def _retry_after_seconds(exc: HTTPError) -> float | None: |
| 779 | """Extract a positive Retry-After delay from delta-seconds or HTTP-date.""" |
| 780 | header = None |
| 781 | try: |
| 782 | header = exc.headers.get("Retry-After") if exc.headers else None |
| 783 | except AttributeError: |
| 784 | header = None |
| 785 | if not header: |
| 786 | return None |
| 787 | try: |
| 788 | value = float(header) |
| 789 | if math.isfinite(value): |
| 790 | return max(value, 1.0) |
| 791 | except (TypeError, ValueError): |
| 792 | pass |
| 793 | try: |
| 794 | retry_at = parsedate_to_datetime(header) |
| 795 | except (TypeError, ValueError): |
| 796 | return None |
| 797 | if retry_at.tzinfo is None: |
| 798 | retry_at = retry_at.replace(tzinfo=timezone.utc) |
| 799 | delay = (retry_at - datetime.now(timezone.utc)).total_seconds() |
| 800 | return max(delay, 1.0) |
| 801 | |
| 802 | |
| 803 | def fetch_feed( |
| 804 | url: str = FEED_URL, |
| 805 | retries: int = DEFAULT_FETCH_RETRIES, |
| 806 | timeout: int = DEFAULT_FETCH_TIMEOUT_SECONDS, |
| 807 | ) -> Any: |
| 808 | """Fetch and parse an RSS feed with bounded retries and exponential backoff. |
| 809 | |
| 810 | Transient failures (network errors, retryable HTTP statuses, empty/bozo |
| 811 | feeds) are retried with exponential backoff + jitter. Permanent HTTP errors |
| 812 | (e.g. 404) fail fast. Feed bytes are always treated as UNTRUSTED content and |
| 813 | are only parsed, never executed. |
| 814 | """ |
| 815 | validate_feed_url(url) |
| 816 | if timeout <= 0: |
| 817 | raise ValueError("RSS fetch timeout must be greater than zero") |
| 818 | feed: Any = None |
| 819 | for attempt in range(retries + 1): |
| 820 | try: |
| 821 | request = Request(url, headers={"User-Agent": "SquadScope RSS crawler"}) |
| 822 | with urlopen(request, timeout=timeout) as response: # nosec B310 |
| 823 | feed = feedparser.parse(response.read()) |
| 824 | setattr(feed, "squad_fetch_attempts", attempt + 1) |
| 825 | setattr(feed, "squad_fetch_timeout_seconds", timeout) |
| 826 | except HTTPError as exc: |
| 827 | # Only retry transient HTTP statuses; permanent errors fail fast. |
| 828 | if exc.code in RETRYABLE_STATUSES and attempt < retries: |
| 829 | _sleep_before_retry(attempt, _retry_after_seconds(exc)) |
| 830 | continue |
| 831 | raise |
| 832 | except (URLError, TimeoutError, OSError): |
| 833 | if attempt < retries: |
| 834 | _sleep_before_retry(attempt) |
| 835 | continue |
| 836 | raise |
| 837 | if feed.bozo and not feed.entries: |
| 838 | if attempt < retries: |
| 839 | _sleep_before_retry(attempt) |
| 840 | continue |
| 841 | # Return partial result even on failure |
| 842 | setattr(feed, "squad_fetch_attempts", attempt + 1) |
| 843 | setattr(feed, "squad_fetch_timeout_seconds", timeout) |
| 844 | return feed |
| 845 | return feed |
| 846 | return feed # pragma: no cover |
| 847 | |
| 848 | |
| 849 | class NewsFeedSource: |
| 850 | """RSS data source following the DataSource protocol.""" |
| 851 | |
| 852 | def __init__(self, config: NewsSourceConfig) -> None: |
| 853 | self.config = config |
| 854 | self.last_attempts = 0 |
| 855 | self.last_timeout_seconds = DEFAULT_FETCH_TIMEOUT_SECONDS |
| 856 | |
| 857 | def get_name(self) -> str: |
| 858 | return self.config.name |
| 859 | |
| 860 | def get_rate_limits(self) -> dict: |
| 861 | return {"requests_per_minute": self.config.requests_per_minute} |
| 862 | |
| 863 | def crawl( |
| 864 | self, |
| 865 | since: datetime, |
| 866 | until: datetime, |
| 867 | feed_url: str | None = None, |
| 868 | ) -> list[dict[str, Any]]: |
| 869 | """Crawl an RSS feed and return structured articles.""" |
| 870 | resolved_feed_url = feed_url or self.config.feed_url |
| 871 | validate_feed_url(resolved_feed_url) |
| 872 | try: |
| 873 | feed = fetch_feed(resolved_feed_url) |
| 874 | except Exception: |
| 875 | self.last_attempts = DEFAULT_FETCH_RETRIES + 1 |
| 876 | self.last_timeout_seconds = DEFAULT_FETCH_TIMEOUT_SECONDS |
| 877 | raise |
| 878 | self.last_attempts = int(getattr(feed, "squad_fetch_attempts", 1)) |
| 879 | self.last_timeout_seconds = int( |
| 880 | getattr(feed, "squad_fetch_timeout_seconds", DEFAULT_FETCH_TIMEOUT_SECONDS) |
| 881 | ) |
| 882 | articles: list[dict[str, Any]] = [] |
| 883 | |
| 884 | for entry in feed.entries: |
| 885 | pub_date = parse_published_date(entry) |
| 886 | if pub_date is None: |
| 887 | continue |
| 888 | if pub_date < since or pub_date >= until: |
| 889 | continue |
| 890 | |
| 891 | # Get content for GitHub URL extraction |
| 892 | content_text = "" |
| 893 | if hasattr(entry, "content") and entry.content: |
| 894 | content_text = entry.content[0].get("value", "") |
| 895 | elif hasattr(entry, "summary"): |
| 896 | content_text = entry.summary or "" |
| 897 | |
| 898 | categories = [tag.term for tag in getattr(entry, "tags", []) if hasattr(tag, "term")] |
| 899 | |
| 900 | summary = getattr(entry, "summary", "") or "" |
| 901 | # Strip HTML tags from summary |
| 902 | summary = re.sub(r"<[^>]+>", "", summary).strip() |
| 903 | if len(summary) > 500: |
| 904 | summary = summary[:497] + "..." |
| 905 | |
| 906 | article: dict[str, Any] = { |
| 907 | "source": self.get_name(), |
| 908 | "title": getattr(entry, "title", ""), |
| 909 | "url": getattr(entry, "link", ""), |
| 910 | "published_at": iso_timestamp(pub_date), |
| 911 | "categories": categories, |
| 912 | "summary": summary, |
| 913 | "github_links": extract_github_urls(content_text), |
| 914 | "entities": extract_entities(getattr(entry, "title", "")), |
| 915 | } |
| 916 | article["relevance_score"] = compute_relevance_score(article) |
| 917 | articles.append(article) |
| 918 | |
| 919 | return articles |
| 920 | |
| 921 | |
| 922 | class TechCrunchSource(NewsFeedSource): |
| 923 | """Backward-compatible TechCrunch RSS source.""" |
| 924 | |
| 925 | def __init__(self) -> None: |
| 926 | super().__init__(NewsSourceConfig("techcrunch", FEED_URL, 10)) |
| 927 | |
| 928 | |
| 929 | def crawl_sources_parallel( |
| 930 | sources: list[NewsSourceConfig], |
| 931 | since: datetime, |
| 932 | until: datetime, |
| 933 | max_workers: int | None = None, |
| 934 | ) -> tuple[list[dict[str, Any]], list[dict[str, str]], list[dict[str, Any]]]: |
| 935 | """Crawl configured RSS sources concurrently and return articles, errors, statuses.""" |
| 936 | if not sources: |
| 937 | return [], [], [] |
| 938 | |
| 939 | if max_workers is not None and max_workers < 1: |
| 940 | raise ValueError("--max-workers must be at least 1") |
| 941 | workers = min(max_workers or len(sources), len(sources), DEFAULT_MAX_WORKERS) |
| 942 | articles: list[dict[str, Any]] = [] |
| 943 | errors: list[dict[str, str]] = [] |
| 944 | statuses: list[dict[str, Any]] = [] |
| 945 | |
| 946 | def crawl_one( |
| 947 | source: NewsSourceConfig, |
| 948 | ) -> tuple[list[dict[str, Any]], dict[str, Any], dict[str, str] | None]: |
| 949 | started = datetime.now(UTC) |
| 950 | source_client = NewsFeedSource(source) |
| 951 | status: dict[str, Any] = { |
| 952 | "source": source.name, |
| 953 | "host": source.host, |
| 954 | "started_at": iso_timestamp(started), |
| 955 | "timeout_seconds": DEFAULT_FETCH_TIMEOUT_SECONDS, |
| 956 | "attempts": 0, |
| 957 | "total_articles": 0, |
| 958 | "relevant_articles": 0, |
| 959 | "github_links_found": 0, |
| 960 | "success": False, |
| 961 | "error_class": "", |
| 962 | "error_message": "", |
| 963 | } |
| 964 | try: |
| 965 | source_articles = source_client.crawl(since, until) |
| 966 | status["attempts"] = source_client.last_attempts or 1 |
| 967 | status["timeout_seconds"] = source_client.last_timeout_seconds |
| 968 | status["total_articles"] = len(source_articles) |
| 969 | status["relevant_articles"] = sum( |
| 970 | 1 for article in source_articles if article.get("relevance_score", 0) >= 0.4 |
| 971 | ) |
| 972 | github_links: set[str] = set() |
| 973 | for article in source_articles: |
| 974 | github_links.update(article.get("github_links", [])) |
| 975 | status["github_links_found"] = len(github_links) |
| 976 | status["success"] = True |
| 977 | return source_articles, status, None |
| 978 | except Exception as exc: # pragma: no cover - defensive around network/parser failures |
| 979 | status["attempts"] = source_client.last_attempts or (DEFAULT_FETCH_RETRIES + 1) |
| 980 | status["error_class"] = exc.__class__.__name__ |
| 981 | status["error_message"] = str(exc) |
| 982 | error = { |
| 983 | "source": source.name, |
| 984 | "error_class": exc.__class__.__name__, |
| 985 | "error": str(exc), |
| 986 | } |
| 987 | return [], status, error |
| 988 | finally: |
| 989 | ended = datetime.now(UTC) |
| 990 | status["ended_at"] = iso_timestamp(ended) |
| 991 | status["duration_seconds"] = round((ended - started).total_seconds(), 3) |
| 992 | |
| 993 | with ThreadPoolExecutor(max_workers=workers) as executor: |
| 994 | futures = {executor.submit(crawl_one, source): source for source in sources} |
| 995 | for future in as_completed(futures): |
| 996 | source_articles, status, error = future.result() |
| 997 | articles.extend(source_articles) |
| 998 | statuses.append(status) |
| 999 | if error: |
| 1000 | errors.append(error) |
| 1001 | |
| 1002 | articles.sort( |
| 1003 | key=lambda article: ( |
| 1004 | article.get("published_at", ""), |
| 1005 | article.get("source", ""), |
| 1006 | article.get("url", ""), |
| 1007 | article.get("title", ""), |
| 1008 | ), |
| 1009 | reverse=True, |
| 1010 | ) |
| 1011 | statuses.sort(key=lambda status: status["source"]) |
| 1012 | errors.sort(key=lambda error: error["source"]) |
| 1013 | for status in statuses: |
| 1014 | state = "ok" if status["success"] else f"failed:{status['error_class']}" |
| 1015 | print( |
| 1016 | "[external-news] " |
| 1017 | f"{status['source']} host={status['host']} status={state} " |
| 1018 | f"duration={status['duration_seconds']:.3f}s attempts={status['attempts']} " |
| 1019 | f"articles={status['total_articles']} relevant={status['relevant_articles']} " |
| 1020 | f"github_links={status['github_links_found']}", |
| 1021 | file=sys.stderr, |
| 1022 | ) |
| 1023 | return articles, errors, statuses |
| 1024 | |
| 1025 | |
| 1026 | def _normalized_article_url(url: str) -> str: |
| 1027 | """Normalize a URL for cross-source dedupe.""" |
| 1028 | if not url: |
| 1029 | return "" |
| 1030 | parsed = urlparse(url.strip()) |
| 1031 | host = (parsed.netloc or "").lower() |
| 1032 | path = parsed.path.rstrip("/") |
| 1033 | return f"{parsed.scheme.lower()}://{host}{path}" |
| 1034 | |
| 1035 | |
| 1036 | def dedupe_articles(articles: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], int]: |
| 1037 | """Deduplicate mirrored articles by URL while preserving provenance.""" |
| 1038 | grouped: dict[str, dict[str, Any]] = {} |
| 1039 | duplicates = 0 |
| 1040 | for article in sorted( |
| 1041 | articles, |
| 1042 | key=lambda item: ( |
| 1043 | item.get("published_at", ""), |
| 1044 | item.get("source", ""), |
| 1045 | item.get("url", ""), |
| 1046 | item.get("title", ""), |
| 1047 | ), |
| 1048 | reverse=True, |
| 1049 | ): |
| 1050 | key = _normalized_article_url(str(article.get("url", ""))) |
| 1051 | if not key: |
| 1052 | key = "|".join( |
| 1053 | [ |
| 1054 | str(article.get("source", "")), |
| 1055 | str(article.get("published_at", "")), |
| 1056 | str(article.get("title", "")).lower(), |
| 1057 | ] |
| 1058 | ) |
| 1059 | if key not in grouped: |
| 1060 | current = dict(article) |
| 1061 | current["sources"] = sorted( |
| 1062 | { |
| 1063 | str(article.get("source", "")) or "unknown", |
| 1064 | *[str(s) for s in article.get("sources", [])], |
| 1065 | } |
| 1066 | ) |
| 1067 | grouped[key] = current |
| 1068 | continue |
| 1069 | duplicates += 1 |
| 1070 | existing = grouped[key] |
| 1071 | existing_sources = set(existing.get("sources", [])) |
| 1072 | existing_sources.add(str(article.get("source", "")) or "unknown") |
| 1073 | existing_sources.update(str(s) for s in article.get("sources", [])) |
| 1074 | existing["sources"] = sorted(existing_sources) |
| 1075 | existing["relevance_score"] = max( |
| 1076 | float(existing.get("relevance_score", 0)), |
| 1077 | float(article.get("relevance_score", 0)), |
| 1078 | ) |
| 1079 | existing_links = list(existing.get("github_links", [])) |
| 1080 | for link in article.get("github_links", []): |
| 1081 | if link not in existing_links: |
| 1082 | existing_links.append(link) |
| 1083 | existing["github_links"] = existing_links |
| 1084 | deduped = list(grouped.values()) |
| 1085 | deduped.sort( |
| 1086 | key=lambda article: ( |
| 1087 | article.get("published_at", ""), |
| 1088 | article.get("source", ""), |
| 1089 | article.get("url", ""), |
| 1090 | article.get("title", ""), |
| 1091 | ), |
| 1092 | reverse=True, |
| 1093 | ) |
| 1094 | return deduped, duplicates |
| 1095 | |
| 1096 | |
| 1097 | def _checksum_payload(output: dict[str, Any]) -> dict[str, Any]: |
| 1098 | """Return the deterministic subset covered by artifact_checksum.""" |
| 1099 | metadata = dict(output.get("metadata", {})) |
| 1100 | metadata.pop("artifact_checksum", None) |
| 1101 | metadata.pop("run_id", None) |
| 1102 | metadata.pop("source_reuse_summary", None) |
| 1103 | metadata.pop("source_artifact_provenance", None) |
| 1104 | payload = dict(output) |
| 1105 | payload["metadata"] = metadata |
| 1106 | payload.pop("crawled_at", None) |
| 1107 | return payload |
| 1108 | |
| 1109 | |
| 1110 | def artifact_checksum(output: dict[str, Any]) -> str: |
| 1111 | """Return a stable checksum for the canonical artifact content.""" |
| 1112 | payload = json.dumps( |
| 1113 | _checksum_payload(output), |
| 1114 | sort_keys=True, |
| 1115 | separators=(",", ":"), |
| 1116 | ensure_ascii=False, |
| 1117 | ) |
| 1118 | return hashlib.sha256(payload.encode("utf-8")).hexdigest() |
| 1119 | |
| 1120 | |
| 1121 | def build_output( |
| 1122 | articles: list[dict[str, Any]], |
| 1123 | crawled_at: datetime, |
| 1124 | *, |
| 1125 | source: str = "techcrunch", |
| 1126 | source_count: int = 1, |
| 1127 | crawl_window: dict[str, str] | None = None, |
| 1128 | source_config_checksum_value: str | None = None, |
| 1129 | requested_sources: list[str] | None = None, |
| 1130 | source_statuses: list[dict[str, Any]] | None = None, |
| 1131 | errors: list[dict[str, str]] | None = None, |
| 1132 | source_reuse_summary: list[dict[str, Any]] | None = None, |
| 1133 | source_artifact_provenance: list[dict[str, Any]] | None = None, |
| 1134 | run_id: str | None = None, |
| 1135 | ) -> dict[str, Any]: |
| 1136 | """Build the final output structure with metadata.""" |
| 1137 | articles, dedupe_count = dedupe_articles(articles) |
| 1138 | relevant = [a for a in articles if a["relevance_score"] >= 0.4] |
| 1139 | all_github_links = set() |
| 1140 | for a in articles: |
| 1141 | all_github_links.update(a.get("github_links", [])) |
| 1142 | |
| 1143 | statuses = source_statuses or [] |
| 1144 | succeeded = [str(status.get("source")) for status in statuses if status.get("success")] |
| 1145 | failed = [str(status.get("source")) for status in statuses if not status.get("success")] |
| 1146 | requested = ( |
| 1147 | requested_sources |
| 1148 | or sorted({str(article.get("source", source)) for article in articles}) |
| 1149 | or [source] |
| 1150 | ) |
| 1151 | by_source = Counter(str(article.get("source", source)) for article in articles) |
| 1152 | output = { |
| 1153 | "schema_version": CANONICAL_SCHEMA_VERSION, |
| 1154 | "week": week_slug(crawled_at), |
| 1155 | "source": source, |
| 1156 | "crawled_at": iso_timestamp(crawled_at), |
| 1157 | "crawl_window": crawl_window or {}, |
| 1158 | "articles": articles, |
| 1159 | "metadata": { |
| 1160 | "run_id": run_id or "local", |
| 1161 | "source_count": source_count, |
| 1162 | "source_config_checksum": source_config_checksum_value or "", |
| 1163 | "schema_checksum": schema_checksum(), |
| 1164 | "sources_requested": sorted(requested), |
| 1165 | "sources_succeeded": sorted(succeeded or requested), |
| 1166 | "sources_failed": sorted(failed), |
| 1167 | "source_status": sorted(statuses, key=lambda status: status["source"]), |
| 1168 | "source_reuse_summary": sorted( |
| 1169 | source_reuse_summary or [], key=lambda item: item["source"] |
| 1170 | ), |
| 1171 | "source_artifact_provenance": sorted( |
| 1172 | source_artifact_provenance or [], key=lambda item: item["source_id"] |
| 1173 | ), |
| 1174 | "sources_with_articles": dict(sorted(by_source.items())), |
| 1175 | "total_articles": len(articles), |
| 1176 | "relevant_articles": len(relevant), |
| 1177 | "github_links_found": len(all_github_links), |
| 1178 | "dedupe_count": dedupe_count, |
| 1179 | "errors": errors or [], |
| 1180 | }, |
| 1181 | } |
| 1182 | output["metadata"]["artifact_checksum"] = artifact_checksum(output) |
| 1183 | for entry in output["metadata"]["source_artifact_provenance"]: |
| 1184 | if not entry.get("artifact_checksum"): |
| 1185 | entry["artifact_checksum"] = output["metadata"]["artifact_checksum"] |
| 1186 | validate_canonical_output(output) |
| 1187 | return output |
| 1188 | |
| 1189 | |
| 1190 | def validate_canonical_output(output: dict[str, Any]) -> None: |
| 1191 | """Validate canonical external-news artifact shape.""" |
| 1192 | if output.get("schema_version") != CANONICAL_SCHEMA_VERSION: |
| 1193 | raise ValueError("External news artifact has unsupported schema_version") |
| 1194 | if output.get("source") == "external_news" and not output.get("crawl_window"): |
| 1195 | raise ValueError("Canonical external news artifact requires crawl_window") |
| 1196 | metadata = output.get("metadata") |
| 1197 | if not isinstance(metadata, dict): |
| 1198 | raise ValueError("External news artifact requires metadata") |
| 1199 | required = { |
| 1200 | "source_config_checksum", |
| 1201 | "schema_checksum", |
| 1202 | "sources_requested", |
| 1203 | "sources_succeeded", |
| 1204 | "sources_failed", |
| 1205 | "source_status", |
| 1206 | "source_reuse_summary", |
| 1207 | "source_artifact_provenance", |
| 1208 | "total_articles", |
| 1209 | "relevant_articles", |
| 1210 | "dedupe_count", |
| 1211 | "errors", |
| 1212 | "artifact_checksum", |
| 1213 | } |
| 1214 | missing = sorted(required - set(metadata)) |
| 1215 | if missing: |
| 1216 | raise ValueError(f"External news artifact missing metadata keys: {missing}") |
| 1217 | expected_checksum = artifact_checksum(output) |
| 1218 | if metadata.get("artifact_checksum") != expected_checksum: |
| 1219 | raise ValueError("External news artifact checksum mismatch") |
| 1220 | if metadata.get("schema_checksum") != schema_checksum(): |
| 1221 | raise ValueError("External news artifact schema checksum mismatch") |
| 1222 | |
| 1223 | |
| 1224 | def main(argv: list[str] | None = None) -> int: |
| 1225 | crawl_started = time.monotonic() |
| 1226 | parser = argparse.ArgumentParser(description="Crawl external news RSS feeds for SquadScope") |
| 1227 | parser.add_argument( |
| 1228 | "--topic", |
| 1229 | default="general", |
| 1230 | help="Topic ID for output path (default: general)", |
| 1231 | ) |
| 1232 | parser.add_argument( |
| 1233 | "--output", |
| 1234 | default=None, |
| 1235 | help="Override output file path", |
| 1236 | ) |
| 1237 | parser.add_argument( |
| 1238 | "--since", |
| 1239 | default=None, |
| 1240 | help="Start date filter (YYYY-MM-DD, default: 7 days ago)", |
| 1241 | ) |
| 1242 | parser.add_argument( |
| 1243 | "--until", |
| 1244 | default=None, |
| 1245 | help="End date filter (YYYY-MM-DD, default: now)", |
| 1246 | ) |
| 1247 | parser.add_argument( |
| 1248 | "--sources", |
| 1249 | default=str(DEFAULT_SOURCES_PATH), |
| 1250 | help="Path to external RSS source config JSON", |
| 1251 | ) |
| 1252 | parser.add_argument( |
| 1253 | "--max-workers", |
| 1254 | type=int, |
| 1255 | default=None, |
| 1256 | help="Maximum parallel RSS fetches (default: one per source, capped at 8)", |
| 1257 | ) |
| 1258 | parser.add_argument( |
| 1259 | "--force-refresh", |
| 1260 | action="store_true", |
| 1261 | help="Refresh all sources even when same-day artifacts are reusable.", |
| 1262 | ) |
| 1263 | parser.add_argument( |
| 1264 | "--force-refresh-source", |
| 1265 | action="append", |
| 1266 | default=[], |
| 1267 | help="Refresh one source by id even when its same-day artifact is reusable. Can be repeated.", |
| 1268 | ) |
| 1269 | parser.add_argument( |
| 1270 | "--reuse-artifact", |
| 1271 | default=None, |
| 1272 | help="Existing external-news artifact to reuse per source when fresh for this run window.", |
| 1273 | ) |
| 1274 | parser.add_argument( |
| 1275 | "--source-refresh-policy", |
| 1276 | choices=["reuse-same-day", "refresh-missing-stale", "force-refresh"], |
| 1277 | default="reuse-same-day", |
| 1278 | help="Source refresh policy for reruns (default: reuse eligible same-day sources).", |
| 1279 | ) |
| 1280 | parser.add_argument( |
| 1281 | "--run-started-at", |
| 1282 | default=None, |
| 1283 | help="UTC run start timestamp used for same-day reuse checks (ISO 8601). Defaults to now.", |
| 1284 | ) |
| 1285 | parser.add_argument( |
| 1286 | "--current-code-sha", |
| 1287 | default=None, |
| 1288 | help="Optional crawler/config fingerprint; reused artifacts with a conflicting fingerprint are stale.", |
| 1289 | ) |
| 1290 | args = parser.parse_args(argv) |
| 1291 | |
| 1292 | now = datetime.now(UTC) |
| 1293 | run_started_at = parse_iso_datetime(args.run_started_at) if args.run_started_at else now |
| 1294 | if run_started_at is None: |
| 1295 | print("--run-started-at must be an ISO 8601 timestamp", file=sys.stderr) |
| 1296 | return 1 |
| 1297 | since = ( |
| 1298 | datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=UTC) |
| 1299 | if args.since |
| 1300 | else now - timedelta(days=7) |
| 1301 | ) |
| 1302 | until = datetime.strptime(args.until, "%Y-%m-%d").replace(tzinfo=UTC) if args.until else now |
| 1303 | |
| 1304 | source_configs = load_source_configs(Path(args.sources)) |
| 1305 | if args.output: |
| 1306 | out_path = Path(args.output) |
| 1307 | else: |
| 1308 | out_dir = raw_dir(args.topic) |
| 1309 | out_dir.mkdir(parents=True, exist_ok=True) |
| 1310 | out_path = out_dir / f"{week_slug(now)}-external-news.json" |
| 1311 | |
| 1312 | config_checksum = source_config_checksum(source_configs) |
| 1313 | source_refresh_policy = "force-refresh" if args.force_refresh else args.source_refresh_policy |
| 1314 | current_code_sha = args.current_code_sha or "" |
| 1315 | force_sources = ( |
| 1316 | {source.name for source in source_configs} |
| 1317 | if source_refresh_policy == "force-refresh" |
| 1318 | else set(args.force_refresh_source or []) |
| 1319 | ) |
| 1320 | reuse_path = Path(args.reuse_artifact) if args.reuse_artifact else out_path |
| 1321 | reused_articles, sources_to_crawl, reuse_summary, provenance, _ = plan_source_reuse( |
| 1322 | reuse_path, |
| 1323 | source_configs, |
| 1324 | now=now, |
| 1325 | since=since, |
| 1326 | until=until, |
| 1327 | config_checksum=config_checksum, |
| 1328 | forced_sources=force_sources, |
| 1329 | source_refresh_policy=source_refresh_policy, |
| 1330 | run_started_at=run_started_at, |
| 1331 | current_code_sha=current_code_sha, |
| 1332 | ) |
| 1333 | refreshed_articles, errors, refreshed_statuses = crawl_sources_parallel( |
| 1334 | sources_to_crawl, since=since, until=until, max_workers=args.max_workers |
| 1335 | ) |
| 1336 | articles = [*reused_articles, *refreshed_articles] |
| 1337 | reuse_summary, provenance = merge_reuse_results( |
| 1338 | reuse_summary, |
| 1339 | provenance, |
| 1340 | refreshed_statuses, |
| 1341 | refreshed_articles, |
| 1342 | now=now, |
| 1343 | since=since, |
| 1344 | until=until, |
| 1345 | config_checksum=config_checksum, |
| 1346 | output_path=out_path, |
| 1347 | run_id=os.environ.get("GITHUB_RUN_ID", "local"), |
| 1348 | ) |
| 1349 | reused_sources = {entry["source_id"] for entry in provenance if entry.get("action") == "reused"} |
| 1350 | previous = _load_json_object(reuse_path) if reuse_path.exists() else None |
| 1351 | previous_metadata = ( |
| 1352 | previous.get("metadata", {}) |
| 1353 | if isinstance(previous, dict) and isinstance(previous.get("metadata"), dict) |
| 1354 | else {} |
| 1355 | ) |
| 1356 | previous_statuses = [ |
| 1357 | {**status, "reused": True} |
| 1358 | for status in previous_metadata.get("source_status", []) |
| 1359 | if isinstance(status, dict) and status.get("source") in reused_sources |
| 1360 | ] |
| 1361 | statuses = [*previous_statuses, *refreshed_statuses] |
| 1362 | output = build_output( |
| 1363 | articles, |
| 1364 | crawled_at=now, |
| 1365 | source="external_news", |
| 1366 | source_count=len(source_configs), |
| 1367 | crawl_window={ |
| 1368 | "since": iso_timestamp(since), |
| 1369 | "until": iso_timestamp(until), |
| 1370 | }, |
| 1371 | source_config_checksum_value=config_checksum, |
| 1372 | requested_sources=[source.name for source in source_configs], |
| 1373 | source_statuses=statuses, |
| 1374 | errors=errors, |
| 1375 | source_reuse_summary=reuse_summary, |
| 1376 | source_artifact_provenance=provenance, |
| 1377 | run_id=os.environ.get("GITHUB_RUN_ID", "local"), |
| 1378 | ) |
| 1379 | output["metadata"]["same_day_reuse"] = ( |
| 1380 | "mixed" |
| 1381 | if reused_articles and refreshed_articles |
| 1382 | else "reused" |
| 1383 | if reused_articles |
| 1384 | else "not_reused" |
| 1385 | ) |
| 1386 | output["metadata"]["source_refresh_policy"] = source_refresh_policy |
| 1387 | output["metadata"]["source_reuse_decisions"] = [ |
| 1388 | { |
| 1389 | "source": item["source"], |
| 1390 | "decision": "reuse" if item["action"] == "reused" else "refresh", |
| 1391 | "reasons": item["reasons"], |
| 1392 | } |
| 1393 | for item in output["metadata"]["source_reuse_summary"] |
| 1394 | ] |
| 1395 | output["metadata"]["crawler_code_sha"] = current_code_sha |
| 1396 | output["metadata"]["artifact_checksum"] = artifact_checksum(output) |
| 1397 | validate_canonical_output(output) |
| 1398 | |
| 1399 | out_path.parent.mkdir(parents=True, exist_ok=True) |
| 1400 | with open(out_path, "w", encoding="utf-8") as f: |
| 1401 | json.dump(output, f, indent=2, ensure_ascii=False) |
| 1402 | |
| 1403 | total_duration_seconds = round(time.monotonic() - crawl_started, 3) |
| 1404 | sampled_durations = [ |
| 1405 | float(status["duration_seconds"]) |
| 1406 | for status in statuses |
| 1407 | if isinstance(status.get("duration_seconds"), (int, float)) |
| 1408 | ] |
| 1409 | if sampled_durations: |
| 1410 | sampled_p95 = duration_p95(sampled_durations) |
| 1411 | sample_count = len(sampled_durations) |
| 1412 | else: |
| 1413 | # Treat overall runtime as a single sample for consistency |
| 1414 | sampled_p95 = total_duration_seconds |
| 1415 | sample_count = 1 |
| 1416 | observability_path = DEFAULT_OBSERVABILITY_DIR / f"{output['week']}-external-news-crawl.json" |
| 1417 | emit_ledger( |
| 1418 | ObservabilityLedger( |
| 1419 | schema_version=METRICS_SCHEMA_VERSION, |
| 1420 | run_id=os.environ.get("GITHUB_RUN_ID", "local"), |
| 1421 | week=output["week"], |
| 1422 | timestamp=output["crawled_at"], |
| 1423 | crawl_metrics=[ |
| 1424 | CrawlMetrics( |
| 1425 | duration_seconds=sampled_p95, |
| 1426 | duration_p95_seconds=sampled_p95, |
| 1427 | duration_sample_count=sample_count, |
| 1428 | api_calls=sum(int(status.get("attempts") or 0) for status in statuses), |
| 1429 | cache_hits=0, |
| 1430 | cache_misses=sum(int(status.get("attempts") or 0) for status in statuses), |
| 1431 | stale_cache_hits=0, |
| 1432 | rate_limit_events=0, |
| 1433 | secondary_rate_limit_hit=False, |
| 1434 | source_type="external-news", |
| 1435 | ) |
| 1436 | ], |
| 1437 | analysis_metrics=None, |
| 1438 | environment={ |
| 1439 | "pipeline": "external-news-crawl", |
| 1440 | "output_path": out_path.as_posix(), |
| 1441 | "source_refresh_policy": source_refresh_policy, |
| 1442 | "same_day_reuse_status": output["metadata"]["same_day_reuse"], |
| 1443 | "sources_requested": list(output["metadata"]["sources_requested"]), |
| 1444 | "sources_succeeded": list(output["metadata"]["sources_succeeded"]), |
| 1445 | "sources_failed": list(output["metadata"]["sources_failed"]), |
| 1446 | "errors": list(errors), |
| 1447 | "total_run_duration_seconds": total_duration_seconds, |
| 1448 | }, |
| 1449 | ), |
| 1450 | observability_path, |
| 1451 | ) |
| 1452 | |
| 1453 | reused_count = sum( |
| 1454 | 1 for item in output["metadata"]["source_reuse_summary"] if item["action"] == "reused" |
| 1455 | ) |
| 1456 | refreshed_count = sum( |
| 1457 | 1 for item in output["metadata"]["source_reuse_summary"] if item["action"] != "reused" |
| 1458 | ) |
| 1459 | |
| 1460 | # Explicit per-source success/failure summary on stdout so a partial crawl |
| 1461 | # (one source failing while others succeed) is diagnosable straight from CI |
| 1462 | # logs. The same detail is persisted in metadata.source_status / sources_failed. |
| 1463 | succeeded_sources = list(output["metadata"]["sources_succeeded"]) |
| 1464 | failed_sources = list(output["metadata"]["sources_failed"]) |
| 1465 | print( |
| 1466 | f"[external-news] per-source summary: " |
| 1467 | f"{len(succeeded_sources)} succeeded, {len(failed_sources)} failed " |
| 1468 | f"(requested {len(output['metadata']['sources_requested'])})" |
| 1469 | ) |
| 1470 | if succeeded_sources: |
| 1471 | print(f"[external-news] succeeded: {', '.join(sorted(succeeded_sources))}") |
| 1472 | if failed_sources: |
| 1473 | status_by_source = { |
| 1474 | str(status.get("source")): status for status in output["metadata"]["source_status"] |
| 1475 | } |
| 1476 | for source_name in sorted(failed_sources): |
| 1477 | status = status_by_source.get(source_name, {}) |
| 1478 | reason = ( |
| 1479 | str(status.get("error_message") or status.get("error_class") or "unknown error") |
| 1480 | .replace("\n", " ") |
| 1481 | .strip() |
| 1482 | ) |
| 1483 | attempts = status.get("attempts", "?") |
| 1484 | print(f"[external-news] FAILED: {source_name} attempts={attempts} reason={reason}") |
| 1485 | print( |
| 1486 | "::warning::external-news crawl completed with partial results; " |
| 1487 | f"{len(failed_sources)} source(s) failed: {', '.join(sorted(failed_sources))}" |
| 1488 | ) |
| 1489 | |
| 1490 | print( |
| 1491 | f"Crawled {output['metadata']['total_articles']} articles " |
| 1492 | f"from {output['metadata']['source_count']} sources " |
| 1493 | f"({output['metadata']['relevant_articles']} relevant, " |
| 1494 | f"{output['metadata']['dedupe_count']} deduped, " |
| 1495 | f"{reused_count} reused, {refreshed_count} refreshed, p95={sampled_p95:.3f}s) " |
| 1496 | f"→ {out_path} [observability={observability_path}]" |
| 1497 | ) |
| 1498 | return 0 |
| 1499 | |
| 1500 | |
| 1501 | if __name__ == "__main__": |
| 1502 | sys.exit(main()) |