feat(bot): please claude i need this my services are kinda homeless

src/utils/__init__.py

@@ -0,0 +1,3 @@
+from .env import env
+
+__all__ = ["env"]

src/utils/db/__init__.py

@@ -0,0 +1,61 @@
+from pathlib import Path
+
+import aiosqlite
+
+from utils.env import env
+
+from .repositories import IncidentRepository, ServiceRepository
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS services (
+    id         INTEGER PRIMARY KEY AUTOINCREMENT,
+    name       TEXT    NOT NULL,
+    url        TEXT    NOT NULL UNIQUE,
+    match_type TEXT    NOT NULL,
+    ok_status  INTEGER NOT NULL,
+    ok_body    TEXT    NOT NULL DEFAULT '',
+    json_path  TEXT,
+    json_value TEXT,
+    is_up      INTEGER NOT NULL DEFAULT 1,
+    created_at TEXT    NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS incidents (
+    id               INTEGER PRIMARY KEY AUTOINCREMENT,
+    service_id       INTEGER NOT NULL REFERENCES services(id) ON DELETE CASCADE,
+    started_at       TEXT    NOT NULL,
+    ended_at         TEXT,
+    reason           TEXT    NOT NULL DEFAULT '',
+    last_reminder_at TEXT    NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_incidents_service ON incidents(service_id);
+"""
+
+
+class Database:
+    def __init__(self, path: str) -> None:
+        self._path = path
+        self._conn: aiosqlite.Connection | None = None
+        self.services: ServiceRepository = None  # type: ignore[assignment]
+        self.incidents: IncidentRepository = None  # type: ignore[assignment]
+
+    async def connect(self) -> None:
+        Path(self._path).parent.mkdir(parents=True, exist_ok=True)
+        self._conn = await aiosqlite.connect(self._path)
+        self._conn.row_factory = aiosqlite.Row
+        await self._conn.execute("PRAGMA foreign_keys = ON")
+        await self._conn.executescript(SCHEMA)
+        await self._conn.commit()
+        self.services = ServiceRepository(self._conn)
+        self.incidents = IncidentRepository(self._conn)
+
+    async def close(self) -> None:
+        if self._conn is not None:
+            await self._conn.close()
+            self._conn = None
+
+
+db = Database(env.db.path)
+
+__all__ = ["Database", "db"]

src/utils/db/models.py

@@ -0,0 +1,40 @@
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class MatchType(StrEnum):
+    STATUS = "status"  # следим только за HTTP-статусом
+    BODY = "body"  # тело ответа должно совпадать целиком
+    JSON = "json"  # конкретное поле JSON должно быть равно эталону
+
+
+@dataclass(slots=True)
+class Service:
+    id: int
+    name: str
+    url: str
+    match_type: MatchType
+    ok_status: int
+    ok_body: str
+    json_path: str | None
+    json_value: str | None
+    is_up: bool
+    created_at: str
+
+    @property
+    def rule(self) -> str:
+        if self.match_type is MatchType.STATUS:
+            return f"HTTP-статус = {self.ok_status}"
+        if self.match_type is MatchType.JSON:
+            return f"<code>{self.json_path}</code> = <code>{self.json_value}</code>"
+        return "тело ответа неизменно"
+
+
+@dataclass(slots=True)
+class Incident:
+    id: int
+    service_id: int
+    started_at: str
+    ended_at: str | None
+    reason: str
+    last_reminder_at: str

src/utils/db/repositories/__init__.py

@@ -0,0 +1,4 @@
+from .incident import IncidentRepository
+from .service import ServiceRepository
+
+__all__ = ["IncidentRepository", "ServiceRepository"]

src/utils/db/repositories/incident.py

@@ -0,0 +1,74 @@
+import aiosqlite
+
+from utils.db.models import Incident
+from utils.format import now
+
+_COLUMNS = "id, service_id, started_at, ended_at, reason, last_reminder_at"
+
+
+def _row_to_incident(row: aiosqlite.Row) -> Incident:
+    return Incident(
+        id=row["id"],
+        service_id=row["service_id"],
+        started_at=row["started_at"],
+        ended_at=row["ended_at"],
+        reason=row["reason"],
+        last_reminder_at=row["last_reminder_at"],
+    )
+
+
+class IncidentRepository:
+    def __init__(self, conn: aiosqlite.Connection) -> None:
+        self._conn = conn
+
+    async def open(self, service_id: int, reason: str) -> Incident:
+        stamp = now().isoformat()
+        cursor = await self._conn.execute(
+            "INSERT INTO incidents "
+            "(service_id, started_at, reason, last_reminder_at) VALUES (?, ?, ?, ?)",
+            (service_id, stamp, reason, stamp),
+        )
+        await self._conn.commit()
+        incident = await self.get(cursor.lastrowid)  # type: ignore[arg-type]
+        assert incident is not None
+        return incident
+
+    async def get(self, incident_id: int) -> Incident | None:
+        cursor = await self._conn.execute(
+            f"SELECT {_COLUMNS} FROM incidents WHERE id = ?",  # noqa: S608
+            (incident_id,),
+        )
+        row = await cursor.fetchone()
+        return _row_to_incident(row) if row else None
+
+    async def current(self, service_id: int) -> Incident | None:
+        cursor = await self._conn.execute(
+            f"SELECT {_COLUMNS} FROM incidents "  # noqa: S608
+            "WHERE service_id = ? AND ended_at IS NULL "
+            "ORDER BY id DESC LIMIT 1",
+            (service_id,),
+        )
+        row = await cursor.fetchone()
+        return _row_to_incident(row) if row else None
+
+    async def close(self, incident_id: int) -> None:
+        await self._conn.execute(
+            "UPDATE incidents SET ended_at = ? WHERE id = ?",
+            (now().isoformat(), incident_id),
+        )
+        await self._conn.commit()
+
+    async def touch_reminder(self, incident_id: int) -> None:
+        await self._conn.execute(
+            "UPDATE incidents SET last_reminder_at = ? WHERE id = ?",
+            (now().isoformat(), incident_id),
+        )
+        await self._conn.commit()
+
+    async def for_service(self, service_id: int) -> list[Incident]:
+        cursor = await self._conn.execute(
+            f"SELECT {_COLUMNS} FROM incidents "  # noqa: S608
+            "WHERE service_id = ? ORDER BY id",
+            (service_id,),
+        )
+        return [_row_to_incident(row) for row in await cursor.fetchall()]

src/utils/db/repositories/service.py

@@ -0,0 +1,92 @@
+import aiosqlite
+
+from utils.db.models import MatchType, Service
+from utils.format import now
+
+_COLUMNS = (
+    "id, name, url, match_type, ok_status, ok_body, "
+    "json_path, json_value, is_up, created_at"
+)
+
+
+def _row_to_service(row: aiosqlite.Row) -> Service:
+    return Service(
+        id=row["id"],
+        name=row["name"],
+        url=row["url"],
+        match_type=MatchType(row["match_type"]),
+        ok_status=row["ok_status"],
+        ok_body=row["ok_body"],
+        json_path=row["json_path"],
+        json_value=row["json_value"],
+        is_up=bool(row["is_up"]),
+        created_at=row["created_at"],
+    )
+
+
+class ServiceRepository:
+    def __init__(self, conn: aiosqlite.Connection) -> None:
+        self._conn = conn
+
+    async def all(self) -> list[Service]:
+        cursor = await self._conn.execute(
+            f"SELECT {_COLUMNS} FROM services ORDER BY id"  # noqa: S608
+        )
+        return [_row_to_service(row) for row in await cursor.fetchall()]
+
+    async def get(self, service_id: int) -> Service | None:
+        cursor = await self._conn.execute(
+            f"SELECT {_COLUMNS} FROM services WHERE id = ?",  # noqa: S608
+            (service_id,),
+        )
+        row = await cursor.fetchone()
+        return _row_to_service(row) if row else None
+
+    async def by_url(self, url: str) -> Service | None:
+        cursor = await self._conn.execute(
+            f"SELECT {_COLUMNS} FROM services WHERE url = ?",  # noqa: S608
+            (url,),
+        )
+        row = await cursor.fetchone()
+        return _row_to_service(row) if row else None
+
+    async def add(  # noqa: PLR0913
+        self,
+        *,
+        name: str,
+        url: str,
+        match_type: MatchType,
+        ok_status: int,
+        ok_body: str = "",
+        json_path: str | None = None,
+        json_value: str | None = None,
+    ) -> Service:
+        cursor = await self._conn.execute(
+            "INSERT INTO services "
+            "(name, url, match_type, ok_status, ok_body, json_path, json_value, "
+            "is_up, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, 1, ?)",
+            (
+                name,
+                url,
+                match_type.value,
+                ok_status,
+                ok_body,
+                json_path,
+                json_value,
+                now().isoformat(),
+            ),
+        )
+        await self._conn.commit()
+        service = await self.get(cursor.lastrowid)  # type: ignore[arg-type]
+        assert service is not None
+        return service
+
+    async def set_up(self, service_id: int, *, is_up: bool) -> None:
+        await self._conn.execute(
+            "UPDATE services SET is_up = ? WHERE id = ?", (int(is_up), service_id)
+        )
+        await self._conn.commit()
+
+    async def delete(self, service_id: int) -> None:
+        await self._conn.execute("DELETE FROM services WHERE id = ?", (service_id,))
+        await self._conn.commit()

src/utils/env.py

@@ -0,0 +1,38 @@
+from pydantic import Field, SecretStr
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class BotSettings(BaseSettings):
+    token: SecretStr
+
+
+class MonitorSettings(BaseSettings):
+    check_interval: int = 30  # как часто пинговать сервисы, сек
+    reminder_interval: int = 1800  # как часто напоминать что всё ещё лежит, сек
+    request_timeout: int = 10  # таймаут одного запроса, сек
+
+
+class DbSettings(BaseSettings):
+    path: str = "data/healthbot.db"
+
+
+class LogSettings(BaseSettings):
+    level: str = "INFO"
+    level_external: str = "WARNING"
+    show_time: bool = False
+    console_width: int = 150
+
+
+class Settings(BaseSettings):
+    admin_id: int
+    bot: BotSettings = Field(default_factory=BotSettings)
+    monitor: MonitorSettings = Field(default_factory=MonitorSettings)
+    db: DbSettings = Field(default_factory=DbSettings)
+    log: LogSettings = Field(default_factory=LogSettings)
+
+    model_config = SettingsConfigDict(
+        case_sensitive=False, env_file=".env", env_nested_delimiter="__", extra="ignore"
+    )
+
+
+env = Settings()

src/utils/format.py

@@ -0,0 +1,32 @@
+from datetime import UTC, datetime
+
+
+def now() -> datetime:
+    return datetime.now(UTC)
+
+
+def parse(ts: str) -> datetime:
+    return datetime.fromisoformat(ts)
+
+
+MINUTE = 60
+MAX_PARTS = 2
+
+
+def human_duration(seconds: float) -> str:
+    seconds = int(seconds)
+    if seconds < MINUTE:
+        return f"{seconds}с"
+    parts: list[str] = []
+    for unit, size in (("д", 86400), ("ч", 3600), ("м", 60)):
+        if seconds >= size:
+            parts.append(f"{seconds // size}{unit}")
+            seconds %= size
+    if seconds and len(parts) < MAX_PARTS:
+        parts.append(f"{seconds}с")
+    return " ".join(parts[:MAX_PARTS])
+
+
+def short(value: object, limit: int = 60) -> str:
+    text = str(value).replace("\n", " ").strip()
+    return text if len(text) <= limit else text[: limit - 1] + "…"

src/utils/logging.py

@@ -0,0 +1,36 @@
+import logging
+
+from rich.console import Console
+from rich.logging import RichHandler
+from rich.traceback import install
+
+from .env import env
+
+console = Console(width=env.log.console_width, color_system="auto", force_terminal=True)
+
+
+def setup_logging() -> None:
+    from aiogram.dispatcher import router  # noqa: PLC0415
+
+    logging.basicConfig(
+        level=env.log.level_external,
+        format="",
+        datefmt=None,
+        handlers=[
+            RichHandler(
+                console=console,
+                markup=True,
+                rich_tracebacks=True,
+                enable_link_path=False,
+                tracebacks_show_locals=True,
+                omit_repeated_times=False,
+                show_time=env.log.show_time,
+                tracebacks_suppress=[router],
+            )
+        ],
+    )
+    install(console=console, show_locals=True)
+
+
+logger = logging.getLogger("healthbot")
+logger.setLevel(env.log.level)