db_manager.py

"""Асинхронный менеджер БД для GitHub Industrial Engine.
Инициализация, миграции, высокоуровневые операции.
"""
from __future__ import annotations

import datetime as dt
import json
import logging
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional, Sequence

from sqlalchemy import select, update, func, text, or_
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from models import Base, Account, Repository, Log

log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Схема-миграции для старых БД
# Формат: (table_name, column_name, sql_type_with_default)
# Добавляй сюда все новые колонки при расширении схемы.
# ---------------------------------------------------------------------------
_NEW_REPO_COLUMNS: list[tuple[str, str, str]] = [
    # accounts: full mapped Account schema for legacy SQLite files
    ("accounts", "username", "VARCHAR(255)"),
    ("accounts", "email", "VARCHAR(255)"),
    ("accounts", "recovery_codes", "JSON"),
    ("accounts", "totp_secret", "VARCHAR(64)"),
    ("accounts", "status", "VARCHAR(32) DEFAULT 'active'"),
    ("accounts", "ban_reason", "TEXT"),
    ("accounts", "banned_at", "DATETIME"),
    ("accounts", "proxy", "VARCHAR(512)"),
    ("accounts", "user_agent", "TEXT"),
    ("accounts", "os_family", "VARCHAR(32)"),
    ("accounts", "profile_path", "VARCHAR(1024)"),
    ("accounts", "created_at", "DATETIME"),
    ("accounts", "last_used_at", "DATETIME"),
    ("accounts", "last_warmup_at", "DATETIME"),
    ("accounts", "cooldown_until", "DATETIME"),
    ("accounts", "last_action_kind", "VARCHAR(32)"),
    ("accounts", "rate_limit_strikes", "INTEGER DEFAULT 0"),
    ("accounts", "warmup_status", "VARCHAR(32) DEFAULT 'none'"),
    ("accounts", "warmup_sessions_done", "INTEGER DEFAULT 0"),
    ("accounts", "private_warmup_repo", "VARCHAR(255)"),
    ("accounts", "private_warmup_has_gha", "INTEGER DEFAULT 0"),
    ("accounts", "notes", "TEXT"),

    # repositories: current mapped Repository schema plus legacy aliases backfill below
    ("repositories", "account_id", "INTEGER"),
    ("repositories", "account_login", "VARCHAR(255)"),
    ("repositories", "url", "VARCHAR(512)"),
    ("repositories", "owner", "VARCHAR(255)"),
    ("repositories", "name", "VARCHAR(255)"),
    ("repositories", "description", "TEXT"),
    ("repositories", "topics", "JSON"),
    ("repositories", "language", "VARCHAR(64)"),
    ("repositories", "readme_path", "VARCHAR(1024)"),
    ("repositories", "status", "VARCHAR(32) DEFAULT 'active'"),
    ("repositories", "ban_reason", "TEXT"),
    ("repositories", "banned_at", "DATETIME"),
    ("repositories", "stars_count", "INTEGER DEFAULT 0"),
    ("repositories", "forks_count", "INTEGER DEFAULT 0"),
    ("repositories", "watchers_count", "INTEGER DEFAULT 0"),
    ("repositories", "issues_count", "INTEGER DEFAULT 0"),
    ("repositories", "created_at", "DATETIME"),
    ("repositories", "last_boosted_at", "DATETIME"),
    ("repositories", "last_seo_at", "DATETIME"),

    # tasks
    ("tasks", "task_type", "VARCHAR(32)"),
    ("tasks", "status", "VARCHAR(16) DEFAULT 'pending'"),
    ("tasks", "account_id", "INTEGER"),
    ("tasks", "repo_url", "VARCHAR(512)"),
    ("tasks", "payload", "JSON"),
    ("tasks", "attempts", "INTEGER DEFAULT 0"),
    ("tasks", "max_attempts", "INTEGER DEFAULT 3"),
    ("tasks", "last_error", "TEXT"),
    ("tasks", "priority", "INTEGER DEFAULT 0"),
    ("tasks", "scheduled_at", "DATETIME"),
    ("tasks", "started_at", "DATETIME"),
    ("tasks", "finished_at", "DATETIME"),
    ("tasks", "created_at", "DATETIME"),
    ("tasks", "notes", "TEXT"),

    # logs: old DBs used created_at; current model reads timestamp
    ("logs", "timestamp", "DATETIME"),
    ("logs", "level", "VARCHAR(16) DEFAULT 'INFO'"),
    ("logs", "source", "VARCHAR(64)"),
    ("logs", "account_login", "VARCHAR(255)"),
    ("logs", "repo_url", "VARCHAR(512)"),
    ("logs", "message", "TEXT"),
    ("logs", "extra", "JSON"),
]


