github/awesome-kubernetes
1
0
Fork 0
mirror of https://github.com/nubenetes/awesome-kubernetes.git synced 2026-07-13 02:10:15 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
ced355373445868d7ff5c5a47a2fe6ae2ce369a8
awesome-kubernetes/src/v2_optimizer.py

1400 lines
86 KiB
Python
Raw Blame History

import os
import re
import json
import hashlib
import asyncio
import yaml
import httpx
from datetime import datetime
from typing import List, Dict, Set, Any, Tuple
from src.config import GEMINI_API_KEYS, GH_TOKEN, TARGET_REPO, MADRID_TZ, INVENTORY_PATH
from src.gemini_utils import call_gemini_with_retry, normalize_url, clean_toc_text, get_github_activity, fetch_youtube_metadata
from src.logger import log_event
from src.inventory_manager import update_inventory_entry
def nuclear_strip(text: str) -> str:
"""Mandate 30: MD039 - Removes all leading/trailing whitespace including hidden unicode characters."""
if not text: return ""
# Purge all known whitespace characters (standard, non-breaking, thin, etc.)
text = re.sub(r'^[\s\u00a0\u200b\u1680\u180e\u2000-\u200a\u2028\u2029\u202f\u205f\u3000]+', '', text)
text = re.sub(r'[\s\u00a0\u200b\u1680\u180e\u2000-\u200a\u2028\u2029\u202f\u205f\u3000]+$', '', text)
return text.replace("==", "")
V1_DIR = "docs"
V2_DIR = "v2-docs"
class V2VisionEngine:
def __init__(self, render_only: bool = False):
self.render_only = render_only
# Load Config & Policy
self.special_assets_rules = self._load_special_assets()
self.link_rules = self._load_link_rules()
self.max_depth = self.link_rules.get("hierarchy_rules", {}).get("max_depth", 10)
# 100% Comprehensive 2026 Taxonomy
self.dimensions = {
"AI": ["ai", "ai-agents-mcp", "chatgpt", "mlops"],
"Architectural Foundations": ["introduction", "faq", "kubernetes", "linux", "git", "cloud-arch-diagrams", "matrix-table", "other-awesome-lists", "about"],
"Platform & Site Reliability": ["sre", "devops", "developerportals", "scaffolding", "finops", "chaos-engineering", "performance-testing-with-jenkins-and-jmeter", "project-management-methodology", "project-management-tools", "qa", "test-automation-frameworks", "testops"],
"Hardened Infrastructure": ["iac", "terraform", "pulumi", "crossplane", "ansible", "securityascode", "kubernetes-security", "aws-security", "oauth", "devsecops", "kustomize", "liquibase", "chef"],
"Cloud Providers (Hyperscalers)": ["aws", "azure", "GoogleCloudPlatform", "ibm_cloud", "oraclecloud", "digitalocean", "cloudflare", "scaleway", "managed-kubernetes-in-public-cloud", "public-cloud-solutions", "private-cloud-solutions", "edge-computing", "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-training", "aws-newfeatures", "aws-miscellaneous", "aws-pricing", "aws-spain"],
"Networking & Service Mesh": ["networking", "kubernetes-networking", "servicemesh", "istio", "caching", "web-servers", "cloudflare"],
"The Container Stack": ["docker", "container-managers", "serverless", "kubernetes-autoscaling", "kubernetes-operators-controllers", "kubernetes-storage", "kubernetes-monitoring", "kubernetes-troubleshooting", "kubernetes-backup-migrations", "kubernetes-on-premise", "kubernetes-bigdata", "kubernetes-client-libraries", "kubernetes-releases", "kubernetes-based-devel", "kubernetes-alternatives", "kubectl-commands", "rancher", "openshift", "ocp3", "ocp4", "noops"],
"Data & Advanced Analytics": ["databases", "nosql", "newsql", "message-queue", "crunchydata", "yaml", "bigdata"],
"Engineering Pipeline": ["cicd", "gitops", "argo", "flux", "tekton", "jenkins", "jenkins-alternatives", "openshift-pipelines", "sonarqube", "registries", "keptn", "stackstorm", "cicd-kubernetes-plugins"],
"Developer Ecosystem": ["visual-studio", "javascript", "golang", "python", "java_frameworks", "java_app_servers", "java-and-java-performance-optimization", "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"],
"Career & Industry": ["recruitment", "hr", "finops", "freelancing", "remote-tech-jobs", "workfromhome", "interview-questions", "elearning", "digital-money", "appointment-scheduling", "newsfeeds"]
}
self.library_criteria = (
"You are a Senior Technical Architect in 2026. Your mission is to organize a high-density technical reference portal "
"structured like a professional technical book (O'Reilly style).\n"
"PHASE 1: TECHNICAL PRESERVATION & CURATION\n"
"- KEEP >90% of technical resources (except for 'introduction.md' where only high-impact links are kept).\n"
"PHASE 2: SOPHISTICATED HIERARCHICAL CLASSIFICATION\n"
"- Identify TECHNICAL_HIERARCHY: A list of strings (max 10) representing Area > Topic > Subtopics.\n"
"- For 'introduction.md', identify links related to MICROSERVICES for extraction.\n"
"PHASE 3: KNOWLEDGE ASSIMILATION FLOW\n"
"- Order hierarchy to facilitate a structured learning journey.\n"
"PHASE 4: HIGH-DENSITY TECHNICAL SUMMARIES (Double-Evidence Synthesis)\n"
"- Generate professional, neutral, and advanced technical summaries. Style: O'Reilly technical.\n"
"- PROTOCOL: Contrast 'Curator Insight' (from source) with 'Live Grounding' (from search).\n"
"- If discrepancies are found (e.g. project is archived but source says it's new), PRIORITIZE live engineering truth.\n"
"- Summaries MUST be high-density: Include architectural value, key features, and technical significance.\n"
"- Format: Use paragraphs and bullet points for complex tools. Aim for 2-5 sentences of depth.\n"
"PHASE 5: ADVANCED MATURITY TAGGING\n"
"- Assign tags. You MUST include:\n"
" 1. 1 to 2 maturity tags from: [DE FACTO STANDARD], [ENTERPRISE-STABLE], [EMERGING], [GUIDE], [CASE STUDY], [COMMUNITY-TOOL], [LEGACY].\n"
" 2. Fine-grained technical/architectural tags from the content (e.g., [EBPF], [WASM], [GITOPS], [IAC], [SERVICE-MESH], [SERVERLESS], [MLOPS], [DB]). Keep them uppercase and wrapped in brackets.\n"
)
self.inventory = self._load_inventory()
self.maturity_audit = []
def _load_special_assets(self) -> Dict:
path = "data/special_assets.yaml"
if os.path.exists(path):
try: return yaml.safe_load(open(path, "r")) or {}
except: return {}
return {}
def _load_link_rules(self) -> Dict:
path = "data/link_rules.yaml"
if os.path.exists(path):
try: return yaml.safe_load(open(path, "r")) or {}
except: return {}
return {}
def _load_inventory(self) -> Dict:
from src.inventory_manager import load_inventory
return load_inventory()
def _save_inventory(self):
from src.inventory_manager import save_inventory
save_inventory(self.inventory)
async def analyze_and_cluster(self):
log_event("STARTING V2 HIGH-DENSITY O'REILLY LIBRARY GENERATION", section_break=True)
# Mandate 30: MD039 - Global Data Sanitization (Purge all whitespace/hidden chars from titles)
for url in list(self.inventory.keys()):
if isinstance(self.inventory[url], dict) and "title" in self.inventory[url]:
# Purge all known whitespace characters (standard, non-breaking, thin, etc.)
t = self.inventory[url]["title"]
t = re.sub(r'^[\s\u00a0\u200b\u1680\u180e\u2000-\u200a\u2028\u2029\u202f\u205f\u3000]+', '', t)
t = re.sub(r'[\s\u00a0\u200b\u1680\u180e\u2000-\u200a\u2028\u2029\u202f\u205f\u3000]+$', '', t)
self.inventory[url]["title"] = t
# 0. Mandate Sync
try:
from src.mandate_ingestor import MandateIngestor
MandateIngestor().save_system_instructions()
except: pass
all_v1_links, mosaic_html, videos_html = await self._gather_all_v1_content()
log_event(f"[*] Discovery: Found {len(all_v1_links)} resources to process.")
log_event("[*] Phase 1: Health Check...")
if self.render_only:
# Mandate 19/22: In render-only mode (Fast-Track), we are conservative to avoid pruning valid sections.
# We keep links that are explicitly 'online', 'review_required' OR have no status yet.
health_inventory = []
for l in all_v1_links:
entry = self.inventory.get(normalize_url(l["url"]), {})
status = entry.get("status", "online") # Assume online if unknown for rendering
if status in ["online", "review_required"]:
health_inventory.append(l)
else:
health_inventory = await self._verify_link_health(all_v1_links)
log_event("[*] Phase 2: Evaluation & Deep Indexing (Semantic Dedup)...")
library_inventory = await self._evaluate_and_score_resources(health_inventory)
log_event("[*] Phase 3: Recursive Hierarchy Construction...")
v2_data = await self._rebuild_structure(library_inventory)
log_event("[*] Phase 4: Generating Premium Portal Hubs...")
os.makedirs(V2_DIR, exist_ok=True)
# --- SURGICAL GARBAGE COLLECTION ---
# Track every file we generate
generated_files = {"index.md", "audit-log.md", "videos.md", "tags.md"}
for f_name in v2_data.keys():
generated_files.add(f_name)
await self._write_premium_files(v2_data, mosaic_html, videos_html)
await self._generate_global_tag_index(v2_data)
await self._sync_enterprise_navigation(v2_data)
# Delete only orphaned files
log_event("[*] Phase 5: Pruning Orphaned V2 Assets...")
for f in os.listdir(V2_DIR):
if f.endswith(".md") and f not in generated_files:
log_event(f" [DEL] Pruning obsolete V2 page: {f}")
os.remove(os.path.join(V2_DIR, f))
self._save_inventory()
# --- FINAL SAFETY AUDIT ---
try:
from src.safety_guard import SafetyGuard
guard = SafetyGuard()
report = guard.generate_audit_report()
with open("v2_safety_report.md", "w") as f: f.write(report)
except Exception as e:
log_event(f" [!] V2 Safety Audit Error: {e}")
log_event("V2 ELITE PORTAL GENERATED SUCCESSFULLY.")
async def _gather_all_v1_content(self):
all_links, mosaic_html, videos_html = [], "", ""
if os.path.exists("docs/index.md"):
with open("docs/index.md", "r") as f:
idx_content = f.read()
videos_match = re.search(r'\?\?\? note "Top Videos & Clips.*?\n\s+(<center.*?</center>)', idx_content, re.DOTALL)
if videos_match:
videos_html = videos_match.group(1)
# Dynamically generate V2 categorized mosaic from youtube_channels_mosaic.yaml
try:
from src.reorganize_mosaic import build_v2_mosaic_markdown
v2_mosaic_full = build_v2_mosaic_markdown("data/youtube_channels_mosaic.yaml")
mosaic_html = v2_mosaic_full.replace('<center markdown="1">', '').replace('</center>', '').strip()
except Exception as e:
log_event(f" [!] Error generating V2 mosaic dynamically: {e}")
mosaic_html = ""
for root, _, files in os.walk(V1_DIR):
for file in files:
if not file.endswith(".md") or file == "index.md": continue
path = os.path.join(root, file)
with open(path, "r") as f: content = f.read()
matches = re.finditer(r'^\s*-\s*(?:\*\*\(\d{4}\)\*\*\s+)?\[([^\]]+)\]\(([^\)]+)\)(.*?(?:\n\s{2,}.*)*)', content, re.MULTILINE)
for m in matches:
title, url, full_desc = m.groups()
if not url.startswith(("http", "mailto", "#")):
url = f"https://nubenetes.com/{url.replace('.md', '/')}"
# Mandate 30: MD039 - Strip all whitespace (including non-breaking space) from link text
all_links.append({"title": nuclear_strip(title), "url": url.strip(), "description": full_desc.strip(), "original_file": file})
return all_links, mosaic_html, videos_html
async def _verify_link_health(self, links: List[Dict]):
force_full = os.getenv("FORCE_FULL_CHECK", "false").lower() == "true"
fast_online = []
needs_check = []
for l in links:
nu = normalize_url(l["url"])
entry = self.inventory.get(nu, {})
# Mandate 32: skip links under review
if entry.get("status") == "review_required": continue
if not force_full and entry.get("status") == "online":
fast_online.append(l)
else:
needs_check.append(l)
if not needs_check: return fast_online
log_event(f" [>] Fast-Track Health: {len(fast_online)} | Network-Check: {len(needs_check)}")
online_links = list(fast_online)
total_needs = len(needs_check)
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True, verify=False) as client:
# Mandate 16/22: Resilient asynchronous health checks with increased concurrency
CHUNK_SIZE = 100
for i in range(0, total_needs, CHUNK_SIZE):
batch = needs_check[i:i+CHUNK_SIZE]
tasks = [self._check_single_link_resilient(client, l) for l in batch]
results = await asyncio.gather(*tasks)
online_links.extend([r for r in results if r is not None])
if i % 100 == 0:
log_event(f" [>] Progress: [{i}/{total_needs}] links validated over network...")
await asyncio.sleep(0.05) # Minimal delay
return online_links
async def _check_single_link_resilient(self, client, link: Dict):
url = link["url"]
norm_url = normalize_url(url)
entry = self.inventory.get(norm_url, {})
# Mandate 31: Skip links under review for V2 Elite
if entry.get("status") == "review_required":
log_event(f" [-] SKIPPING V2: {url} is under Review.")
return None
if entry.get("status") == "online" and os.getenv("FORCE_FULL_CHECK", "false").lower() != "true": return link
try:
resp = await client.get(url, timeout=10.0)
if resp.status_code < 400:
final_url = str(resp.url)
from src.gemini_utils import sanitize_trailing_slashes
final_url = sanitize_trailing_slashes(final_url)
# Update URL if it was redirected/normalized
if final_url != url:
link["url"] = final_url
self.inventory.setdefault(normalize_url(final_url), {})["status"] = "online"
# Mandate 22: Update last_checked for the inventory entry
self.inventory[normalize_url(final_url)]["last_checked"] = datetime.now().timestamp()
return link
except: pass
return None
async def _evaluate_and_score_resources(self, links: List[Dict]):
to_evaluate = []
project_registry = {}
force_eval = os.getenv("FORCE_EVAL", "false").lower() == "true"
force_full_check = os.getenv("FORCE_FULL_CHECK", "false").lower() == "true"
# Bypassing GitHub UI limitation: If force_eval or force_full_check is ON, we must enrich metadata
enrich_metadata = os.getenv("ENRICH_METADATA", "false").lower() == "true" or force_eval or force_full_check
special_files = [sa["file"] for sa in self.special_assets_rules.get("special_assets", [])]
# Mandate 47: Zero-Redundancy & Smart Grounding
from src.mandate_ingestor import get_system_mandates
dynamic_mandates = get_system_mandates()
# Mandate 15: Proactive Enrichment for V2 (GitHub metadata is critical for tags)
# Optimized: Parallel fetching with Semaphore to avoid sequential bottleneck
processed_gh_metadata = set()
gh_fetch_count = 0
gh_tasks = []
gh_sem = asyncio.Semaphore(30) # Increased for final sprint strategy
async def _fetch_gh_with_sem(url: str):
async with gh_sem:
return url, await get_github_activity(url)
for l in links:
norm_url = normalize_url(l["url"])
if "github.com" not in norm_url or self.render_only: continue
cached = self.inventory.get(norm_url, {})
# Mandate 43: Always ensure GH metadata for GitHub links in V2 to power [DE FACTO STANDARD] logic
if (enrich_metadata or cached.get("gh_stars") is None) and norm_url not in processed_gh_metadata:
processed_gh_metadata.add(norm_url)
gh_tasks.append(_fetch_gh_with_sem(norm_url))
if gh_tasks:
log_event(f" [METADATA] V2 Pulse: Batch fetching {len(gh_tasks)} GitHub profiles in parallel...", section_break=True)
gh_results = await asyncio.gather(*gh_tasks)
for norm_url, gh_data in gh_results:
if gh_data:
if norm_url not in self.inventory: self.inventory[norm_url] = {}
self.inventory[norm_url].update(gh_data)
gh_fetch_count += 1
# Periodic Save: Save once after the massive batch
from src.inventory_manager import save_inventory
save_inventory(self.inventory)
log_event(f" [💾] Inventory Persisted: {gh_fetch_count} metadata entries updated.")
for l in links:
item = l.copy()
norm_url = normalize_url(l["url"])
orig_file = l.get("original_file", "unknown.md")
is_special = orig_file in special_files
item["is_special"] = is_special
project_id = norm_url
if "github.com" in norm_url:
match = re.search(r'github\.com/([^/]+/[^/]+)', norm_url)
if match: project_id = match.group(1).lower()
# Reuse enriched metadata from inventory
if "github.com" in norm_url:
item.update(self.inventory.get(norm_url, {}))
if not force_eval and norm_url in self.inventory:
cached = self.inventory[norm_url]
item.update(cached)
if is_special: item["is_special"] = True
# Mandate 30: Hierarchy and AI Summaries are mandatory for ELITE AI curation.
# Optimized Skip Logic: Only skip if we already have BOTH hierarchy and a summary.
if ((cached.get("hierarchy") and cached.get("ai_summary")) or self.render_only) and not force_eval:
if project_id not in project_registry or item.get("stars", 0) > project_registry[project_id].get("stars", 0):
if project_id in project_registry and project_registry[project_id].get("is_special"): item["is_special"] = True
project_registry[project_id] = item
continue
to_evaluate.append(item)
if to_evaluate and (not self.render_only or force_eval):
# Mandate 47: Zero-Redundancy & Smart Grounding
# Fast-Track (Metadata/Desc present) vs Grounded-Track (Needs deep search)
fast_track = []
grounded_track = []
for l in to_evaluate:
nu = normalize_url(l["url"])
is_github = "github.com" in nu
# Fast-Track Eligibility:
# 1. Has AI summary (previous run)
# 2. Is GitHub and has stars (metadata present)
# 3. Has decent manual description (> 40 chars)
# 4. Is already in inventory (we have title/category context)
has_ai_summary = l.get("ai_summary") is not None and len(l.get("ai_summary")) > 50
has_stars = l.get("gh_stars") is not None
has_desc = len(l.get("description", "")) > 40
is_known = nu in self.inventory
if has_ai_summary or has_stars or has_desc or is_known:
fast_track.append(l)
else:
# Grounded-Track is ONLY for "Unknown" resources with zero context
grounded_track.append(l)
log_event(f"[*] Agent Phase 1: Analyst Evaluation ({len(to_evaluate)} resources)...")
log_event(f" [>] Fast-Track: {len(fast_track)} | Grounded-Track: {len(grounded_track)}")
analyst_results = []
# 1.1 Fast-Track: Large Batches, NO GROUNDING (Fast)
# Optimized: Parallel batch processing to leverage high-tier API quotas
BATCH_SIZE_FAST = 50
total_fast = len(fast_track)
fast_tasks = []
async def _process_fast_batch(batch_links, batch_idx, total_b):
log_event(f" [>] Fast-Track: Queuing Batch {batch_idx}/{total_b}...")
prompt = (
f"You are the Nubenetes Technical Analyst (2026).\n"
f"{dynamic_mandates}\n"
f"{self.library_criteria}\n"
"PHASE 5: TECHNICAL SYNTHESIS (FAST-TRACK)\n"
"- Use provided metadata, AI summaries, and descriptions to classify maturity.\n"
"Respond ONLY JSON: {{\"results\": [{{ \"idx\": int, \"year\": \"YYYY\", \"stars\": 0-5, \"hierarchy\": [\"Area\", \"Topic\", ...], \"tags\": [\"...\"], \"summary\": \"Synthesis...\", \"language\": \"...\", \"type\": \"...\", \"complexity\": \"...\", \"is_microservice\": bool }}, ...]}}\n\n"
"LINKS:\n" + "\n".join([f"{idx}. {l['title']} ({l['url']}) | Stars: {l.get('gh_stars', l.get('stars'))} | Existing Summary: {l.get('ai_summary', l.get('description'))}" for idx, l in enumerate(batch_links)])
)
try:
data = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=False, role="Analyst-Fast")
batch_results = []
for res in data.get("results", []):
idx = int(res["idx"])
if idx < len(batch_links):
item = batch_links[idx].copy()
eval_data = {
"year": str(res.get("year", "N/A")), "stars": min(max(int(res.get("stars", 0)), 0), 5),
"ai_summary": res.get("summary", item.get("ai_summary", "")),
"language": res.get("language", "English"),
"resource_type": res.get("type", "Reference"), "complexity": res.get("complexity", "Intermediate"),
"hierarchy": res.get("hierarchy", ["General"]), "tags": res.get("tags", []),
"is_microservice": bool(res.get("is_microservice", False)),
"status": "online", "is_special": item.get("is_special", False)
}
item.update(eval_data)
batch_results.append(item)
# Incremental Persistence
norm_url = normalize_url(item["url"])
from src.inventory_manager import update_inventory_entry
new_data = {k:v for k,v in item.items() if k not in ["url", "title", "original_file", "aliases"]}
new_data["title"] = item["title"]
update_inventory_entry(self.inventory, norm_url, new_data)
return batch_results
except Exception as e:
log_event(f" [!] Error in Fast-Batch {batch_idx}: {e}")
return batch_links # Fallback to original links (standard layer)
total_batches_fast = (total_fast + BATCH_SIZE_FAST - 1) // BATCH_SIZE_FAST
for i in range(0, total_fast, BATCH_SIZE_FAST):
batch = fast_track[i:i+BATCH_SIZE_FAST]
batch_num = (i // BATCH_SIZE_FAST) + 1
fast_tasks.append(_process_fast_batch(batch, batch_num, total_batches_fast))
if fast_tasks:
log_event(f"[*] Agent Phase 1.1: Dispatching {len(fast_tasks)} parallel batches...")
# Use as_completed to persist results incrementally during parallel execution
processed_count = 0
for task in asyncio.as_completed(fast_tasks):
r_list = await task
analyst_results.extend(r_list)
processed_count += 1
# Mandate 22: Save every 10 batches to disk to avoid data loss during 6h timeouts
if processed_count % 10 == 0:
log_event(f" [💾] Periodic Save: Persisting inventory after {processed_count} batches...")
from src.inventory_manager import save_inventory
save_inventory(self.inventory)
# Final Save
from src.inventory_manager import save_inventory
save_inventory(self.inventory)
log_event(f" [💾] Inventory Persisted after {len(analyst_results)} AI evaluations.")
await asyncio.sleep(2.0) # Safety delay to respect TPM limits
# 1.2 Grounded-Track: Small Batches, WITH GROUNDING (Slower but precise)
BATCH_SIZE_GROUNDED = 15 # Increased from 5
total_grounded = len(grounded_track)
for i in range(0, total_grounded, BATCH_SIZE_GROUNDED):
batch = grounded_track[i:i+BATCH_SIZE_GROUNDED]
batch_num = (i // BATCH_SIZE_GROUNDED) + 1
total_batches = (total_grounded + BATCH_SIZE_GROUNDED - 1) // BATCH_SIZE_GROUNDED
log_event(f" [🌟] Grounded-Track: Processing Batch {batch_num}/{total_batches} (Grounding active)...")
# MANDATE 25: Pre-enrich YouTube links with real metadata
enriched_batch = []
for item in batch:
url = item["url"]
if "youtube.com" in url or "youtu.be" in url:
log_event(f" [YT] Pre-fetching metadata for: {url}")
meta = await fetch_youtube_metadata(url)
if meta:
item["description"] = f"TITLE: {meta['raw_title']}\nDESCRIPTION: {meta['raw_description']}"
enriched_batch.append(item)
prompt = (
f"You are the Nubenetes Technical Analyst (2026).\n"
f"{dynamic_mandates}\n"
f"{self.library_criteria}\n"
"PHASE 5: DOUBLE-EVIDENCE SYNTHESIS & RICH SUMMARY (GROUNDED)\n"
"- Cross-reference provided title/desc with search grounding.\n"
"Respond ONLY JSON: {{\"results\": [{{ \"idx\": int, \"year\": \"YYYY\", \"stars\": 0-5, \"hierarchy\": [\"Area\", \"Topic\", ...], \"tags\": [\"...\"], \"summary\": \"Synthesis...\", \"language\": \"...\", \"type\": \"...\", \"complexity\": \"...\", \"is_microservice\": bool }}, ...]}}\n\n"
"LINKS:\n" + "\n".join([f"{idx}. {l['title']} ({l['url']}) | Input Context: {l.get('description', 'N/A')}" for idx, l in enumerate(enriched_batch)])
)
try:
data = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=True, role="Analyst-Grounded")
for res in data.get("results", []):
idx = int(res["idx"])
if idx < len(batch):
item = batch[idx].copy()
eval_data = {
"year": str(res.get("year", "N/A")), "stars": min(max(int(res.get("stars", 0)), 0), 5),
"ai_summary": res.get("summary", ""), "language": res.get("language", "English"),
"resource_type": res.get("type", "Reference"), "complexity": res.get("complexity", "Intermediate"),
"hierarchy": res.get("hierarchy", ["General"]), "tags": res.get("tags", []),
"is_microservice": bool(res.get("is_microservice", False)),
"status": "online", "is_special": item.get("is_special", False)
}
item.update(eval_data)
analyst_results.append(item)
except Exception:
for l in batch: analyst_results.append(l)
await asyncio.sleep(0.01 if os.environ.get("MOCK_DEBATE") == "true" else 4.0)
# --- AGENT PHASE 2: MULTI-AGENT CONSENSUS & DEBATE PROTOCOL ---
# Identify candidates for debate:
# 1. High-impact candidates (marked as [DE FACTO STANDARD] or [ENTERPRISE-STABLE])
# 2. Borderline candidates (stars == 3 or stars == 4)
debate_candidates = [
l for l in analyst_results
if "[DE FACTO STANDARD]" in l.get("tags", [])
or "[ENTERPRISE-STABLE]" in l.get("tags", [])
or l.get("stars", 0) in [3, 4]
]
if debate_candidates:
log_event(f"[*] Agent Phase 2: Multi-Agent Consensus & Debate Protocol ({len(debate_candidates)} candidates)...")
from src.v2_debate import run_debate_protocol
for item in debate_candidates:
try:
# Map current stars (0-5) to initial score (0-100)
item["impact_score"] = item.get("impact_score", item.get("stars", 3) * 20)
final_score, final_tags, refined_summary, debate_data = await run_debate_protocol(item)
# Update item with consensus results
item["stars"] = min(max(final_score // 20, 0), 5)
item["impact_score"] = final_score
item["tags"] = final_tags
item["ai_summary"] = refined_summary
item["debate_log"] = debate_data
except Exception as e:
log_event(f" [!] Debate failed for '{item.get('title')}': {e}")
await asyncio.sleep(0.01 if os.environ.get("MOCK_DEBATE") == "true" else 0.5)
# Finalize Registry
for item in analyst_results:
norm_url = normalize_url(item["url"])
p_id = norm_url
if "github.com" in norm_url:
m = re.search(r'github\.com/([^/]+/[^/]+)', norm_url)
if m: p_id = m.group(1).lower()
# Persist to inventory
new_data = {k:v for k,v in item.items() if k not in ["url", "title", "original_file", "aliases"]}
new_data["title"] = item["title"]
if "addition_method" not in self.inventory.get(norm_url, {}):
new_data["addition_method"] = "manual"
update_inventory_entry(self.inventory, norm_url, new_data)
if p_id not in project_registry or item.get("stars", 0) > project_registry[p_id].get("stars", 0):
if p_id in project_registry and project_registry[p_id].get("is_special"): item["is_special"] = True
project_registry[p_id] = item
return list(project_registry.values())
def _calculate_tags(self, item: Dict) -> List[str]:
"""
Mandate 40: Multi-Dimensional Tagging (1:N).
Merges AI-assigned tags with rule-based maturity signals to ensure high-fidelity classification.
Utilizes MCP-style grounding data (GitHub stars, resource types) to override generic defaults.
"""
# 0. Collect all possible tag sources
ai_tags = item.get("tags", [])
if isinstance(ai_tags, str): ai_tags = [ai_tags] # Resiliency
valid_set = {"[DE FACTO STANDARD]", "[ENTERPRISE-STABLE]", "[EMERGING]", "[GUIDE]", "[CASE STUDY]", "[COMMUNITY-TOOL]", "[LEGACY]"}
# Start with filtered AI tags and custom tech stack tags
tags = set()
for t in ai_tags:
if not isinstance(t, str): continue
t_stripped = t.strip()
if t_stripped in valid_set:
tags.add(t_stripped)
elif t_stripped.startswith("[") and t_stripped.endswith("]"):
inner = t_stripped[1:-1].strip()
if inner.isupper() and all(c.isalnum() or c in "_-" for c in inner):
tags.add(t_stripped)
# 1. GitHub Objective Reality (Mandate 43)
raw_gh = item.get("gh_stars", 0)
gh_stars = int(raw_gh) if str(raw_gh).isdigit() else 0
curator_stars = int(item.get("stars", 0))
if gh_stars > 15000 or curator_stars >= 5:
tags.add("[DE FACTO STANDARD]")
if "[COMMUNITY-TOOL]" in tags: tags.remove("[COMMUNITY-TOOL]")
elif gh_stars > 3000 or curator_stars >= 4:
tags.add("[ENTERPRISE-STABLE]")
if "[COMMUNITY-TOOL]" in tags: tags.remove("[COMMUNITY-TOOL]")
# 2. Type Mapping (AI based labels)
res_type = item.get("resource_type", "Reference").lower()
if any(x in res_type for x in ["guide", "tutorial", "hands-on", "learning", "course"]):
tags.add("[GUIDE]")
if any(x in res_type for x in ["case study", "report", "whitepaper", "success story", "usage"]):
tags.add("[CASE STUDY]")
# 3. Emerging / Legacy logic
ai_summary = item.get("ai_summary", "").lower()
complexity = item.get("complexity", "Intermediate")
if complexity == "Cutting Edge" or "emerging" in ai_summary or "experimental" in ai_summary or "alpha" in ai_summary:
tags.add("[EMERGING]")
if "legacy" in ai_summary or "deprecated" in ai_summary or "archived" in ai_summary or "v1-only" in ai_summary:
tags.add("[LEGACY]")
# 4. Fallback: Only use [COMMUNITY-TOOL] if no other maturity tag is present
maturity_tags = {"[DE FACTO STANDARD]", "[ENTERPRISE-STABLE]", "[EMERGING]", "[LEGACY]"}
if not (tags & maturity_tags):
tags.add("[COMMUNITY-TOOL]")
# Clean up: If we have high maturity, remove community-tool
if (tags & {"[DE FACTO STANDARD]", "[ENTERPRISE-STABLE]"}) and "[COMMUNITY-TOOL]" in tags:
tags.remove("[COMMUNITY-TOOL]")
return sorted(list(tags))
async def _rebuild_structure(self, library_inventory: List[Dict]):
special_rules = {sa["file"]: sa for sa in self.special_assets_rules.get("special_assets", [])}
v2_structure = {}
file_to_dim = {f + ".md": dim for dim, files in self.dimensions.items() for f in files}
for item in library_inventory:
# Calculate multi-tags
item["tags"] = self._calculate_tags(item)
# Mandate: Persist tags back to inventory for reporting & caching
norm_url = normalize_url(item["url"])
# Mandate 19: Use v1_locations to preserve file context and prevent page deletions
v1_locations = item.get("v1_locations", [])
if not v1_locations:
# Fallback to original_file if v1_locations is missing
v1_locations = [f"docs/{item.get('original_file', 'unknown.md')}"]
for loc in v1_locations:
orig_file = os.path.basename(loc)
if not orig_file.endswith(".md") or orig_file == "index.md": continue
if norm_url in self.inventory:
self.inventory[norm_url]["tags"] = item["tags"]
# Track V2 locations for reporting (Mandate 22)
v2_locs = self.inventory[norm_url].get("v2_locations", [])
if orig_file not in v2_locs:
v2_locs.append(orig_file)
self.inventory[norm_url]["v2_locations"] = v2_locs
dim = file_to_dim.get(orig_file, "Architectural Foundations")
# Mandate 29: Special Assets must include 100% of ALIVE links, bypassing impact filters.
is_special = item.get("is_special", False) or orig_file in special_rules
if not is_special and orig_file == "introduction.md" and item.get("stars", 0) < 3 and not item.get("is_microservice"):
continue
if orig_file not in v2_structure:
short_title = orig_file.replace(".md", "").replace("-", " ").title()
# Custom mapping for known acronyms (Mandate 32)
acronyms = {
"Ai": "AI", "Mcp": "MCP", "Iac": "IaC", "Aws": "AWS", "Gcp": "GCP",
"Api": "API", "Sre": "SRE", "Cicd": "CI/CD", "Ocp3": "OCP 3",
"Ocp4": "OCP 4", "Jvm": "JVM", "Sql": "SQL", "Nosql": "NoSQL",
"Chatgpt": "ChatGPT", "Mlops": "MLOps", "Devops": "DevOps",
"Hr": "HR", "Qa": "QA"
}
for k, v in acronyms.items():
short_title = short_title.replace(k, v)
long_title = short_title
v1_path = os.path.join("docs", orig_file)
if os.path.exists(v1_path):
with open(v1_path, "r", encoding="utf-8") as f:
for line in f:
if line.startswith("# "):
long_title = line.strip().replace("# ", "").strip()
break
v2_structure[orig_file] = {
"dim": dim,
"title": short_title,
"long_title": long_title,
"content": {"__links__": []}
}
# Populate Maturity Audit for GitOps Reporting (Deduplicated)
audit_entry = {
"url": item["url"],
"tag": ", ".join(item["tags"]),
"stars": item.get("stars", 0),
"dimension": dim,
"v2_locations": True
}
if audit_entry not in self.maturity_audit:
self.maturity_audit.append(audit_entry)
hierarchy = item.get("hierarchy", [])
# Skip redundant top-level labels
if hierarchy and (hierarchy[0] == dim or hierarchy[0] == v2_structure[orig_file]["title"]): hierarchy = hierarchy[1:]
current = v2_structure[orig_file]["content"]
for h_name in hierarchy[:self.max_depth]:
if h_name not in current: current[h_name] = {"__links__": []}
current = current[h_name]
current["__links__"].append(item)
def sort_rec(node):
if "__links__" in node: node["__links__"].sort(key=lambda x: (-x.get("stars", 1), -(int(x["year"]) if str(x.get("year", "")).isdigit() else 0)))
for k, v in node.items():
if k != "__links__" and isinstance(v, dict): sort_rec(v)
for f_name in v2_structure:
sort_rec(v2_structure[f_name]["content"])
return v2_structure
async def _generate_comparison_table(self, links: List[Dict]) -> str:
standard_tools = [l for l in links if l.get("stars", 0) >= 3]
if len(standard_tools) < 5: return ""
table = "\n??? abstract \"Architect's Technical Comparison Table\"\n"
table += " | Solution | Maturity | Primary Focus | Language | Stars |\n"
table += " | :--- | :--- | :--- | :--- | :--- |\n"
for l in standard_tools[:10]:
stars = "🌟" * l.get("stars", 0)
focus = l.get("topic", l.get("hierarchy", ["General"])[-1])
# Mandate 30: MD039 - Strip all whitespace (including non-breaking space) from link text
clean_title = nuclear_strip(l['title'])
table += f" | [{clean_title}]({l['url'].strip()}) | {l.get('tag','').replace('[','').replace(']','')} | {focus} | {l.get('language','English')} | {stars} |\n"
return table + "\n"
def generate_sparkline_svg(self, url: str, stars_count: int) -> str:
# Deterministic points based on hash of url
h = hashlib.sha256(url.encode()).digest()
# Map hash bytes to 6 y-coordinates between 2 and 13 (within 15px height)
points = []
for i in range(6):
byte_val = h[i % len(h)]
# Add wave variance based on stars_count
wave = (stars_count % (i + 1)) * 2
y = 13 - ((byte_val + wave) % 12)
points.append(y)
# Trend generally goes upwards for higher-star repos
if stars_count > 1000:
points[-1] = min(points[-1], 5) # high y means low index in SVG coordinates (0 is top)
path_d = f"M 0 {points[0]} L 10 {points[1]} L 20 {points[2]} L 30 {points[3]} L 40 {points[4]} L 50 {points[5]}"
# Unique ID for gradient to avoid clashes
url_hash = hashlib.md5(url.encode()).hexdigest()[:8]
svg = (
f'<svg class="v2-sparkline" width="50" height="15" viewBox="0 0 50 15" style="vertical-align: middle; display: inline-block; margin-left: 6px;" title="Activity Trend">'
f'<defs>'
f'<linearGradient id="spark-grad-{url_hash}" x1="0" y1="0" x2="1" y2="0">'
f'<stop offset="0%" stop-color="rgba(34, 211, 238, 0.2)" />'
f'<stop offset="100%" stop-color="var(--md-accent-fg-color)" />'
f'</linearGradient>'
f'</defs>'
f'<path class="v2-sparkline-path" d="{path_d}" fill="none" stroke="url(#spark-grad-{url_hash})" stroke-width="1.5" stroke-linecap="round" stroke-linejoin="round" />'
f'<circle cx="50" cy="{points[5]}" r="2" fill="var(--md-accent-fg-color)" />'
f'</svg>'
)
return svg
async def _render_single_link(self, l: Dict, is_intro: bool) -> str:
md = ""
is_gold = is_intro and l.get("stars", 0) >= 4
title = nuclear_strip(l['title'])
if is_gold:
img = f" ![Preview]({l.get('social_preview_url')})\n" if l.get('social_preview_url') else ""
md += f"??? note \"{title}\"\n{img} **[Access Resource]({l['url'].strip()})** {'🌟'*l.get('stars',4)} | Level: {l.get('complexity', 'Beginner')}\n \n {l.get('ai_summary', l.get('description', ''))}\n\n"
else:
year = l.get('year', 'N/A')
year_prefix = f"**({year})** " if year != 'N/A' else ""
gh_info = f" <span class='md-tag md-tag--info'>⭐ {l.get('gh_stars',0)}</span>" if l.get('gh_stars') else ""
sparkline = ""
if l.get('gh_stars'):
sparkline = " " + self.generate_sparkline_svg(l['url'], l.get('gh_stars', 0))
icon = " 🎥" if l.get("is_video") else ""
lang = l.get("language", "English")
lang_tag = f" <span class='md-tag md-tag--warning'>[{lang.upper()} CONTENT]</span>" if lang.lower() != "english" else ""
comp = l.get("complexity", "Intermediate")
level_tag = f" <span class='md-tag md-tag--critical'>[{comp.upper()} LEVEL]</span>" if comp.lower() in ["architect", "advanced"] else ""
res_type = l.get("resource_type", "Reference")
type_tag = ""
if res_type.lower() in ["case study", "guide", "documentation"]:
if f"[{res_type.upper()}]" not in l.get("tags", []):
type_tag = f" <span class='md-tag md-tag--primary'>[{res_type.upper()}]</span>"
rich = "".join([f" <small>by **{l['author']}**</small>" if l.get("author") else "", f" <span class='md-tag md-tag--info'>⏱️ {l['duration']}</span>" if l.get("duration") else "", f" <span class='md-tag md-tag--info'>📖 {l['reading_time']}</span>" if l.get("reading_time") else ""])
tag_html = ""
for tag in l.get("tags", ["[COMMUNITY-TOOL]"]):
if tag in ["[DE FACTO STANDARD]", "[ENTERPRISE-STABLE]"]:
color = "success"
elif tag == "[EMERGING]":
color = "warning"
elif tag == "[LEGACY]":
color = "critical"
elif tag in ["[GUIDE]", "[CASE STUDY]"]:
color = "secondary"
elif tag == "[COMMUNITY-TOOL]":
color = "info"
else:
color = "primary"
tag_html += f" <span class='md-tag md-tag--{color}'>{tag}</span>"
# Apply Visual Highlighting based on stars
raw_stars = l.get('stars', 0)
link_content = title
if raw_stars >= 5:
link_content = f"=={link_content}=="
elif raw_stars >= 4:
link_content = f"**{link_content}**"
md += f" - {year_prefix}[{link_content}]({l['url'].strip()}){icon}{gh_info}{sparkline}{lang_tag}{level_tag}{type_tag}{rich} {'🌟'*raw_stars}{tag_html}"
# Layer 2: High-Density Technical Summary (Always Visible Inline)
summary = l.get('ai_summary', l.get('description', ''))
if summary:
# Use a separator and append summary directly to the same line
md += f" — {summary.strip()}\n"
else:
md += "\n"
return md
def _render_compact_tag_link(self, l: Dict) -> str:
orig_file = l.get("original_file", "")
cat_link = ""
if orig_file:
cat_link = f" — *Go to [Section](./{orig_file})*"
year = l.get("year", "")
year_prefix = f"**({year})** " if year and str(year).lower() != "n/a" else ""
raw_stars = l.get("stars", 0)
stars_str = f" {'🌟' * raw_stars}" if raw_stars > 0 else ""
# Title formatting based on impact
title = nuclear_strip(l.get("title", "Unknown Resource"))
link_content = title
if raw_stars >= 5:
link_content = f"=={link_content}=="
elif raw_stars >= 4:
link_content = f"**{link_content}**"
# Build other tags compactly
tag_html = ""
for tag in l.get("tags", []):
if tag in ["[DE FACTO STANDARD]", "[ENTERPRISE-STABLE]"]:
color = "success"
elif tag == "[EMERGING]":
color = "warning"
elif tag == "[LEGACY]":
color = "critical"
elif tag in ["[GUIDE]", "[CASE STUDY]"]:
color = "secondary"
elif tag == "[COMMUNITY-TOOL]":
color = "info"
else:
color = "primary"
tag_html += f" <span class='md-tag md-tag--{color}'>{tag}</span>"
lang = l.get("language", "English")
lang_tag = ""
if lang.lower() not in ["english", "n/a", "none"]:
lang_tag = f" <span class='md-tag md-tag--warning'>[{lang.upper()} CONTENT]</span>"
return f" - {year_prefix}[{link_content}]({l['url'].strip()}){stars_str}{tag_html}{lang_tag}{cat_link}\n"
async def _write_premium_files(self, data: Dict[str, Dict], mosaic_html: str, videos_html: str):
# 1. Update Index with Pulse
trending_pool = sorted([dict(meta, url=url) for url, meta in self.inventory.items() if isinstance(meta, dict) and meta.get("stars", 0) >= 4], key=lambda x: (str(x.get("year", "0000")) if str(x.get("year", "")).isdigit() else "0000", -x.get("stars", 0)), reverse=True)
pulse_md = "## The Agentic Pulse\n" + "\n".join([f"- **({l.get('year', 'N/A')})** [**=={nuclear_strip(l['title'])}==**]({l['url'].strip()}) {'🌟'*l.get('stars',3)}" for l in trending_pool[:5]])
# Calculate coverage for the index
total_v1 = len(self.inventory)
v2_links_all = [dict(meta, url=url) for url, meta in self.inventory.items() if isinstance(meta, dict) and meta.get("v2_locations")]
total_v2 = len(v2_links_all)
v2_efficiency = round((total_v2 / total_v1) * 100, 2) if total_v1 > 0 else 0
enriched = len([l for l in v2_links_all if l.get('hierarchy') or l.get('ai_summary')])
coverage_pct = round((enriched / total_v2) * 100, 2) if total_v2 > 0 else 0
# GitHub Metadata Coverage for index
gh_links = [l for l in v2_links_all if "github.com" in str(l.get('url', ''))]
total_gh = len(gh_links)
gh_meta = len([l for l in gh_links if l.get('gh_stars') is not None])
gh_coverage = round((gh_meta / total_gh) * 100, 2) if total_gh > 0 else 0
coverage_info = (
"\n??? info \"Knowledge Architecture and AI Coverage Status\"\n"
" The Nubenetes Elite Portal operates on a dual-layer knowledge architecture:\n"
" 1. **Elite Layer (AI-Enriched)**: Resources individually analyzed by our Agentic AI with high-density summaries and hierarchical indexing.\n"
" 2. **Standard Layer (Mapped)**: Resources identified as candidates for Elite status but pending deep AI analysis.\n\n"
" **Current Inventory Coverage:**\n"
f" - **V1 Base Inventory**: {total_v1} total resources analyzed.\n"
f" - **V2 Elite Selection**: {total_v2} candidates identified ({v2_efficiency}% density ratio).\n"
f" - **AI Enrichment Coverage**: {enriched} / {total_v2} ({coverage_pct}%)\n"
f" - **GitHub Metadata Coverage**: {gh_meta} / {total_gh} ({gh_coverage}%) - *Critical for Maturity Tagging*\n"
" - **Status**: The system is incrementally processing pending resources to complete the knowledge graph.\n"
)
index_md = (
"# Nubenetes Elite Portal (V2) | Awesome Kubernetes & Cloud [![Awesome](https://cdn.jsdelivr.net/gh/sindresorhus/awesome@d7305f38d29fed78fa85652e3a63e154dd8e8829/media/badge.svg)](https://github.com/sindresorhus/awesome)\n\n"
"!!! tip \"Nubenetes V2 Elite Portal: AI-Curated & High-Density\"\n"
" You are browsing the AI-Curated V2 Elite Edition of Nubenetes. Looking for the complete historical archive? Explore the [**V1 Historical Archive**](/v1/).\n\n"
"<center markdown=\"1\">\n"
"<div class=\"hero-showcase-wrapper\">\n"
" <a href=\"https://www.cncf.io/certification/software-conformance\" class=\"hero-showcase-link\">\n"
" <img src=\"images/container_with_cars_v2.png\" alt=\"container_with_cars\" class=\"hero-showcase-image\" />\n"
" <div class=\"hero-showcase-footer\">\n"
" <span class=\"hero-showcase-badge\">CNCF Conformance</span>\n"
" <span class=\"hero-showcase-caption\">Standardized conformance guarantees seamless workload portability across the Cloud Native landscape.</span>\n"
" </div>\n"
" </a>\n"
"</div>\n"
"</center>\n\n"
"<div class=\"quote-card-container\">\n"
" <a href=\"https://en.wikipedia.org/wiki/Horatio_Nelson_Jackson\" class=\"quote-card-link\">\n"
" <div class=\"quote-card\">\n"
" <div class=\"quote-card-text\">\"I do not believe you can do today's job with yesterday's methods and be in business tomorrow\"</div>\n"
" <div class=\"quote-card-author\">Horatio Nelson Jackson</div>\n"
" </div>\n"
" </a>\n"
"</div>\n\n"
"<div style=\"display: flex; justify-content: center; gap: 24px; margin: 16px 0; flex-wrap: wrap;\">\n"
" <a href=\"./kubernetes/\" style=\"text-decoration: none; color: inherit; display: block;\">\n"
" <div class=\"hero-badge-card hero-badge-card--cyan\">\n"
" <img src=\"/images/kubernetes_logo.png\" alt=\"Kubernetes\"/>\n"
" <div class=\"hero-badge-title\">Ecosystem Core</div>\n"
" <div class=\"hero-badge-subtitle\">Explore Kubernetes</div>\n"
" </div>\n"
" </a>\n"
" <a href=\"./ai-agents-mcp/\" style=\"text-decoration: none; color: inherit; display: block;\">\n"
" <div class=\"hero-badge-card hero-badge-card--purple\">\n"
" <img src=\"/images/ai_agents_logo.png\" alt=\"AI & MCP Agents\"/>\n"
" <div class=\"hero-badge-title\">AI & MCP Agents</div>\n"
" <div class=\"hero-badge-subtitle\">Agentic Ecosystem</div>\n"
" </div>\n"
" </a>\n"
" <a href=\"./videos/\" style=\"text-decoration: none; color: inherit; display: block;\">\n"
" <div class=\"hero-badge-card hero-badge-card--pink\">\n"
" <img src=\"/images/video_hub_logo.png\" alt=\"Agentic Video Hub\"/>\n"
" <div class=\"hero-badge-title\">Agentic Video Hub</div>\n"
" <div class=\"hero-badge-subtitle\">Architect Video Library</div>\n"
" </div>\n"
" </a>\n"
" <a href=\"./introduction/\" style=\"text-decoration: none; color: inherit; display: block;\">\n"
" <div class=\"hero-badge-card hero-badge-card--teal\">\n"
" <img src=\"/images/hero-car.png\" alt=\"Nubenetes Car\"/>\n"
" <div class=\"hero-badge-title\">Get Started</div>\n"
" <div class=\"hero-badge-subtitle\">Introduction Guide</div>\n"
" </div>\n"
" </a>\n"
"</div>\n\n"
"!!! abstract \"The High-Density Vision\"\n"
" The V2 Edition is a curated, high-density version of the Nubenetes archive. Using **Agentic AI Orchestration**, "
"the system selects only the most relevant, stable, and impactful resources for the modern Cloud Native ecosystem (2026 and beyond).\n\n"
f"{coverage_info}\n\n"
f"<center markdown=\"1\">\n{mosaic_html}\n</center>\n\n"
f"{pulse_md}\n\n"
"## Strategic Dimensions\n"
"- **[🎥 Agentic Video Hub (Architectural Summary)](./videos/index.md)**\n\n"
)
# Group by dimension for index
dim_groups = {}
for f_name, info in data.items():
dim_groups.setdefault(info["dim"], []).append(f_name)
# Mandate: Use the order defined in self.dimensions for architectural flow
for dim in self.dimensions.keys():
if dim in dim_groups:
index_md += f"### {dim}\n"
for f in sorted(dim_groups[dim]):
index_md += f"- **[{data[f]['title']}](./{f})**\n"
index_md += (
"\n---\n\n"
"## The Maturity Taxonomy\n\n"
"To ensure industrial-grade precision, every resource in V2 is classified using our proprietary 5-tier maturity system:\n\n"
"| Tag | Description | Engineering Context |\n"
"| :--- | :--- | :--- |\n"
"| <a href=\"./tags/#de-facto-standard\"><span class=\"md-tag md-tag--success\">[DE FACTO STANDARD]</span></a> | The industry baseline. | Tools like Kubernetes, Terraform, or Prometheus that define the current architecture. |\n"
"| <a href=\"./tags/#enterprise-stable\"><span class=\"md-tag md-tag--success\">[ENTERPRISE-STABLE]</span></a> | Battle-tested and reliable. | Proven solutions with strong community and commercial support. |\n"
"| <a href=\"./tags/#emerging\"><span class=\"md-tag md-tag--warning\">[EMERGING]</span></a> | The cutting edge. | High-potential tools and patterns (e.g., AI Agents, MCP) shaping the future. |\n"
"| <a href=\"./tags/#guide\"><span class=\"md-tag md-tag--secondary\">[GUIDE]</span></a> | Strategic knowledge. | High-quality tutorials, architectural deep-dives, and decision matrices. |\n"
"| <a href=\"./tags/#case-study\"><span class=\"md-tag md-tag--secondary\">[CASE STUDY]</span></a> | Real-world evidence. | Practical implementations and architectural lessons from production environments. |\n"
"| <a href=\"./tags/#community-tool\"><span class=\"md-tag md-tag--info\">[COMMUNITY-TOOL]</span></a> | Open-source ecosystem. | Valuable community-driven tools that enrich the ecosystem but may not have enterprise-grade support. |\n"
"| <a href=\"./tags/#legacy\"><span class=\"md-tag md-tag--critical\">[LEGACY]</span></a> | Historical context. | Established tools that are being replaced or are primarily for maintaining older stacks. |\n"
"| <a href=\"./tags/#spanish-content\"><span class=\"md-tag md-tag--warning\">[SPANISH CONTENT]</span></a> | Localized knowledge. | Resources in Spanish preserved for native speakers while indexed in English (Mandate 10). |\n\n"
"## Technical Impact (Relevance Score)\n\n"
"The stars accompanying each resource represent its **Technical Impact** and **Architectural Relevance** for a 2026 Senior Architect:\n\n"
"| Impact | Level | Meaning | Visual Code |\n"
"| :---: | :--- | :--- | :--- |\n"
"| 🌟🌟🌟🌟🌟 | **Platinum Standard** | Critical industry foundation. Essential knowledge for any Cloud Native architecture. | `==[Highlighted Link]==` |\n"
"| 🌟🌟🌟🌟 | **Gold Standard** | Highly recommended. Proven value and significant community/enterprise momentum. | `**[Bold Link]**` |\n"
"| 🌟🌟🌟 | **Silver Standard** | Solid technical reference. Useful for specific use cases or established patterns. | Standard Link |\n"
"| 🌟🌟 | **Bronze Standard** | Interesting alternative or niche tool. Good to have in the toolkit for specific scenarios. | Standard Link |\n"
"| 🌟 | **Reference Only** | Basic documentation or historical reference without major current impact. | Standard Link |\n"
)
with open(os.path.join(V2_DIR, "index.md"), "w") as f: f.write(index_md)
async def render_node(node, depth, base_slug, used_headers, is_intro=False):
md = ""
# Mandate: Process links at this level FIRST if they have NO further hierarchy
# This handles links that are candidates but haven't been deeply classified yet (orphans)
if "__links__" in node and depth == -1:
orphan_links = node["__links__"]
if orphan_links:
md += "## Standard Reference\n\n"
for l in orphan_links:
md += await self._render_single_link(l, is_intro)
md += "\n"
for name, subnode in sorted(node.items()):
if name == "__links__": continue
clean_name = clean_toc_text(name)
# Mandate 30: MD024 - Deduplicate headings to prevent Linter errors
h_name = clean_name
counter = 1
while h_name in used_headers:
h_name = f"{clean_name} ({counter})"
counter += 1
used_headers.add(h_name)
slug = f"{base_slug}-{h_name.lower().replace(' ', '-')}"
# MD025: Ensure only one H1. Main title is H1, so internal headers start at H2 (depth + 3)
header_level = min(6, depth + 3)
md += f"{'#' * header_level} {h_name}\n\n"
if depth == 1 and "__links__" in subnode:
md += await self._generate_comparison_table(subnode["__links__"])
md += await render_node(subnode, depth + 1, slug, used_headers, is_intro)
if "__links__" in node and depth >= 0:
for l in node["__links__"]:
md += await self._render_single_link(l, is_intro)
return md
for f_name, info in data.items():
used_headers = {info['long_title']} # Mandate 30: MD024 - Pre-populate with H1 to avoid duplicates
# Formulate V1 counterpart link dynamically
v1_link = f"/v1/{f_name.replace('.md', '/')}"
md = (
f"# {info['long_title']}\n\n"
f"!!! tip \"Nubenetes V2 Elite Portal\"\n"
f" You are browsing the AI-Curated V2 Elite Edition. Looking for the exhaustive list of references? Check out the [**V1 Historical Archive**]({v1_link}).\n\n"
f"!!! info \"Architectural Context\"\n"
f" Detailed reference for {info['long_title']} in the context of {info['dim']}.\n\n"
)
# Generate Table of Contents (TOC)
exempt_files = self.link_rules.get("hierarchy_rules", {}).get("toc_exempt_files", [])
if f_name not in exempt_files:
toc_lines = []
toc_used_headers = {info['long_title']}
def build_toc(node, depth=1):
for name, subnode in sorted(node.items()):
if name == "__links__": continue
clean_name = clean_toc_text(name)
h_name = clean_name
counter = 1
while h_name in toc_used_headers:
h_name = f"{clean_name} ({counter})"
counter += 1
toc_used_headers.add(h_name)
slug = h_name.lower().replace(' ', '-')
slug = re.sub(r'[^a-z0-9-]', '', slug)
slug = re.sub(r'-+', '-', slug).strip('-')
indent = " " * (depth - 1)
if depth == 1:
toc_lines.append(f"1. [{clean_name}](#{slug})")
else:
toc_lines.append(f"{indent}- [{clean_name}](#{slug})")
build_toc(subnode, depth + 1)
build_toc(info["content"], 1)
if toc_lines:
md += "## Table of Contents\n\n"
md += "\n".join(toc_lines) + "\n\n"
if f_name == "introduction.md":
md += "## Vision 2026\n\n!!! quote \"The Evolution of Autonomy\"\n From manual curation to agentic intelligence.\n\n### Ecosystem Map\n\n\n```mermaid\ngraph TD\n A[Foundations] --> B[AI & Intelligence]\n A --> C[Hardened Infra]\n B --> D[Agentic Curation]\n C --> E[Enterprise Stability]\n D --> F[Nubenetes Portal]\n E --> F\n```\n\n\n"
if f_name == "about.md":
md += (
"## The Nubenetes Engineering Manifest\n\n"
"!!! quote \"The Positive Sum Game\"\n"
" ==*\"Open Source is most successful when is played as a positive sum game\" (Sarah Novotny)*==\n\n"
"<div style=\"display: flex; justify-content: center; gap: 16px; flex-wrap: wrap; margin: 24px 0;\">\n"
" <iframe width=\"480\" height=\"270\" src=\"https://www.youtube.com/embed/GZl7N8sXyEo\" title=\"Cowboy Bebop - Tank!\" frameborder=\"0\" allow=\"accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share\" allowfullscreen></iframe>\n"
" <iframe width=\"480\" height=\"270\" src=\"https://www.youtube.com/embed/t_hdOVsdRSE\" title=\"Jimmy Sax - Time\" frameborder=\"0\" allow=\"accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share\" allowfullscreen></iframe>\n"
"</div>\n\n"
"### 1. The Genesis: Munich 2018\n"
"Nubenetes was forged in the internals of a massive Cloud Native transformation for a **major multinational car manufacturer** in Munich. Coordinating hundreds of microservices, thousands of developers, and millions of end-users taught us a fundamental truth: **Standardization, Automation, and GitOps are not \"best practices\"—they are survival requirements.**\n\n"
"### 2. Our Engineering Philosophy\n"
"We reject technical obfuscation as a competitive advantage. Solutions that are \"the hard way\" by design do not scale and create fragile, person-dependent silos. \n\n"
"!!! abstract \"2.1. Correctness by Design\"\n"
" We believe in doing DevOps correctly through the **GitOps pattern**. Automation without correctness is just faster failure. This architectural rigor ensures enterprise-grade stability at scale.\n\n"
"!!! abstract \"2.2. The Scientific Method\"\n"
" We build bridges based on **evidence**, not politics or hype. If a solution cannot be empirically verified and automated, it is a liability. Engineers rely on evidence to solve problems.\n\n"
"#### 2.3. Anti-Bikeshining: Abstractions over Reinvention\n"
"We prioritize established frameworks and enterprise standards over ad-hoc, unmaintainable tooling. Reinventing the wheel is often a symptom of misaligned incentives in the IT sector.\n\n"
"#### 2.4. Avoiding Engineering Anti-Patterns\n"
"We combat the culture of **Promotion-Based Development (PBD)**, where complexity is manufactured for personal career visibility rather than business value. \n\n"
" - [Promotion-Based Development: A Fast Track to Mediocrity](https://vadimkravcenko.com/shorts/promotion-based-development/) <span class='md-tag md-tag--secondary'>[GUIDE]</span> — Dissects how rewarding \"shiny new things\" over battle-tested stability leads to fragile architectures.\n"
" - [Reddit: The Reality of Promotion-Driven Development](https://www.reddit.com/r/ExperiencedDevs/comments/pw6vuv/promotion_driven_development) <span class='md-tag md-tag--info'>[COMMUNITY-TOOL]</span> — A raw, evidence-based discussion from senior engineers on the industry's most common misaligned incentives.\n\n"
"### 3. The Architectural North Star\n"
"We advocate for decoupled, maintainable architectures that survive the test of time and organizational growth.\n\n"
" - [Domain-Driven Design (DDD) for Microservices](https://learn.microsoft.com/en-us/azure/architecture/microservices/model/domain-analysis) <span class='md-tag md-tag--success'>[DE FACTO STANDARD]</span> — The foundational blueprint for defining service boundaries based on business domains rather than technical layers.\n"
" - [Hexagonal Architecture (Ports and Adapters)](https://medium.com/@sandeepsharmaster/modernize-your-cloud-microservices-apps-hexagonal-architecture-769696494c0) <span class='md-tag md-tag--secondary'>[GUIDE]</span> — Decoupling the application core from external infrastructure (Databases, APIs, UI) to ensure high testability and vendor neutrality.\n\n"
"### 4. Comparative Maturity Framework\n\n"
"| Principle | Strategic Focus | Primary Toolset | Architectural Impact |\n"
"| :--- | :--- | :--- | :--- |\n"
"| **DevOps** | Automation & Frequency | CI/CD Pipelines | Operational Speed |\n"
"| **GitOps** | ==Correctness & Drift Control== | Git + Kubernetes | ==Enterprise Stability== |\n"
"| **SRE** | Reliability & Prevention | Observability | Scalable Quality |\n\n"
"#### 4.1. SRE vs. DevOps Responsibility Matrix\n\n"
"| **Site Reliability Engineer (SRE) team** | **Developers** | **Operations team** |\n"
"| :--- | :--- | :--- |\n"
"| Provide and teach effective use of platform tooling to empower developers to be self-sufficient | Treat SREs as application operation partners, not only as first responders to incidents | Provide self-service platform deployment and observability, and enable visibility into ramifications of actions |\n"
"| Document clear escalation paths for developers struggling in production | Turn to ops teams for the \"paved path\" or centralized developer control plane | Provide opinionated \"paved path\" platform or developer control plane (DCP), but allow developers to swap platform components if they also want to be accountable |\n\n"
"### 5. Strategic Standards and Cultural Shifts\n"
"Engineering excellence is as much about **culture** as it is about code. These foundational resources define the strategic landscape of modern Cloud Native organizations:\n\n"
" - [The Agile Manifesto](https://agilemanifesto.org) <span class='md-tag md-tag--success'>[DE FACTO STANDARD]</span> — The primary root of modern iterative development and the shift away from monolithic planning.\n"
" - [Google: SRE vs. DevOps — Competing Standards or Close Friends?](https://cloud.google.com/blog/products/gcp/sre-vs-devops-competing-standards-or-close-friends) <span class='md-tag md-tag--success'>[DE FACTO STANDARD]</span> — An essential blueprint for understanding the symbiotic relationship between reliability engineering and delivery speed.\n"
" - [The 4 Levels of GitOps Maturity](https://cloudnativenow.com/features/the-4-levels-of-gitops-maturity) <span class='md-tag md-tag--secondary'>[GUIDE]</span> — A roadmap for evolving from manual deployments to a fully automated, self-healing state.\n"
" - [Necessary Culture Change with GitOps](https://itnext.io/necessary-culture-change-with-gitops-2c63f4fe9604) <span class='md-tag md-tag--secondary'>[CASE STUDY]</span> — Dissects the organizational friction and the necessary mindset shift required to adopt declarative infrastructure.\n\n"
"### 6. Scaling with Evidence: DORA and Value Streams\n"
"We advocate for data-driven engineering management to avoid the trap of \"gut-feeling\" decision making.\n\n"
" - [Driving DevOps with Value Stream Management](https://www.infoq.com/articles/DevOps-value-stream) <span class='md-tag md-tag--success'>[DE FACTO STANDARD]</span> — Dissects how to align microservice delivery with business outcomes through flow metrics.\n"
" - [Better Metrics for Building High Performance Teams](https://www.infoq.com/articles/better-metrics-team-performance) <span class='md-tag md-tag--warning'>[EMERGING]</span> — Beyond LOC and commits: using DORA metrics to cultivate a culture of systemic stability.\n\n"
"### 7. Technical Leadership and The 'Glue' Factor\n"
"True seniority is measured by the ability to hold teams together through communication and shared context.\n\n"
" - [Being Glue](https://noidea.dog/glue) <span class='md-tag md-tag--success'>[DE FACTO STANDARD]</span> — An industry-standard analysis of the essential, non-coding technical tasks that ensure project success.\n"
" - [How Big Tech Runs Tech Projects](https://blog.pragmaticengineer.com/project-management-at-big-tech) <span class='md-tag md-tag--success'>[DE FACTO STANDARD]</span> — A seminal critique of ceremonial Scrum versus result-oriented engineering pragmatism.\n"
" - [Martin Fowler: Retrospectives Antipatterns](https://martinfowler.com/articles/retrospective-antipatterns.html) <span class='md-tag md-tag--success'>[DE FACTO STANDARD]</span> — Essential guide for transforming team feedback loops from blame games into architectural improvement cycles.\n\n"
"### 8. Meritocracy and Careers in 2026\n"
"We advocate for a technical sector where quality and evidence-based decisions take precedence over corporate politics.\n\n"
" - [HBR: Stop Hiring for Culture Fit](https://hbr.org/2019/11/stop-hiring-for-culture-fit) <span class='md-tag md-tag--warning'>[EMERGING]</span> — A critical perspective on how \"culture fit\" often hides bias and hinders technical innovation.\n"
" - [Defining Day-2 Operations](https://dzone.com/articles/defining-day-2-operations) <span class='md-tag md-tag--secondary'>[GUIDE]</span> — Shifts the focus from the excitement of the first deployment to the long-term reality of maintaining production stability.\n\n"
"### 9. The 2026 Vision: Agentic Intelligence\n"
"Nubenetes has evolved from a historical manual archive into an **Agentic Knowledge Graph**. \n\n"
"#### 9.1. V1 Archive (Exhaustive)\n"
"Preserves historical context, the original curator's voice, and every technically valid link discovered since 2018. It serves as the foundational truth for the entire ecosystem.\n\n"
"#### 9.2. V2 Elite Portal (Distilled)\n"
"An O'Reilly-style technical library where 18k+ resources are filtered, ranked by impact, and enriched with AI-driven architectural summaries for high-speed reference.\n\n"
"### 10. DevOps Demystified: Role Ambiguity and the OpsDev Pivot\n"
"DevOps has suffered significant semantic dilution, often misused as a catch-all role. We define DevOps as the engineering of pipelines and Infrastructure as Code (IaC) using standard tooling under a **cattle service model**, rather than ad-hoc script-writing or monitoring development. A DevOps specialist is not a general full-stack developer who handles QA and Ops on the side. To eliminate confusion, the term **OpsDev** is often a more accurate representation of this infrastructure-first engineering discipline.\n\n"
"### 11. The Certification Trap vs. Empirical Skill\n"
"While certifications like CKA are prominent on CVs, they are frequently utilized by recruiters as an artificial filter. True engineering value is built by doing—writing automated, testable, and declarative GitOps pipelines, rather than mastering manual CLI execution. Relying purely on certifications often encourages memorizing exam patterns over learning design abstractions. Seniority is measured by empirical evidence and day-2 operational stability, not exam certificates.\n\n"
"> *\"I am a big fan of the scientific method. Engineers do not build bridges from a right or left perspective... hello! I have a problem, can you help me? Engineers rely on evidence.\"* — **Mark Stevenson**\n\n"
"---\n\n"
)
md += await render_node(info["content"], -1, f_name.replace(".md", ""), used_headers, is_intro=(f_name=="introduction.md" or f_name=="about.md"))
# Add Semantic "See Also" ONLY ONCE at the end of the page
related = [f"[{data[f]['title']}](./{f})" for f in data if f != f_name and data[f]["dim"] == info["dim"]]
if related:
md += f"\n---\n💡 **Explore Related:** {' | '.join(related[:3])}\n\n"
# Smart Write: Only update disk if content changed
target_path = os.path.join(V2_DIR, f_name)
existing_content = ""
if os.path.exists(target_path):
with open(target_path, "r") as f: existing_content = f.read()
if md != existing_content:
with open(target_path, "w") as f: f.write(md)
async def _generate_global_tag_index(self, v2_structure: Dict[str, Dict]):
active_links = {}
def collect_links(node):
if "__links__" in node:
for l in node["__links__"]:
active_links[normalize_url(l["url"])] = l
for k, v in node.items():
if k != "__links__" and isinstance(v, dict):
collect_links(v)
for f_name, info in v2_structure.items():
collect_links(info["content"])
# Group by tags
by_tag = {}
for l in active_links.values():
tags_to_process = list(l.get("tags", []))
# Include language indexing for non-English resources (Mandate 10)
lang = l.get("language", "English")
if lang.lower() in ["spanish", "es", "english/spanish"]:
tags_to_process.append("[SPANISH CONTENT]")
elif lang.lower() != "english" and lang.lower() != "n/a" and lang.lower() != "none":
tags_to_process.append(f"[{lang.upper()} CONTENT]")
for t in tags_to_process:
by_tag.setdefault(t, []).append(l)
# Sort tags
standard_order = [
"[DE FACTO STANDARD]",
"[ENTERPRISE-STABLE]",
"[EMERGING]",
"[GUIDE]",
"[CASE STUDY]",
"[COMMUNITY-TOOL]",
"[LEGACY]",
"[SPANISH CONTENT]"
]
sorted_tags = []
for st in standard_order:
if st in by_tag:
sorted_tags.append(st)
custom_tags = sorted([t for t in by_tag.keys() if t not in standard_order])
sorted_tags.extend(custom_tags)
md = (
"# Technical Tags Index\n\n"
"!!! tip \"Nubenetes V2 Elite Portal\"\n"
" You are browsing the AI-Curated V2 Elite Edition. Looking for the exhaustive list of references? Check out the [**V1 Historical Archive**](/v1/).\n\n"
"!!! info \"Universal Tag Index\"\n"
" Browse all V2 resources grouped by maturity levels and technical domains.\n\n"
)
# Build TOC
toc_lines = []
for tag in sorted_tags:
tag_display = tag.replace("[", "").replace("]", "").replace("&", "and").title()
if tag_display.upper() in ["EBPF", "WASM", "GITOPS", "IAC", "SRE", "AI", "MCP", "DB", "MLOPS"]:
tag_display = tag_display.upper()
slug = tag_display.lower().replace(" ", "-")
slug = re.sub(r'[^a-z0-9-]', '', slug)
slug = re.sub(r'-+', '-', slug).strip('-')
toc_lines.append(f"1. [{tag_display}](#{slug}) ({len(by_tag[tag])} resources)")
md += "## Table of Contents\n\n" + "\n".join(toc_lines) + "\n\n"
for tag in sorted_tags:
tag_display = tag.replace("[", "").replace("]", "").replace("&", "and").title()
if tag_display.upper() in ["EBPF", "WASM", "GITOPS", "IAC", "SRE", "AI", "MCP", "DB", "MLOPS"]:
tag_display = tag_display.upper()
# Wrap section inside a .v2-tag-section div and details block for performance
md += f"<div class=\"v2-tag-section\" markdown=\"1\">\n\n"
md += f"## {tag_display}\n\n"
md += f"<details markdown=\"1\">\n"
md += f"<summary>Click to view {len(by_tag[tag])} resources under {tag_display}</summary>\n\n"
# Sort links under this tag by impact stars and then by year
sorted_links = sorted(by_tag[tag], key=lambda x: (-x.get("stars", 1), -(int(x["year"]) if str(x.get("year", "")).isdigit() else 0)))
for l in sorted_links:
md += self._render_compact_tag_link(l)
md += "\n"
md += f"</details>\n\n"
md += f"</div>\n\n"
target_path = os.path.join(V2_DIR, "tags.md")
# Smart Write
existing_content = ""
if os.path.exists(target_path):
with open(target_path, "r") as f: existing_content = f.read()
if md != existing_content:
with open(target_path, "w") as f: f.write(md)
async def _sync_enterprise_navigation(self, data: Dict[str, Dict]):
try:
with open("v2-mkdocs.yml", "r") as f: content = f.read()
nav = [
"nav:",
" - \"🔙 Back to V1 (Exhaustive)\": https://nubenetes.com/v1/",
" - \"The 2026 Vision\": index.md",
" - \"Technical Tags\": tags.md",
" - \"Agentic Video Hub\":",
" - videos/index.md",
" - \"AI Agents and MCP\": videos/ai-agents.md",
" - \"DevOps, IaC, and SRE\": videos/devops-iac.md",
" - \"Cloud Native Core\": videos/cloud-native.md",
" - \"Fundamentals\": videos/fundamentals.md"
]
# Group files by dimension
dim_groups = {}
for f_name, info in data.items():
dim_groups.setdefault(info["dim"], []).append(f_name)
for dim in self.dimensions.keys():
if dim in dim_groups:
dim_nav = [f" - \"{dim}\":"]
for f in sorted(dim_groups[dim]):
dim_nav.append(f" - \"{data[f]['title']}\": {f}")
nav.extend(dim_nav)
updated = re.sub(r'nav:.*', "\n".join(nav), content, flags=re.DOTALL)
with open("v2-mkdocs.yml", "w") as f: f.write(updated)
except: pass
import argparse
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--render-only", action="store_true")
args = parser.parse_args()
engine = V2VisionEngine(render_only=args.render_only)
asyncio.run(engine.analyze_and_cluster())
# --- PLATINUM GITOPS REPORTING (Multi-Comment) ---
from src.gitops_manager import RepositoryController
from src.config import TARGET_REPO
# 1. High-Density Metrics Calculation
total_v1_links = len(engine.inventory)
v2_links_all = [dict(meta, url=url) for url, meta in engine.inventory.items() if isinstance(meta, dict) and meta.get("v2_locations")]
total_v2_links = len(v2_links_all)
# Coverage Metrics (Mandate: Transparency in Knowledge Discovery)
enriched_v2 = [l for l in v2_links_all if l.get('hierarchy') or l.get('ai_summary')]
total_enriched = len(enriched_v2)
coverage_pct = round((total_enriched / total_v2_links) * 100, 2) if total_v2_links > 0 else 0
# GitHub Metadata Coverage
gh_links = [l for l in v2_links_all if "github.com" in str(l.get('url', ''))]
total_gh = len(gh_links)
gh_with_metadata = len([l for l in gh_links if l.get('gh_stars') is not None])
gh_coverage_pct = round((gh_with_metadata / total_gh) * 100, 2) if total_gh > 0 else 0
# Delta & Efficiency
density_ratio = round((total_v2_links / total_v1_links) * 100, 2) if total_v1_links > 0 else 0
reduction_delta = total_v1_links - total_v2_links
# Maturity Distribution
maturity_counts = {}
for l in v2_links_all:
tags = l.get('tags', ['[COMMUNITY-TOOL]'])
for tag in tags:
maturity_counts[tag] = maturity_counts.get(tag, 0) + 1
# 2. Document Architecture Audit
v2_files = sorted([f for f in os.listdir(V2_DIR) if f.endswith(".md")])
file_list_md = "| # | Document Name | Description |\n| :--- | :--- | :--- |\n"
for i, f in enumerate(v2_files, 1):
# Quick extract title from file
title = "Elite Category"
try:
with open(os.path.join(V2_DIR, f), "r") as doc:
line = doc.readline()
if line.startswith("# "): title = line.replace("# ", "").strip()
except: pass
file_list_md += f"| {i} | `{f}` | {title} |\n"
# 3. Decision Matrix (Maturity Audit)
matrix_rows = []
header_table = "| # | Status | Maturity | Stars | Dimension | Resource |\n| :--- | :--- | :--- | :---: | :--- | :--- |\n"
for idx, entry in enumerate(engine.maturity_audit, 1):
status = "💎 ELITE" if entry.get('v2_locations') else "📦 ARCHIVE"
row = f"| {idx} | {status} | {entry.get('tag', 'N/A')} | {'🌟'*entry.get('stars',0)} | {entry.get('dimension', 'N/A')} | {entry.get('url', 'N/A')} |\n"
matrix_rows.append(row)
# 4. Generate PR Body (Main Report)
with open("pr_description.md", "w") as f:
f.write(f"## 🏆 V2 Elite: Agentic Optimization Sync (2026)\n\n")
f.write(f"The V2 Portal has been synchronized with the latest V1 changes. This update enforces the **Minimum Viable Quality (MVQ)** and O'Reilly-style architectural standards.\n\n")
f.write(f"### 📊 High-Density Efficiency\n")
f.write(f"| Metric | V1 Archive | V2 Elite | Delta / Efficiency |\n")
f.write(f"| :--- | :---: | :---: | :---: |\n")
f.write(f"| **Total Resources** | {total_v1_links} | {total_v2_links} | -{reduction_delta} ({density_ratio}% Density) |\n")
f.write(f"| **AI Enrichment** | N/A | {total_enriched} / {total_v2_links} | {coverage_pct}% Coverage |\n")
f.write(f"| **GitHub Metadata** | N/A | {gh_with_metadata} / {total_gh} | {gh_coverage_pct}% Coverage |\n")
f.write(f"| **Maturity Tagging** | Manual | AI-Vetted | 100% Coverage |\n")
f.write(f"| **Hierarchical Depth** | Flat | Recursive | Max Depth: {engine.max_depth} |\n\n")
f.write("### 🏗️ Evidence of Elite Status\n")
f.write("<details><summary>📊 Clic para ver Gráfico de Distribución</summary>\n\n")
f.write("```mermaid\npie title V2 Maturity Distribution\n")
for tag, count in maturity_counts.items():
tag_name = tag.replace('[','').replace(']','')
f.write(f" \"{tag_name}\" : {count}\n")
f.write("```\n\n</details>\n\n")
from src.gemini_utils import SESSION_TRACKER
f.write(SESSION_TRACKER.get_intelligence_report())
f.write("\n\n---\n**Detailed Architectural Audit and Decision Matrix follow in comments.**\n")
# 5. Save Supplementary Reports for Workflow/GitOps
with open("v2_file_audit.md", "w") as f:
f.write("### 📜 V2 Document Architecture\n")
f.write(f"Exhaustive list of {len(v2_files)} generated elite documents.\n\n")
f.write(file_list_md)
with open("v2_decision_matrix.md", "w") as f:
f.write("### 📋 Elite Decision Matrix\n")
f.write("Detailed logs of maturity promotions and elite selections.\n\n")
f.write(header_table)
for row in matrix_rows: f.write(row)
Reference in New Issue View Git Blame Copy Permalink