from __future__ import annotations import logging from app.clients.base import BaseHTTPClient from app.config import Settings from app.models.sources import AdGuardSnapshot logger = logging.getLogger(__name__) class AdGuardClient(BaseHTTPClient): """ Reads DNS statistics from AdGuard Home's /control/stats endpoint. Requires Basic Auth (username + password). """ def __init__(self, settings: Settings) -> None: super().__init__(settings, "adguard", settings.adguard_base_url) async def fetch_stats(self) -> AdGuardSnapshot: snapshot = AdGuardSnapshot() if not self.base_url: logger.info("adguard skipped: base URL missing") return snapshot if not self.settings.adguard_username or not self.settings.adguard_password: logger.info("adguard skipped: no credentials configured") return snapshot data = await self._request_json( "GET", "/control/stats", auth=(self.settings.adguard_username, self.settings.adguard_password), ) if data is None: logger.warning("adguard: empty or failed response") return snapshot total = int(data.get("num_dns_queries") or 0) blocked = int(data.get("num_blocked_filtering") or 0) avg_ms = round(float(data.get("avg_processing_time") or 0.0) * 1000, 2) blocked_pct = round((blocked / total * 100), 1) if total > 0 else 0.0 result = AdGuardSnapshot( source_status="online", total_queries=total, blocked_queries=blocked, blocked_percent=blocked_pct, avg_processing_ms=avg_ms, ) logger.info("adguard stats: %s", result.model_dump()) return result