"""Overview stats and SSE activity stream.""" import asyncio import json import subprocess import sqlite3 from datetime import date from fastapi import APIRouter from sse_starlette.sse import EventSourceResponse import httpx import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from lib_bridge import ( portainer_list_containers, ENDPOINTS, ollama_available, GMAIL_DB, DVISH_DB, PROTON_DB, RESTART_DB, LOG_DIR, OLLAMA_URL, ) from log_parser import get_recent_events, tail_logs, get_new_lines router = APIRouter(tags=["overview"]) def _count_today_emails(db_path: Path) -> int: """Count emails processed today from a processed.db file.""" if not db_path.exists(): return 0 try: today = date.today().isoformat() conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) cur = conn.execute( "SELECT COUNT(*) FROM processed WHERE processed_at LIKE ?", (f"{today}%",), ) count = cur.fetchone()[0] conn.close() return count except Exception: return 0 def _count_unhealthy(db_path: Path) -> int: """Count unhealthy containers from stack-restart.db.""" if not db_path.exists(): return 0 try: conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) cur = conn.execute("SELECT COUNT(*) FROM unhealthy_tracking") count = cur.fetchone()[0] conn.close() return count except Exception: return 0 def _gpu_info() -> dict: """Get GPU info from olares via SSH.""" try: result = subprocess.run( ["ssh", "-o", "ConnectTimeout=3", "olares", "nvidia-smi --query-gpu=temperature.gpu,power.draw,power.limit," "memory.used,memory.total,utilization.gpu --format=csv,noheader,nounits"], capture_output=True, text=True, timeout=10, ) if result.returncode != 0: return {"available": False} parts = [p.strip() for p in result.stdout.strip().split(",")] def _f(v): try: return float(v) except (ValueError, TypeError): return None if len(parts) >= 6: return { "available": True, "temp_c": _f(parts[0]), "power_draw_w": _f(parts[1]), "power_limit_w": _f(parts[2]), "memory_used_mb": _f(parts[3]), "memory_total_mb": _f(parts[4]), "utilization_pct": _f(parts[5]), } except Exception: pass return {"available": False} @router.get("/stats/overview") def stats_overview(): """Aggregate overview stats.""" # Container counts container_counts = {} total = 0 for ep_name in ENDPOINTS: try: containers = portainer_list_containers(ep_name) running = sum(1 for c in containers if c.get("State") == "running") container_counts[ep_name] = {"total": len(containers), "running": running} total += len(containers) except Exception: container_counts[ep_name] = {"total": 0, "running": 0, "error": True} # GPU gpu = _gpu_info() # Email counts email_today = { "gmail": _count_today_emails(GMAIL_DB), "dvish": _count_today_emails(DVISH_DB), "proton": _count_today_emails(PROTON_DB), } email_today["total"] = sum(email_today.values()) # Unhealthy unhealthy = _count_unhealthy(RESTART_DB) # Ollama ollama_up = ollama_available(OLLAMA_URL) return { "containers": {"total": total, "by_endpoint": container_counts}, "gpu": gpu, "email_today": email_today, "unhealthy_count": unhealthy, "ollama_available": ollama_up, } @router.get("/activity") async def activity_stream(): """SSE stream of today's automation events.""" async def event_generator(): # Send initial batch events = get_recent_events(LOG_DIR) yield {"event": "init", "data": json.dumps(events)} # Poll for new events positions = tail_logs(LOG_DIR) while True: await asyncio.sleep(5) new_events, positions = get_new_lines(LOG_DIR, positions) if new_events: yield {"event": "update", "data": json.dumps(new_events)} return EventSourceResponse(event_generator()) @router.post("/actions/pause-organizers") def pause_organizers(): """Pause all email organizer cron jobs.""" result = subprocess.run( ["/home/homelab/organized/repos/homelab/scripts/gmail-organizer-ctl.sh", "stop"], capture_output=True, text=True, timeout=10, ) return {"success": result.returncode == 0, "output": result.stdout.strip()} @router.post("/actions/resume-organizers") def resume_organizers(): """Resume all email organizer cron jobs.""" result = subprocess.run( ["/home/homelab/organized/repos/homelab/scripts/gmail-organizer-ctl.sh", "start"], capture_output=True, text=True, timeout=10, ) return {"success": result.returncode == 0, "output": result.stdout.strip()} @router.get("/actions/organizer-status") def organizer_status(): """Check if organizers are running or paused.""" result = subprocess.run( ["/home/homelab/organized/repos/homelab/scripts/gmail-organizer-ctl.sh", "status"], capture_output=True, text=True, timeout=10, ) return {"output": result.stdout.strip()} @router.get("/calendar") def get_calendar_events(): """Fetch upcoming events from Baikal CalDAV.""" import re from datetime import datetime, timezone BAIKAL_URL = "http://192.168.0.200:12852/dav.php/calendars/vish/default/" BAIKAL_USER = "vish" BAIKAL_PASS = "REDACTED_PASSWORD" today = datetime.now(timezone.utc).strftime("%Y%m%dT000000Z") body = f''' ''' try: auth = httpx.DigestAuth(BAIKAL_USER, BAIKAL_PASS) with httpx.Client(timeout=10) as client: r = client.request("REPORT", BAIKAL_URL, content=body, headers={"Content-Type": "application/xml", "Depth": "1"}, auth=auth) r.raise_for_status() # Parse iCal events summaries = re.findall(r'SUMMARY:(.*?)(?:\r?\n)', r.text) starts = re.findall(r'DTSTART[^:]*:(.*?)(?:\r?\n)', r.text) locations = re.findall(r'LOCATION:(.*?)(?:\r?\n)', r.text) events = [] now = datetime.now(timezone.utc) for i, (start, summary) in enumerate(zip(starts, summaries)): # Parse date — handle both date and datetime formats try: if len(start) == 8: dt = datetime.strptime(start, "%Y%m%d").replace(tzinfo=timezone.utc) else: clean = start.replace("Z", "") dt = datetime.strptime(clean[:15], "%Y%m%dT%H%M%S").replace(tzinfo=timezone.utc) except ValueError: continue # Only future events if dt < now: continue # Clean up summary (unescape iCal) clean_summary = summary.replace("\\,", ",").replace("\\;", ";").replace("&", "&") events.append({ "summary": clean_summary, "start": dt.isoformat(), "date": dt.strftime("%b %d"), "time": dt.strftime("%I:%M %p") if len(start) > 8 else "All day", "location": locations[i].replace("\\,", ",").replace("\\n", ", ") if i < len(locations) else None, }) # Sort by date, limit to next 8 events.sort(key=lambda e: e["start"]) return {"events": events[:8], "total": len(events)} except Exception as e: return {"events": [], "error": str(e)} def _search_repo_docs(query: str, max_chars: int = 2000) -> str: """Search repo docs/scripts for relevant snippets. Lightweight keyword match.""" import re repo = Path("/app/scripts").parent if Path("/app/scripts").exists() else Path(__file__).parent.parent.parent.parent search_dirs = [repo / "docs" / "services" / "individual", repo / "scripts", repo / "docs"] keywords = [w.lower() for w in re.findall(r'\w{3,}', query) if w.lower() not in { "the", "how", "what", "does", "can", "are", "this", "that", "have", "many", "much", "about", "from", "with", "your", "there", "which", }] if not keywords: return "" # Add aliases so related terms find each other aliases = {"tailscale": "headscale", "headscale": "tailscale", "gpu": "nvidia", "jellyfin": "olares", "containers": "portainer", "dns": "adguard"} extra = [aliases[k] for k in keywords if k in aliases] keywords = list(set(keywords + extra)) scored = [] for search_dir in search_dirs: if not search_dir.exists(): continue for f in search_dir.rglob("*.md"): try: text = f.read_text(errors="ignore")[:8000] score = sum(text.lower().count(kw) for kw in keywords) if score > 0: scored.append((score, f, text)) except Exception: continue for f in search_dir.rglob("*.py"): if f.name.startswith("__"): continue try: # Only read the docstring/header, not full scripts text = f.read_text(errors="ignore")[:1000] score = sum(text.lower().count(kw) for kw in keywords) if score > 0: scored.append((score, f, text)) except Exception: continue if not scored: return "" scored.sort(key=lambda x: -x[0]) snippets = [] total = 0 for _, path, text in scored[:2]: # max 2 files # Trim to relevant section — find paragraphs with keywords lines = text.split("\n") relevant = [] for i, line in enumerate(lines): if any(kw in line.lower() for kw in keywords): start = max(0, i - 2) end = min(len(lines), i + 5) relevant.extend(lines[start:end]) snippet = "\n".join(dict.fromkeys(relevant))[:1000] # dedup, cap at 1K if not snippet.strip(): snippet = text[:500] snippets.append(f"[{path.name}]\n{snippet}") total += len(snippet) if total >= max_chars: break return "\n\n".join(snippets) @router.post("/chat") def chat_with_ollama(body: dict): """Chat with Ollama using live homelab context + repo docs.""" message = body.get("message", "") if not message: return {"error": "No message provided"} # Gather live context from multiple sources context_parts = [] try: overview = get_overview() containers = overview.get("containers", {}) gpu = overview.get("gpu", {}) context_parts.append( f"Containers: {containers.get('total', '?')} total across endpoints: " + ", ".join(f"{k} ({v.get('total','?')} containers, {v.get('running','?')} running)" for k, v in containers.get("by_endpoint", {}).items()) ) if gpu.get("available"): context_parts.append( f"GPU: {gpu.get('name','RTX 5090')}, {gpu.get('temp_c','?')}°C, " f"{gpu.get('memory_used_mb','?')}/{gpu.get('memory_total_mb','?')} MB VRAM, " f"{gpu.get('utilization_pct','?')}% util" ) email_data = overview.get("email_today", {}) if isinstance(email_data, dict): context_parts.append(f"Emails today: {email_data.get('total', 0)} (dvish: {email_data.get('dvish', 0)}, proton: {email_data.get('proton', 0)})") context_parts.append(f"Ollama: {'online' if overview.get('ollama_available') else 'offline'}") context_parts.append(f"Unhealthy containers: {overview.get('unhealthy_count', 0)}") except Exception: context_parts.append("(could not fetch live stats)") # Fetch Headscale nodes if question mentions network/tailscale/headscale/nodes msg_lower = message.lower() if any(kw in msg_lower for kw in ["tailscale", "headscale", "node", "mesh", "vpn", "network"]): try: import json as _json hs_result = subprocess.run( ["ssh", "-o", "ConnectTimeout=3", "calypso", "/usr/local/bin/docker exec headscale headscale nodes list -o json"], capture_output=True, text=True, timeout=10, ) if hs_result.returncode == 0: nodes = _json.loads(hs_result.stdout) online = [n for n in nodes if n.get("online")] node_names = ", ".join(n.get("givenName") or n.get("name", "?") for n in nodes) context_parts.append(f"Headscale/Tailscale: {len(nodes)} nodes ({len(online)} online): {node_names}") else: context_parts.append("Headscale: 26 nodes (could not fetch live list, but documented as 26)") except Exception: context_parts.append("Headscale: 26 nodes (documented, could not fetch live)") # Fetch Jellyfin status if question mentions media/jellyfin/streaming if any(kw in msg_lower for kw in ["jellyfin", "media", "stream", "movie", "tv", "playing"]): try: from routers.media import jellyfin_status jf = jellyfin_status() libs = ", ".join(f"{l['name']} ({l['type']})" for l in jf.get("libraries", [])) active = jf.get("active_sessions", []) playing = ", ".join(f"{s['title']} by {s['user']}" for s in active) if active else "nothing" context_parts.append(f"Jellyfin v{jf.get('version','?')}: libraries={libs}. Now playing: {playing}") except Exception: pass # Fetch AdGuard stats if question mentions dns/adguard/blocked if any(kw in msg_lower for kw in ["dns", "adguard", "blocked", "queries", "domain"]): try: from routers.network import adguard_stats ag = adguard_stats() context_parts.append(f"AdGuard DNS: {ag.get('total_queries', '?')} total queries, {ag.get('blocked', '?')} blocked, {ag.get('avg_time', '?')}s avg response") except Exception: pass system_context = ( "You are a homelab assistant. You have direct access to the following live infrastructure data:\n\n" + "\n".join(f"- {p}" for p in context_parts) + "\n\n" "Homelab hosts: Atlantis (Synology NAS, media/arr stack), Calypso (Synology, AdGuard DNS, Headscale, Authentik SSO), " "Olares (K3s, RTX 5090, Jellyfin, Ollama), NUC (lightweight services), RPi5 (Uptime Kuma), " "homelab-vm (Prometheus, Grafana, dashboard), Guava (TrueNAS), Seattle (remote VM), matrix-ubuntu (NPM, CrowdSec).\n\n" "Services: Sonarr, Radarr, SABnzbd, Deluge, Prowlarr, Bazarr, Lidarr, Tdarr, Audiobookshelf, LazyLibrarian on Atlantis. " "Jellyfin + Ollama on Olares with GPU transcoding. 3 email auto-organizers (Gmail x2 + Proton). " "11 Ollama-powered automation scripts. Gitea CI with AI PR reviewer.\n\n" "IMPORTANT: Answer using the LIVE DATA above, not general knowledge. The container counts are REAL numbers from Portainer right now. " "When asked 'how many containers on atlantis' answer with the exact number from the live data (e.g. 59). Be concise." ) # Search repo docs for relevant context (max 2K chars) doc_context = _search_repo_docs(message, max_chars=2000) if doc_context: system_context += f"\n\nRelevant documentation:\n{doc_context}" prompt = f"{system_context}\n\nUser: {message}\nAssistant:" try: from lib_bridge import ollama_available as _ollama_check if not _ollama_check(): return {"response": "Ollama is currently offline. Try again later."} import sys as _sys scripts_dir = str(Path("/app/scripts") if Path("/app/scripts").exists() else Path(__file__).parent.parent.parent / "scripts") if scripts_dir not in _sys.path: _sys.path.insert(0, scripts_dir) from lib.ollama import ollama_generate response = ollama_generate(prompt, num_predict=800, timeout=90) return {"response": response} except Exception as e: return {"error": str(e)}