from __future__ import annotations from datetime import UTC, datetime from pathlib import Path from typing import Any from fastapi import APIRouter, Depends, Query, Request from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import Settings, get_settings from app.db import get_session from app.db.models import ProbeResult, Tracker from app.db.queries import get_app_state, get_tracker, overview_stats, trackers_query from app.i18n import ( html_lang, lang_redirect, render_bootstrap_note, resolve_lang, translate, translator, ) from app.services.scoring import load_metrics from app.services.smart_list import SmartListFilters, build_smart_list, format_smart_list_txt router = APIRouter() TEMPLATES_DIR = Path(__file__).resolve().parent.parent / "templates" templates = Jinja2Templates(directory=str(TEMPLATES_DIR)) def _fmt_pct(value: float | None) -> str: if value is None: return "—" return f"{value * 100:.1f}%" def _fmt_ms(value: float | None) -> str: if value is None: return "—" return f"{value:.0f} ms" def _fmt_dt(value: datetime | None) -> str: if value is None: return "—" if value.tzinfo is None: value = value.replace(tzinfo=UTC) return value.strftime("%Y-%m-%d %H:%M UTC") templates.env.filters["pct"] = _fmt_pct templates.env.filters["ms"] = _fmt_ms templates.env.filters["dt"] = _fmt_dt def _base_ctx(request: Request, settings: Settings, **extra: Any) -> dict[str, Any]: lang = resolve_lang(request) ctx = { "request": request, "app_name": settings.app_name, "app_description": translate(lang, "tagline"), "app_version": settings.app_version, "repository_url": settings.repository_url, "lang": lang, "html_lang": html_lang(lang), "t": translator(lang), } ctx.update(extra) return ctx @router.get("/lang/{code}") async def set_lang(code: str, request: Request) -> RedirectResponse: return lang_redirect(request, code) @router.get("/", response_class=HTMLResponse) async def home( request: Request, session: AsyncSession = Depends(get_session), settings: Settings = Depends(get_settings), q: str | None = None, scheme: str | None = None, status: str | None = None, source: str | None = None, sort: str = "checked", order: str = "desc", page: int = Query(1, ge=1), ) -> HTMLResponse: per_page = 50 stats = await overview_stats(session) lang = resolve_lang(request) bootstrap_note = render_bootstrap_note(lang, await get_app_state(session, "bootstrap_note")) stmt = trackers_query(q=q, scheme=scheme, status=status, source=source, sort=sort, order=order) total = int( await session.scalar(select(func.count()).select_from(stmt.order_by(None).subquery())) or 0 ) rows = await session.execute(stmt.offset((page - 1) * per_page).limit(per_page)) trackers = list(rows.scalars().unique().all()) table_rows = [] for t in trackers: metrics = await load_metrics(session, t) alias_count = 0 if t.infrastructure_fingerprint: alias_count = int( await session.scalar( select(func.count()) .select_from(Tracker) .where( Tracker.infrastructure_fingerprint == t.infrastructure_fingerprint, Tracker.id != t.id, ) ) or 0 ) table_rows.append( { "tracker": t, "metrics": metrics, "alias_count": alias_count, "sources": ", ".join(sorted({s.source_name for s in t.sources})), } ) return templates.TemplateResponse( request, "index.html", _base_ctx( request, settings, stats=stats, bootstrap_note=bootstrap_note, rows=table_rows, q=q or "", scheme=scheme or "", status=status or "", source=source or "", sort=sort, order=order, page=page, total=total, per_page=per_page, pages=max(1, (total + per_page - 1) // per_page), active_nav="trackers", ), ) @router.get("/trackers/{tracker_id}", response_class=HTMLResponse) async def tracker_page( tracker_id: int, request: Request, session: AsyncSession = Depends(get_session), settings: Settings = Depends(get_settings), ) -> HTMLResponse: tracker = await get_tracker(session, tracker_id) if tracker is None: return HTMLResponse(translate(resolve_lang(request), "not_found"), status_code=404) metrics = await load_metrics(session, tracker) history = await session.execute( select(ProbeResult) .where(ProbeResult.tracker_id == tracker.id) .order_by(ProbeResult.checked_at.desc()) .limit(40) ) probes = list(history.scalars().all()) aliases: list[Tracker] = [] if tracker.infrastructure_fingerprint: alias_rows = await session.execute( select(Tracker) .where( Tracker.infrastructure_fingerprint == tracker.infrastructure_fingerprint, Tracker.id != tracker.id, ) .limit(20) ) aliases = list(alias_rows.scalars().all()) # oldest→newest so the sparkline draws left to right spark = [p.latency_ms or 0 for p in reversed(probes) if p.latency_ms is not None][-24:] return templates.TemplateResponse( request, "tracker_detail.html", _base_ctx( request, settings, tracker=tracker, metrics=metrics, probes=probes, aliases=aliases, spark=spark, active_nav="trackers", ), ) @router.get("/lists", response_class=HTMLResponse) async def lists_page( request: Request, session: AsyncSession = Depends(get_session), settings: Settings = Depends(get_settings), protocol: str | None = None, min_uptime: float = 0.95, max_latency: float | None = None, ip: str = "any", limit: int = Query(20, ge=1, le=50), min_age_days: float = 0.0, diversity: bool = True, ) -> HTMLResponse: filters = SmartListFilters( protocol=protocol or None, min_uptime=min_uptime, max_latency_ms=max_latency, ip_version=ip, limit=limit, min_age_days=min_age_days, diversity=diversity, ) entries = await build_smart_list(session, filters) txt = format_smart_list_txt(entries) qs = str(request.url.query) api_txt = f"/api/v1/lists/smart.txt?{qs}" if qs else "/api/v1/lists/smart.txt" api_json = f"/api/v1/lists/smart.json?{qs}" if qs else "/api/v1/lists/smart.json" return templates.TemplateResponse( request, "lists.html", _base_ctx( request, settings, entries=entries, txt=txt, protocol=protocol or "", min_uptime=min_uptime, max_latency=max_latency if max_latency is not None else "", ip=ip, limit=limit, min_age_days=min_age_days, diversity=diversity, api_txt=api_txt, api_json=api_json, active_nav="lists", ), )