nessi
d694c8b8e3
Add .env.example with configuration for database, Redis, security, SMTP, workers, and plugins. Add .gitignore for Python, Node.js, Next.js, Docker volumes, and IDE files. Add MIT License. Update README.md with feature overview, quick start guide, architecture description, plugin system documentation, security details, backup/restore instructions, and developer setup. Add Alembic configuration files and placeholder directories for API, web, worker, and plugin components
58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
from typing import Any
|
|
|
|
from apps.api.src.plugins.base import PluginConnector
|
|
from apps.api.src.plugins.connectors._shared import safe_json
|
|
|
|
|
|
class AdGuardHomeConnector(PluginConnector):
|
|
id = "adguard-home"
|
|
name = "AdGuard Home"
|
|
|
|
def _url(self, path: str) -> str:
|
|
base = self.context.connection.base_url.rstrip("/") if self.context.connection else ""
|
|
return f"{base}/control{path}"
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
headers = {}
|
|
creds = self.context.credentials
|
|
if creds.get("username") and creds.get("password"):
|
|
from base64 import b64encode
|
|
token = b64encode(f"{creds['username']}:{creds['password']}".encode()).decode()
|
|
headers["Authorization"] = f"Basic {token}"
|
|
return headers
|
|
|
|
async def healthcheck(self) -> dict[str, Any]:
|
|
try:
|
|
resp = await self.get_client().get(self._url("/status"), headers=self._headers())
|
|
ok = resp.status_code == 200
|
|
return {"status": "ok" if ok else "error", "status_code": resp.status_code}
|
|
except Exception as e:
|
|
return {"status": "error", "message": str(e)}
|
|
|
|
async def fetch_widget_data(self, widget_type: str, settings: dict[str, Any]) -> dict[str, Any]:
|
|
if widget_type == "dns-overview":
|
|
return await self._get_stats()
|
|
if widget_type == "block-rate-chart":
|
|
return await self._get_stats()
|
|
if widget_type == "top-clients":
|
|
return await self._get_top_clients()
|
|
if widget_type == "top-blocked-domains":
|
|
return await self._get_top_blocked()
|
|
return {"error": "Unsupported widget type"}
|
|
|
|
async def _get_stats(self) -> dict[str, Any]:
|
|
resp = await self.get_client().get(self._url("/stats"), headers=self._headers())
|
|
return await safe_json(resp)
|
|
|
|
async def _get_top_clients(self) -> dict[str, Any]:
|
|
resp = await self.get_client().get(
|
|
self._url("/stats/top_clients?limit=10"), headers=self._headers()
|
|
)
|
|
return await safe_json(resp)
|
|
|
|
async def _get_top_blocked(self) -> dict[str, Any]:
|
|
resp = await self.get_client().get(
|
|
self._url("/stats/top_blocked_domains?limit=10"), headers=self._headers()
|
|
)
|
|
return await safe_json(resp)
|