github/awesome-kubernetes
1
0
Fork 0
mirror of https://github.com/nubenetes/awesome-kubernetes.git synced 2026-07-28 17:41:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(health-checker): add checkpointing and fix GH CLI issue triage checks

Browse Source
This commit is contained in:
Inaki Fernandez
2026-07-12 23:21:48 +02:00
parent a4e5b049fd
commit 158c2e6656
2 changed files with 544 additions and 483 deletions
Download Patch File Download Diff File Expand all files Collapse all files

238
.github/workflows/02.1.intelligent_link_cleaner.yml vendored
View File

@@ -1,118 +1,120 @@
name: 02.1. Intelligent Link Cleaner & Dedup
on:
schedule:
- cron: '0 0 1 */3 *' # Quarterly: 1st of Jan, Apr, Jul, Oct
workflow_dispatch:
inputs:
force_full_check:
description: 'Force full re-validation (bypasses 21-day cache)'
type: boolean
default: false
permissions:
contents: write
pull-requests: write
issues: write
concurrency:
group: link-cleaner-${{ github.ref }}
cancel-in-progress: true
jobs:
intelligent-clean-sequential:
runs-on: ubuntu-latest
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true
steps:
- name: Repository Synchronization
uses: actions/checkout@v6
with:
ref: develop
- name: Python 3.11 Environment Provisioning
uses: actions/setup-python@v6
with:
python-version: '3.11'
cache: 'pip'
- name: Dependencies Installation
run: |
python -m pip install --upgrade pip
pip install --no-cache-dir pydantic PyGithub aiohttp beautifulsoup4 httpx fake-useragent pytz python-dotenv playwright PyYAML tenacity
- name: Cache Playwright Binaries
uses: actions/cache@v5
id: playwright-cache
with:
path: ~/.cache/ms-playwright
key: ${{ runner.os }}-playwright-cleaner
- name: Install Playwright Browsers
if: steps.playwright-cache.outputs.cache-hit != 'true'
run: playwright install chromium --with-deps
- name: Restore Link Checker Cache
uses: actions/cache/restore@v4
id: link-checker-cache
with:
path: |
shard_result.json
triage_report.md
key: link-cleaner-results-${{ github.run_id }}-${{ github.run_attempt }}
restore-keys: |
link-cleaner-results-${{ github.run_id }}-
link-cleaner-results-
- name: Global Intelligent Cleaning Execution (Sequential Fast-Track)
env:
GEMINI_API_KEY_1: ${{ secrets.GEMINI_API_KEY_1 }}
GEMINI_API_KEY_2: ${{ secrets.GEMINI_API_KEY_2 }}
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
FORCE_FULL_CHECK: ${{ github.event.inputs.force_full_check || 'false' }}
PYTHONPATH: ${{ github.workspace }}
PYTHONUNBUFFERED: 1
run: |
if [ -f shard_result.json ]; then
echo "shard_result.json found in cache/restored files. Skipping health checker execution."
else
python src/intelligent_health_checker.py
fi
- name: Run Inventory Reducer & Apply Changes
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
PYTHONPATH: ${{ github.workspace }}
run: |
# Since it's sequential, the script produces shard_result.json (default name for no shard)
python src/inventory_reducer.py
- name: Open Triage Issue for High-Value Links
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
if [ -f triage_report.md ]; then
ISSUE_TITLE="🚨 Manual Triage Required: High-Value Links Failing ($(date +'%Y-%m-%d'))"
EXISTING_ISSUE=$(gh issue list --search "$ISSUE_TITLE" --state open --json number --jq '.[0].number')
if [ -z "$EXISTING_ISSUE" ]; then
gh issue create --title "$ISSUE_TITLE" --body-file triage_report.md --label "triage, automated-report" || gh issue create --title "$ISSUE_TITLE" --body-file triage_report.md
else
echo "A triage issue already exists (#$EXISTING_ISSUE). Updating with new report..."
gh issue comment $EXISTING_ISSUE --body-file triage_report.md
fi
fi
- name: Ensure Cache Files Exist
if: always()
run: |
touch shard_result.json triage_report.md
- name: Save Link Checker Cache
if: always()
uses: actions/cache/save@v4
with:
path: |
shard_result.json
triage_report.md
key: link-cleaner-results-${{ github.run_id }}-${{ github.run_attempt }}
name: 02.1. Intelligent Link Cleaner & Dedup
on:
schedule:
- cron: '0 0 1 */3 *' # Quarterly: 1st of Jan, Apr, Jul, Oct
workflow_dispatch:
inputs:
force_full_check:
description: 'Force full re-validation (bypasses 21-day cache)'
type: boolean
default: false
permissions:
contents: write
pull-requests: write
issues: write
concurrency:
group: link-cleaner-${{ github.ref }}
cancel-in-progress: true
jobs:
intelligent-clean-sequential:
runs-on: ubuntu-latest
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true
steps:
- name: Repository Synchronization
uses: actions/checkout@v6
with:
ref: develop
- name: Python 3.11 Environment Provisioning
uses: actions/setup-python@v6
with:
python-version: '3.11'
cache: 'pip'
- name: Dependencies Installation
run: |
python -m pip install --upgrade pip
pip install --no-cache-dir pydantic PyGithub aiohttp beautifulsoup4 httpx fake-useragent pytz python-dotenv playwright PyYAML tenacity
- name: Cache Playwright Binaries
uses: actions/cache@v5
id: playwright-cache
with:
path: ~/.cache/ms-playwright
key: ${{ runner.os }}-playwright-cleaner
- name: Install Playwright Browsers
if: steps.playwright-cache.outputs.cache-hit != 'true'
run: playwright install chromium --with-deps
- name: Restore Link Checker Cache
uses: actions/cache/restore@v4
id: link-checker-cache
with:
path: |
shard_result.json
triage_report.md
health_check_checkpoint.json
key: link-cleaner-results-${{ github.run_id }}-${{ github.run_attempt }}
restore-keys: |
link-cleaner-results-${{ github.run_id }}-
link-cleaner-results-
- name: Global Intelligent Cleaning Execution (Sequential Fast-Track)
env:
GEMINI_API_KEY_1: ${{ secrets.GEMINI_API_KEY_1 }}
GEMINI_API_KEY_2: ${{ secrets.GEMINI_API_KEY_2 }}
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
FORCE_FULL_CHECK: ${{ github.event.inputs.force_full_check || 'false' }}
PYTHONPATH: ${{ github.workspace }}
PYTHONUNBUFFERED: 1
run: |
if [ -s shard_result.json ]; then
echo "shard_result.json found in cache/restored files and has content. Skipping health checker execution."
else
python src/intelligent_health_checker.py
fi
- name: Run Inventory Reducer & Apply Changes
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
PYTHONPATH: ${{ github.workspace }}
run: |
# Since it's sequential, the script produces shard_result.json (default name for no shard)
python src/inventory_reducer.py
- name: Open Triage Issue for High-Value Links
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
if [ -f triage_report.md ] && [ -s triage_report.md ]; then
ISSUE_TITLE="🚨 Manual Triage Required: High-Value Links Failing ($(date +'%Y-%m-%d'))"
EXISTING_ISSUE=$(gh issue list --search "$ISSUE_TITLE" --state open --json number --jq '.[0] | select(. != null) | .number')
if [ -z "$EXISTING_ISSUE" ] || [ "$EXISTING_ISSUE" = "null" ]; then
gh issue create --title "$ISSUE_TITLE" --body-file triage_report.md --label "triage, automated-report" || gh issue create --title "$ISSUE_TITLE" --body-file triage_report.md
else
echo "A triage issue already exists (#$EXISTING_ISSUE). Updating with new report..."
gh issue comment $EXISTING_ISSUE --body-file triage_report.md
fi
fi
- name: Ensure Cache Files Exist
if: always()
run: |
touch shard_result.json triage_report.md health_check_checkpoint.json
- name: Save Link Checker Cache
if: always()
uses: actions/cache/save@v4
with:
path: |
shard_result.json
triage_report.md
health_check_checkpoint.json
key: link-cleaner-results-${{ github.run_id }}-${{ github.run_attempt }}