class DatabaseManager:
    """Единственная точка доступа к БД. Работает только в async-контексте."""

    def __init__(self, db_path: str | Path = "data/engine.sqlite", echo: bool = False):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)

        # SQLAlchemy URL для aiosqlite
        url = f"sqlite+aiosqlite:///{self.db_path.as_posix()}"
        self.engine = create_async_engine(url, echo=echo, future=True)
        self._session_factory = async_sessionmaker(
            self.engine,
            expire_on_commit=False,
            class_=AsyncSession,
        )

    # ------------------------------------------------------------------
    # Init / migrate
    # ------------------------------------------------------------------
    async def init_db(self) -> None:
        """Создаёт таблицы и накатывает миграции. Вызывать один раз при старте."""
        async with self.engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        await self._migrate_sqlite_schema()
        log.info("[DB] Initialized + migrated at %s", self.db_path)

    async def _migrate_sqlite_schema(self) -> None:
        """Безопасно добавляет недостающие колонки в существующие таблицы. Идемпотентно."""
        async with self.engine.begin() as conn:
            for table, column, coltype in _NEW_REPO_COLUMNS:
                result = await conn.execute(text(f"PRAGMA table_info({table})"))
                existing = {row[1] for row in result.fetchall()}
                if column in existing:
                    log.debug("[DB MIGRATE] %s.%s already exists, skip", table, column)
                    continue
                try:
                    await conn.execute(
                        text(f"ALTER TABLE {table} ADD COLUMN {column} {coltype}")
                    )
                    log.info("[DB MIGRATE] ✅ Added %s.%s (%s)", table, column, coltype)
                except Exception as e:
                    log.warning(
                        "[DB MIGRATE] ⚠️ %s.%s: %s: %s",
                        table, column, type(e).__name__, e,
                    )
            await self._backfill_legacy_sqlite_columns(conn)

    async def _table_columns(self, conn, table: str) -> set[str]:
        result = await conn.execute(text(f"PRAGMA table_info({table})"))
        return {row[1] for row in result.fetchall()}

    async def _normalize_json_column(self, conn, table: str, column: str, fallback) -> None:
        columns = await self._table_columns(conn, table)
        if not {"id", column} <= columns:
            return
        fallback_json = json.dumps(fallback, ensure_ascii=False)
        rows = (await conn.execute(text(f"SELECT id, {column} FROM {table}"))).fetchall()
        for row_id, value in rows:
            if value in (None, ""):
                new_value = fallback_json
            elif isinstance(value, str):
                try:
                    json.loads(value)
                    continue
                except Exception:
                    new_value = fallback_json
            else:
                new_value = json.dumps(value, ensure_ascii=False)
            await conn.execute(
                text(f"UPDATE {table} SET {column} = :value WHERE id = :id"),
                {"value": new_value, "id": row_id},
            )

    async def _backfill_legacy_sqlite_columns(self, conn) -> None:
        """Copy data from old column names into the current mapped schema."""
        accounts = await self._table_columns(conn, "accounts")
        repositories = await self._table_columns(conn, "repositories")
        tasks = await self._table_columns(conn, "tasks")
        logs = await self._table_columns(conn, "logs")

        if {"profile_path", "profile_dir"} <= accounts:
            await conn.execute(text(
                "UPDATE accounts SET profile_path = COALESCE(profile_path, profile_dir)"
            ))
        if {"last_used_at", "last_login_at"} <= accounts:
            await conn.execute(text(
                "UPDATE accounts SET last_used_at = COALESCE(last_used_at, last_login_at)"
            ))

        if {"url", "repo_url"} <= repositories:
            await conn.execute(text(
                "UPDATE repositories SET url = COALESCE(url, repo_url)"
            ))
        if {"name", "repo_name"} <= repositories:
            await conn.execute(text(
                "UPDATE repositories SET name = COALESCE(name, repo_name)"
            ))
        if {"topics", "keywords", "id"} <= repositories:
            rows = (await conn.execute(text(
                "SELECT id, keywords FROM repositories "
                "WHERE (topics IS NULL OR topics = '') AND keywords IS NOT NULL"
            ))).fetchall()
            for repo_id, keywords in rows:
                if isinstance(keywords, str):
                    topics = [t.strip() for t in keywords.split(",") if t.strip()]
                elif isinstance(keywords, list):
                    topics = keywords
                else:
                    topics = []
                await conn.execute(
                    text("UPDATE repositories SET topics = :topics WHERE id = :id"),
                    {"topics": json.dumps(topics, ensure_ascii=False), "id": repo_id},
                )
        if {"account_id", "account_login"} <= repositories and "login" in accounts:
            await conn.execute(text(
                "UPDATE repositories "
                "SET account_id = COALESCE("
                "account_id, "
                "(SELECT accounts.id FROM accounts "
                " WHERE accounts.login = repositories.account_login LIMIT 1)"
                ")"
            ))

        if {"task_type", "type"} <= tasks:
            await conn.execute(text(
                "UPDATE tasks SET task_type = COALESCE(task_type, type)"
            ))
            await conn.execute(text(
                "UPDATE tasks SET type = COALESCE(type, task_type)"
            ))
        if {"timestamp", "created_at"} <= logs:
            await conn.execute(text(
                "UPDATE logs SET timestamp = COALESCE(timestamp, created_at)"
            ))

        await self._normalize_json_column(conn, "accounts", "recovery_codes", [])
        await self._normalize_json_column(conn, "repositories", "topics", [])
        await self._normalize_json_column(conn, "tasks", "payload", {})
        await self._normalize_json_column(conn, "logs", "extra", {})

    async def close(self) -> None:
        await self.engine.dispose()

    # ------------------------------------------------------------------
    # Session
    # ------------------------------------------------------------------
    @asynccontextmanager
    async def async_session(self):
        """Контекст для AsyncSession. Коммиты делаешь сам, роллбек автоматически при исключении."""
        session: AsyncSession = self._session_factory()
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

    # ------------------------------------------------------------------
    # Accounts
    # ------------------------------------------------------------------
    async def get_account_by_login(self, login: str) -> Optional[Account]:
        async with self.async_session() as session:
            result = await session.execute(
                select(Account).where(Account.login == login)
            )
            return result.scalar_one_or_none()

    async def get_active_accounts(
        self,
        limit: int | None = None,
        respect_cooldown: bool = False,
    ) -> list[Account]:
        async with self.async_session() as session:
            stmt = select(Account).where(Account.status == "active")
            if respect_cooldown:
                now = dt.datetime.utcnow()
                stmt = stmt.where(
                    or_(
                        Account.cooldown_until.is_(None),
                        Account.cooldown_until <= now,
                    )
                )
            if limit:
                stmt = stmt.limit(limit)
            result = await session.execute(stmt)
            return list(result.scalars().all())

    # ── Cooldown / throttling ──────────────────────────────────────────
    # Default cooldown_hours per action kind. Pick conservatively — лучше
    # «слишком осторожно» чем shadow-ban. Override через config/env по
    # необходимости. Главная цель — чтобы один аккаунт не выполнял два
    # крупных действия (create_repo+boost+warmup) подряд за <6h.
    COOLDOWN_HOURS: dict[str, float] = {
        "create_repo": 24.0,
        "create_repo_failed": 6.0,
        "warmup": 6.0,
        "boost": 1.0,
        "fork": 2.0,
        "humanize": 0.5,
        "rate_limit": 6.0,        # после 429/403 от GitHub
    }

    async def set_account_cooldown(
        self,
        login: str,
        kind: str,
        hours: float | None = None,
    ) -> None:
        """Поставить cooldown_until = now + hours для аккаунта.
        Если hours не задан — используется COOLDOWN_HOURS[kind] или 1h."""
        h = hours if hours is not None else self.COOLDOWN_HOURS.get(kind, 1.0)
        until = dt.datetime.utcnow() + dt.timedelta(hours=h)
        async with self.async_session() as session:
            await session.execute(
                update(Account)
                .where(Account.login == login)
                .values(
                    cooldown_until=until,
                    last_action_kind=kind,
                    last_used_at=dt.datetime.utcnow(),
                )
            )
            await session.commit()
        log.info(
            "[DB] cooldown set: %s kind=%s until=%s (~%.1fh)",
            login, kind, until.isoformat(timespec="seconds"), h,
        )

    async def increment_rate_limit_strike(self, login: str) -> int:
        """Увеличить счётчик rate-limit ошибок аккаунта.
        После 3 strike — переводим в quarantine (sticky proxy скорее всего
        в чёрном списке у GitHub). Возвращает новое значение."""
        async with self.async_session() as session:
            row = (await session.execute(
                select(Account).where(Account.login == login)
            )).scalar_one_or_none()
            if not row:
                return 0
            row.rate_limit_strikes = (row.rate_limit_strikes or 0) + 1
            new_value = row.rate_limit_strikes
            if new_value >= 3:
                row.status = "cooldown"
                row.cooldown_until = (
                    dt.datetime.utcnow() + dt.timedelta(hours=12)
                )
                log.warning(
                    "[DB] %s hit %d rate-limit strikes -> status=cooldown 12h",
                    login, new_value,
                )
            else:
                row.cooldown_until = (
                    dt.datetime.utcnow()
                    + dt.timedelta(hours=self.COOLDOWN_HOURS["rate_limit"])
                )
            await session.commit()
            return new_value

    async def reset_rate_limit_strikes(self, login: str) -> None:
        async with self.async_session() as session:
            await session.execute(
                update(Account)
                .where(Account.login == login)
                .values(rate_limit_strikes=0)
            )
            await session.commit()

    async def update_account_status(
        self,
        login: str,
        status: str,
        reason: str | None = None,
    ) -> None:
        async with self.async_session() as session:
            values = {"status": status}
            if status == "banned":
                values["banned_at"] = dt.datetime.utcnow()
                if reason:
                    values["ban_reason"] = reason
            await session.execute(
                update(Account).where(Account.login == login).values(**values)
            )
            await session.commit()
        log.info("[DB] Account %s -> %s%s", login, status, f" ({reason})" if reason else "")

    async def touch_account(self, login: str) -> None:
        """Обновить last_used_at."""
        async with self.async_session() as session:
            await session.execute(
                update(Account)
                .where(Account.login == login)
                .values(last_used_at=dt.datetime.utcnow())
            )
            await session.commit()

    async def update_account_fingerprint(self, login: str, **kwargs) -> None:
        values = {}
        for field in ("user_agent", "os_family", "profile_path"):
            if kwargs.get(field) is not None:
                values[field] = kwargs[field]
        if kwargs.get("last_login_at") is not None:
            values["last_used_at"] = kwargs["last_login_at"]
        if kwargs.get("last_used_at") is not None:
            values["last_used_at"] = kwargs["last_used_at"]
        if not values:
            return

        async with self.async_session() as session:
            await session.execute(
                update(Account).where(Account.login == login).values(**values)
            )
            await session.commit()

    # ------------------------------------------------------------------
    # Repositories
    # ------------------------------------------------------------------
    async def register_repository(
        self,
        account_id,
        owner: str | None = None,
        repo_name: str | None = None,
        url: str | None = None,
        status: str = "active",
        description: str | None = None,
        topics: list[str] | str | None = None,
        language: str | None = None,
        ban_reason: str | None = None,
    ) -> Repository:
        """Create/update repo; accepts both new and legacy worker signatures."""
        account_login = None
        if not isinstance(account_id, int):
            account_login = str(account_id)
            old_repo_name = owner
            old_url = repo_name
            old_owner = url
            old_description = status
            old_topics = description
            old_status = topics if isinstance(topics, str) else "active"
            owner = old_owner
            repo_name = old_repo_name
            url = old_url
            description = old_description
            topics = old_topics
            status = old_status or "active"

        if isinstance(topics, str):
            topics = [t.strip() for t in topics.split(",") if t.strip()]

        async with self.async_session() as session:
            if account_login is not None:
                account = (await session.execute(
                    select(Account).where(Account.login == account_login)
                )).scalar_one_or_none()
                if account is None:
                    raise ValueError(f"Account not found: {account_login}")
                account_id = account.id
            else:
                account = (await session.execute(
                    select(Account).where(Account.id == int(account_id))
                )).scalar_one_or_none()
                account_login = account.login if account else None

            if not owner and url:
                parts = url.replace("https://github.com/", "").strip("/").split("/")
                owner = parts[0] if len(parts) >= 2 else account_login
            if not repo_name and url:
                parts = url.replace("https://github.com/", "").strip("/").split("/")
                repo_name = parts[1] if len(parts) >= 2 else "repo"
            if not url and owner and repo_name:
                url = f"https://github.com/{owner}/{repo_name}"

            existing = (await session.execute(
                select(Repository).where(Repository.url == url)
            )).scalar_one_or_none()

            if existing is not None:
                existing.account_login = account_login
                existing.owner = owner or existing.owner
                existing.name = repo_name or existing.name
                existing.status = status
                if description is not None:
                    existing.description = description
                if topics is not None:
                    existing.topics = topics
                if language is not None:
                    existing.language = language
                if status == "banned":
                    existing.banned_at = dt.datetime.utcnow()
                    if ban_reason:
                        existing.ban_reason = ban_reason
                await session.commit()
                await session.refresh(existing)
                return existing

            repo = Repository(
                account_id=account_id,
                account_login=account_login,
                owner=owner,
                name=repo_name,
                url=url,
                status=status,
                description=description,
                topics=topics or [],
                language=language,
                ban_reason=ban_reason,
                banned_at=dt.datetime.utcnow() if status == "banned" else None,
            )
            session.add(repo)
            await session.commit()
            await session.refresh(repo)
            return repo

    async def get_account_repositories(
        self,
        account_id: int,
        status: str | None = None,
    ) -> list[Repository]:
        async with self.async_session() as session:
            stmt = select(Repository).where(Repository.account_id == account_id)
            if status:
                stmt = stmt.where(Repository.status == status)
            result = await session.execute(stmt)
            return list(result.scalars().all())

    async def count_banned_repos_for_account(self, account_id: int) -> int:
        """Используется ban_checker'ом для пороговой проверки (3+ = аккаунт banned)."""
        async with self.async_session() as session:
            result = await session.execute(
                select(func.count(Repository.id))
                .where(Repository.account_id == account_id)
                .where(Repository.status == "banned")
            )
            return int(result.scalar_one() or 0)

    async def increment_repo_counter(
        self,
        repo_url: str,
        field: str,
        delta: int = 1,
    ) -> None:
        """Безопасный инкремент счётчика. field ∈ stars_count|forks_count|watchers_count|issues_count."""
        allowed = {"stars_count", "forks_count", "watchers_count", "issues_count"}
        if field not in allowed:
            raise ValueError(f"Unsupported counter field: {field!r}")
        async with self.async_session() as session:
            repo = (await session.execute(
                select(Repository).where(Repository.url == repo_url)
            )).scalar_one_or_none()
            if repo is None:
                log.warning("[DB] increment_repo_counter: repo not found %s", repo_url)
                return
            current = getattr(repo, field) or 0
            setattr(repo, field, current + delta)
            repo.last_boosted_at = dt.datetime.utcnow()
            await session.commit()

    async def mark_repo_banned(self, repo_url: str | int, reason: str | None = None) -> None:
        async with self.async_session() as session:
            condition = (
                Repository.id == repo_url
                if isinstance(repo_url, int)
                else Repository.url == repo_url
            )
            await session.execute(
                update(Repository)
                .where(condition)
                .values(
                    status="banned",
                    banned_at=dt.datetime.utcnow(),
                    ban_reason=reason,
                )
            )
            await session.commit()
        log.info("[DB] Repo %s -> banned (%s)", repo_url, reason)

    # ------------------------------------------------------------------
    # Logs
    # ------------------------------------------------------------------
    async def add_log(
        self,
        message: str,
        level: str = "INFO",
        source: str | None = None,
        account: str | None = None,
        repo_url: str | None = None,
        extra: dict | None = None,
    ) -> None:
        known_levels = {"DEBUG", "INFO", "WARN", "WARNING", "ERROR", "CRITICAL"}
        if str(message).upper() in known_levels and str(level).upper() not in known_levels:
            message, level = level, message
        if str(level).upper() == "WARNING":
            level = "WARN"
        async with self.async_session() as session:
            entry = Log(
                level=level.upper(),
                source=source,
                account_login=account,
                repo_url=repo_url,
                message=message,
                extra=extra or {},
            )
            session.add(entry)
            await session.commit()

    async def add_account(
        self,
        login: str,
        password: str | None = None,
        token: str | None = None,
        status: str = "active",
        **kwargs,
    ) -> Account:
        async with self.async_session() as session:
            existing = (await session.execute(
                select(Account).where(Account.login == login)
            )).scalar_one_or_none()
            if existing:
                if password is not None:
                    existing.password = password
                if token is not None:
                    existing.token = token
                for key, value in kwargs.items():
                    if hasattr(existing, key) and value is not None:
                        setattr(existing, key, value)
                await session.commit()
                await session.refresh(existing)
                return existing

            account = Account(
                login=login,
                password=password,
                token=token,
                status=status,
                **{k: v for k, v in kwargs.items() if hasattr(Account, k)},
            )
            session.add(account)
            await session.commit()
            await session.refresh(account)
            return account

    async def get_recent_repos(self, limit: int = 20) -> list[Repository]:
        async with self.async_session() as session:
            result = await session.execute(
                select(Repository).order_by(Repository.created_at.desc()).limit(limit)
            )
            return list(result.scalars().all())

    async def get_all_repos(self, status: str | None = None) -> list[Repository]:
        async with self.async_session() as session:
            stmt = select(Repository).order_by(Repository.created_at.desc())
            if status:
                stmt = stmt.where(Repository.status == status)
            result = await session.execute(stmt)
            return list(result.scalars().all())

    async def update_repo_status(
        self,
        account_login: str,
        repo_name: str,
        status: str,
        new_keywords: str | list[str] | None = None,
    ) -> None:
        values = {"status": status}
        if status == "banned":
            values["banned_at"] = dt.datetime.utcnow()
        if new_keywords is not None:
            values["topics"] = (
                [t.strip() for t in new_keywords.split(",") if t.strip()]
                if isinstance(new_keywords, str)
                else new_keywords
            )
        async with self.async_session() as session:
            await session.execute(
                update(Repository)
                .where(Repository.account_login == account_login)
                .where(Repository.name == repo_name)
                .values(**values)
            )
            await session.commit()

    async def delete_repo(self, account_login: str, repo_name: str) -> bool:
        async with self.async_session() as session:
            result = await session.execute(
                text(
                    "DELETE FROM repositories "
                    "WHERE account_login = :login AND name = :name"
                ),
                {"login": account_login, "name": repo_name},
            )
            await session.commit()
            return bool(result.rowcount)

    async def delete_account_and_repos(self, login: str) -> bool:
        async with self.async_session() as session:
            account = (await session.execute(
                select(Account).where(Account.login == login)
            )).scalar_one_or_none()
            if not account:
                return False
            await session.delete(account)
            await session.commit()
            return True

    async def get_stats_snapshot(self) -> dict:
        from models import Task as _Task
        async with self.async_session() as session:
            now = dt.datetime.utcnow()
            day_ago = now - dt.timedelta(hours=24)
            week_ago = now - dt.timedelta(days=7)

            active_accounts = await session.scalar(
                select(func.count(Account.id)).where(Account.status == "active")
            )
            in_cooldown = await session.scalar(
                select(func.count(Account.id))
                .where(Account.status == "active")
                .where(Account.cooldown_until.isnot(None))
                .where(Account.cooldown_until > now)
            )
            ready_now = (active_accounts or 0) - (in_cooldown or 0)
            banned_accounts = await session.scalar(
                select(func.count(Account.id)).where(Account.status == "banned")
            )
            locked_accounts = await session.scalar(
                select(func.count(Account.id)).where(Account.status == "locked")
            )
            invalid_accounts = await session.scalar(
                select(func.count(Account.id)).where(Account.status == "invalid")
            )
            quarantine_2fa = await session.scalar(
                select(func.count(Account.id)).where(Account.status == "quarantine_2fa")
            )
            cooldown_total = await session.scalar(
                select(func.count(Account.id)).where(Account.status == "cooldown")
            )
            repos_total = await session.scalar(select(func.count(Repository.id)))
            repos_banned = await session.scalar(
                select(func.count(Repository.id)).where(Repository.status == "banned")
            )
            repos_24h = await session.scalar(
                select(func.count(Repository.id))
                .where(Repository.created_at >= day_ago)
            )
            repos_week = await session.scalar(
                select(func.count(Repository.id))
                .where(Repository.created_at >= week_ago)
            )
            stars_total = await session.scalar(
                select(func.coalesce(func.sum(Repository.stars_count), 0))
            )
            stars_24h = await session.scalar(
                select(func.coalesce(func.sum(Repository.stars_count), 0))
                .where(Repository.last_boosted_at >= day_ago)
            )
            forks_24h = await session.scalar(
                select(func.coalesce(func.sum(Repository.forks_count), 0))
                .where(Repository.last_boosted_at >= day_ago)
            )
            queue_pending = await session.scalar(
                select(func.count(_Task.id)).where(_Task.status == "pending")
            )
            queue_running = await session.scalar(
                select(func.count(_Task.id)).where(_Task.status == "running")
            )
            queue_failed_24h = await session.scalar(
                select(func.count(_Task.id))
                .where(_Task.status == "failed")
                .where(_Task.finished_at >= day_ago)
            )
        return {
            "active_accounts": active_accounts or 0,
            "ready_now": ready_now,
            "in_cooldown": in_cooldown or 0,
            "banned_accounts": banned_accounts or 0,
            "locked_accounts": locked_accounts or 0,
            "invalid_accounts": invalid_accounts or 0,
            "quarantine_2fa": quarantine_2fa or 0,
            "cooldown_total": cooldown_total or 0,
            "repos_total": repos_total or 0,
            "repos_banned": repos_banned or 0,
            "repos_24h": repos_24h or 0,
            "repos_week": repos_week or 0,
            "stars_total": stars_total or 0,
            "stars_24h": stars_24h or 0,
            "forks_24h": forks_24h or 0,
            "queue_pending": queue_pending or 0,
            "queue_running": queue_running or 0,
            "queue_failed_24h": queue_failed_24h or 0,
        }

    async def get_recent_logs(
        self,
        limit: int = 100,
        level: str | None = None,
        account: str | None = None,
    ) -> list[Log]:
        async with self.async_session() as session:
            stmt = select(Log).order_by(Log.timestamp.desc()).limit(limit)
            if level:
                stmt = stmt.where(Log.level == level.upper())
            if account:
                stmt = stmt.where(Log.account_login == account)
            result = await session.execute(stmt)
            return list(result.scalars().all())

    async def purge_old_logs(self, keep_days: int = 30) -> int:
        """Удалить логи старше keep_days. Возвращает количество удалённых записей."""
        cutoff = dt.datetime.utcnow() - dt.timedelta(days=keep_days)
        async with self.async_session() as session:
            result = await session.execute(
                text("DELETE FROM logs WHERE timestamp < :cutoff"),
                {"cutoff": cutoff.isoformat()},
            )
            await session.commit()
            return result.rowcount or 0


DBManager = DatabaseManager