# -*- coding: utf-8 -*-
"""P0 数据访问层：收藏夹、视频与黑名单。

业务状态迁移集中在这里，服务层只负责请求与编排。所有写操作经 ``Database``
的单写队列串行执行。
"""

from __future__ import annotations

import sqlite3
import time
from dataclasses import dataclass
from typing import Any, Iterable, Mapping

from .config import FolderConfig
from .database import Database


@dataclass(frozen=True)
class UpsertOutcome:
    aid: int
    status: str
    previous_folder_id: int | None
    moved_folder: bool


@dataclass(frozen=True)
class FolderRow:
    media_id: int
    year: int
    title: str
    media_count: int | None
    last_ids_at: int | None
    last_full_at: int | None
    enabled: bool


class Repository:
    def __init__(self, db: Database, *, clock=time.time) -> None:
        self.db = db
        self._clock = clock

    def _now(self) -> int:
        return int(self._clock())

    # ------------------------------------------------------------ folder

    def ensure_folder(self, folder: FolderConfig, *, now: int | None = None) -> None:
        now = self._now() if now is None else int(now)
        def run(conn: sqlite3.Connection) -> None:
            conn.execute(
                """
                INSERT INTO folder(media_id, year, title, media_count,
                                   last_ids_at, last_full_at, enabled)
                VALUES (?, ?, ?, NULL, NULL, NULL, 1)
                ON CONFLICT(media_id) DO UPDATE SET
                  year = excluded.year,
                  title = COALESCE(NULLIF(excluded.title, ''), folder.title),
                  enabled = 1
                """,
                (folder.media_id, folder.year, folder.title),
            )
        self.db.write(run)

    def disable_folders_except(self, media_ids: Iterable[int]) -> None:
        ids = [int(mid) for mid in media_ids]
        def run(conn: sqlite3.Connection) -> None:
            if ids:
                placeholders = ",".join("?" for _ in ids)
                conn.execute(
                    f"UPDATE folder SET enabled = 0 WHERE media_id NOT IN ({placeholders})",
                    ids,
                )
            else:
                conn.execute("UPDATE folder SET enabled = 0")
        self.db.write(run)

    def finish_folder_sync(
        self,
        folder: FolderConfig,
        *,
        media_count: int,
        full_refresh: bool,
        now: int | None = None,
    ) -> None:
        now = self._now() if now is None else int(now)
        def run(conn: sqlite3.Connection) -> None:
            conn.execute(
                """
                INSERT INTO folder(media_id, year, title, media_count,
                                   last_ids_at, last_full_at, enabled)
                VALUES (?, ?, ?, ?, ?, ?, 1)
                ON CONFLICT(media_id) DO UPDATE SET
                  year = excluded.year,
                  title = COALESCE(NULLIF(excluded.title, ''), folder.title),
                  media_count = excluded.media_count,
                  last_ids_at = excluded.last_ids_at,
                  last_full_at = CASE
                    WHEN ? THEN excluded.last_full_at
                    ELSE folder.last_full_at
                  END,
                  enabled = 1
                """,
                (
                    folder.media_id, folder.year, folder.title, int(media_count),
                    now, now if full_refresh else None, 1 if full_refresh else 0,
                ),
            )
        self.db.write(run)

    def get_folder(self, media_id: int) -> FolderRow | None:
        def run(conn: sqlite3.Connection) -> FolderRow | None:
            row = conn.execute(
                "SELECT media_id, year, title, media_count, last_ids_at, "
                "last_full_at, enabled FROM folder WHERE media_id = ?",
                (int(media_id),),
            ).fetchone()
            if row is None:
                return None
            return FolderRow(
                media_id=int(row["media_id"]),
                year=int(row["year"]),
                title=str(row["title"] or ""),
                media_count=row["media_count"],
                last_ids_at=row["last_ids_at"],
                last_full_at=row["last_full_at"],
                enabled=bool(row["enabled"]),
            )
        return self.db.read(run)

    def list_folders(self) -> list[FolderRow]:
        def run(conn: sqlite3.Connection) -> list[FolderRow]:
            rows = conn.execute(
                "SELECT media_id, year, title, media_count, last_ids_at, "
                "last_full_at, enabled FROM folder ORDER BY year, media_id"
            ).fetchall()
            return [
                FolderRow(
                    media_id=int(row["media_id"]),
                    year=int(row["year"]),
                    title=str(row["title"] or ""),
                    media_count=row["media_count"],
                    last_ids_at=row["last_ids_at"],
                    last_full_at=row["last_full_at"],
                    enabled=bool(row["enabled"]),
                )
                for row in rows
            ]
        return self.db.read(run)

    def folder_active_aids(self, media_id: int) -> set[int]:
        def run(conn: sqlite3.Connection) -> set[int]:
            rows = conn.execute(
                "SELECT aid FROM folder_resource WHERE media_id = ? AND active = 1",
                (int(media_id),),
            ).fetchall()
            return {int(row["aid"]) for row in rows}
        return self.db.read(run)

    def touch_folder_resources(self, media_id: int, *, now: int | None = None) -> None:
        now = self._now() if now is None else int(now)
        self.db.write(lambda conn: conn.execute(
            "UPDATE folder_resource SET last_seen = ? WHERE media_id = ? AND active = 1",
            (now, int(media_id)),
        ))

    # ------------------------------------------------------------ video

    @staticmethod
    def _recompute_primary_folder(conn: sqlite3.Connection, aid: int) -> int | None:
        """按有效收藏关系重算主归属，避免依赖收藏夹遍历顺序。

        优先使用逐夹 ``fav_time``（后加入者优先），其次 ``first_seen``，
        最后用年度与 media_id 做确定性决胜。
        """
        row = conn.execute(
            """
            SELECT fr.media_id
            FROM folder_resource fr
            JOIN folder f ON f.media_id = fr.media_id
            WHERE fr.aid = ? AND fr.active = 1 AND f.enabled = 1
            ORDER BY COALESCE(fr.fav_time, 0) DESC,
                     fr.first_seen DESC,
                     f.year DESC,
                     fr.media_id DESC
            LIMIT 1
            """,
            (int(aid),),
        ).fetchone()
        if row is None:
            return None
        folder_id = int(row["media_id"])
        conn.execute(
            "UPDATE video SET folder_id = ? WHERE aid = ?", (folder_id, int(aid))
        )
        return folder_id

    def upsert_from_fav(
        self,
        resource,
        folder_id: int,
        *,
        was_added: bool,
        now: int | None = None,
    ) -> UpsertOutcome:
        now = self._now() if now is None else int(now)
        folder_id = int(folder_id)
        invalid = bool(getattr(resource, "invalid", False))

        def run(conn: sqlite3.Connection) -> UpsertOutcome:
            previous = conn.execute(
                "SELECT status, folder_id FROM video WHERE aid = ?", (resource.aid,)
            ).fetchone()
            previous_status = str(previous["status"]) if previous else None
            previous_folder = (
                int(previous["folder_id"])
                if previous and previous["folder_id"] is not None else None
            )

            if previous_status is None:
                status = "removed" if invalid else "active"
            elif previous_status == "dropped":
                status = "active" if (was_added and not invalid) else "dropped"
            elif invalid:
                status = "removed"
            else:
                status = "active"

            conn.execute(
                """
                INSERT INTO video(aid, bvid, title, up_mid, up_name, pubdate,
                                  duration, pages, cover, folder_id, fav_time, status,
                                  source, first_seen, last_sync, attr)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(aid) DO UPDATE SET
                  bvid = COALESCE(excluded.bvid, video.bvid),
                  title = CASE
                    WHEN (excluded.attr = 9 OR excluded.status = 'removed')
                         AND video.title != ''
                      THEN video.title
                    WHEN excluded.title != '' THEN excluded.title
                    ELSE video.title
                  END,
                  up_mid = COALESCE(excluded.up_mid, video.up_mid),
                  up_name = COALESCE(excluded.up_name, video.up_name),
                  pubdate = CASE
                    WHEN excluded.pubdate > 0 THEN excluded.pubdate
                    ELSE video.pubdate
                  END,
                  duration = COALESCE(excluded.duration, video.duration),
                  pages = COALESCE(excluded.pages, video.pages),
                  cover = COALESCE(excluded.cover, video.cover),
                  folder_id = CASE
                    WHEN video.folder_id IS NULL THEN excluded.folder_id
                    ELSE video.folder_id
                  END,
                  fav_time = COALESCE(excluded.fav_time, video.fav_time),
                  status = excluded.status,
                  source = COALESCE(video.source, excluded.source),
                  last_sync = excluded.last_sync,
                  attr = excluded.attr
                """,
                (
                    resource.aid,
                    resource.bvid,
                    resource.title or "(无标题)",
                    resource.up_mid,
                    resource.up_name,
                    int(resource.pubdate or 0),
                    resource.duration,
                    resource.pages,
                    resource.cover,
                    folder_id,
                    resource.fav_time,
                    status,
                    f"fav:{folder_id}",
                    now,
                    now,
                    resource.attr,
                ),
            )
            conn.execute(
                """
                INSERT INTO folder_resource(media_id, aid, first_seen, active,
                                            last_seen, fav_time)
                VALUES (?, ?, ?, 1, ?, ?)
                ON CONFLICT(media_id, aid) DO UPDATE SET
                  active = 1,
                  last_seen = excluded.last_seen,
                  fav_time = COALESCE(excluded.fav_time, folder_resource.fav_time)
                """,
                (folder_id, resource.aid, now, now, resource.fav_time),
            )
            primary = self._recompute_primary_folder(conn, resource.aid)
            return UpsertOutcome(
                aid=resource.aid,
                status=status,
                previous_folder_id=previous_folder,
                moved_folder=previous_folder is not None and previous_folder != primary,
            )
        return self.db.write(run)

    def mark_missing_resource(
        self,
        aid: int,
        *,
        folder_id: int,
        bvid: str | None = None,
        now: int | None = None,
    ) -> str:
        """处理 infos 静默缺失：写入/更新为 removed，保留历史。"""
        now = self._now() if now is None else int(now)
        aid = int(aid)
        folder_id = int(folder_id)

        def run(conn: sqlite3.Connection) -> str:
            row = conn.execute("SELECT status FROM video WHERE aid = ?", (aid,)).fetchone()
            previous_status = str(row["status"]) if row else None
            status = "dropped" if previous_status == "dropped" else "removed"
            conn.execute(
                """
                INSERT INTO video(aid, bvid, title, up_mid, up_name, pubdate,
                                  duration, cover, folder_id, fav_time, status,
                                  source, first_seen, last_sync, attr)
                VALUES (?, ?, '(资源已失效)', NULL, NULL, 0, NULL, NULL, ?, NULL,
                        ?, ?, ?, ?, 9)
                ON CONFLICT(aid) DO UPDATE SET
                  bvid = COALESCE(video.bvid, excluded.bvid),
                  status = ?,
                  last_sync = excluded.last_sync,
                  attr = 9
                """,
                (aid, bvid, folder_id, status, f"fav:{folder_id}", now, now, status),
            )
            conn.execute(
                """
                INSERT INTO folder_resource(media_id, aid, first_seen, active, last_seen)
                VALUES (?, ?, ?, 1, ?)
                ON CONFLICT(media_id, aid) DO UPDATE SET
                  active = 1,
                  last_seen = excluded.last_seen
                """,
                (folder_id, aid, now, now),
            )
            self._recompute_primary_folder(conn, aid)
            return status
        return self.db.write(run)

    def mark_resource_removed(
        self, media_id: int, aid: int, *, now: int | None = None
    ) -> str | None:
        """资源从收藏夹移除：active -> unfaved，其他（dropped/removed）保持不变。"""
        now = self._now() if now is None else int(now)
        media_id, aid = int(media_id), int(aid)

        def run(conn: sqlite3.Connection) -> str | None:
            conn.execute(
                "UPDATE folder_resource SET active = 0, last_seen = ? "
                "WHERE media_id = ? AND aid = ?",
                (now, media_id, aid),
            )
            self._recompute_primary_folder(conn, aid)
            other = conn.execute(
                """
                SELECT 1
                FROM folder_resource fr
                JOIN folder f ON f.media_id = fr.media_id
                WHERE fr.aid = ? AND fr.active = 1 AND f.enabled = 1
                LIMIT 1
                """,
                (aid,),
            ).fetchone()
            if other is None:
                conn.execute(
                    "UPDATE video SET status = 'unfaved', last_sync = ? "
                    "WHERE aid = ? AND status IN ('active', 'unfaved')",
                    (now, aid),
                )
            row = conn.execute("SELECT status FROM video WHERE aid = ?", (aid,)).fetchone()
            return str(row["status"]) if row else None
        return self.db.write(run)

    def set_video_status(self, aid: int, status: str, *, now: int | None = None) -> None:
        now = self._now() if now is None else int(now)
        self.db.write(lambda conn: conn.execute(
            "UPDATE video SET status = ?, last_sync = ? WHERE aid = ?",
            (status, now, int(aid)),
        ))

    def get_video(self, aid: int) -> dict[str, Any] | None:
        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
            row = conn.execute("SELECT * FROM video WHERE aid = ?", (int(aid),)).fetchone()
            return dict(row) if row else None
        return self.db.read(run)

    def list_videos(self) -> list[dict[str, Any]]:
        def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
            rows = conn.execute("SELECT * FROM video ORDER BY aid").fetchall()
            return [dict(row) for row in rows]
        return self.db.read(run)

    # ------------------------------------------------------- up profile

    def apply_up_follower(
        self,
        *,
        mid: int,
        name: str | None,
        follower: int,
        following: int | None = None,
        now: int | None = None,
        backfilled: bool = False,
    ) -> int:
        """写入 UP 粉丝快照，并补齐该 UP 尚未记录粉丝数的视频。

        ``follower_first`` 只在首次插入时写入；``follower_latest`` 每次刷新都会
        更新。视频上的 ``up_follower`` 是入库时快照，已有值不会被覆盖。返回本次
        补齐的视频数。
        """
        now = self._now() if now is None else int(now)
        mid = int(mid)
        follower = max(0, int(follower))
        following_value = None if following is None else max(0, int(following))
        backfill_flag = 1 if backfilled else 0

        def run(conn: sqlite3.Connection) -> int:
            conn.execute(
                """
                INSERT INTO up_profile(mid, name, follower_first, following_first,
                                       first_captured_at, follower_latest, last_refreshed)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(mid) DO UPDATE SET
                  name = COALESCE(NULLIF(excluded.name, ''), up_profile.name),
                  follower_latest = excluded.follower_latest,
                  last_refreshed = excluded.last_refreshed
                """,
                (mid, name, follower, following_value, now, follower, now),
            )
            cursor = conn.execute(
                """
                UPDATE video
                SET up_follower = ?, up_follower_at = ?, up_follower_backfilled = ?
                WHERE up_mid = ? AND up_follower IS NULL
                """,
                (follower, now, backfill_flag, mid),
            )
            return max(cursor.rowcount, 0)

        return self.db.write(run)

    def get_up_profile(self, mid: int) -> dict[str, Any] | None:
        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
            row = conn.execute(
                "SELECT * FROM up_profile WHERE mid = ?", (int(mid),)
            ).fetchone()
            return dict(row) if row else None
        return self.db.read(run)

    def up_mids_missing_follower(self) -> list[dict[str, Any]]:
        """追踪中但仍有视频缺少入库粉丝数的 UP。"""
        def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
            rows = conn.execute(
                """
                SELECT up_mid, COALESCE(MAX(up_name), '') AS up_name,
                       COUNT(*) AS missing
                FROM video
                WHERE status IN ('active', 'unfaved')
                  AND up_mid IS NOT NULL
                  AND up_follower IS NULL
                GROUP BY up_mid
                ORDER BY up_mid
                """
            ).fetchall()
            return [dict(row) for row in rows]
        return self.db.read(run)

    def up_follower_coverage(self) -> dict[str, int]:
        """返回追踪视频的粉丝快照覆盖情况。"""
        def run(conn: sqlite3.Connection) -> dict[str, int]:
            row = conn.execute(
                """
                SELECT COUNT(*) AS total,
                       SUM(CASE WHEN up_follower IS NOT NULL THEN 1 ELSE 0 END) AS covered,
                       COUNT(DISTINCT CASE WHEN up_follower IS NOT NULL
                                           THEN up_mid END) AS ups_covered
                FROM video
                WHERE status IN ('active', 'unfaved')
                  AND up_mid IS NOT NULL
                """
            ).fetchone()
            total = int(row["total"] or 0)
            covered = int(row["covered"] or 0)
            return {
                "total": total,
                "covered": covered,
                "missing": max(0, total - covered),
                "ups_covered": int(row["ups_covered"] or 0),
            }
        return self.db.read(run)

    def folder_resources(self, media_id: int) -> list[dict[str, Any]]:
        def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
            rows = conn.execute(
                "SELECT * FROM folder_resource WHERE media_id = ? ORDER BY aid",
                (int(media_id),),
            ).fetchall()
            return [dict(row) for row in rows]
        return self.db.read(run)

    # ------------------------------------------------------------ snapshot

    def video_aids_for_snapshot(self) -> list[int]:
        def run(conn: sqlite3.Connection) -> list[int]:
            rows = conn.execute(
                "SELECT aid FROM video WHERE status IN ('active', 'unfaved') "
                "ORDER BY aid"
            ).fetchall()
            return [int(row["aid"]) for row in rows]
        return self.db.read(run)

    def pending_snapshot_aids(self, stat_date: str) -> list[int]:
        """当天仍缺成功快照的追踪视频；用于新入库/恢复后的同日补采。"""
        def run(conn: sqlite3.Connection) -> list[int]:
            rows = conn.execute(
                """
                SELECT v.aid
                FROM video v
                WHERE v.status IN ('active', 'unfaved')
                  AND NOT EXISTS (
                    SELECT 1 FROM snapshot s
                    WHERE s.stat_date = ? AND s.aid = v.aid AND s.ok = 1
                  )
                ORDER BY v.aid
                """,
                (stat_date,),
            ).fetchall()
            return [int(row["aid"]) for row in rows]
        return self.db.read(run)

    def get_snapshot(self, stat_date: str, aid: int) -> dict[str, Any] | None:
        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
            row = conn.execute(
                "SELECT * FROM snapshot WHERE stat_date = ? AND aid = ?",
                (stat_date, int(aid)),
            ).fetchone()
            return dict(row) if row else None
        return self.db.read(run)

    def upsert_snapshot(
        self,
        stat_date: str,
        aid: int,
        *,
        metrics: Mapping[str, int] | None,
        ok: bool,
        err: str | None,
        captured_at: int,
        src: str = "view",
    ) -> None:
        values = metrics or {}
        def run(conn: sqlite3.Connection) -> None:
            conn.execute(
                """
                INSERT INTO snapshot(stat_date, aid, "view", "like", favorite, coin,
                                     share, reply, danmaku, captured_at, ok, err, src)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(stat_date, aid) DO UPDATE SET
                  "view" = excluded."view",
                  "like" = excluded."like",
                  favorite = excluded.favorite,
                  coin = excluded.coin,
                  share = excluded.share,
                  reply = excluded.reply,
                  danmaku = excluded.danmaku,
                  captured_at = excluded.captured_at,
                  ok = excluded.ok,
                  err = excluded.err,
                  src = excluded.src
                """,
                (
                    stat_date,
                    int(aid),
                    values.get("view"),
                    values.get("like"),
                    values.get("favorite"),
                    values.get("coin"),
                    values.get("share"),
                    values.get("reply"),
                    values.get("danmaku"),
                    int(captured_at),
                    1 if ok else 0,
                    err,
                    src,
                ),
            )
        self.db.write(run)

    def snapshot_counts(self, stat_date: str) -> dict[str, int]:
        def run(conn: sqlite3.Connection) -> dict[str, int]:
            rows = conn.execute(
                "SELECT ok, COUNT(*) AS n FROM snapshot WHERE stat_date = ? GROUP BY ok",
                (stat_date,),
            ).fetchall()
            by_ok = {int(row["ok"]): int(row["n"]) for row in rows}
            return {
                "total": sum(by_ok.values()),
                "ok": by_ok.get(1, 0),
                "failed": by_ok.get(0, 0),
            }
        return self.db.read(run)

    def upsert_up_stat(
        self,
        stat_date: str,
        mid: int,
        *,
        follower: int | None,
        following: int | None,
        ok: bool,
        captured_at: int,
    ) -> None:
        self.db.write(lambda conn: conn.execute(
            """
            INSERT INTO up_stat(stat_date, mid, follower, following, captured_at, ok)
            VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT(stat_date, mid) DO UPDATE SET
              follower = excluded.follower,
              following = excluded.following,
              captured_at = excluded.captured_at,
              ok = excluded.ok
            """,
            (stat_date, int(mid), follower, following, int(captured_at), 1 if ok else 0),
        ))

    def get_up_stat(self, stat_date: str, mid: int) -> dict[str, Any] | None:
        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
            row = conn.execute(
                "SELECT * FROM up_stat WHERE stat_date = ? AND mid = ?",
                (stat_date, int(mid)),
            ).fetchone()
            return dict(row) if row else None
        return self.db.read(run)

    # ------------------------------------------------------------ kv

    def kv_get(self, key: str) -> str | None:
        def run(conn: sqlite3.Connection) -> str | None:
            row = conn.execute("SELECT v FROM kv WHERE k = ?", (key,)).fetchone()
            return None if row is None else str(row["v"])
        return self.db.read(run)

    def kv_set(self, key: str, value: str) -> None:
        self.db.write(lambda conn: conn.execute(
            "INSERT INTO kv(k, v) VALUES (?, ?) "
            "ON CONFLICT(k) DO UPDATE SET v = excluded.v",
            (key, str(value)),
        ))

    # ------------------------------------------------------------ cache version

    @staticmethod
    def cache_version_key(group: str) -> str:
        return f"cache:v:{group}"

    def cache_version(self, group: str) -> int:
        def run(conn: sqlite3.Connection) -> int:
            row = conn.execute(
                "SELECT v FROM kv WHERE k = ?", (self.cache_version_key(group),)
            ).fetchone()
            if row is None:
                return 0
            try:
                return int(str(row["v"]))
            except ValueError:
                return 0
        return self.db.read(run)

    def bump_cache_version(self, group: str, *, now: int | None = None) -> int:
        now = self._now() if now is None else int(now)
        key = self.cache_version_key(group)

        def run(conn: sqlite3.Connection) -> int:
            row = conn.execute("SELECT v FROM kv WHERE k = ?", (key,)).fetchone()
            try:
                value = int(str(row["v"])) + 1 if row is not None else 1
            except ValueError:
                value = 1
            conn.execute(
                "INSERT INTO kv(k, v) VALUES (?, ?) "
                "ON CONFLICT(k) DO UPDATE SET v = excluded.v",
                (key, str(value)),
            )
            return value
        return self.db.write(run)

    # ------------------------------------------------------------ candidate

    def known_aid(self, aid: int) -> bool:
        def run(conn: sqlite3.Connection) -> bool:
            row = conn.execute(
                "SELECT 1 FROM video WHERE aid = ? "
                "UNION ALL SELECT 1 FROM seen_aid WHERE aid = ? LIMIT 1",
                (int(aid), int(aid)),
            ).fetchone()
            return row is not None
        return self.db.read(run)

    def record_seen(
        self, aid: int, *, source: str, now: int | None = None
    ) -> None:
        now = self._now() if now is None else int(now)
        self.db.write(lambda conn: conn.execute(
            "INSERT INTO seen_aid(aid, first_seen, source) VALUES (?, ?, ?) "
            "ON CONFLICT(aid) DO NOTHING",
            (int(aid), now, source),
        ))

    def upsert_candidate(
        self,
        *,
        aid: int,
        bvid: str | None,
        title: str,
        up_mid: int | None,
        up_name: str | None,
        pubdate: int,
        cover: str | None,
        score: int,
        channel: str,
        duration: int = 0,
        play: int = 0,
        like: int = 0,
        favorite: int = 0,
        danmaku: int = 0,
        review: int = 0,
        now: int | None = None,
    ) -> dict[str, Any]:
        now = self._now() if now is None else int(now)

        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            conn.execute(
                """
                INSERT INTO candidate(aid, bvid, title, up_mid, up_name, pubdate,
                                      cover, score, channel, first_seen,
                                      notified_at, decision, decided_at,
                                      duration, play, like, favorite, danmaku, review)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL,
                        ?, ?, ?, ?, ?, ?)
                ON CONFLICT(aid) DO UPDATE SET
                  bvid = COALESCE(excluded.bvid, candidate.bvid),
                  title = COALESCE(NULLIF(excluded.title, ''), candidate.title),
                  up_mid = COALESCE(excluded.up_mid, candidate.up_mid),
                  up_name = COALESCE(excluded.up_name, candidate.up_name),
                  pubdate = CASE WHEN excluded.pubdate > 0
                                 THEN excluded.pubdate ELSE candidate.pubdate END,
                  cover = COALESCE(excluded.cover, candidate.cover),
                  score = excluded.score,
                  channel = excluded.channel,
                  duration = excluded.duration,
                  play = excluded.play,
                  "like" = excluded."like",
                  favorite = excluded.favorite,
                  danmaku = excluded.danmaku,
                  review = excluded.review
                """,
                (
                    int(aid), bvid, title, up_mid, up_name, int(pubdate),
                    cover, float(score), channel, now,
                    int(duration or 0), int(play or 0), int(like or 0),
                    int(favorite or 0), int(danmaku or 0), int(review or 0),
                ),
            )
            row = conn.execute(
                "SELECT * FROM candidate WHERE aid = ?", (int(aid),)
            ).fetchone()
            return dict(row) if row else {}
        return self.db.write(run)

    def get_candidate(self, aid: int) -> dict[str, Any] | None:
        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
            row = conn.execute(
                "SELECT * FROM candidate WHERE aid = ?", (int(aid),)
            ).fetchone()
            return dict(row) if row else None
        return self.db.read(run)

    def list_candidates(self, decision: str | None = None) -> list[dict[str, Any]]:
        def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
            if decision is None:
                rows = conn.execute(
                    "SELECT * FROM candidate ORDER BY COALESCE(score, 0) DESC, "
                    "first_seen DESC, aid"
                ).fetchall()
            else:
                rows = conn.execute(
                    "SELECT * FROM candidate WHERE decision = ? "
                    "ORDER BY COALESCE(score, 0) DESC, first_seen DESC, aid",
                    (decision,),
                ).fetchall()
            return [dict(row) for row in rows]
        return self.db.read(run)

    def mark_candidate_decision(
        self, aid: int, decision: str, *, now: int | None = None
    ) -> None:
        now = self._now() if now is None else int(now)
        self.db.write(lambda conn: conn.execute(
            "UPDATE candidate SET decision = ?, decided_at = ? WHERE aid = ?",
            (decision, now, int(aid)),
        ))

    def reject_pending_for_up(self, up_mid: int, *, now: int | None = None) -> int:
        now = self._now() if now is None else int(now)
        def run(conn: sqlite3.Connection) -> int:
            cur = conn.execute(
                "UPDATE candidate SET decision = 'rejected', decided_at = ? "
                "WHERE up_mid = ? AND decision IS NULL",
                (now, int(up_mid)),
            )
            return max(cur.rowcount, 0)
        return self.db.write(run)

    def delete_candidate(self, aid: int) -> None:
        self.db.write(lambda conn: conn.execute(
            "DELETE FROM candidate WHERE aid = ?", (int(aid),)
        ))

    def delete_expired_candidates(
        self, *, ttl_days: int, now: int | None = None
    ) -> int:
        now = self._now() if now is None else int(now)
        cutoff = now - int(ttl_days) * 86400
        def run(conn: sqlite3.Connection) -> int:
            cur = conn.execute(
                "DELETE FROM candidate WHERE decision IS NULL AND first_seen < ?",
                (cutoff,),
            )
            return max(cur.rowcount, 0)
        return self.db.write(run)

    def mark_notified(
        self,
        aid: int,
        *,
        channel: str,
        batch_id: str | None,
        ok: bool,
        err: str | None = None,
        now: int | None = None,
    ) -> None:
        now = self._now() if now is None else int(now)

        def run(conn: sqlite3.Connection) -> None:
            conn.execute(
                """
                INSERT INTO notify(aid, channel, batch_id, sent_at, ok, err)
                VALUES (?, ?, ?, ?, ?, ?)
                ON CONFLICT(aid, channel) DO UPDATE SET
                  batch_id = excluded.batch_id,
                  sent_at = excluded.sent_at,
                  ok = excluded.ok,
                  err = excluded.err
                """,
                (int(aid), channel, batch_id, now, 1 if ok else 0, err),
            )
            if ok:
                conn.execute(
                    "UPDATE candidate SET notified_at = ? WHERE aid = ?",
                    (now, int(aid)),
                )
        self.db.write(run)

    # ------------------------------------------------------------ fav action

    def video_metadata(self, aids: Iterable[int]) -> dict[int, dict[str, Any]]:
        ids = [int(aid) for aid in aids]
        if not ids:
            return {}

        def run(conn: sqlite3.Connection) -> dict[int, dict[str, Any]]:
            placeholders = ",".join("?" for _ in ids)
            rows = conn.execute(
                f"SELECT aid, up_mid, pubdate FROM video WHERE aid IN ({placeholders})",
                ids,
            ).fetchall()
            return {
                int(row["aid"]): {
                    "up_mid": row["up_mid"],
                    "pubdate": int(row["pubdate"] or 0),
                }
                for row in rows
            }
        return self.db.read(run)

    def folder_for_up(self, up_mid: int) -> int | None:
        def run(conn: sqlite3.Connection) -> int | None:
            row = conn.execute(
                "SELECT folder_id FROM video "
                "WHERE up_mid = ? AND folder_id IS NOT NULL "
                "AND status IN ('active', 'unfaved') "
                "ORDER BY pubdate ASC, aid ASC LIMIT 1",
                (int(up_mid),),
            ).fetchone()
            return int(row["folder_id"]) if row else None
        return self.db.read(run)

    def begin_fav_action(
        self,
        aid: int,
        folder_id: int,
        *,
        now: int | None = None,
        ttl: int = 86400,
    ) -> tuple[str, str]:
        now = self._now() if now is None else int(now)
        token = f"fav:{int(aid)}"

        def run(conn: sqlite3.Connection) -> tuple[str, str]:
            row = conn.execute(
                "SELECT state, expires_at FROM fav_action WHERE token = ?",
                (token,),
            ).fetchone()
            if row is not None:
                state = str(row["state"])
                if state == "done":
                    return token, "done"
                if state == "processing" and int(row["expires_at"]) > now:
                    return token, "busy"
            conn.execute(
                """
                INSERT INTO fav_action(token, aid, folder_id, state, created_at,
                                       expires_at, done_at, position, err)
                VALUES (?, ?, ?, 'processing', ?, ?, NULL, NULL, NULL)
                ON CONFLICT(token) DO UPDATE SET
                  folder_id = excluded.folder_id,
                  state = 'processing',
                  expires_at = excluded.expires_at,
                  done_at = NULL,
                  position = NULL,
                  err = NULL
                """,
                (token, int(aid), int(folder_id), now, now + int(ttl)),
            )
            return token, "processing"
        return self.db.write(run)

    def finish_fav_action(
        self, token: str, *, position: int, now: int | None = None
    ) -> None:
        now = self._now() if now is None else int(now)
        self.db.write(lambda conn: conn.execute(
            "UPDATE fav_action SET state = 'done', done_at = ?, position = ?, err = NULL "
            "WHERE token = ?",
            (now, int(position), token),
        ))

    def fail_fav_action(
        self, token: str, *, error: str, now: int | None = None
    ) -> None:
        now = self._now() if now is None else int(now)
        self.db.write(lambda conn: conn.execute(
            "UPDATE fav_action SET state = 'failed', done_at = ?, err = ? "
            "WHERE token = ?",
            (now, error, token),
        ))

    def get_fav_action(self, token: str) -> dict[str, Any] | None:
        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
            row = conn.execute(
                "SELECT * FROM fav_action WHERE token = ?", (token,)
            ).fetchone()
            return dict(row) if row else None
        return self.db.read(run)

    def expire_fav_actions(self, *, now: int | None = None) -> int:
        now = self._now() if now is None else int(now)
        def run(conn: sqlite3.Connection) -> int:
            cur = conn.execute(
                "UPDATE fav_action SET state = 'failed', done_at = ?, "
                "err = '处理超时' WHERE state = 'processing' AND expires_at < ?",
                (now, now),
            )
            return max(cur.rowcount, 0)
        return self.db.write(run)

    # ------------------------------------------------------------ blacklist

    def seed_blacklist(self, mids: Iterable[int], *, now: int | None = None) -> int:
        now = self._now() if now is None else int(now)
        def run(conn: sqlite3.Connection) -> int:
            inserted = 0
            for mid in mids:
                cur = conn.execute(
                    "INSERT OR IGNORE INTO up_blacklist(mid, name, reason, created_at) "
                    "VALUES (?, NULL, 'seed', ?)",
                    (int(mid), now),
                )
                inserted += max(cur.rowcount, 0)
            return inserted
        return self.db.write(run)

    def add_blacklist(
        self,
        mid: int,
        *,
        name: str | None = None,
        reason: str | None = None,
        now: int | None = None,
    ) -> None:
        now = self._now() if now is None else int(now)
        self.db.write(lambda conn: conn.execute(
            """
            INSERT INTO up_blacklist(mid, name, reason, created_at)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(mid) DO UPDATE SET
              name = COALESCE(excluded.name, up_blacklist.name),
              reason = COALESCE(excluded.reason, up_blacklist.reason)
            """,
            (int(mid), name, reason, now),
        ))

    def is_blacklisted(self, mid: int) -> bool:
        def run(conn: sqlite3.Connection) -> bool:
            row = conn.execute(
                "SELECT 1 FROM up_blacklist WHERE mid = ?", (int(mid),)
            ).fetchone()
            return row is not None
        return self.db.read(run)

    def blacklist(self) -> list[dict[str, Any]]:
        def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
            rows = conn.execute(
                "SELECT mid, name, reason, created_at FROM up_blacklist ORDER BY created_at, mid"
            ).fetchall()
            return [dict(row) for row in rows]
        return self.db.read(run)

    # ------------------------------------------------------------ summary

    def counts(self) -> dict[str, Any]:
        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            status_rows = conn.execute(
                "SELECT status, COUNT(*) AS n FROM video GROUP BY status"
            ).fetchall()
            by_status = {str(row["status"]): int(row["n"]) for row in status_rows}
            folder_count = conn.execute("SELECT COUNT(*) FROM folder").fetchone()[0]
            blacklist_count = conn.execute("SELECT COUNT(*) FROM up_blacklist").fetchone()[0]
            candidate_count = conn.execute("SELECT COUNT(*) FROM candidate").fetchone()[0]
            return {
                "video_total": sum(by_status.values()),
                "video_by_status": by_status,
                "folders": int(folder_count),
                "blacklist": int(blacklist_count),
                "candidates": int(candidate_count),
            }
        return self.db.read(run)