mirror of
https://github.com/nubenetes/awesome-kubernetes.git
synced 2026-07-19 21:29:56 +00:00
b321169459
Inventory entries can have stars=null or discovered_at=null. Replace
.get("stars", 0) with .get("stars") or 0 in all three modules so that
None values are safely coerced to 0/"" before numeric/string comparison.
Fixes TypeError crashes in:
- v2_optimizer.py:632 _calculate_tags
- dedup.py:74,108 title dedup and entry scoring
- news_digest.py:322 category pool sorting
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
419 lines
16 KiB
Python
419 lines
16 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
import json
|
||
import asyncio
|
||
from datetime import datetime, timedelta
|
||
from typing import Dict, List, Any
|
||
|
||
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"
|
||
|
||
|
||
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": [],
|
||
"España": [],
|
||
"Asia-Pacific": [],
|
||
}
|
||
|
||
GEO_CATEGORIES: Dict[str, str] = {
|
||
"Americas": "americas",
|
||
"Europe": "europe",
|
||
"España": "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) -> str | None:
|
||
"""Return the geo digest category if ``geo_region`` matches."""
|
||
region = entry.get("geo_region", "")
|
||
for geo_name, geo_val in self.GEO_CATEGORIES.items():
|
||
if region == geo_val:
|
||
return geo_name
|
||
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]
|
||
]
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Core generation loop #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
async def generate_digest(self) -> dict:
|
||
"""Generate the full digest for all categories and time periods.
|
||
|
||
Returns a nested dict::
|
||
|
||
{
|
||
"3_months": { "Kubernetes & Orchestration": [...], ... },
|
||
"6_months": { ... },
|
||
"12_months": { ... }
|
||
}
|
||
"""
|
||
digest: Dict[str, Dict[str, List[dict]]] = {}
|
||
|
||
for period_name, days in self.PERIODS.items():
|
||
cutoff = (
|
||
datetime.now(MADRID_TZ) - timedelta(days=days)
|
||
).isoformat()
|
||
digest[period_name] = {}
|
||
|
||
# Bucket entries into their digest categories
|
||
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
|
||
|
||
# Tech / topic category
|
||
cat = self._get_entry_category(entry)
|
||
if cat:
|
||
category_pools.setdefault(cat, []).append(
|
||
dict(entry, url=url)
|
||
)
|
||
|
||
# Geo category (entry may belong to both)
|
||
geo = self._get_entry_geo(entry)
|
||
if geo:
|
||
category_pools.setdefault(geo, []).append(
|
||
dict(entry, url=url)
|
||
)
|
||
|
||
# Rank each category pool
|
||
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)
|
||
|
||
if len(entries) < 3:
|
||
digest[period_name][cat_name] = self._fallback_items(
|
||
entries, cat_name, limit=max_items
|
||
)
|
||
continue
|
||
|
||
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]
|
||
log_event(
|
||
f" [Digest] {period_name}/{cat_name}: "
|
||
f"{len(ranked)} items ranked"
|
||
)
|
||
|
||
except Exception as exc:
|
||
log_event(
|
||
f" [Digest WARN] {period_name}/{cat_name}: "
|
||
f"Gemini failed ({str(exc)[:80]}), "
|
||
"using star-based fallback"
|
||
)
|
||
digest[period_name][cat_name] = self._fallback_items(
|
||
entries, cat_name, limit=max_items
|
||
)
|
||
|
||
# Respect Gemini rate limits
|
||
await asyncio.sleep(1.0)
|
||
|
||
return digest
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Persistence #
|
||
# ------------------------------------------------------------------ #
|
||
|
||
@staticmethod
|
||
def save_digest(digest: dict) -> None:
|
||
"""Serialise *digest* 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)
|
||
log_event(f"[Digest] Saved to {DIGEST_OUTPUT_PATH}")
|
||
|
||
|
||
# ====================================================================== #
|
||
# 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())
|