789
src/intelligent_health_checker.py
View File

@@ -1,365 +1,424 @@
import asyncio
import json
import os
import re
import httpx
import random
import yaml
import hashlib
from datetime import datetime
import argparse
from typing import Dict, List, Set, Tuple, Optional, Any
from src.config import GH_TOKEN, TARGET_REPO, GEMINI_API_KEY, NUBENETES_CATEGORIES, MADRID_TZ
from src.gitops_manager import RepositoryController
from src.markdown_ast import MarkdownSanitizer
from src.agentic_curator import AgenticCurator
from src.logger import log_event
from src.gemini_utils import call_gemini_with_retry, normalize_url
from src.inventory_manager import load_inventory, get_shard_name, TOTAL_SHARDS
# Configuración de Excepciones
CORE_FILES = ["docs/index.md", "README.md", "docs/about.md"]
MEMORY_FILE = "src/memory/health_learning.json"
class IntelligentLinkCleaner:
def __init__(self, shard_index: int = None, total_shards: int = TOTAL_SHARDS):
self.shard_index = shard_index
self.total_shards = total_shards
self.git_controller = RepositoryController(GH_TOKEN, TARGET_REPO)
self.sanitizer = MarkdownSanitizer()
self.curator = AgenticCurator()
self.ai_semaphore = asyncio.Semaphore(5)
self.link_registry: Dict[str, List[Dict]] = {}
self.dead_links: Dict[str, Tuple[str, str]] = {}
self.learning_data = self._load_memory()
# Determine the shard file if sharded mode
self.shard_file = f"shard_{self.shard_index:02d}.yaml" if self.shard_index is not None else None
self.inventory = load_inventory(self.shard_file)
self.full_report_metrics = []
self.detailed_stats = {"total_scanned": 0, "skipped_recent": 0, "by_file": {}, "operation_types": {"removals": 0, "consolidated": 0, "healed": 0, "enriched": 0}}
self.stats = {"total_links": 0, "dead_links_removed": 0, "orphans_fixed": 0, "enriched_descriptions": 0}
def _load_memory(self) -> Dict:
if os.path.exists(MEMORY_FILE):
try: return json.load(open(MEMORY_FILE, 'r'))
except Exception as e:
log_event(f"[WARN] load health learning memory: {str(e)[:100]}")
return {"domains": {}, "known_soft_404_patterns": []}
def _save_memory(self):
os.makedirs(os.path.dirname(MEMORY_FILE), exist_ok=True)
json.dump(self.learning_data, open(MEMORY_FILE, "w"), indent=2)
async def execute_clean_cycle(self):
log_event("STARTING INTELLIGENT CLEANING CYCLE", section_break=True)
# 1. Map all links and Importance Markers
for root, _, files in os.walk("docs"):
for f in files:
if f.endswith(".md"):
path = os.path.join(root, f)
content = open(path, "r").read()
# Mandate 23: Detect Mosaic boundaries in index.md
mosaic_lines = set()
if path == "docs/index.md":
# Find all center blocks and identify the one that looks like the YouTube mosaic
mosaics = list(re.finditer(r'<center markdown="1">\s*\n(.*?)\n\s*</center>', content, re.DOTALL))
target_mosaic = None
max_yt_links = 0
for m in mosaics:
yt_count = m.group(1).count("youtube.com")
if yt_count > max_yt_links:
max_yt_links = yt_count
target_mosaic = m
if target_mosaic:
mosaic_text = target_mosaic.group(1)
# We find the start line of this specific block
start_pos = target_mosaic.start(1)
start_line = content[:start_pos].count('\n')
line_count = mosaic_text.count('\n') + 1
for i in range(start_line, start_line + line_count):
mosaic_lines.add(i)
log_event(f"[*] YouTube Mosaic identified in docs/index.md (Lines {start_line}-{start_line+line_count})")
lines = content.splitlines()
for idx, line in enumerate(lines):
# Detect standard markdown links
matches = list(re.finditer(r'(\*\*|==)?\s*\[(.*?)\]\((https?://.*?)\)\s*(\*\*|==)?\s*(.*)', line))
# MANDATE 23: Also detect YouTube links in iframes for the videos section
if path == "docs/index.md" and "iframe" in line and "youtube.com" in line:
iframe_match = re.search(r'src="(https?://www\.youtube\.com/.*?)"', line)
if iframe_match:
url = iframe_match.group(1)
nu = normalize_url(url)
# We treat iframes as non-important by default but they must be checked
self.link_registry.setdefault(nu, []).append({
"file": path, "line_index": idx, "url": url,
"is_important": False, "is_iframe": True
})
for m in matches:
fmt_pre, title, url, fmt_post, desc = m.groups()
nu = normalize_url(url)
# MANDATE 23: Skip YouTube Mosaic links in docs/index.md
if path == "docs/index.md" and idx in mosaic_lines and "youtube.com" in nu:
continue
is_important = False
if fmt_pre or fmt_post: is_important = True
if "🌟" in title or "🌟" in (desc or ""): is_important = True
if desc and len(desc.strip()) > 100: is_important = True
if path in CORE_FILES: is_important = True
self.link_registry.setdefault(nu, []).append({
"file": path, "line_index": idx, "url": url,
"is_important": is_important
})
unique_urls = list(self.link_registry.keys())
random.shuffle(unique_urls)
to_check = []
for u in unique_urls:
nu = normalize_url(u)
# Shard filtering
if self.shard_index is not None:
if get_shard_name(nu) != self.shard_file:
continue
entry = self.inventory.get(nu, {})
is_suspicious = False
if entry.get("status") == "online":
path = nu.split("://")[-1].rstrip("/")
if path.count("/") <= 1 or any(kw in path for kw in ["/about", "/products", "/index", "/whats-new"]):
is_suspicious = True
if is_suspicious or entry.get("needs_ai_refresh") or os.getenv("FORCE_FULL_CHECK") == "true":
to_check.append(u)
elif (datetime.now().timestamp() - (entry.get("last_checked") or 0)) > (86400 * 21):
to_check.append(u)
total_to_check = len(to_check)
log_event(f"[*] Queue: {total_to_check} links prioritized for validation.")
# 2. Parallel Network Checks (Optimized with Mandate 22 & 33)
BATCH_SIZE = 15 # Slightly smaller batch to accommodate deep fetch
check_results = {}
for i in range(0, total_to_check, BATCH_SIZE):
batch = to_check[i:i+BATCH_SIZE]
tasks = [self._check_url_logic(url) for url in batch]
results = await asyncio.gather(*tasks)
for url, res in zip(batch, results): check_results[url] = res
if i % 45 == 0: log_event(f" [>] Progress: [{i}/{total_to_check}] links validated...")
# 2.5. UNIVERSAL AI RESCUE
to_rescue = [u for u, res in check_results.items() if not res[0] or res[1] == "generic_redirect_loss"]
if to_rescue:
AI_BATCH_SIZE = 10
for i in range(0, len(to_rescue), AI_BATCH_SIZE):
batch = to_rescue[i:i+AI_BATCH_SIZE]
batch_info = []
for u in batch:
entry = self.inventory.get(normalize_url(u), {})
batch_info.append({"url": u, "title": entry.get("title", u), "context": entry.get("description", "")})
prompt = (
"You act as a Technical Librarian. Rescue missing resources based on title and description.\n"
"Consider acquisitions (Ansible->RedHat, Nginx->F5) and cross-domain moves.\n"
"Return ONLY JSON list: [{\"old_url\": \"...\", \"new_url\": \"...\"}, ...]\n\n"
"RESOURCES:\n" + "\n".join([f"- {d['title']} | {d['url']}" for d in batch_info])
)
try:
async with self.ai_semaphore:
# Mandate 48: Use Flash/Lite for high-volume rescue to avoid Rate-Limits
ai_results = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=True, role="Link-Rescue")
if isinstance(ai_results, list):
res_map = {normalize_url(r.get("old_url", "")): r.get("new_url") for r in ai_results}
for u in batch:
new_loc = res_map.get(normalize_url(u))
if new_loc and new_loc.startswith("http") and "NONE" not in new_loc.upper():
try:
async with httpx.AsyncClient(timeout=10, follow_redirects=True, verify=False) as client:
resp_new = await client.get(new_loc)
if resp_new.status_code < 400:
final_new = str(resp_new.url)
u_p = u.split("://")[-1].rstrip("/")
f_p = final_new.split("://")[-1].rstrip("/")
if u_p.count("/") >= 3 and (f_p.count("/") <= 2 or any(kw in f_p for kw in ["/about", "/products", "/home", "/learn"])):
log_event(f" [⚠️] AI proposed a generic redirect for {u} -> {new_loc}. Rejecting.")
else:
log_event(f" [✨] RESCUED: {u} -> {new_loc}")
check_results[u] = (True, "resurrected", new_loc)
except Exception as e:
log_event(f"[WARN] verify rescued URL {new_loc[:50]}: {str(e)[:100]}")
except Exception as e:
log_event(f"[WARN] AI rescue batch: {str(e)[:100]}")
# 2.8. Finalize Status
log_event("FINALIZING STATUS AND METRICS...", section_break=True)
for url, (alive, reason, final) in check_results.items():
nu = normalize_url(url); entry = self.inventory.get(nu, {})
score = entry.get("health_score")
if score is None:
score = 100.0
score = (score * 0.8) + (100 if alive else 0) * 0.2
entry["health_score"] = round(score, 1); entry["last_checked"] = datetime.now().timestamp()
# Mandate 22 & 33 were handled in _check_url_logic to avoid sequential bottlenecks
is_important = any(occ.get("is_important") for occ in self.link_registry.get(nu, []))
if (entry.get("stars") or 0) >= 3: is_important = True
status, final_reason = ("INCLUDED", reason) if alive else ("FILTERED", reason)
if not alive or reason == "generic_redirect_loss":
if is_important:
entry["status"] = "review_required"
entry["review_metadata"] = {
"original_url": url, "proposed_url": final if final else "NONE",
"reason": f"High-Value Protection: {reason}", "timestamp": datetime.now().isoformat()
}
log_event(f" [⚠️] REVIEW STORED: {url}")
status, final_reason = "INCLUDED", f"Preserved for Review ({reason})"
elif score < 20:
entry["status"] = "dead"; self.dead_links[url] = (None, reason)
status, final_reason = "FILTERED", f"Dead: {reason}"
elif final and alive:
self.dead_links[url] = (f"CANONICAL:{final}", "Redirect/Resurrection")
final_reason = "Updated (Canonical)"
self.full_report_metrics.append({
"url": url, "status": status, "reason": final_reason,
"category": entry.get("category", "N/A"), "post_date": entry.get("pub_date"),
"source": "Health Checker", "impact_score": (entry.get("stars") or 0) * 20,
"language": entry.get("language", "EN"), "type": entry.get("resource_type", "Ref")
})
self.inventory[nu] = entry
await self.apply_changes()
async def apply_changes(self):
# Export JSON artifact for the reducer instead of applying Git changes directly
output_payload = {
"shard_index": self.shard_index,
"inventory_updates": self.inventory,
"dead_links": self.dead_links,
"full_report_metrics": self.full_report_metrics
}
out_name = f"shard_result_{self.shard_index:02d}.json" if self.shard_index is not None else "shard_result.json"
with open(out_name, "w") as f:
json.dump(output_payload, f, indent=2)
log_event(f"EXPORTED SHARD RESULTS TO {out_name}", section_break=True)
# --- AUTOMATED TRIAGE REPORT GENERATION ---
triage_links = []
for url, meta in self.inventory.items():
if meta.get('status') == 'review_required':
triage_links.append({"url": url, "stars": meta.get('stars') or 0, "desc": meta.get('description', 'N/A')})
if triage_links:
# Sort by stars (impact) DESC
triage_links.sort(key=lambda x: x['stars'], reverse=True)
with open("triage_report.md", "w") as f:
f.write(f"### 🚨 Manual Triage Required ({len(triage_links)} High-Value Links)\n\n")
f.write("The following resources were flagged for manual review because they failed health checks but are considered high-value assets.\n\n")
f.write("| Impact | Resource | Description |\n| :---: | :--- | :--- |\n")
for item in triage_links:
f.write(f"| {'🌟'*item['stars']} | {item['url']} | {item['desc']} |\n")
async def _check_url_logic(self, url: str) -> Tuple[bool, str, Optional[str]]:
headers = {"User-Agent": "Mozilla/5.0", "Accept-Language": "en-US,en;q=0.5"}
parked = ["buy this domain", "parked free", "domain is for sale"]
nu = normalize_url(url); entry = self.inventory.get(nu, {})
try:
async with httpx.AsyncClient(headers=headers, follow_redirects=True, timeout=12) as client:
resp = await client.get(url)
if resp.status_code < 400:
text = resp.text.lower()
final_url = str(resp.url)
from src.gemini_utils import sanitize_trailing_slashes
final_url = sanitize_trailing_slashes(final_url)
if any(kw in text for kw in parked): return False, "parked", None
# Mandate 22: Content Drift detection (SHA256) - Integrated here for speed
new_hash = hashlib.sha256(resp.text.encode()).hexdigest()
old_hash = entry.get("content_hash", "N/A")
if old_hash != "N/A" and new_hash != old_hash:
log_event(f" [!] DRIFT DETECTED: {url}")
entry["needs_ai_refresh"] = True
entry["content_hash"] = new_hash
# Mandate 33: License Guard for GitHub
if "github.com" in url:
try:
from src.agentic_curator import _get_github_activity
gh_meta = await _get_github_activity(url)
if gh_meta.get("gh_license"):
old_lic = entry.get("gh_license", "N/A")
new_lic = gh_meta["gh_license"]
if old_lic != "N/A" and old_lic != new_lic:
if any(x in new_lic.upper() for x in ["BSL", "SSPL", "PROPRIETARY"]):
log_event(f" [⚖️] LICENSE ALERT: {url} -> {new_lic}")
entry["status"] = "review_required"
entry["gh_license"] = new_lic
except Exception as e:
log_event(f"[WARN] license guard check for {url}: {str(e)[:100]}")
if final_url != url:
u_p = url.split("://")[-1].rstrip("/"); f_p = final_url.split("://")[-1].rstrip("/")
if u_p == f_p: return True, "normalized_slashes", final_url
if u_p.count("/") >= 3 and (f_p.count("/") <= 2 or any(kw in f_p for kw in ["/about", "/products", "/home", "/learn"])):
return False, "generic_redirect_loss", None
return True, "OK", final_url if final_url != url else None
if resp.status_code in [404, 410]:
if "github.com" in url:
if "/master/" in url:
h = url.replace("/master/", "/main/")
try:
if (await client.get(h)).status_code < 400: return True, "healed", h
except Exception as e:
log_event(f"[WARN] heal master->main for {url}: {str(e)[:100]}")
m = re.search(r'(https?://github\.com/[^/]+/[^/]+)', url)
if m and (await client.get(m.group(1))).status_code < 400: return True, "consolidated", m.group(1)
return False, "404", None
return True, f"Soft Block {resp.status_code}", None
except Exception as e:
log_event(f"[WARN] URL check logic for {url}: {str(e)[:100]}")
return True, "Error", None
async def prune_orphaned_metadata(self):
valid_map = {}
for root, _, files in os.walk("docs"):
for f in files:
if f.endswith(".md"):
p = os.path.join(root, f); c = open(p, "r").read()
for u in re.findall(r'\[.*?\]\((https?://.*?)\)', c): valid_map.setdefault(normalize_url(u), []).append(p)
new_inv = {}
for u, m in self.inventory.items():
if u.startswith("INTRO:") or u in valid_map:
if u in valid_map: m["v1_locations"] = sorted(list(set(valid_map[u])))
new_inv[u] = m
self.inventory = new_inv
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--shard-index", type=int, default=None, help="Index of the shard (0-63)")
parser.add_argument("--total-shards", type=int, default=TOTAL_SHARDS, help="Total number of shards")
args = parser.parse_args()
cleaner = IntelligentLinkCleaner(shard_index=args.shard_index, total_shards=args.total_shards)
asyncio.run(cleaner.execute_clean_cycle())
import asyncio
import json
import os
import re
import httpx
import random
import yaml
import hashlib
from datetime import datetime
import argparse
from typing import Dict, List, Set, Tuple, Optional, Any
from src.config import GH_TOKEN, TARGET_REPO, GEMINI_API_KEY, NUBENETES_CATEGORIES, MADRID_TZ
from src.gitops_manager import RepositoryController
from src.markdown_ast import MarkdownSanitizer
from src.agentic_curator import AgenticCurator
from src.logger import log_event
from src.gemini_utils import call_gemini_with_retry, normalize_url
from src.inventory_manager import load_inventory, get_shard_name, TOTAL_SHARDS
# Configuración de Excepciones
CORE_FILES = ["docs/index.md", "README.md", "docs/about.md"]
MEMORY_FILE = "src/memory/health_learning.json"
class IntelligentLinkCleaner:
def __init__(self, shard_index: int = None, total_shards: int = TOTAL_SHARDS):
self.shard_index = shard_index
self.total_shards = total_shards
self.git_controller = RepositoryController(GH_TOKEN, TARGET_REPO)
self.sanitizer = MarkdownSanitizer()
self.curator = AgenticCurator()
self.ai_semaphore = asyncio.Semaphore(5)
self.link_registry: Dict[str, List[Dict]] = {}
self.dead_links: Dict[str, Tuple[str, str]] = {}
self.learning_data = self._load_memory()
# Determine the shard file if sharded mode
self.shard_file = f"shard_{self.shard_index:02d}.yaml" if self.shard_index is not None else None
self.inventory = load_inventory(self.shard_file)
# Determine checkpoint file and load it
self.checkpoint_file = f"checkpoint_{self.shard_index:02d}.json" if self.shard_index is not None else "health_check_checkpoint.json"
self.checkpoint_data = self._load_checkpoint()
self.full_report_metrics = []
self.detailed_stats = {"total_scanned": 0, "skipped_recent": 0, "by_file": {}, "operation_types": {"removals": 0, "consolidated": 0, "healed": 0, "enriched": 0}}
self.stats = {"total_links": 0, "dead_links_removed": 0, "orphans_fixed": 0, "enriched_descriptions": 0}
def _load_memory(self) -> Dict:
if os.path.exists(MEMORY_FILE):
try: return json.load(open(MEMORY_FILE, 'r'))
except Exception as e:
log_event(f"[WARN] load health learning memory: {str(e)[:100]}")
return {"domains": {}, "known_soft_404_patterns": []}
def _save_memory(self):
os.makedirs(os.path.dirname(MEMORY_FILE), exist_ok=True)
json.dump(self.learning_data, open(MEMORY_FILE, "w"), indent=2)
def _load_checkpoint(self) -> Dict:
if os.path.exists(self.checkpoint_file) and os.path.getsize(self.checkpoint_file) > 0:
try:
with open(self.checkpoint_file, "r") as f:
data = json.load(f)
log_event(f"[*] Resuming from checkpoint file: {self.checkpoint_file}")
# Restore checkpoint data into self variables
self.inventory.update(data.get("inventory_updates", {}))
return data
except Exception as e:
log_event(f"[WARN] Failed to load checkpoint {self.checkpoint_file}: {str(e)[:100]}")
return {
"checked_urls": {}, # url -> [alive, reason, final_url]
"inventory_updates": {}, # url -> inventory_entry
"rescued_urls": [] # list of URLs that we attempted to rescue
}
def _save_checkpoint(self):
try:
with open(self.checkpoint_file, "w") as f:
json.dump(self.checkpoint_data, f, indent=2)
except Exception as e:
log_event(f"[WARN] Failed to save checkpoint {self.checkpoint_file}: {str(e)[:100]}")
def _delete_checkpoint(self):
if os.path.exists(self.checkpoint_file):
try:
os.remove(self.checkpoint_file)
log_event(f"[*] Cleaned up checkpoint file: {self.checkpoint_file}")
except Exception as e:
log_event(f"[WARN] Failed to delete checkpoint {self.checkpoint_file}: {str(e)[:100]}")
async def execute_clean_cycle(self):
log_event("STARTING INTELLIGENT CLEANING CYCLE", section_break=True)
# 1. Map all links and Importance Markers
for root, _, files in os.walk("docs"):
for f in files:
if f.endswith(".md"):
path = os.path.join(root, f)
content = open(path, "r").read()
# Mandate 23: Detect Mosaic boundaries in index.md
mosaic_lines = set()
if path == "docs/index.md":
# Find all center blocks and identify the one that looks like the YouTube mosaic
mosaics = list(re.finditer(r'<center markdown="1">\s*\n(.*?)\n\s*</center>', content, re.DOTALL))
target_mosaic = None
max_yt_links = 0
for m in mosaics:
yt_count = m.group(1).count("youtube.com")
if yt_count > max_yt_links:
max_yt_links = yt_count
target_mosaic = m
if target_mosaic:
mosaic_text = target_mosaic.group(1)
# We find the start line of this specific block
start_pos = target_mosaic.start(1)
start_line = content[:start_pos].count('\n')
line_count = mosaic_text.count('\n') + 1
for i in range(start_line, start_line + line_count):
mosaic_lines.add(i)
log_event(f"[*] YouTube Mosaic identified in docs/index.md (Lines {start_line}-{start_line+line_count})")
lines = content.splitlines()
for idx, line in enumerate(lines):
# Detect standard markdown links
matches = list(re.finditer(r'(\*\*|==)?\s*\[(.*?)\]\((https?://.*?)\)\s*(\*\*|==)?\s*(.*)', line))
# MANDATE 23: Also detect YouTube links in iframes for the videos section
if path == "docs/index.md" and "iframe" in line and "youtube.com" in line:
iframe_match = re.search(r'src="(https?://www\.youtube\.com/.*?)"', line)
if iframe_match:
url = iframe_match.group(1)
nu = normalize_url(url)
# We treat iframes as non-important by default but they must be checked
self.link_registry.setdefault(nu, []).append({
"file": path, "line_index": idx, "url": url,
"is_important": False, "is_iframe": True
})
for m in matches:
fmt_pre, title, url, fmt_post, desc = m.groups()
nu = normalize_url(url)
# MANDATE 23: Skip YouTube Mosaic links in docs/index.md
if path == "docs/index.md" and idx in mosaic_lines and "youtube.com" in nu:
continue
is_important = False
if fmt_pre or fmt_post: is_important = True
if "🌟" in title or "🌟" in (desc or ""): is_important = True
if desc and len(desc.strip()) > 100: is_important = True
if path in CORE_FILES: is_important = True
self.link_registry.setdefault(nu, []).append({
"file": path, "line_index": idx, "url": url,
"is_important": is_important
})
unique_urls = list(self.link_registry.keys())
random.shuffle(unique_urls)
to_check = []
for u in unique_urls:
nu = normalize_url(u)
# Shard filtering
if self.shard_index is not None:
if get_shard_name(nu) != self.shard_file:
continue
# Checkpoint filtering: skip if already checked in checkpoint
if nu in self.checkpoint_data["checked_urls"]:
continue
entry = self.inventory.get(nu, {})
is_suspicious = False
if entry.get("status") == "online":
path = nu.split("://")[-1].rstrip("/")
if path.count("/") <= 1 or any(kw in path for kw in ["/about", "/products", "/index", "/whats-new"]):
is_suspicious = True
if is_suspicious or entry.get("needs_ai_refresh") or os.getenv("FORCE_FULL_CHECK") == "true":
to_check.append(u)
elif (datetime.now().timestamp() - (entry.get("last_checked") or 0)) > (86400 * 21):
to_check.append(u)
total_to_check = len(to_check)
log_event(f"[*] Queue: {total_to_check} links prioritized for validation (skipped {len(self.checkpoint_data['checked_urls'])} already checked).")
# 2. Parallel Network Checks (Optimized with Mandate 22 & 33)
BATCH_SIZE = 15 # Slightly smaller batch to accommodate deep fetch
check_results = {}
for nu, res in self.checkpoint_data["checked_urls"].items():
check_results[nu] = tuple(res)
for i in range(0, total_to_check, BATCH_SIZE):
batch = to_check[i:i+BATCH_SIZE]
tasks = [self._check_url_logic(url) for url in batch]
results = await asyncio.gather(*tasks)
for url, res in zip(batch, results):
check_results[url] = res
nu = normalize_url(url)
self.checkpoint_data["checked_urls"][nu] = list(res)
self.checkpoint_data["inventory_updates"][nu] = self.inventory.get(nu)
self._save_checkpoint()
if i % 45 == 0: log_event(f" [>] Progress: [{i}/{total_to_check}] links validated...")
# 2.5. UNIVERSAL AI RESCUE
to_rescue = [u for u, res in check_results.items() if not res[0] or res[1] == "generic_redirect_loss"]
to_rescue = [u for u in to_rescue if normalize_url(u) not in self.checkpoint_data["rescued_urls"]]
if to_rescue:
AI_BATCH_SIZE = 10
for i in range(0, len(to_rescue), AI_BATCH_SIZE):
batch = to_rescue[i:i+AI_BATCH_SIZE]
batch_info = []
for u in batch:
entry = self.inventory.get(normalize_url(u), {})
batch_info.append({"url": u, "title": entry.get("title", u), "context": entry.get("description", "")})
prompt = (
"You act as a Technical Librarian. Rescue missing resources based on title and description.\n"
"Consider acquisitions (Ansible->RedHat, Nginx->F5) and cross-domain moves.\n"
"Return ONLY JSON list: [{\"old_url\": \"...\", \"new_url\": \"...\"}, ...]\n\n"
"RESOURCES:\n" + "\n".join([f"- {d['title']} | {d['url']}" for d in batch_info])
)
try:
async with self.ai_semaphore:
# Mandate 48: Use Flash/Lite for high-volume rescue to avoid Rate-Limits
ai_results = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=True, role="Link-Rescue")
if isinstance(ai_results, list):
res_map = {normalize_url(r.get("old_url", "")): r.get("new_url") for r in ai_results}
for u in batch:
nu = normalize_url(u)
if nu not in self.checkpoint_data["rescued_urls"]:
self.checkpoint_data["rescued_urls"].append(nu)
new_loc = res_map.get(nu)
if new_loc and new_loc.startswith("http") and "NONE" not in new_loc.upper():
try:
async with httpx.AsyncClient(timeout=10, follow_redirects=True, verify=False) as client:
resp_new = await client.get(new_loc)
if resp_new.status_code < 400:
final_new = str(resp_new.url)
u_p = u.split("://")[-1].rstrip("/")
f_p = final_new.split("://")[-1].rstrip("/")
if u_p.count("/") >= 3 and (f_p.count("/") <= 2 or any(kw in f_p for kw in ["/about", "/products", "/home", "/learn"])):
log_event(f" [⚠️] AI proposed a generic redirect for {u} -> {new_loc}. Rejecting.")
else:
log_event(f" [✨] RESCUED: {u} -> {new_loc}")
check_results[u] = (True, "resurrected", new_loc)
self.checkpoint_data["checked_urls"][nu] = [True, "resurrected", new_loc]
except Exception as e:
log_event(f"[WARN] verify rescued URL {new_loc[:50]}: {str(e)[:100]}")
self._save_checkpoint()
except Exception as e:
log_event(f"[WARN] AI rescue batch: {str(e)[:100]}")
# 2.8. Finalize Status
log_event("FINALIZING STATUS AND METRICS...", section_break=True)
for url, (alive, reason, final) in check_results.items():
nu = normalize_url(url); entry = self.inventory.get(nu, {})
score = entry.get("health_score")
if score is None:
score = 100.0
score = (score * 0.8) + (100 if alive else 0) * 0.2
entry["health_score"] = round(score, 1); entry["last_checked"] = datetime.now().timestamp()
# Mandate 22 & 33 were handled in _check_url_logic to avoid sequential bottlenecks
is_important = any(occ.get("is_important") for occ in self.link_registry.get(nu, []))
if (entry.get("stars") or 0) >= 3: is_important = True
status, final_reason = ("INCLUDED", reason) if alive else ("FILTERED", reason)
if not alive or reason == "generic_redirect_loss":
if is_important:
entry["status"] = "review_required"
entry["review_metadata"] = {
"original_url": url, "proposed_url": final if final else "NONE",
"reason": f"High-Value Protection: {reason}", "timestamp": datetime.now().isoformat()
}
log_event(f" [⚠️] REVIEW STORED: {url}")
status, final_reason = "INCLUDED", f"Preserved for Review ({reason})"
elif score < 20:
entry["status"] = "dead"; self.dead_links[url] = (None, reason)
status, final_reason = "FILTERED", f"Dead: {reason}"
elif final and alive:
self.dead_links[url] = (f"CANONICAL:{final}", "Redirect/Resurrection")
final_reason = "Updated (Canonical)"
self.full_report_metrics.append({
"url": url, "status": status, "reason": final_reason,
"category": entry.get("category", "N/A"), "post_date": entry.get("pub_date"),
"source": "Health Checker", "impact_score": (entry.get("stars") or 0) * 20,
"language": entry.get("language", "EN"), "type": entry.get("resource_type", "Ref")
})
self.inventory[nu] = entry
await self.apply_changes()
self._delete_checkpoint()
async def apply_changes(self):
# Export JSON artifact for the reducer instead of applying Git changes directly
output_payload = {
"shard_index": self.shard_index,
"inventory_updates": self.inventory,
"dead_links": self.dead_links,
"full_report_metrics": self.full_report_metrics
}
out_name = f"shard_result_{self.shard_index:02d}.json" if self.shard_index is not None else "shard_result.json"
with open(out_name, "w") as f:
json.dump(output_payload, f, indent=2)
log_event(f"EXPORTED SHARD RESULTS TO {out_name}", section_break=True)
# --- AUTOMATED TRIAGE REPORT GENERATION ---
triage_links = []
for url, meta in self.inventory.items():
if meta.get('status') == 'review_required':
triage_links.append({"url": url, "stars": meta.get('stars') or 0, "desc": meta.get('description', 'N/A')})
if triage_links:
# Sort by stars (impact) DESC
triage_links.sort(key=lambda x: x['stars'], reverse=True)
with open("triage_report.md", "w") as f:
f.write(f"### 🚨 Manual Triage Required ({len(triage_links)} High-Value Links)\n\n")
f.write("The following resources were flagged for manual review because they failed health checks but are considered high-value assets.\n\n")
f.write("| Impact | Resource | Description |\n| :---: | :--- | :--- |\n")
for item in triage_links:
f.write(f"| {'🌟'*item['stars']} | {item['url']} | {item['desc']} |\n")
async def _check_url_logic(self, url: str) -> Tuple[bool, str, Optional[str]]:
headers = {"User-Agent": "Mozilla/5.0", "Accept-Language": "en-US,en;q=0.5"}
parked = ["buy this domain", "parked free", "domain is for sale"]
nu = normalize_url(url); entry = self.inventory.get(nu, {})
try:
async with httpx.AsyncClient(headers=headers, follow_redirects=True, timeout=12) as client:
resp = await client.get(url)
if resp.status_code < 400:
text = resp.text.lower()
final_url = str(resp.url)
from src.gemini_utils import sanitize_trailing_slashes
final_url = sanitize_trailing_slashes(final_url)
if any(kw in text for kw in parked): return False, "parked", None
# Mandate 22: Content Drift detection (SHA256) - Integrated here for speed
new_hash = hashlib.sha256(resp.text.encode()).hexdigest()
old_hash = entry.get("content_hash", "N/A")
if old_hash != "N/A" and new_hash != old_hash:
log_event(f" [!] DRIFT DETECTED: {url}")
entry["needs_ai_refresh"] = True
entry["content_hash"] = new_hash
# Mandate 33: License Guard for GitHub
if "github.com" in url:
try:
from src.agentic_curator import _get_github_activity
gh_meta = await _get_github_activity(url)
if gh_meta.get("gh_license"):
old_lic = entry.get("gh_license", "N/A")
new_lic = gh_meta["gh_license"]
if old_lic != "N/A" and old_lic != new_lic:
if any(x in new_lic.upper() for x in ["BSL", "SSPL", "PROPRIETARY"]):
log_event(f" [⚖️] LICENSE ALERT: {url} -> {new_lic}")
entry["status"] = "review_required"
entry["gh_license"] = new_lic
except Exception as e:
log_event(f"[WARN] license guard check for {url}: {str(e)[:100]}")
if final_url != url:
u_p = url.split("://")[-1].rstrip("/"); f_p = final_url.split("://")[-1].rstrip("/")
if u_p == f_p: return True, "normalized_slashes", final_url
if u_p.count("/") >= 3 and (f_p.count("/") <= 2 or any(kw in f_p for kw in ["/about", "/products", "/home", "/learn"])):
return False, "generic_redirect_loss", None
return True, "OK", final_url if final_url != url else None
if resp.status_code in [404, 410]:
if "github.com" in url:
if "/master/" in url:
h = url.replace("/master/", "/main/")
try:
if (await client.get(h)).status_code < 400: return True, "healed", h
except Exception as e:
log_event(f"[WARN] heal master->main for {url}: {str(e)[:100]}")
m = re.search(r'(https?://github\.com/[^/]+/[^/]+)', url)
if m and (await client.get(m.group(1))).status_code < 400: return True, "consolidated", m.group(1)
return False, "404", None
return True, f"Soft Block {resp.status_code}", None
except Exception as e:
log_event(f"[WARN] URL check logic for {url}: {str(e)[:100]}")
return True, "Error", None
async def prune_orphaned_metadata(self):
valid_map = {}
for root, _, files in os.walk("docs"):
for f in files:
if f.endswith(".md"):
p = os.path.join(root, f); c = open(p, "r").read()
for u in re.findall(r'\[.*?\]\((https?://.*?)\)', c): valid_map.setdefault(normalize_url(u), []).append(p)
new_inv = {}
for u, m in self.inventory.items():
if u.startswith("INTRO:") or u in valid_map:
if u in valid_map: m["v1_locations"] = sorted(list(set(valid_map[u])))
new_inv[u] = m
self.inventory = new_inv
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--shard-index", type=int, default=None, help="Index of the shard (0-63)")
parser.add_argument("--total-shards", type=int, default=TOTAL_SHARDS, help="Total number of shards")
args = parser.parse_args()
cleaner = IntelligentLinkCleaner(shard_index=args.shard_index, total_shards=args.total_shards)
asyncio.run(cleaner.execute_clean_cycle())