| 1 | #!/usr/bin/env python3 |
| 2 | """Cross-source correlation engine for SquadScope. |
| 3 | |
| 4 | Matches external news articles to GitHub repo activity using fuzzy matching |
| 5 | heuristics to identify press-correlated repositories. |
| 6 | |
| 7 | Usage: |
| 8 | python scripts/correlate.py [--raw data/raw/ai-ml/2026-W21.json] \ |
| 9 | [--techcrunch data/raw/ai-ml/2026-W21-external-news.json] \ |
| 10 | [--output data/analyzed/ai-ml/2026-W21-correlations.json] \ |
| 11 | [--topic ai-ml] |
| 12 | """ |
| 13 | |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | import argparse |
| 17 | import json |
| 18 | import re |
| 19 | import sys |
| 20 | from difflib import SequenceMatcher |
| 21 | from pathlib import Path |
| 22 | from typing import Any |
| 23 | |
| 24 | from scripts.sanitize_repo_content import sanitize_text |
| 25 | from scripts.topic_paths import analyzed_dir, raw_dir |
| 26 | |
| 27 | MAX_ARTICLES_FOR_CORRELATION = 80 |
| 28 | MAX_CORRELATIONS = 50 |
| 29 | MAX_MATCHED_ARTICLES_PER_REPO = 5 |
| 30 | MAX_DIVERGENCE_ARTICLES = 30 |
| 31 | WEAK_MATCH_TYPES = {"category", "project_name"} |
| 32 | |
| 33 | # Length caps for sanitized correlation output fields |
| 34 | _CITATION_TITLE_MAX = 200 |
| 35 | _CITATION_URL_MAX = 300 |
| 36 | _CITATION_SOURCE_MAX = 100 |
| 37 | _REPO_NAME_MAX = 200 |
| 38 | |
| 39 | |
| 40 | def log(message: str) -> None: |
| 41 | print(f"[correlate] {message}", file=sys.stderr) |
| 42 | |
| 43 | |
| 44 | # --------------------------------------------------------------------------- |
| 45 | # Heuristic matchers |
| 46 | # --------------------------------------------------------------------------- |
| 47 | |
| 48 | |
| 49 | def match_direct_link(repo: dict[str, Any], articles: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| 50 | """Match articles that contain a direct GitHub link to the repo.""" |
| 51 | repo_url = (repo.get("url") or "").rstrip("/").lower() |
| 52 | full_name = (repo.get("full_name") or "").lower() |
| 53 | if not repo_url and not full_name: |
| 54 | return [] |
| 55 | |
| 56 | matches = [] |
| 57 | for article in articles: |
| 58 | for link in article.get("github_links", []): |
| 59 | normalized = link.rstrip("/").lower() |
| 60 | if normalized == repo_url or normalized.endswith(f"/{full_name}"): |
| 61 | matches.append(article) |
| 62 | break |
| 63 | return matches |
| 64 | |
| 65 | |
| 66 | def match_org_name(repo: dict[str, Any], articles: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| 67 | """Match articles whose entities contain the repo owner name.""" |
| 68 | owner = (repo.get("owner") or "").lower() |
| 69 | if not owner or len(owner) < 2: |
| 70 | return [] |
| 71 | |
| 72 | matches = [] |
| 73 | for article in articles: |
| 74 | entities = [e.lower() for e in article.get("entities", [])] |
| 75 | if owner in entities: |
| 76 | matches.append(article) |
| 77 | return matches |
| 78 | |
| 79 | |
| 80 | def _token_overlap_ratio(a: str, b: str) -> float: |
| 81 | """Compute token overlap ratio between two strings.""" |
| 82 | tokens_a = set(re.split(r"[\s\-_]+", a.lower())) |
| 83 | tokens_b = set(re.split(r"[\s\-_]+", b.lower())) |
| 84 | tokens_a.discard("") |
| 85 | tokens_b.discard("") |
| 86 | if not tokens_a or not tokens_b: |
| 87 | return 0.0 |
| 88 | intersection = tokens_a & tokens_b |
| 89 | return len(intersection) / min(len(tokens_a), len(tokens_b)) |
| 90 | |
| 91 | |
| 92 | def fuzzy_name_score(repo_name: str, text: str) -> float: |
| 93 | """Compute fuzzy match score between repo name and text.""" |
| 94 | if not repo_name or not text: |
| 95 | return 0.0 |
| 96 | # SequenceMatcher ratio |
| 97 | seq_score = SequenceMatcher(None, repo_name.lower(), text.lower()).ratio() |
| 98 | # Token overlap |
| 99 | token_score = _token_overlap_ratio(repo_name, text) |
| 100 | return max(seq_score, token_score) |
| 101 | |
| 102 | |
| 103 | def match_project_name( |
| 104 | repo: dict[str, Any], articles: list[dict[str, Any]], threshold: float = 0.6 |
| 105 | ) -> list[dict[str, Any]]: |
| 106 | """Match articles by fuzzy matching repo name against title/entities.""" |
| 107 | repo_name = repo.get("name") or "" |
| 108 | if not repo_name or len(repo_name) < 3: |
| 109 | return [] |
| 110 | |
| 111 | matches = [] |
| 112 | for article in articles: |
| 113 | title = article.get("title") or "" |
| 114 | entities = article.get("entities", []) |
| 115 | |
| 116 | # Check title |
| 117 | if fuzzy_name_score(repo_name, title) >= threshold: |
| 118 | matches.append(article) |
| 119 | continue |
| 120 | |
| 121 | # Check individual entities |
| 122 | for entity in entities: |
| 123 | if fuzzy_name_score(repo_name, entity) >= threshold: |
| 124 | matches.append(article) |
| 125 | break |
| 126 | return matches |
| 127 | |
| 128 | |
| 129 | def match_category(repo: dict[str, Any], articles: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| 130 | """Match articles whose categories overlap with repo topics.""" |
| 131 | topics = {t.lower() for t in (repo.get("topics") or [])} |
| 132 | if not topics: |
| 133 | return [] |
| 134 | |
| 135 | matches = [] |
| 136 | for article in articles: |
| 137 | categories = {c.lower() for c in (article.get("categories") or [])} |
| 138 | if topics & categories: |
| 139 | matches.append(article) |
| 140 | return matches |
| 141 | |
| 142 | |
| 143 | def _normalized_article_url(url: str) -> str: |
| 144 | """Normalize an article URL for dedupe and citation joins.""" |
| 145 | if not url: |
| 146 | return "" |
| 147 | from urllib.parse import urlparse |
| 148 | |
| 149 | parsed = urlparse(url.strip()) |
| 150 | return f"{parsed.scheme.lower()}://{parsed.netloc.lower()}{parsed.path.rstrip('/')}" |
| 151 | |
| 152 | |
| 153 | def dedupe_articles(articles: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], int]: |
| 154 | """Deduplicate cross-source/mirrored stories by normalized URL.""" |
| 155 | grouped: dict[str, dict[str, Any]] = {} |
| 156 | duplicates = 0 |
| 157 | for article in sorted( |
| 158 | articles, |
| 159 | key=lambda item: ( |
| 160 | item.get("published_at", ""), |
| 161 | item.get("source", ""), |
| 162 | item.get("url", ""), |
| 163 | item.get("title", ""), |
| 164 | ), |
| 165 | reverse=True, |
| 166 | ): |
| 167 | key = _normalized_article_url(str(article.get("url", ""))) |
| 168 | if not key: |
| 169 | key = str(article.get("title", "")).strip().lower() |
| 170 | if key not in grouped: |
| 171 | current = dict(article) |
| 172 | current["sources"] = sorted( |
| 173 | { |
| 174 | str(current.get("source", "")) or "unknown", |
| 175 | *[str(source) for source in current.get("sources", [])], |
| 176 | } |
| 177 | ) |
| 178 | grouped[key] = current |
| 179 | continue |
| 180 | duplicates += 1 |
| 181 | existing = grouped[key] |
| 182 | sources = set(existing.get("sources", [])) |
| 183 | sources.add(str(article.get("source", "")) or "unknown") |
| 184 | sources.update(str(source) for source in article.get("sources", [])) |
| 185 | existing["sources"] = sorted(sources) |
| 186 | existing["relevance_score"] = max( |
| 187 | float(existing.get("relevance_score", 0)), |
| 188 | float(article.get("relevance_score", 0)), |
| 189 | ) |
| 190 | existing_links = list(existing.get("github_links", [])) |
| 191 | for link in article.get("github_links", []): |
| 192 | if link not in existing_links: |
| 193 | existing_links.append(link) |
| 194 | existing["github_links"] = existing_links |
| 195 | deduped = list(grouped.values()) |
| 196 | deduped.sort( |
| 197 | key=lambda item: ( |
| 198 | item.get("published_at", ""), |
| 199 | item.get("source", ""), |
| 200 | item.get("url", ""), |
| 201 | item.get("title", ""), |
| 202 | ), |
| 203 | reverse=True, |
| 204 | ) |
| 205 | return deduped, duplicates |
| 206 | |
| 207 | |
| 208 | def _article_citation(article: dict[str, Any]) -> dict[str, Any]: |
| 209 | """Return bounded, sanitized citation fields for downstream renderers.""" |
| 210 | return { |
| 211 | "title": sanitize_text( |
| 212 | article.get("title", ""), |
| 213 | max_length=_CITATION_TITLE_MAX, |
| 214 | label="article title", |
| 215 | ), |
| 216 | "url": sanitize_text( |
| 217 | article.get("url", ""), |
| 218 | max_length=_CITATION_URL_MAX, |
| 219 | label="article url", |
| 220 | ), |
| 221 | "source": sanitize_text( |
| 222 | article.get("source", "unknown"), |
| 223 | max_length=_CITATION_SOURCE_MAX, |
| 224 | label="article source", |
| 225 | ), |
| 226 | "sources": [ |
| 227 | sanitize_text(s, max_length=_CITATION_SOURCE_MAX, label="article source") |
| 228 | for s in article.get("sources", [article.get("source", "unknown")]) |
| 229 | ], |
| 230 | "published_at": article.get("published_at", ""), |
| 231 | "relevance_score": article.get("relevance_score", 0), |
| 232 | } |
| 233 | |
| 234 | |
| 235 | def _unique_articles(articles: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| 236 | """Return URL-deduped articles preserving order.""" |
| 237 | seen: set[str] = set() |
| 238 | unique: list[dict[str, Any]] = [] |
| 239 | for article in articles: |
| 240 | key = _normalized_article_url(str(article.get("url", ""))) or str(article) |
| 241 | if key in seen: |
| 242 | continue |
| 243 | seen.add(key) |
| 244 | unique.append(article) |
| 245 | return unique |
| 246 | |
| 247 | |
| 248 | def correlation_strength( |
| 249 | match_type: str, |
| 250 | matched_articles: list[dict[str, Any]], |
| 251 | *, |
| 252 | temporal_spike: bool, |
| 253 | ) -> str: |
| 254 | """Label strong vs weak correlations without letting fuzzy/category inflate claims.""" |
| 255 | source_names = { |
| 256 | source |
| 257 | for article in matched_articles |
| 258 | for source in article.get("sources", [article.get("source", "unknown")]) |
| 259 | } |
| 260 | corroborated = len(source_names) >= 2 or len(_unique_articles(matched_articles)) >= 2 |
| 261 | if match_type in WEAK_MATCH_TYPES: |
| 262 | return "weak" |
| 263 | if match_type in {"direct_link", "org_name"} or temporal_spike or corroborated: |
| 264 | return "strong" |
| 265 | return "weak" |
| 266 | |
| 267 | |
| 268 | def has_temporal_spike(repo: dict[str, Any], stars_threshold: int = 10) -> bool: |
| 269 | """Check if repo had a stars_gained spike in the same week.""" |
| 270 | stars_gained = repo.get("stars_gained") |
| 271 | if stars_gained is None: |
| 272 | return False |
| 273 | return stars_gained >= stars_threshold |
| 274 | |
| 275 | |
| 276 | # --------------------------------------------------------------------------- |
| 277 | # Hype risk assessment |
| 278 | # --------------------------------------------------------------------------- |
| 279 | |
| 280 | |
| 281 | def assess_hype_risk(confidence: float, stars_gained: int | None) -> str: |
| 282 | """Assess hype risk based on correlation confidence and star velocity.""" |
| 283 | if confidence >= 0.8: |
| 284 | if stars_gained is not None and stars_gained > 100: |
| 285 | return "high" |
| 286 | return "medium" |
| 287 | if confidence >= 0.6: |
| 288 | return "medium" |
| 289 | if confidence >= 0.4: |
| 290 | return "low" |
| 291 | return "none" |
| 292 | |
| 293 | |
| 294 | # --------------------------------------------------------------------------- |
| 295 | # Main correlation logic |
| 296 | # --------------------------------------------------------------------------- |
| 297 | |
| 298 | |
| 299 | def correlate_repo(repo: dict[str, Any], articles: list[dict[str, Any]]) -> dict[str, Any] | None: |
| 300 | """Correlate a single repo against all articles. Returns correlation or None.""" |
| 301 | best_confidence = 0.0 |
| 302 | best_type = "" |
| 303 | matched_articles: list[str] = [] |
| 304 | matched_article_objs: list[dict[str, Any]] = [] |
| 305 | |
| 306 | # Priority 1: Direct link match (confidence 1.0) |
| 307 | direct = match_direct_link(repo, articles) |
| 308 | if direct: |
| 309 | best_confidence = 1.0 |
| 310 | best_type = "direct_link" |
| 311 | matched_article_objs = _unique_articles(direct)[:MAX_MATCHED_ARTICLES_PER_REPO] |
| 312 | matched_articles = [a["url"] for a in matched_article_objs] |
| 313 | |
| 314 | # Priority 2: Org name match (confidence 0.8) |
| 315 | if not matched_articles: |
| 316 | org = match_org_name(repo, articles) |
| 317 | if org: |
| 318 | best_confidence = 0.8 |
| 319 | best_type = "org_name" |
| 320 | matched_article_objs = _unique_articles(org)[:MAX_MATCHED_ARTICLES_PER_REPO] |
| 321 | matched_articles = [a["url"] for a in matched_article_objs] |
| 322 | |
| 323 | # Priority 3: Project name fuzzy match (confidence 0.6) |
| 324 | if not matched_articles: |
| 325 | fuzzy = match_project_name(repo, articles) |
| 326 | if fuzzy: |
| 327 | best_confidence = 0.6 |
| 328 | best_type = "project_name" |
| 329 | matched_article_objs = _unique_articles(fuzzy)[:MAX_MATCHED_ARTICLES_PER_REPO] |
| 330 | matched_articles = [a["url"] for a in matched_article_objs] |
| 331 | |
| 332 | # Priority 4: Category correlation (confidence 0.4) |
| 333 | if not matched_articles: |
| 334 | cat = match_category(repo, articles) |
| 335 | if cat: |
| 336 | best_confidence = 0.4 |
| 337 | best_type = "category" |
| 338 | matched_article_objs = _unique_articles(cat)[:MAX_MATCHED_ARTICLES_PER_REPO] |
| 339 | matched_articles = [a["url"] for a in matched_article_objs] |
| 340 | |
| 341 | if not matched_articles: |
| 342 | return None |
| 343 | |
| 344 | # Priority 5: Temporal lag bonus |
| 345 | press_correlated = True |
| 346 | temporal_spike = has_temporal_spike(repo) |
| 347 | if temporal_spike and best_type not in WEAK_MATCH_TYPES: |
| 348 | best_confidence = min(best_confidence + 0.2, 1.0) |
| 349 | press_correlated = True |
| 350 | strength = correlation_strength( |
| 351 | best_type, |
| 352 | matched_article_objs, |
| 353 | temporal_spike=temporal_spike, |
| 354 | ) |
| 355 | |
| 356 | raw_repo_key = repo.get("full_name") or f"{repo.get('owner')}/{repo.get('name')}" |
| 357 | repo_name = sanitize_text( |
| 358 | raw_repo_key, |
| 359 | max_length=_REPO_NAME_MAX, |
| 360 | label="correlation repo name", |
| 361 | ) |
| 362 | |
| 363 | return { |
| 364 | "repo": repo_name, |
| 365 | "repo_key": raw_repo_key, |
| 366 | "press_correlated": press_correlated, |
| 367 | "correlation_confidence": round(best_confidence, 2), |
| 368 | "matched_articles": matched_articles, |
| 369 | "matched_article_details": [_article_citation(article) for article in matched_article_objs], |
| 370 | "match_type": best_type, |
| 371 | "correlation_strength": strength, |
| 372 | "confidence_label": strength, |
| 373 | "temporal_spike": temporal_spike, |
| 374 | "hype_risk": assess_hype_risk(best_confidence, repo.get("stars_gained")), |
| 375 | } |
| 376 | |
| 377 | |
| 378 | def _extract_article_topic(article: dict[str, Any]) -> str: |
| 379 | """Extract a representative topic string from an article.""" |
| 380 | categories = article.get("categories", []) |
| 381 | if categories: |
| 382 | return categories[0] |
| 383 | entities = article.get("entities", []) |
| 384 | if entities: |
| 385 | return entities[0] |
| 386 | title = article.get("title", "") |
| 387 | # Use first few meaningful words from title as fallback |
| 388 | words = [w for w in re.split(r"\s+", title) if len(w) > 3] |
| 389 | return " ".join(words[:3]) if words else "unknown" |
| 390 | |
| 391 | |
| 392 | def _extract_repo_topic(repo: dict[str, Any]) -> str: |
| 393 | """Extract a representative topic string from a repo.""" |
| 394 | topics = repo.get("topics", []) |
| 395 | if topics: |
| 396 | return topics[0] |
| 397 | description = repo.get("description") or "" |
| 398 | words = [w for w in re.split(r"\s+", description) if len(w) > 3] |
| 399 | return " ".join(words[:3]) if words else repo.get("name", "unknown") |
| 400 | |
| 401 | |
| 402 | def detect_divergences( |
| 403 | repos: list[dict[str, Any]], |
| 404 | articles: list[dict[str, Any]], |
| 405 | correlations: list[dict[str, Any]], |
| 406 | ) -> dict[str, Any]: |
| 407 | """Detect divergences — gaps between press coverage and dev activity. |
| 408 | |
| 409 | Returns two lists: |
| 410 | - uncovered_tech_trends: articles/topics with no matching GitHub activity |
| 411 | - unpublicized_dev_activity: repos/trends with no matching press coverage |
| 412 | """ |
| 413 | # Find article URLs that were matched by at least one correlation |
| 414 | matched_article_urls: set[str] = set() |
| 415 | for corr in correlations: |
| 416 | matched_article_urls.update(corr.get("matched_articles", [])) |
| 417 | |
| 418 | # Unmatched articles → uncovered tech trends |
| 419 | unmatched_articles = [a for a in articles if a.get("url") not in matched_article_urls][ |
| 420 | :MAX_DIVERGENCE_ARTICLES |
| 421 | ] |
| 422 | |
| 423 | # Group unmatched articles by topic |
| 424 | topic_articles: dict[str, list[dict[str, Any]]] = {} |
| 425 | for article in unmatched_articles: |
| 426 | topic = _extract_article_topic(article) |
| 427 | topic_articles.setdefault(topic, []).append(article) |
| 428 | |
| 429 | uncovered_tech_trends = [ |
| 430 | { |
| 431 | "topic": topic, |
| 432 | "news_articles": [{"title": a.get("title", ""), "url": a.get("url", "")} for a in arts], |
| 433 | "techcrunch_articles": [ |
| 434 | {"title": a.get("title", ""), "url": a.get("url", "")} for a in arts |
| 435 | ], |
| 436 | "signal": "No matching GitHub activity", |
| 437 | } |
| 438 | for topic, arts in sorted(topic_articles.items(), key=lambda x: -len(x[1])) |
| 439 | ] |
| 440 | |
| 441 | # Find repos that had no correlation match |
| 442 | correlated_repo_names: set[str] = {c.get("repo_key", c.get("repo", "")) for c in correlations} |
| 443 | unmatched_repos = [ |
| 444 | r |
| 445 | for r in repos |
| 446 | if (r.get("full_name") or f"{r.get('owner')}/{r.get('name')}") not in correlated_repo_names |
| 447 | ] |
| 448 | |
| 449 | # Group unmatched repos by topic |
| 450 | topic_repos: dict[str, list[dict[str, Any]]] = {} |
| 451 | for repo in unmatched_repos: |
| 452 | topic = _extract_repo_topic(repo) |
| 453 | topic_repos.setdefault(topic, []).append(repo) |
| 454 | |
| 455 | unpublicized_dev_activity = [ |
| 456 | { |
| 457 | "topic": topic, |
| 458 | "github_repos": [ |
| 459 | { |
| 460 | "full_name": r.get("full_name") or f"{r.get('owner')}/{r.get('name')}", |
| 461 | "stars": r.get("stars", 0), |
| 462 | "stars_gained": r.get("stars_gained"), |
| 463 | } |
| 464 | for r in reps |
| 465 | ], |
| 466 | "signal": "No external press coverage", |
| 467 | } |
| 468 | for topic, reps in sorted(topic_repos.items(), key=lambda x: -len(x[1])) |
| 469 | ] |
| 470 | |
| 471 | return { |
| 472 | "uncovered_tech_trends": uncovered_tech_trends, |
| 473 | "unpublicized_dev_activity": unpublicized_dev_activity, |
| 474 | } |
| 475 | |
| 476 | |
| 477 | def correlate_all( |
| 478 | repos: list[dict[str, Any]], articles: list[dict[str, Any]], week: str |
| 479 | ) -> dict[str, Any]: |
| 480 | """Run correlation engine across all repos and articles.""" |
| 481 | articles, dedupe_count = dedupe_articles(articles) |
| 482 | articles = articles[:MAX_ARTICLES_FOR_CORRELATION] |
| 483 | correlations: list[dict[str, Any]] = [] |
| 484 | uncorrelated: list[str] = [] |
| 485 | |
| 486 | for repo in repos: |
| 487 | result = correlate_repo(repo, articles) |
| 488 | if result: |
| 489 | correlations.append(result) |
| 490 | else: |
| 491 | name = repo.get("full_name") or f"{repo.get('owner')}/{repo.get('name')}" |
| 492 | uncorrelated.append(name) |
| 493 | |
| 494 | # Sort by confidence descending |
| 495 | correlations.sort( |
| 496 | key=lambda c: ( |
| 497 | c.get("correlation_strength") != "strong", |
| 498 | -c["correlation_confidence"], |
| 499 | c.get("repo", ""), |
| 500 | ) |
| 501 | ) |
| 502 | correlations = correlations[:MAX_CORRELATIONS] |
| 503 | |
| 504 | articles_matched = len({url for c in correlations for url in c["matched_articles"]}) |
| 505 | |
| 506 | # Detect divergences |
| 507 | divergences = detect_divergences(repos, articles, correlations) |
| 508 | |
| 509 | return { |
| 510 | "week": week, |
| 511 | "correlations": correlations, |
| 512 | "divergences": divergences, |
| 513 | "uncorrelated_repos": uncorrelated, |
| 514 | "metadata": { |
| 515 | "repos_analyzed": len(repos), |
| 516 | "articles_analyzed": len(articles), |
| 517 | "correlations_found": len(correlations), |
| 518 | "strong_correlations": sum( |
| 519 | 1 for corr in correlations if corr.get("correlation_strength") == "strong" |
| 520 | ), |
| 521 | "weak_correlations": sum( |
| 522 | 1 for corr in correlations if corr.get("correlation_strength") == "weak" |
| 523 | ), |
| 524 | "articles_matched": articles_matched, |
| 525 | "dedupe_count": dedupe_count, |
| 526 | "limits": { |
| 527 | "max_articles": MAX_ARTICLES_FOR_CORRELATION, |
| 528 | "max_correlations": MAX_CORRELATIONS, |
| 529 | "max_matched_articles_per_repo": MAX_MATCHED_ARTICLES_PER_REPO, |
| 530 | }, |
| 531 | "uncovered_tech_trends": len(divergences["uncovered_tech_trends"]), |
| 532 | "unpublicized_dev_activity": len(divergences["unpublicized_dev_activity"]), |
| 533 | }, |
| 534 | } |
| 535 | |
| 536 | |
| 537 | # --------------------------------------------------------------------------- |
| 538 | # File discovery |
| 539 | # --------------------------------------------------------------------------- |
| 540 | |
| 541 | |
| 542 | def find_latest_file(directory: Path, pattern: str) -> Path | None: |
| 543 | """Find the latest file matching a glob pattern in directory.""" |
| 544 | if not directory.exists(): |
| 545 | return None |
| 546 | files = sorted(directory.glob(pattern), reverse=True) |
| 547 | return files[0] if files else None |
| 548 | |
| 549 | |
| 550 | def load_json(path: Path) -> dict[str, Any]: |
| 551 | """Load and return parsed JSON from a file.""" |
| 552 | with open(path, encoding="utf-8") as f: |
| 553 | return json.load(f) |
| 554 | |
| 555 | |
| 556 | def extract_news_metadata(news_data: dict[str, Any] | list[dict[str, Any]]) -> dict[str, Any]: |
| 557 | """Extract source/failure metadata from canonical or legacy news payloads.""" |
| 558 | if isinstance(news_data, list): |
| 559 | return { |
| 560 | "schema_version": 1, |
| 561 | "sources_requested": ["techcrunch"], |
| 562 | "sources_succeeded": ["techcrunch"], |
| 563 | "sources_failed": [], |
| 564 | "source_status": [], |
| 565 | "errors": [], |
| 566 | } |
| 567 | metadata = news_data.get("metadata", {}) |
| 568 | return { |
| 569 | "schema_version": news_data.get("schema_version", 1), |
| 570 | "source_config_checksum": metadata.get("source_config_checksum", ""), |
| 571 | "sources_requested": metadata.get( |
| 572 | "sources_requested", [news_data.get("source", "techcrunch")] |
| 573 | ), |
| 574 | "sources_succeeded": metadata.get("sources_succeeded", []), |
| 575 | "sources_failed": metadata.get("sources_failed", []), |
| 576 | "source_status": metadata.get("source_status", []), |
| 577 | "errors": metadata.get("errors", []), |
| 578 | "artifact_checksum": metadata.get("artifact_checksum", ""), |
| 579 | } |
| 580 | |
| 581 | |
| 582 | def extract_week_from_filename(path: Path) -> str: |
| 583 | """Extract week slug from filename like '2026-W21.json'.""" |
| 584 | match = re.search(r"(\d{4}-W\d{2})", path.name) |
| 585 | return match.group(1) if match else "unknown" |
| 586 | |
| 587 | |
| 588 | # --------------------------------------------------------------------------- |
| 589 | # CLI |
| 590 | # --------------------------------------------------------------------------- |
| 591 | |
| 592 | |
| 593 | def main(argv: list[str] | None = None) -> int: |
| 594 | parser = argparse.ArgumentParser(description="Cross-source correlation engine for SquadScope") |
| 595 | parser.add_argument( |
| 596 | "--raw", |
| 597 | default=None, |
| 598 | help="Path to raw GitHub repos JSON file", |
| 599 | ) |
| 600 | parser.add_argument( |
| 601 | "--techcrunch", |
| 602 | default=None, |
| 603 | help="Path to external news articles JSON file", |
| 604 | ) |
| 605 | parser.add_argument( |
| 606 | "--output", |
| 607 | default=None, |
| 608 | help="Output file path for correlations", |
| 609 | ) |
| 610 | parser.add_argument( |
| 611 | "--topic", |
| 612 | default="general", |
| 613 | help="Topic ID for path resolution (default: general)", |
| 614 | ) |
| 615 | |
| 616 | args = parser.parse_args(argv) |
| 617 | topic = args.topic |
| 618 | |
| 619 | # Resolve raw repos file |
| 620 | if args.raw: |
| 621 | raw_path = Path(args.raw) |
| 622 | else: |
| 623 | raw_path = find_latest_file(raw_dir(topic), "[0-9]*-W[0-9]*.json") |
| 624 | if raw_path is None: |
| 625 | log(f"No raw data found in {raw_dir(topic)}") |
| 626 | return 1 |
| 627 | |
| 628 | if not raw_path.exists(): |
| 629 | log(f"Raw file not found: {raw_path}") |
| 630 | return 1 |
| 631 | |
| 632 | # Resolve external news file, with TechCrunch-only legacy fallback. |
| 633 | if args.techcrunch: |
| 634 | tc_path = Path(args.techcrunch) |
| 635 | else: |
| 636 | tc_path = find_latest_file(raw_dir(topic), "*-external-news.json") |
| 637 | if tc_path is None: |
| 638 | tc_path = find_latest_file(raw_dir(topic), "*-techcrunch.json") |
| 639 | |
| 640 | # Load repos |
| 641 | raw_data = load_json(raw_path) |
| 642 | if isinstance(raw_data, list): |
| 643 | repos = raw_data |
| 644 | else: |
| 645 | # The crawl output stores repos under "new_repos" and "trending_repos" |
| 646 | repos = raw_data.get("repos", raw_data.get("repositories", [])) |
| 647 | if not repos: |
| 648 | new_repos = raw_data.get("new_repos", []) |
| 649 | trending_repos = raw_data.get("trending_repos", []) |
| 650 | repos = new_repos + trending_repos |
| 651 | |
| 652 | # Load articles (graceful if missing) |
| 653 | articles: list[dict[str, Any]] = [] |
| 654 | news_metadata: dict[str, Any] = {} |
| 655 | if tc_path and tc_path.exists(): |
| 656 | tc_data = load_json(tc_path) |
| 657 | news_metadata = extract_news_metadata(tc_data) |
| 658 | articles = tc_data if isinstance(tc_data, list) else tc_data.get("articles", []) |
| 659 | else: |
| 660 | log("No external news data found; producing empty correlations") |
| 661 | |
| 662 | # Determine week |
| 663 | week = extract_week_from_filename(raw_path) |
| 664 | |
| 665 | # Run correlation |
| 666 | result = correlate_all(repos, articles, week) |
| 667 | if news_metadata: |
| 668 | result["metadata"]["news_sources"] = news_metadata |
| 669 | |
| 670 | # Write output |
| 671 | if args.output: |
| 672 | output_path = Path(args.output) |
| 673 | else: |
| 674 | out_dir = analyzed_dir(topic) |
| 675 | out_dir.mkdir(parents=True, exist_ok=True) |
| 676 | output_path = out_dir / f"{week}-correlations.json" |
| 677 | |
| 678 | output_path.parent.mkdir(parents=True, exist_ok=True) |
| 679 | with open(output_path, "w", encoding="utf-8") as f: |
| 680 | json.dump(result, f, indent=2, ensure_ascii=False) |
| 681 | f.write("\n") |
| 682 | |
| 683 | log( |
| 684 | f"Wrote {output_path}: {result['metadata']['correlations_found']} correlations from {result['metadata']['repos_analyzed']} repos" |
| 685 | ) |
| 686 | return 0 |
| 687 | |
| 688 | |
| 689 | if __name__ == "__main__": |
| 690 | sys.exit(main()) |