mirror of
https://github.com/nubenetes/awesome-kubernetes.git
synced 2026-07-13 18:30:44 +00:00
48e8f71b6d
_get_entry_geo() called entry.get("url","") before the url key was
injected into the dict, so TLD inference always received an empty
string and all 304 geo-eligible entries were silently dropped.
Fix: add url= parameter to _get_entry_geo and pass the dict key from
generate_digest. Also renames España→Spain in DIGEST_CATEGORIES,
GEO_CATEGORIES, TLD map, and v2_optimizer geo_cats list.
news_digest.json now includes Americas/Europe/Spain/Asia-Pacific data
(star-based fallback; Gemini will re-rank on next Monday CI run).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
571 lines
23 KiB
Python
571 lines
23 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
import json
|
||
import hashlib
|
||
import asyncio
|
||
from datetime import datetime, timedelta
|
||
from typing import Dict, List, Any, Tuple
|
||
|
||
from src.inventory_manager import load_inventory
|
||
from src.gemini_utils import call_gemini_with_retry
|
||
from src.config import MADRID_TZ
|
||
from src.logger import log_event
|
||
|
||
DIGEST_OUTPUT_PATH = "data/news_digest.json"
|
||
|
||
# Refresh a cell even when hash matches if last_analyzed is older than this.
|
||
# Ensures Gemini rankings stay fresh even for stable categories.
|
||
MAX_STALENESS_DAYS = 30
|
||
|
||
|
||
class NewsDigestEngine:
|
||
"""Generates a curated news digest by filtering inventory entries by
|
||
recency and using Gemini AI to rank the most relevant ones per category.
|
||
|
||
Three time windows (3 / 6 / 12 months) are produced in a single run.
|
||
Each window contains up to 10 AI-ranked items per digest category.
|
||
"""
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# 26 Digest Categories – mapped from V2 category slugs #
|
||
# ------------------------------------------------------------------ #
|
||
DIGEST_CATEGORIES: Dict[str, List[str]] = {
|
||
# --- TECH CORE (9) ---
|
||
"Kubernetes & Orchestration": [
|
||
"kubernetes", "kubernetes-tools", "kubernetes-tutorials",
|
||
"kubectl-commands", "kubernetes-releases", "kubernetes-autoscaling",
|
||
"kubernetes-operators-controllers", "kubernetes-based-devel",
|
||
"kubernetes-alternatives", "kubernetes-client-libraries",
|
||
"kubernetes-bigdata", "managed-kubernetes-in-public-cloud", "helm",
|
||
],
|
||
"Containers & Runtime": [
|
||
"docker", "container-managers", "serverless", "noops", "registries",
|
||
],
|
||
"Networking & Service Mesh": [
|
||
"networking", "kubernetes-networking", "servicemesh", "istio",
|
||
"caching", "web-servers", "cloudflare",
|
||
],
|
||
"Architecture & Microservices": [
|
||
"introduction", "faq", "cloud-arch-diagrams", "matrix-table",
|
||
"other-awesome-lists", "about",
|
||
],
|
||
"Data, Messaging & Storage": [
|
||
"databases", "nosql", "newsql", "message-queue", "crunchydata",
|
||
"yaml", "kubernetes-storage", "kubernetes-backup-migrations",
|
||
],
|
||
"AI & Agents": [
|
||
"ai", "ai-agents-mcp", "chatgpt",
|
||
],
|
||
"MLOps & Data Science": [
|
||
"mlops",
|
||
],
|
||
"Python, Java & Developer Ecosystem": [
|
||
"python", "golang", "java_frameworks", "java_app_servers",
|
||
"java-and-java-performance-optimization", "javascript", "dotnet",
|
||
"angular", "react", "web3", "api",
|
||
"swagger-code-generator-for-rest-apis", "postman",
|
||
"lowcode-nocode", "devel-sites", "dom", "linux-dev-env",
|
||
"ChromeDevTools", "xamarin", "jvm-parameters-matrix-table",
|
||
"maven-gradle", "embedded-servlet-containers", "visual-studio",
|
||
],
|
||
"Linux & System Foundations": [
|
||
"linux", "git",
|
||
],
|
||
# --- PLATFORM & OPS (8) ---
|
||
"Security & Compliance": [
|
||
"securityascode", "kubernetes-security", "aws-security", "oauth",
|
||
"devsecops",
|
||
],
|
||
"Infrastructure as Code": [
|
||
"iac", "terraform", "pulumi", "crossplane", "ansible",
|
||
"kustomize", "chef", "liquibase",
|
||
],
|
||
"CI/CD & GitOps": [
|
||
"cicd", "gitops", "argo", "flux", "tekton", "jenkins",
|
||
"jenkins-alternatives", "sonarqube", "cicd-kubernetes-plugins",
|
||
"openshift-pipelines", "stackstorm", "keptn",
|
||
],
|
||
"Observability, SRE & Testing": [
|
||
"sre", "monitoring", "prometheus", "grafana",
|
||
"kubernetes-monitoring", "chaos-engineering", "qa",
|
||
"test-automation-frameworks", "testops",
|
||
"performance-testing-with-jenkins-and-jmeter",
|
||
"kubernetes-troubleshooting",
|
||
],
|
||
"DevOps & Culture": [
|
||
"devops", "devops-tools", "project-management-methodology",
|
||
"project-management-tools",
|
||
],
|
||
"Platform Engineering & DevEx": [
|
||
"developerportals", "scaffolding", "mkdocs",
|
||
],
|
||
"FinOps & Cloud Cost": [
|
||
"finops", "aws-pricing",
|
||
],
|
||
"Certification & Training": [
|
||
"elearning", "interview-questions", "aws-training", "cheatsheets",
|
||
"demos",
|
||
],
|
||
# --- CLOUD & ENTERPRISE (5) ---
|
||
"AWS": [
|
||
"aws", "aws-architecture", "aws-security", "aws-networking",
|
||
"aws-databases", "aws-storage", "aws-monitoring", "aws-iac",
|
||
"aws-tools-scripts", "aws-messaging", "aws-data", "aws-devops",
|
||
"aws-serverless", "aws-containers", "aws-backup",
|
||
"aws-newfeatures", "aws-miscellaneous", "aws-spain",
|
||
],
|
||
"Azure": [
|
||
"azure",
|
||
],
|
||
"GCP, OCI & Others": [
|
||
"GoogleCloudPlatform", "ibm_cloud", "oraclecloud",
|
||
"digitalocean", "scaleway", "edge-computing",
|
||
"public-cloud-solutions",
|
||
],
|
||
"OpenShift / Red Hat": [
|
||
"openshift", "ocp3", "ocp4", "openshift-pipelines", "rancher",
|
||
],
|
||
"Virtualization & Private Cloud": [
|
||
"kubernetes-on-premise", "kubernetes-alternatives",
|
||
"private-cloud-solutions",
|
||
],
|
||
# --- INDUSTRY / GEO (4) – resolved via geo_region, not slugs ---
|
||
"Americas": [],
|
||
"Europe": [],
|
||
"Spain": [],
|
||
"Asia-Pacific": [],
|
||
}
|
||
|
||
GEO_CATEGORIES: Dict[str, str] = {
|
||
"Americas": "americas",
|
||
"Europe": "europe",
|
||
"Spain": "spain",
|
||
"Asia-Pacific": "asia_pacific",
|
||
}
|
||
|
||
PERIODS: Dict[str, int] = {
|
||
"3_months": 90,
|
||
"6_months": 180,
|
||
"12_months": 365,
|
||
}
|
||
|
||
ITEMS_PER_PERIOD: Dict[str, int] = {
|
||
"3_months": 10,
|
||
"6_months": 15,
|
||
"12_months": 20,
|
||
}
|
||
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def __init__(self) -> None:
|
||
self.inventory: Dict[str, Any] = load_inventory()
|
||
|
||
# Reverse map: v2_category_slug -> digest_category_name
|
||
self.category_map: Dict[str, str] = {}
|
||
for digest_cat, slugs in self.DIGEST_CATEGORIES.items():
|
||
for slug in slugs:
|
||
self.category_map[slug] = digest_cat
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Classification helpers #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def _get_entry_category(self, entry: dict) -> str | None:
|
||
"""Determine which digest category an entry belongs to.
|
||
|
||
Checks ``v2_locations`` paths first (more specific), then falls
|
||
back to the ``category`` field.
|
||
"""
|
||
for loc in entry.get("v2_locations", []):
|
||
slug = loc.replace(".md", "")
|
||
if slug in self.category_map:
|
||
return self.category_map[slug]
|
||
cat = entry.get("category", "")
|
||
if cat in self.category_map:
|
||
return self.category_map[cat]
|
||
return None
|
||
|
||
def _get_entry_geo(self, entry: dict, url: str = "") -> str | None:
|
||
"""Return the geo digest category using geo_region field, falling back to URL TLD inference."""
|
||
region = entry.get("geo_region", "")
|
||
for geo_name, geo_val in self.GEO_CATEGORIES.items():
|
||
if region == geo_val:
|
||
return geo_name
|
||
# Fallback: infer from URL TLD (entry.get("url") populated only after url injection)
|
||
return self._infer_geo_from_url(entry.get("url", "") or url)
|
||
|
||
@staticmethod
|
||
def _infer_geo_from_url(url: str) -> str | None:
|
||
"""Infer geo category from URL TLD. Returns GEO_CATEGORIES key or None."""
|
||
try:
|
||
from urllib.parse import urlparse
|
||
host = urlparse(url).hostname or ""
|
||
# Ordered longest-first to avoid .uk matching before .co.uk
|
||
tld_to_region = [
|
||
(".com.au", "Asia-Pacific"), (".co.uk", "Europe"), (".co.jp", "Asia-Pacific"),
|
||
(".co.kr", "Asia-Pacific"), (".com.br", "Americas"), (".com.mx", "Americas"),
|
||
(".es", "Spain"), (".de", "Europe"), (".fr", "Europe"), (".it", "Europe"),
|
||
(".pt", "Europe"), (".nl", "Europe"), (".be", "Europe"), (".se", "Europe"),
|
||
(".dk", "Europe"), (".fi", "Europe"), (".no", "Europe"), (".ch", "Europe"),
|
||
(".at", "Europe"), (".pl", "Europe"), (".cz", "Europe"), (".uk", "Europe"),
|
||
(".ie", "Europe"), (".eu", "Europe"), (".cn", "Asia-Pacific"),
|
||
(".jp", "Asia-Pacific"), (".kr", "Asia-Pacific"), (".sg", "Asia-Pacific"),
|
||
(".in", "Asia-Pacific"), (".au", "Asia-Pacific"), (".nz", "Asia-Pacific"),
|
||
(".ca", "Americas"), (".mx", "Americas"), (".br", "Americas"),
|
||
]
|
||
for tld, region in tld_to_region:
|
||
if host.endswith(tld):
|
||
return region
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
@staticmethod
|
||
def _is_within_period(entry: dict, cutoff_iso: str) -> bool:
|
||
"""Check if entry falls within the time period using discovered_at,
|
||
with year field as fallback for backfilled entries."""
|
||
discovered = entry.get("discovered_at", "")
|
||
if discovered:
|
||
try:
|
||
if discovered >= cutoff_iso:
|
||
return True
|
||
except Exception:
|
||
pass
|
||
year = entry.get("year", "")
|
||
if year and isinstance(year, str) and year.isdigit():
|
||
cutoff_year = cutoff_iso[:4] if len(cutoff_iso) >= 4 else "2020"
|
||
return year >= cutoff_year
|
||
return False
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Prompt builder #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
@staticmethod
|
||
def _build_ranking_prompt(
|
||
category: str, entries: List[dict], period: str
|
||
) -> str:
|
||
"""Assemble the Gemini prompt that asks for a ranked TOP-10."""
|
||
period_label = period.replace("_", " ")
|
||
lines: List[str] = []
|
||
for i, e in enumerate(entries[:50]):
|
||
summary_fragment = (e.get("ai_summary", "") or "")[:200]
|
||
lines.append(
|
||
f'{i}. "{e.get("title", "Unknown")}" '
|
||
f'({e.get("url", "")}) | '
|
||
f'Stars: {e.get("stars", 0)} | '
|
||
f'Year: {e.get("year", "N/A")} | '
|
||
f'Summary: {summary_fragment}'
|
||
)
|
||
entries_text = "\n".join(lines)
|
||
|
||
return (
|
||
"You are a Senior Technical Curator for a Cloud Native "
|
||
"knowledge portal.\n"
|
||
f'Given these resources discovered in the last {period_label} '
|
||
f'for "{category}", select the TOP 10 most relevant.\n\n'
|
||
"SCORING CRITERIA:\n"
|
||
"- Industry Impact (30%): Does this change how teams "
|
||
"build/operate?\n"
|
||
"- Technical Novelty (25%): New capability, paradigm shift, "
|
||
"major release?\n"
|
||
"- Enterprise Adoption (20%): GA releases, production-ready?\n"
|
||
"- Community Signal (15%): CNCF graduations, major blog posts?\n"
|
||
"- Nubenetes Relevance (10%): Directly related to cloud native?\n\n"
|
||
'Respond ONLY JSON: {"items": [{"idx": int, '
|
||
'"impact": "critical|high|medium", '
|
||
'"why": "1 sentence explaining why this matters"}]}\n\n'
|
||
f"RESOURCES:\n{entries_text}"
|
||
)
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Star-based fallback (used when Gemini is unavailable / < 3 entries) #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
@staticmethod
|
||
def _fallback_items(
|
||
entries: List[dict], cat_name: str, limit: int = 10
|
||
) -> List[dict]:
|
||
"""Return up to *limit* items using a deterministic star-based
|
||
ranking (no AI call required)."""
|
||
return [
|
||
{
|
||
"url": e["url"],
|
||
"title": e.get("title", "Unknown"),
|
||
"date": e.get("discovered_at", "")[:10],
|
||
"stars": e.get("stars") or 0,
|
||
"impact": "high" if (e.get("stars") or 0) >= 4 else "medium",
|
||
"why": (e.get("ai_summary", "") or "")[:200],
|
||
"category": cat_name,
|
||
}
|
||
for e in entries[:limit]
|
||
]
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Incremental cache helpers #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
@staticmethod
|
||
def _compute_pool_hash(entries: List[dict]) -> str:
|
||
"""Stable fingerprint of a category entry pool.
|
||
|
||
Uses sorted URLs so the hash only changes when the actual set of
|
||
entries changes — not when their order or metadata is updated.
|
||
Returns first 16 hex chars (64-bit fingerprint, collision-safe).
|
||
"""
|
||
sorted_urls = sorted(e["url"] for e in entries if e.get("url"))
|
||
return hashlib.sha256("|".join(sorted_urls).encode()).hexdigest()[:16]
|
||
|
||
def _load_existing_digest(self) -> Tuple[dict, dict]:
|
||
"""Load existing digest + meta block from disk.
|
||
|
||
Returns ``(periods_dict, meta_dict)``. Both are empty dicts if the
|
||
file does not exist or cannot be parsed. The ``_meta`` key is
|
||
stripped from *periods_dict* so callers only see period data.
|
||
"""
|
||
if not os.path.exists(DIGEST_OUTPUT_PATH):
|
||
return {}, {}
|
||
try:
|
||
with open(DIGEST_OUTPUT_PATH, encoding="utf-8") as f:
|
||
raw = json.load(f)
|
||
meta = raw.pop("_meta", {})
|
||
return raw, meta
|
||
except Exception as e:
|
||
log_event(
|
||
f"[Digest WARN] Could not load existing digest: {str(e)[:100]}"
|
||
)
|
||
return {}, {}
|
||
|
||
def _is_cache_valid(
|
||
self, period: str, cat: str, pool_hash: str, meta: dict
|
||
) -> bool:
|
||
"""Return True when the stored result can be reused without Gemini.
|
||
|
||
Conditions (both must hold):
|
||
1. Entry-pool hash matches (same URLs, same period).
|
||
2. Result was analysed no more than MAX_STALENESS_DAYS ago.
|
||
"""
|
||
cell = meta.get(period, {}).get(cat)
|
||
if not cell:
|
||
return False
|
||
if cell.get("entry_hash") != pool_hash:
|
||
return False
|
||
last = cell.get("last_analyzed", "")
|
||
if not last:
|
||
return False
|
||
try:
|
||
age = (
|
||
datetime.now(MADRID_TZ) - datetime.fromisoformat(last)
|
||
).days
|
||
return age <= MAX_STALENESS_DAYS
|
||
except Exception:
|
||
return False
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Core generation loop #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
async def generate_digest(self) -> dict:
|
||
"""Incrementally generate the digest for all categories / periods.
|
||
|
||
For each (category, period) cell the engine checks whether the
|
||
inventory pool hash has changed since the last run. If the hash
|
||
is identical *and* the result is not stale, the existing ranked
|
||
list is reused with zero Gemini API calls. Gemini is only invoked
|
||
for cells whose entry pool has actually changed or exceeded
|
||
MAX_STALENESS_DAYS.
|
||
|
||
Returns a nested dict::
|
||
|
||
{
|
||
"3_months": { "Kubernetes & Orchestration": [...], ... },
|
||
"6_months": { ... },
|
||
"12_months": { ... },
|
||
"_meta": { ... } ← persistence metadata, internal use
|
||
}
|
||
"""
|
||
existing, old_meta = self._load_existing_digest()
|
||
new_meta: Dict[str, Dict[str, dict]] = {}
|
||
digest: Dict[str, Dict[str, List[dict]]] = {}
|
||
|
||
n_cached = n_refreshed = n_fallback = 0
|
||
|
||
for period_name, days in self.PERIODS.items():
|
||
cutoff = (
|
||
datetime.now(MADRID_TZ) - timedelta(days=days)
|
||
).isoformat()
|
||
digest[period_name] = {}
|
||
new_meta[period_name] = {}
|
||
|
||
# Build per-category entry pools for this period
|
||
category_pools: Dict[str, List[dict]] = {}
|
||
for url, entry in self.inventory.items():
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
if not self._is_within_period(entry, cutoff):
|
||
continue
|
||
cat = self._get_entry_category(entry)
|
||
if cat:
|
||
category_pools.setdefault(cat, []).append(
|
||
dict(entry, url=url)
|
||
)
|
||
geo = self._get_entry_geo(entry, url=url)
|
||
if geo:
|
||
category_pools.setdefault(geo, []).append(
|
||
dict(entry, url=url)
|
||
)
|
||
|
||
for cat_name, entries in category_pools.items():
|
||
entries.sort(
|
||
key=lambda x: (
|
||
x.get("stars") or 0,
|
||
x.get("discovered_at") or "",
|
||
),
|
||
reverse=True,
|
||
)
|
||
max_items = self.ITEMS_PER_PERIOD.get(period_name, 10)
|
||
pool_hash = self._compute_pool_hash(entries)
|
||
|
||
# ── Cache hit: reuse without calling Gemini ──
|
||
if self._is_cache_valid(period_name, cat_name, pool_hash, old_meta):
|
||
cached_items = existing.get(period_name, {}).get(cat_name, [])
|
||
if cached_items:
|
||
digest[period_name][cat_name] = cached_items
|
||
new_meta[period_name][cat_name] = (
|
||
old_meta[period_name][cat_name]
|
||
)
|
||
n_cached += 1
|
||
continue
|
||
|
||
# ── Fewer than 3 entries: star-based fallback (no AI) ──
|
||
if len(entries) < 3:
|
||
digest[period_name][cat_name] = self._fallback_items(
|
||
entries, cat_name, limit=max_items
|
||
)
|
||
new_meta[period_name][cat_name] = {
|
||
"last_analyzed": datetime.now(MADRID_TZ).isoformat(),
|
||
"entry_hash": pool_hash,
|
||
"entry_count": len(entries),
|
||
"method": "fallback_small",
|
||
}
|
||
n_fallback += 1
|
||
continue
|
||
|
||
# ── Gemini ranking ──
|
||
try:
|
||
prompt = self._build_ranking_prompt(
|
||
cat_name, entries, period_name
|
||
)
|
||
result = await call_gemini_with_retry(
|
||
prompt, prefer_flash=True, role="Digest-Analyst"
|
||
)
|
||
ranked: List[dict] = []
|
||
for item in result.get("items", []):
|
||
idx = int(item.get("idx", -1))
|
||
if 0 <= idx < len(entries):
|
||
e = entries[idx]
|
||
ranked.append(
|
||
{
|
||
"url": e["url"],
|
||
"title": e.get("title", "Unknown"),
|
||
"date": e.get("discovered_at", "")[:10],
|
||
"stars": e.get("stars", 0),
|
||
"impact": item.get("impact", "medium"),
|
||
"why": item.get("why", ""),
|
||
"category": cat_name,
|
||
}
|
||
)
|
||
digest[period_name][cat_name] = ranked[:max_items]
|
||
new_meta[period_name][cat_name] = {
|
||
"last_analyzed": datetime.now(MADRID_TZ).isoformat(),
|
||
"entry_hash": pool_hash,
|
||
"entry_count": len(entries),
|
||
"method": "gemini",
|
||
}
|
||
n_refreshed += 1
|
||
log_event(
|
||
f" [Digest] {period_name}/{cat_name}: "
|
||
f"{len(ranked)} items ranked (Gemini)"
|
||
)
|
||
except Exception as exc:
|
||
log_event(
|
||
f" [Digest WARN] {period_name}/{cat_name}: "
|
||
f"Gemini failed ({str(exc)[:80]}), using fallback"
|
||
)
|
||
digest[period_name][cat_name] = self._fallback_items(
|
||
entries, cat_name, limit=max_items
|
||
)
|
||
new_meta[period_name][cat_name] = {
|
||
"last_analyzed": datetime.now(MADRID_TZ).isoformat(),
|
||
"entry_hash": pool_hash,
|
||
"entry_count": len(entries),
|
||
"method": "fallback_error",
|
||
}
|
||
n_fallback += 1
|
||
|
||
await asyncio.sleep(1.0)
|
||
|
||
log_event(
|
||
f"[Digest] {n_cached} cells reused (0 API calls), "
|
||
f"{n_refreshed} refreshed via Gemini, "
|
||
f"{n_fallback} fallback"
|
||
)
|
||
new_meta["last_updated"] = datetime.now(MADRID_TZ).isoformat()
|
||
digest["_meta"] = new_meta
|
||
return digest
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Persistence #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
@staticmethod
|
||
def save_digest(digest: dict) -> None:
|
||
"""Serialise digest (including ``_meta``) to ``data/news_digest.json``."""
|
||
os.makedirs(os.path.dirname(DIGEST_OUTPUT_PATH), exist_ok=True)
|
||
with open(DIGEST_OUTPUT_PATH, "w", encoding="utf-8") as fh:
|
||
json.dump(digest, fh, indent=2, ensure_ascii=False)
|
||
# Count only period keys for the total-items stat
|
||
total_items = sum(
|
||
len(items)
|
||
for key, period in digest.items()
|
||
if key != "_meta" and isinstance(period, dict)
|
||
for items in period.values()
|
||
if isinstance(items, list)
|
||
)
|
||
gemini_cells = sum(
|
||
1
|
||
for period_meta in digest.get("_meta", {}).values()
|
||
if isinstance(period_meta, dict)
|
||
for cell in period_meta.values()
|
||
if isinstance(cell, dict) and cell.get("method") == "gemini"
|
||
)
|
||
log_event(
|
||
f"[Digest] Saved to {DIGEST_OUTPUT_PATH} — "
|
||
f"{total_items} items, {gemini_cells} Gemini calls this run"
|
||
)
|
||
|
||
|
||
# ====================================================================== #
|
||
# CLI / CI entry point #
|
||
# ====================================================================== #
|
||
|
||
|
||
async def run_news_digest() -> None:
|
||
"""Entry point for the CI pipeline."""
|
||
log_event("STARTING NEWS DIGEST GENERATION", section_break=True)
|
||
engine = NewsDigestEngine()
|
||
digest = await engine.generate_digest()
|
||
engine.save_digest(digest)
|
||
|
||
total_items = sum(
|
||
len(items) for period in digest.values() for items in period.values()
|
||
)
|
||
log_event(
|
||
f"NEWS DIGEST COMPLETE: {total_items} total items across all periods"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(run_news_digest())
|