| 1 | #!/usr/bin/env python3 |
| 2 | """Generate append-only monthly and yearly rollup pages from weekly analyses. |
| 3 | |
| 4 | Monthly pages include: |
| 5 | - Month Synthesis section with cross-links to weekly and yearly pages |
| 6 | - SEO-friendly editorial titles (max 70 chars) derived from theme trajectories |
| 7 | - Structured frontmatter: summary, themes, persistent/accelerating/weakening themes, |
| 8 | key_gaps, top_repos, weeks_covered |
| 9 | - Per-week entries for overview, top repos, trends, and key takeaways |
| 10 | |
| 11 | Yearly pages are delegated to generate_yearly_narrative.build_yearly_narrative_pages(). |
| 12 | A rolling 4-week context report can also be generated with --rolling. |
| 13 | """ |
| 14 | |
| 15 | from __future__ import annotations |
| 16 | |
| 17 | import argparse |
| 18 | import re |
| 19 | import sys |
| 20 | from collections import Counter, defaultdict |
| 21 | from dataclasses import dataclass |
| 22 | from pathlib import Path |
| 23 | from typing import Any |
| 24 | |
| 25 | if __package__ in {None, ""}: |
| 26 | sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) |
| 27 | |
| 28 | import scripts.analysis_gate as analysis_gate |
| 29 | from scripts.generate_yearly_narrative import build_yearly_narrative_pages |
| 30 | from scripts.month_synthesis import ensure_month_synthesis |
| 31 | |
| 32 | PROJECT_ROOT = Path(__file__).resolve().parent.parent |
| 33 | SUMMARY_SUFFIX = "-summary.md" |
| 34 | WEEK_PATTERN = re.compile(r"^(?P<year>\d{4})-W(?P<week>\d{2})$") |
| 35 | REPO_LINK_PATTERN = re.compile(r"https://github\.com/(?P<repo>[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)") |
| 36 | NO_UPDATES_PLACEHOLDER = "_No updates yet._" |
| 37 | MONTHLY_SECTIONS = [ |
| 38 | "Month Synthesis", |
| 39 | "Month Overview", |
| 40 | "Top Repos This Month", |
| 41 | "Trends Observed", |
| 42 | "Key Takeaways", |
| 43 | ] |
| 44 | YEARLY_SECTIONS = [ |
| 45 | "Year in Review", |
| 46 | ] |
| 47 | MONTH_NAMES = { |
| 48 | 1: "January", |
| 49 | 2: "February", |
| 50 | 3: "March", |
| 51 | 4: "April", |
| 52 | 5: "May", |
| 53 | 6: "June", |
| 54 | 7: "July", |
| 55 | 8: "August", |
| 56 | 9: "September", |
| 57 | 10: "October", |
| 58 | 11: "November", |
| 59 | 12: "December", |
| 60 | } |
| 61 | SECTION_HEADING_PATTERN = re.compile(r"(?m)^##\s+(.+?)\s*$") |
| 62 | SUBSECTION_HEADING_TEMPLATE = r"(?ms)^###\s+{title}\s*$\n(.*?)(?=^###\s+|\Z)" |
| 63 | |
| 64 | |
| 65 | class RollupError(ValueError): |
| 66 | pass |
| 67 | |
| 68 | |
| 69 | @dataclass(frozen=True) |
| 70 | class WeeklySummary: |
| 71 | source_path: Path |
| 72 | title: str |
| 73 | date: object |
| 74 | week: str |
| 75 | year: int |
| 76 | month: int |
| 77 | tags: tuple[str, ...] |
| 78 | repos_featured: int |
| 79 | top_repo: str |
| 80 | featured_repos: tuple[str, ...] |
| 81 | summary: str |
| 82 | signal: str |
| 83 | noise: str |
| 84 | gaps: str |
| 85 | conclusion: str |
| 86 | |
| 87 | @property |
| 88 | def week_number(self) -> int: |
| 89 | match = WEEK_PATTERN.fullmatch(self.week) |
| 90 | if not match: |
| 91 | raise RollupError(f"Invalid week slug: {self.week}") |
| 92 | return int(match.group("week")) |
| 93 | |
| 94 | @property |
| 95 | def week_title(self) -> str: |
| 96 | return f"Week {self.week_number}, {self.year}" |
| 97 | |
| 98 | @property |
| 99 | def week_link(self) -> str: |
| 100 | return f"/weekly/{self.year}/W{self.week_number:02d}/" |
| 101 | |
| 102 | @property |
| 103 | def month_slug(self) -> str: |
| 104 | return f"{self.year}-{self.month:02d}" |
| 105 | |
| 106 | @property |
| 107 | def month_title(self) -> str: |
| 108 | return f"{MONTH_NAMES[self.month]} {self.year}" |
| 109 | |
| 110 | @property |
| 111 | def month_link(self) -> str: |
| 112 | return f"/monthly/{self.year}/{self.month:02d}/" |
| 113 | |
| 114 | |
| 115 | @dataclass(frozen=True) |
| 116 | class RollupEntry: |
| 117 | marker: str |
| 118 | text: str |
| 119 | |
| 120 | |
| 121 | @dataclass(frozen=True) |
| 122 | class RollupPage: |
| 123 | path: Path |
| 124 | frontmatter: dict[str, Any] |
| 125 | sections: dict[str, list[RollupEntry]] |
| 126 | section_order: list[str] |
| 127 | replace_existing_sections: bool = False |
| 128 | preserve_unknown_sections: bool = True |
| 129 | replace_sections: frozenset[str] = frozenset() |
| 130 | |
| 131 | |
| 132 | ACRONYMS = {"ai", "mcp", "ci", "cd", "api", "sdk", "llm", "rag", "ml"} |
| 133 | |
| 134 | |
| 135 | def _titlecase_tag(tag: str) -> str: |
| 136 | words = tag.replace("-", " ").split() |
| 137 | return " ".join(w.upper() if w.lower() in ACRONYMS else w.title() for w in words) |
| 138 | |
| 139 | |
| 140 | def generate_monthly_title(synthesis: Any, month: int, year: int) -> str: |
| 141 | """Generate an SEO-friendly editorial title (max 70 chars) from synthesis data.""" |
| 142 | month_year = f"{MONTH_NAMES[month]} {year}" |
| 143 | accel = [_titlecase_tag(t) for t in synthesis.accelerating_themes[:2]] |
| 144 | weak = [_titlecase_tag(t) for t in synthesis.weakening_themes[:1]] |
| 145 | themes = [_titlecase_tag(t) for t in synthesis.themes[:2]] |
| 146 | |
| 147 | if len(accel) >= 2: |
| 148 | title = f"{accel[0]} and {accel[1]} Surge — {month_year}" |
| 149 | elif accel and weak: |
| 150 | title = f"{accel[0]} Surges While {weak[0]} Fades — {month_year}" |
| 151 | elif accel: |
| 152 | title = f"{accel[0]} Takes Center Stage — {month_year}" |
| 153 | elif len(themes) >= 2: |
| 154 | title = f"{themes[0]} and {themes[1]} Define the Month — {month_year}" |
| 155 | elif themes: |
| 156 | title = f"{themes[0]} Leads the Month — {month_year}" |
| 157 | else: |
| 158 | title = f"Trends Shift and Settle — {month_year}" |
| 159 | |
| 160 | if len(title) > 70: |
| 161 | if accel: |
| 162 | title = f"{accel[0]} Surges — {month_year}" |
| 163 | elif themes: |
| 164 | title = f"{themes[0]} Leads — {month_year}" |
| 165 | if len(title) > 70: |
| 166 | title = title[:67] + "…" |
| 167 | |
| 168 | return title |
| 169 | |
| 170 | |
| 171 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 172 | parser = argparse.ArgumentParser( |
| 173 | description="Generate append-only monthly and yearly rollups from weekly analyses. Defaults resolve from the repository root." |
| 174 | ) |
| 175 | parser.add_argument( |
| 176 | "--analyzed-dir", |
| 177 | type=Path, |
| 178 | default=PROJECT_ROOT / "data" / "analyzed", |
| 179 | help="Directory containing weekly summary markdown files.", |
| 180 | ) |
| 181 | parser.add_argument( |
| 182 | "--content-root", |
| 183 | type=Path, |
| 184 | default=PROJECT_ROOT / "content", |
| 185 | help="Root content directory for generated rollups.", |
| 186 | ) |
| 187 | parser.add_argument( |
| 188 | "--rolling", |
| 189 | action="store_true", |
| 190 | default=False, |
| 191 | help="Generate the rolling 4-week context report after rollups.", |
| 192 | ) |
| 193 | return parser.parse_args(argv) |
| 194 | |
| 195 | |
| 196 | def yaml_quote(value: str) -> str: |
| 197 | return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' |
| 198 | |
| 199 | |
| 200 | def yaml_value(value: Any) -> str: |
| 201 | if isinstance(value, str): |
| 202 | return yaml_quote(value) |
| 203 | if isinstance(value, bool): |
| 204 | return "true" if value else "false" |
| 205 | if isinstance(value, int): |
| 206 | return str(value) |
| 207 | if isinstance(value, list): |
| 208 | rendered = ", ".join(yaml_value(item) for item in value) |
| 209 | return f"[{rendered}]" |
| 210 | return str(value) |
| 211 | |
| 212 | |
| 213 | def render_frontmatter(frontmatter: dict[str, Any]) -> str: |
| 214 | lines = ["---"] |
| 215 | for key, value in frontmatter.items(): |
| 216 | lines.append(f"{key}: {yaml_value(value)}") |
| 217 | lines.extend(["---", "", ""]) |
| 218 | return "\n".join(lines) |
| 219 | |
| 220 | |
| 221 | def split_sections(body: str) -> tuple[str, dict[str, str]]: |
| 222 | matches = list(SECTION_HEADING_PATTERN.finditer(body)) |
| 223 | if not matches: |
| 224 | return body.rstrip(), {} |
| 225 | |
| 226 | intro = body[: matches[0].start()].rstrip() |
| 227 | sections: dict[str, str] = {} |
| 228 | for index, match in enumerate(matches): |
| 229 | start = match.end() |
| 230 | end = matches[index + 1].start() if index + 1 < len(matches) else len(body) |
| 231 | sections[match.group(1).strip()] = body[start:end].strip("\n") |
| 232 | return intro, sections |
| 233 | |
| 234 | |
| 235 | def get_section_text(body: str, heading: str) -> str: |
| 236 | matches = list(SECTION_HEADING_PATTERN.finditer(body)) |
| 237 | for index, match in enumerate(matches): |
| 238 | if match.group(1).strip() != heading: |
| 239 | continue |
| 240 | start = match.end() |
| 241 | end = matches[index + 1].start() if index + 1 < len(matches) else len(body) |
| 242 | return body[start:end].strip() |
| 243 | return "" |
| 244 | |
| 245 | |
| 246 | def get_subsection_text(section_body: str, heading: str) -> str: |
| 247 | pattern = re.compile(SUBSECTION_HEADING_TEMPLATE.format(title=re.escape(heading))) |
| 248 | match = pattern.search(section_body) |
| 249 | return re.sub(r"\s+", " ", match.group(1).strip()) if match else "" |
| 250 | |
| 251 | |
| 252 | def normalize_text(value: str) -> str: |
| 253 | return re.sub(r"\s+", " ", value.strip()) |
| 254 | |
| 255 | |
| 256 | def repo_markdown(repo: str) -> str: |
| 257 | return f"[{repo}](https://github.com/{repo})" |
| 258 | |
| 259 | |
| 260 | def extract_featured_repos(body: str, top_repo: str) -> tuple[str, ...]: |
| 261 | repos = [top_repo] |
| 262 | seen = {top_repo} |
| 263 | for match in REPO_LINK_PATTERN.finditer(body): |
| 264 | repo = match.group("repo") |
| 265 | if repo in seen: |
| 266 | continue |
| 267 | seen.add(repo) |
| 268 | repos.append(repo) |
| 269 | return tuple(repos) |
| 270 | |
| 271 | |
| 272 | def load_summary(path: Path) -> WeeklySummary: |
| 273 | frontmatter, body = analysis_gate.extract_frontmatter(path.read_text(encoding="utf-8")) |
| 274 | week = str(frontmatter.get("week", "")) |
| 275 | match = WEEK_PATTERN.fullmatch(week) |
| 276 | if not match: |
| 277 | raise RollupError(f"Invalid weekly summary filename: {path.name}") |
| 278 | |
| 279 | date = analysis_gate.parse_datetime(frontmatter["date"]) |
| 280 | month = date.month |
| 281 | |
| 282 | # New structure: try new heading names first, fall back to old for backward compat |
| 283 | signal_noise_section = get_section_text(body, "Signal & Noise") |
| 284 | if signal_noise_section: |
| 285 | signal = normalize_text(signal_noise_section) |
| 286 | noise = "" |
| 287 | else: |
| 288 | trend_analysis = get_section_text(body, "Trend Analysis") |
| 289 | signal = get_subsection_text(trend_analysis, "Signal") |
| 290 | noise = get_subsection_text(trend_analysis, "Noise") |
| 291 | |
| 292 | blind_spots = get_section_text(body, "Blind Spots") |
| 293 | if blind_spots: |
| 294 | gaps = normalize_text(blind_spots) |
| 295 | else: |
| 296 | gaps = get_subsection_text(get_section_text(body, "What's Missing"), "Gaps") |
| 297 | |
| 298 | week_ahead = get_section_text(body, "The Week Ahead") |
| 299 | if week_ahead: |
| 300 | conclusion = normalize_text(week_ahead) |
| 301 | else: |
| 302 | conclusion = normalize_text(get_section_text(body, "Conclusion")) |
| 303 | top_repo = str(frontmatter["top_repo"]) |
| 304 | |
| 305 | return WeeklySummary( |
| 306 | source_path=path, |
| 307 | title=str(frontmatter["title"]), |
| 308 | date=date, |
| 309 | week=week, |
| 310 | year=int(frontmatter["year"]), |
| 311 | month=month, |
| 312 | tags=tuple(str(tag) for tag in frontmatter.get("tags", [])), |
| 313 | repos_featured=int(frontmatter["repos_featured"]), |
| 314 | top_repo=top_repo, |
| 315 | featured_repos=extract_featured_repos(body, top_repo), |
| 316 | summary=normalize_text(str(frontmatter["summary"])), |
| 317 | signal=signal, |
| 318 | noise=noise, |
| 319 | gaps=gaps, |
| 320 | conclusion=conclusion, |
| 321 | ) |
| 322 | |
| 323 | |
| 324 | def load_weekly_summaries(analyzed_dir: Path) -> list[WeeklySummary]: |
| 325 | summaries = [load_summary(path) for path in sorted(analyzed_dir.glob(f"*{SUMMARY_SUFFIX}"))] |
| 326 | return sorted(summaries, key=lambda item: (item.date, item.week)) |
| 327 | |
| 328 | |
| 329 | def monthly_entries(weekly: WeeklySummary, tags_counter: Counter[str]) -> dict[str, RollupEntry]: |
| 330 | common_tags = ", ".join(tag for tag, _ in tags_counter.most_common(3)) or "none yet" |
| 331 | repo_link = repo_markdown(weekly.top_repo) |
| 332 | page_link = f"[{weekly.week_title}]({weekly.week_link})" |
| 333 | marker = f"### Week {weekly.week}" |
| 334 | return { |
| 335 | "Month Overview": RollupEntry( |
| 336 | marker=marker, |
| 337 | text=( |
| 338 | f"{marker} — {page_link}\n" |
| 339 | f"- Summary: {weekly.summary}\n" |
| 340 | f"- Repositories featured this week: {weekly.repos_featured}\n" |
| 341 | f"- Recurring themes so far: {common_tags}." |
| 342 | ), |
| 343 | ), |
| 344 | "Top Repos This Month": RollupEntry( |
| 345 | marker=marker, |
| 346 | text=( |
| 347 | f"{marker} — {page_link}\n" |
| 348 | f"- {repo_link} led the published weekly analysis for {weekly.week}.\n" |
| 349 | f"- Detailed breakdown: {page_link}." |
| 350 | ), |
| 351 | ), |
| 352 | "Trends Observed": RollupEntry( |
| 353 | marker=marker, |
| 354 | text=( |
| 355 | f"{marker} — {page_link}\n" |
| 356 | f"- Signal: {weekly.signal}" |
| 357 | + (f"\n- Noise: {weekly.noise}" if weekly.noise else "") |
| 358 | ), |
| 359 | ), |
| 360 | "Key Takeaways": RollupEntry( |
| 361 | marker=marker, |
| 362 | text=( |
| 363 | f"{marker} — {page_link}\n" |
| 364 | f"- Gap to watch: {weekly.gaps}\n" |
| 365 | f"- Closing read: {weekly.conclusion}" |
| 366 | ), |
| 367 | ), |
| 368 | } |
| 369 | |
| 370 | |
| 371 | def _build_monthly_crosslinks(year: int, _month: int, items: list[WeeklySummary]) -> str: |
| 372 | """Build navigation cross-links for a monthly page.""" |
| 373 | links = [] |
| 374 | links.append(f"[{year} Year in Review](/yearly/{year}/)") |
| 375 | for item in items: |
| 376 | links.append(f"[{item.week_title}]({item.week_link})") |
| 377 | return f"*Part of {links[0]}* · Weekly: {' · '.join(links[1:])}\n" |
| 378 | |
| 379 | |
| 380 | def build_monthly_pages( |
| 381 | summaries: list[WeeklySummary], content_root: Path, analyzed_dir: Path |
| 382 | ) -> list[RollupPage]: |
| 383 | grouped: dict[tuple[int, int], list[WeeklySummary]] = defaultdict(list) |
| 384 | for summary in summaries: |
| 385 | grouped[(summary.year, summary.month)].append(summary) |
| 386 | |
| 387 | pages: list[RollupPage] = [] |
| 388 | for (year, month), items in sorted(grouped.items()): |
| 389 | items = sorted(items, key=lambda item: (item.date, item.week)) |
| 390 | tags_counter: Counter[str] = Counter() |
| 391 | page_entries: dict[str, list[RollupEntry]] = {section: [] for section in MONTHLY_SECTIONS} |
| 392 | |
| 393 | synthesis = ensure_month_synthesis(items, analyzed_dir) |
| 394 | |
| 395 | crosslinks = _build_monthly_crosslinks(year, month, items) |
| 396 | synthesis_text = crosslinks + "\n" + synthesis.narrative |
| 397 | if synthesis.trend_arc: |
| 398 | synthesis_text += f"\n\n### Trend Arc\n\n{synthesis.trend_arc}" |
| 399 | page_entries["Month Synthesis"].append( |
| 400 | RollupEntry(marker="month-synthesis", text=synthesis_text) |
| 401 | ) |
| 402 | |
| 403 | for item in items: |
| 404 | tags_counter.update(item.tags) |
| 405 | for section, entry in monthly_entries(item, tags_counter).items(): |
| 406 | page_entries[section].append(entry) |
| 407 | |
| 408 | title = generate_monthly_title(synthesis, month, year) |
| 409 | |
| 410 | pages.append( |
| 411 | RollupPage( |
| 412 | path=content_root / "monthly" / str(year) / f"{month:02d}.md", |
| 413 | frontmatter={ |
| 414 | "title": title, |
| 415 | "date": items[-1].date.isoformat(), |
| 416 | "month": month, |
| 417 | "year": year, |
| 418 | "categories": ["monthly"], |
| 419 | "weeks_covered": [item.week for item in items], |
| 420 | "total_repos_featured": len( |
| 421 | {repo for item in items for repo in item.featured_repos} |
| 422 | ), |
| 423 | "summary": synthesis.summary, |
| 424 | "themes": list(synthesis.themes), |
| 425 | "persistent_themes": list(synthesis.persistent_themes), |
| 426 | "accelerating_themes": list(synthesis.accelerating_themes), |
| 427 | "weakening_themes": list(synthesis.weakening_themes), |
| 428 | "key_gaps": list(synthesis.key_gaps), |
| 429 | "top_repos": list(synthesis.top_repos), |
| 430 | }, |
| 431 | sections=page_entries, |
| 432 | section_order=MONTHLY_SECTIONS, |
| 433 | replace_sections=frozenset({"Month Synthesis"}), |
| 434 | ) |
| 435 | ) |
| 436 | return pages |
| 437 | |
| 438 | |
| 439 | def build_yearly_pages(summaries: list[WeeklySummary], content_root: Path) -> list[RollupPage]: |
| 440 | pages: list[RollupPage] = [] |
| 441 | target_years = sorted({summary.year for summary in summaries}) |
| 442 | for page in build_yearly_narrative_pages(content_root, target_years): |
| 443 | pages.append( |
| 444 | RollupPage( |
| 445 | path=page.path, |
| 446 | frontmatter=page.frontmatter, |
| 447 | sections={ |
| 448 | "Year in Review": [ |
| 449 | RollupEntry(marker=f"{page.year}-year-in-review", text=page.narrative) |
| 450 | ], |
| 451 | }, |
| 452 | section_order=YEARLY_SECTIONS, |
| 453 | replace_existing_sections=True, |
| 454 | preserve_unknown_sections=False, |
| 455 | ) |
| 456 | ) |
| 457 | return pages |
| 458 | |
| 459 | |
| 460 | def merge_sections( |
| 461 | path: Path, |
| 462 | section_order: list[str], |
| 463 | new_entries: dict[str, list[RollupEntry]], |
| 464 | *, |
| 465 | replace_existing_sections: bool = False, |
| 466 | preserve_unknown_sections: bool = True, |
| 467 | replace_sections: frozenset[str] = frozenset(), |
| 468 | ) -> str: |
| 469 | intro = "" |
| 470 | existing_sections: dict[str, str] = {} |
| 471 | if path.exists(): |
| 472 | existing_text = path.read_text(encoding="utf-8") |
| 473 | try: |
| 474 | _, body = analysis_gate.extract_frontmatter(existing_text) |
| 475 | except ValueError: |
| 476 | body = "" |
| 477 | intro, existing_sections = split_sections(body) |
| 478 | |
| 479 | rendered_sections: list[str] = [] |
| 480 | for section in section_order: |
| 481 | if replace_existing_sections or section in replace_sections: |
| 482 | content = "" |
| 483 | else: |
| 484 | content = existing_sections.get(section, "") |
| 485 | if content.strip() == NO_UPDATES_PLACEHOLDER: |
| 486 | content = "" |
| 487 | for entry in new_entries[section]: |
| 488 | if entry.marker in content: |
| 489 | continue |
| 490 | content = f"{content.rstrip()}\n\n{entry.text}" if content.strip() else entry.text |
| 491 | section_body = content.strip() or NO_UPDATES_PLACEHOLDER |
| 492 | rendered_sections.append(f"## {section}\n\n{section_body}") |
| 493 | |
| 494 | if preserve_unknown_sections: |
| 495 | for section, content in existing_sections.items(): |
| 496 | if section in section_order: |
| 497 | continue |
| 498 | section_body = content.strip() or NO_UPDATES_PLACEHOLDER |
| 499 | rendered_sections.append(f"## {section}\n\n{section_body}") |
| 500 | |
| 501 | if intro.strip(): |
| 502 | return intro.rstrip() + "\n\n" + "\n\n".join(rendered_sections) + "\n" |
| 503 | return "\n\n".join(rendered_sections) + "\n" |
| 504 | |
| 505 | |
| 506 | def write_rollup(page: RollupPage) -> None: |
| 507 | page.path.parent.mkdir(parents=True, exist_ok=True) |
| 508 | body = merge_sections( |
| 509 | page.path, |
| 510 | page.section_order, |
| 511 | page.sections, |
| 512 | replace_existing_sections=page.replace_existing_sections, |
| 513 | preserve_unknown_sections=page.preserve_unknown_sections, |
| 514 | replace_sections=page.replace_sections, |
| 515 | ) |
| 516 | page.path.write_text(render_frontmatter(page.frontmatter) + body, encoding="utf-8") |
| 517 | |
| 518 | |
| 519 | def generate_rollups(analyzed_dir: Path, content_root: Path) -> list[Path]: |
| 520 | summaries = load_weekly_summaries(analyzed_dir) |
| 521 | if not summaries: |
| 522 | return [] |
| 523 | |
| 524 | written: list[Path] = [] |
| 525 | monthly_pages = build_monthly_pages(summaries, content_root, analyzed_dir) |
| 526 | for page in monthly_pages: |
| 527 | write_rollup(page) |
| 528 | written.append(page.path) |
| 529 | |
| 530 | for page in build_yearly_pages(summaries, content_root): |
| 531 | write_rollup(page) |
| 532 | written.append(page.path) |
| 533 | return written |
| 534 | |
| 535 | |
| 536 | ROLLING_SECTIONS = [ |
| 537 | "Active Trends", |
| 538 | "Trend Velocity", |
| 539 | "Open Predictions", |
| 540 | "Noise Patterns", |
| 541 | ] |
| 542 | |
| 543 | VELOCITY_LABELS = ("accelerating", "new", "decelerating", "dying") |
| 544 | |
| 545 | |
| 546 | def _load_weekly_content(content_root: Path) -> list[WeeklySummary]: |
| 547 | """Load weekly summaries directly from content/weekly/ (published pages). |
| 548 | |
| 549 | Falls back to data/analyzed/ format. Handles content/weekly files that |
| 550 | may lack 'year' frontmatter by extracting it from the 'week' field. |
| 551 | """ |
| 552 | weekly_dir = content_root / "weekly" |
| 553 | if not weekly_dir.exists(): |
| 554 | return [] |
| 555 | paths = sorted(weekly_dir.rglob("W*.md")) |
| 556 | summaries: list[WeeklySummary] = [] |
| 557 | for path in paths: |
| 558 | if path.name == "_index.md": |
| 559 | continue |
| 560 | try: |
| 561 | summaries.append(_load_weekly_summary(path)) |
| 562 | except (RollupError, KeyError, ValueError): |
| 563 | continue |
| 564 | return sorted(summaries, key=lambda s: (s.year, s.week_number)) |
| 565 | |
| 566 | |
| 567 | def _load_weekly_summary(path: Path) -> WeeklySummary: |
| 568 | """Load a weekly summary from content/weekly/, tolerating missing 'year'.""" |
| 569 | frontmatter, body = analysis_gate.extract_frontmatter(path.read_text(encoding="utf-8")) |
| 570 | week = str(frontmatter.get("week", "")) |
| 571 | match = WEEK_PATTERN.fullmatch(week) |
| 572 | if not match: |
| 573 | raise RollupError(f"Invalid week slug in {path.name}") |
| 574 | |
| 575 | year = int(frontmatter.get("year", match.group("year"))) |
| 576 | date = analysis_gate.parse_datetime(frontmatter["date"]) |
| 577 | |
| 578 | signal_noise_section = get_section_text(body, "Signal & Noise") |
| 579 | if signal_noise_section: |
| 580 | signal = normalize_text(signal_noise_section) |
| 581 | noise = "" |
| 582 | else: |
| 583 | trend_analysis = get_section_text(body, "Trend Analysis") |
| 584 | signal = get_subsection_text(trend_analysis, "Signal") |
| 585 | noise = get_subsection_text(trend_analysis, "Noise") |
| 586 | |
| 587 | # Pull structured noise from frontmatter if available (newer format) |
| 588 | signal_noise_fm = frontmatter.get("signal_noise") |
| 589 | if signal_noise_fm and isinstance(signal_noise_fm, dict): |
| 590 | noise_items = signal_noise_fm.get("noise", []) |
| 591 | if noise_items and isinstance(noise_items, list): |
| 592 | noise = "; ".join(str(n) for n in noise_items) |
| 593 | |
| 594 | blind_spots = get_section_text(body, "Blind Spots") |
| 595 | if blind_spots: |
| 596 | gaps = normalize_text(blind_spots) |
| 597 | else: |
| 598 | gaps = get_subsection_text(get_section_text(body, "What's Missing"), "Gaps") |
| 599 | |
| 600 | week_ahead = get_section_text(body, "The Week Ahead") |
| 601 | if week_ahead: |
| 602 | conclusion = normalize_text(week_ahead) |
| 603 | else: |
| 604 | conclusion = normalize_text(get_section_text(body, "Conclusion")) |
| 605 | |
| 606 | top_repo = str(frontmatter.get("top_repo", "")) |
| 607 | |
| 608 | return WeeklySummary( |
| 609 | source_path=path, |
| 610 | title=str(frontmatter.get("title", "")), |
| 611 | date=date, |
| 612 | week=week, |
| 613 | year=year, |
| 614 | month=date.month, |
| 615 | tags=tuple(str(tag) for tag in frontmatter.get("tags", [])), |
| 616 | repos_featured=int(frontmatter.get("repos_featured", 0)), |
| 617 | top_repo=top_repo, |
| 618 | featured_repos=extract_featured_repos(body, top_repo) if top_repo else (), |
| 619 | summary=normalize_text(str(frontmatter.get("summary", ""))), |
| 620 | signal=signal, |
| 621 | noise=noise, |
| 622 | gaps=gaps, |
| 623 | conclusion=conclusion, |
| 624 | ) |
| 625 | |
| 626 | |
| 627 | def _classify_velocity(tags_by_week: list[tuple[str, set[str]]]) -> dict[str, str]: |
| 628 | """Classify trend velocity based on tag presence across weeks. |
| 629 | |
| 630 | Returns a mapping of tag -> velocity label. |
| 631 | """ |
| 632 | if len(tags_by_week) < 2: |
| 633 | all_tags = set() |
| 634 | for _, tags in tags_by_week: |
| 635 | all_tags.update(tags) |
| 636 | return {tag: "new" for tag in all_tags} |
| 637 | |
| 638 | all_tags: set[str] = set() |
| 639 | for _, tags in tags_by_week: |
| 640 | all_tags.update(tags) |
| 641 | |
| 642 | velocity: dict[str, str] = {} |
| 643 | for tag in sorted(all_tags): |
| 644 | presence = [tag in tags for _, tags in tags_by_week] |
| 645 | first_seen = next((i for i, p in enumerate(presence) if p), len(presence)) |
| 646 | last_seen = next( |
| 647 | (len(presence) - 1 - i for i, p in enumerate(reversed(presence)) if p), |
| 648 | 0, |
| 649 | ) |
| 650 | |
| 651 | if first_seen >= len(presence) - 1: |
| 652 | velocity[tag] = "new" |
| 653 | elif last_seen < len(presence) - 2: |
| 654 | velocity[tag] = "dying" |
| 655 | elif sum(presence[len(presence) // 2 :]) >= sum(presence[: len(presence) // 2]): |
| 656 | velocity[tag] = "accelerating" |
| 657 | else: |
| 658 | velocity[tag] = "decelerating" |
| 659 | return velocity |
| 660 | |
| 661 | |
| 662 | def _synthesize_rolling_report(summaries: list[WeeklySummary]) -> str: |
| 663 | """Synthesize a compact rolling report from up to 4 weekly summaries.""" |
| 664 | week_labels = [f"W{s.week_number}" for s in summaries] |
| 665 | current_week = summaries[-1].week |
| 666 | |
| 667 | # Frontmatter |
| 668 | lines = [ |
| 669 | "---", |
| 670 | "title: Rolling 4-Week Context", |
| 671 | f"updated: {current_week}", |
| 672 | f"weeks: [{', '.join(week_labels)}]", |
| 673 | "---", |
| 674 | "", |
| 675 | ] |
| 676 | |
| 677 | # Active Trends: synthesize from signals across weeks |
| 678 | lines.append("## Active Trends") |
| 679 | lines.append("") |
| 680 | seen_signals: list[str] = [] |
| 681 | for s in summaries: |
| 682 | if s.signal and s.signal not in seen_signals: |
| 683 | # Truncate long signals to keep report compact |
| 684 | signal_text = s.signal[:200] + "…" if len(s.signal) > 200 else s.signal |
| 685 | seen_signals.append(signal_text) |
| 686 | # Keep only most recent/relevant signals (last 4-5) |
| 687 | for signal in seen_signals[-5:]: |
| 688 | lines.append(f"- {signal}") |
| 689 | lines.append("") |
| 690 | |
| 691 | # Trend Velocity: classify tags by presence pattern |
| 692 | lines.append("## Trend Velocity") |
| 693 | lines.append("") |
| 694 | tags_by_week = [(s.week, set(s.tags)) for s in summaries] |
| 695 | velocity = _classify_velocity(tags_by_week) |
| 696 | for label in VELOCITY_LABELS: |
| 697 | tags_for_label = [t for t, v in velocity.items() if v == label] |
| 698 | if tags_for_label: |
| 699 | lines.append(f"- **{label.capitalize()}:** {', '.join(tags_for_label)}") |
| 700 | lines.append("") |
| 701 | |
| 702 | # Open Predictions: synthesize from conclusions/gaps |
| 703 | lines.append("## Open Predictions") |
| 704 | lines.append("") |
| 705 | for s in summaries[-3:]: |
| 706 | if s.gaps: |
| 707 | gap_text = s.gaps[:150] + "…" if len(s.gaps) > 150 else s.gaps |
| 708 | lines.append(f"- [W{s.week_number}] {gap_text}") |
| 709 | lines.append("") |
| 710 | |
| 711 | # Noise Patterns: synthesize from noise fields |
| 712 | lines.append("## Noise Patterns") |
| 713 | lines.append("") |
| 714 | seen_noise: list[str] = [] |
| 715 | for s in summaries: |
| 716 | if s.noise and s.noise not in seen_noise: |
| 717 | noise_text = s.noise[:200] + "…" if len(s.noise) > 200 else s.noise |
| 718 | seen_noise.append(noise_text) |
| 719 | for noise in seen_noise[-4:]: |
| 720 | lines.append(f"- {noise}") |
| 721 | if not seen_noise: |
| 722 | lines.append("- No distinct noise patterns isolated in structured data.") |
| 723 | lines.append("") |
| 724 | |
| 725 | return "\n".join(lines) |
| 726 | |
| 727 | |
| 728 | def generate_rolling_report(content_root: Path) -> Path | None: |
| 729 | """Generate a rolling last-4-weeks report from content/weekly/. |
| 730 | |
| 731 | Reads the most recent 4 weekly summaries, synthesizes them into a compact |
| 732 | context report, and writes to content/rolling/last-month.md (overwritten |
| 733 | each week). |
| 734 | |
| 735 | Returns the path written, or None if insufficient data. |
| 736 | """ |
| 737 | summaries = _load_weekly_content(content_root) |
| 738 | if not summaries: |
| 739 | return None |
| 740 | |
| 741 | # Take the last 4 weeks |
| 742 | recent = summaries[-4:] |
| 743 | if not recent: |
| 744 | return None |
| 745 | |
| 746 | report = _synthesize_rolling_report(recent) |
| 747 | |
| 748 | rolling_dir = content_root / "rolling" |
| 749 | rolling_dir.mkdir(parents=True, exist_ok=True) |
| 750 | output_path = rolling_dir / "last-month.md" |
| 751 | output_path.write_text(report, encoding="utf-8") |
| 752 | return output_path |
| 753 | |
| 754 | |
| 755 | def main(argv: list[str] | None = None) -> int: |
| 756 | args = parse_args(argv) |
| 757 | written = generate_rollups(args.analyzed_dir, args.content_root) |
| 758 | |
| 759 | if args.rolling: |
| 760 | rolling_path = generate_rolling_report(args.content_root) |
| 761 | if rolling_path: |
| 762 | written.append(rolling_path) |
| 763 | else: |
| 764 | print("No weekly content found for rolling report.", file=sys.stderr) |
| 765 | |
| 766 | if not written: |
| 767 | print( |
| 768 | f"No weekly summaries found in {args.analyzed_dir}; skipping rollup generation.", |
| 769 | file=sys.stderr, |
| 770 | ) |
| 771 | return 0 |
| 772 | for path in written: |
| 773 | print(f"Generated {path}") |
| 774 | return 0 |
| 775 | |
| 776 | |
| 777 | if __name__ == "__main__": |
| 778 | raise SystemExit(main()) |