notes_service.py

from __future__ import annotations

import base64
import hashlib
import json
import math
import mimetypes
import re
import sqlite3
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from pathlib import Path
from typing import Callable, Iterator, Literal


RecordKind = Literal["note", "expense"]
RecordScope = Literal["active", "deleted"]


class ServiceError(Exception):
    """Base class for privacy-safe service errors."""


class ValidationError(ServiceError):
    pass


class NotFoundError(ServiceError):
    pass


class InvalidCursorError(ValidationError):
    pass


class DatabaseBusyError(ServiceError):
    pass


class AttachmentAccessError(ServiceError):
    pass


@dataclass(frozen=True, slots=True)
class NoteCreate:
    content: str
    raw_text: str = ""
    category: str = "其他"
    tags: str = ""
    source: str = "admin"
    priority: int = 0
    pinned: bool = False
    note_date: str | None = None
    note_time: str | None = None
    attachment_summary: str = ""


@dataclass(frozen=True, slots=True)
class ExpenseCreate:
    amount: Decimal | str | float | int
    expense_type: str = "expense"
    category: str = "其他"
    content: str = ""
    raw_text: str = ""
    payer: str = ""
    people: str = ""
    split_with: int = 0
    source: str = "admin"
    exp_date: str | None = None
    exp_time: str | None = None


@dataclass(frozen=True, slots=True)
class RecordQuery:
    kind: RecordKind | None = None
    scope: RecordScope = "active"
    search: str = ""
    category: str = ""
    direction: str = ""
    start_date: str | None = None
    end_date: str | None = None
    has_attachment: bool | None = None
    limit: int = 20
    cursor: str | None = None


@dataclass(frozen=True, slots=True)
class RecordView:
    kind: RecordKind
    id: int
    display_at: str
    title: str
    summary: str
    category: str
    tags: tuple[str, ...]
    amount: Decimal | None
    direction: str
    split_with: int
    split_amount: Decimal
    payer: str
    people: tuple[str, ...]
    has_attachment: bool
    attachment_count: int
    deleted: bool
    pinned: bool
    source: str
    raw_text: str
    record_date: str
    record_time: str
    priority: int
    attachment_summary: str


@dataclass(frozen=True, slots=True)
class RecordPage:
    items: tuple[RecordView, ...]
    next_cursor: str | None


@dataclass(frozen=True, slots=True)
class DashboardSummary:
    note_count: int
    expense_total: Decimal
    income_total: Decimal
    attachment_count: int


@dataclass(frozen=True, slots=True)
class CalendarDaySummary:
    date: str
    note_count: int
    expense_count: int


@dataclass(frozen=True, slots=True)
class MutationResult:
    kind: RecordKind
    id: int
    changed: bool


@dataclass(frozen=True, slots=True)
class AuditEvent:
    id: int
    action: str
    record_type: RecordKind
    record_id: int
    result: str
    created_at: str


@dataclass(frozen=True, slots=True)
class AttachmentRef:
    path: Path
    download_name: str
    mime_type: str
    inline: bool


NOTE_COLUMN_MIGRATIONS: dict[str, str] = {
    "raw_text": "TEXT DEFAULT ''",
    "category": "TEXT DEFAULT ''",
    "tags": "TEXT DEFAULT ''",
    "source": "TEXT DEFAULT 'telegram'",
    "priority": "INTEGER DEFAULT 0",
    "pinned": "INTEGER DEFAULT 0",
    "note_date": "TEXT DEFAULT ''",
    "note_time": "TEXT DEFAULT ''",
    "has_attachment": "INTEGER DEFAULT 0",
    "attachments": "TEXT DEFAULT ''",
    "attachment_summary": "TEXT DEFAULT ''",
    "ocr_text": "TEXT DEFAULT ''",
    "search_text": "TEXT DEFAULT ''",
    "ocr_status": "TEXT DEFAULT 'none'",
    "ocr_engine": "TEXT DEFAULT ''",
    "ocr_updated_at": "TIMESTAMP",
    "receipt_text": "TEXT DEFAULT ''",
    "receipt_prices": "TEXT DEFAULT ''",
    "receipt_total": "REAL",
    "receipt_ocr_status": "TEXT DEFAULT 'none'",
    "receipt_ocr_engine": "TEXT DEFAULT ''",
    "gps_lat": "REAL",
    "gps_lon": "REAL",
    "gps_alt": "REAL",
    "location_name": "TEXT DEFAULT ''",
    "location_address": "TEXT DEFAULT ''",
    "location_source": "TEXT DEFAULT ''",
    "location_maps_url": "TEXT DEFAULT ''",
    "location_updated_at": "TIMESTAMP",
    "deleted": "INTEGER DEFAULT 0",
    "deleted_at": "TIMESTAMP",
    "created_at": "TIMESTAMP",
    "updated_at": "TIMESTAMP",
}

EXPENSE_COLUMN_MIGRATIONS: dict[str, str] = {
    "expense_type": "TEXT DEFAULT 'expense'",
    "category": "TEXT DEFAULT '其他'",
    "content": "TEXT DEFAULT ''",
    "raw_text": "TEXT DEFAULT ''",
    "payer": "TEXT DEFAULT ''",
    "people": "TEXT DEFAULT ''",
    "split_with": "INTEGER DEFAULT 0",
    "split_amount": "REAL DEFAULT 0",
    "source": "TEXT DEFAULT 'telegram'",
    "exp_date": "TEXT DEFAULT ''",
    "exp_time": "TEXT DEFAULT ''",
    "deleted": "INTEGER DEFAULT 0",
    "deleted_at": "TIMESTAMP",
    "created_at": "TIMESTAMP",
    "updated_at": "TIMESTAMP",
}


class QuickNotesService:
    SCHEMA_VERSION = 1
    MAX_NOTE_LENGTH = 5000
    MAX_TEXT_LENGTH = 1000
    MAX_AMOUNT = Decimal("999999999999.99")
    SAFE_INLINE_MIME_TYPES = {
        "image/png",
        "image/jpeg",
        "image/webp",
        "image/gif",
        "application/pdf",
    }

    def __init__(
        self,
        root: str | Path,
        *,
        busy_timeout_ms: int = 5000,
        now_fn: Callable[[], datetime] | None = None,
    ) -> None:
        self.root = Path(root).resolve()
        self.db_path = self.root / "notes.db"
        self.attachments_dir = self.root / "attachments"
        self.exports_dir = self.root / "exports"
        self.busy_timeout_ms = max(1, int(busy_timeout_ms))
        self._now_fn = now_fn or datetime.now
        self._ready = False

    def _ensure_dirs(self) -> None:
        self.root.mkdir(parents=True, exist_ok=True)
        self.attachments_dir.mkdir(parents=True, exist_ok=True)
        self.exports_dir.mkdir(parents=True, exist_ok=True)

    @contextmanager
    def _connect(self) -> Iterator[sqlite3.Connection]:
        self._ensure_dirs()
        con = sqlite3.connect(
            self.db_path,
            timeout=self.busy_timeout_ms / 1000,
            isolation_level=None,
        )
        con.row_factory = sqlite3.Row
        try:
            con.execute("PRAGMA foreign_keys=ON")
            con.execute(f"PRAGMA busy_timeout={self.busy_timeout_ms}")
            yield con
        finally:
            con.close()

    @contextmanager
    def _transaction(self) -> Iterator[sqlite3.Connection]:
        with self._connect() as con:
            try:
                con.execute("BEGIN IMMEDIATE")
                yield con
                con.commit()
            except sqlite3.OperationalError as exc:
                con.rollback()
                if self._is_busy_error(exc):
                    raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
                raise
            except Exception:
                con.rollback()
                raise

    @staticmethod
    def _is_busy_error(exc: sqlite3.OperationalError) -> bool:
        text = str(exc).lower()
        return "locked" in text or "busy" in text

    @contextmanager
    def _read_connection(self) -> Iterator[sqlite3.Connection]:
        try:
            with self._connect() as con:
                yield con
        except sqlite3.OperationalError as exc:
            if self._is_busy_error(exc):
                raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
            raise

    @staticmethod
    def _execute_statements(con: sqlite3.Connection, script: str) -> None:
        """Execute a semicolon-terminated SQL batch without implicit commits."""
        pending: list[str] = []
        for line in script.splitlines(keepends=True):
            pending.append(line)
            statement = "".join(pending).strip()
            if statement and sqlite3.complete_statement(statement):
                con.execute(statement)
                pending.clear()
        if "".join(pending).strip():
            raise ValueError("migration SQL contains an incomplete statement")

    @staticmethod
    def _ensure_column(con: sqlite3.Connection, table: str, column: str, definition: str) -> None:
        existing = {row["name"] for row in con.execute(f"PRAGMA table_info({table})")}
        if column not in existing:
            con.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}")

    def migrate(
        self,
        *,
        before_migrate: Callable[[], object] | None = None,
    ) -> object | None:
        prepared: object | None = None
        with self._transaction() as con:
            if before_migrate is not None:
                prepared = before_migrate()
            self._execute_statements(con,
                """
                CREATE TABLE IF NOT EXISTS notes (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    content TEXT NOT NULL,
                    raw_text TEXT DEFAULT '',
                    category TEXT DEFAULT '',
                    tags TEXT DEFAULT '',
                    source TEXT DEFAULT 'telegram',
                    priority INTEGER DEFAULT 0,
                    pinned INTEGER DEFAULT 0,
                    note_date TEXT NOT NULL,
                    note_time TEXT NOT NULL,
                    has_attachment INTEGER DEFAULT 0,
                    attachments TEXT DEFAULT '',
                    attachment_summary TEXT DEFAULT '',
                    ocr_text TEXT DEFAULT '',
                    search_text TEXT DEFAULT '',
                    ocr_status TEXT DEFAULT 'none',
                    ocr_engine TEXT DEFAULT '',
                    ocr_updated_at TIMESTAMP,
                    receipt_text TEXT DEFAULT '',
                    receipt_prices TEXT DEFAULT '',
                    receipt_total REAL,
                    receipt_ocr_status TEXT DEFAULT 'none',
                    receipt_ocr_engine TEXT DEFAULT '',
                    gps_lat REAL,
                    gps_lon REAL,
                    gps_alt REAL,
                    location_name TEXT DEFAULT '',
                    location_address TEXT DEFAULT '',
                    location_source TEXT DEFAULT '',
                    location_maps_url TEXT DEFAULT '',
                    location_updated_at TIMESTAMP,
                    deleted INTEGER DEFAULT 0,
                    deleted_at TIMESTAMP,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );

                CREATE TABLE IF NOT EXISTS expenses (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    amount REAL NOT NULL,
                    expense_type TEXT DEFAULT 'expense',
                    category TEXT DEFAULT '其他',
                    content TEXT DEFAULT '',
                    raw_text TEXT DEFAULT '',
                    payer TEXT DEFAULT '',
                    people TEXT DEFAULT '',
                    split_with INTEGER DEFAULT 0,
                    split_amount REAL DEFAULT 0,
                    source TEXT DEFAULT 'telegram',
                    exp_date TEXT NOT NULL,
                    exp_time TEXT NOT NULL,
                    deleted INTEGER DEFAULT 0,
                    deleted_at TIMESTAMP,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );

                CREATE TABLE IF NOT EXISTS admin_audit (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    action TEXT NOT NULL,
                    record_type TEXT NOT NULL,
                    record_id INTEGER NOT NULL,
                    result TEXT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );

                CREATE TABLE IF NOT EXISTS schema_migrations (
                    version INTEGER PRIMARY KEY,
                    name TEXT NOT NULL,
                    applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );
                """
            )
            for column, definition in NOTE_COLUMN_MIGRATIONS.items():
                self._ensure_column(con, "notes", column, definition)
            for column, definition in EXPENSE_COLUMN_MIGRATIONS.items():
                self._ensure_column(con, "expenses", column, definition)
            self._execute_statements(con,
                """
                CREATE INDEX IF NOT EXISTS idx_date ON notes(note_date);
                CREATE INDEX IF NOT EXISTS idx_cat ON notes(category);
                CREATE INDEX IF NOT EXISTS idx_pinned ON notes(pinned);
                CREATE INDEX IF NOT EXISTS idx_tags ON notes(tags);
                CREATE INDEX IF NOT EXISTS idx_deleted ON notes(deleted);
                CREATE INDEX IF NOT EXISTS idx_search_text ON notes(search_text);
                CREATE INDEX IF NOT EXISTS idx_ocr_status ON notes(ocr_status);
                CREATE INDEX IF NOT EXISTS idx_exp_date ON expenses(exp_date);
                CREATE INDEX IF NOT EXISTS idx_exp_cat ON expenses(category);
                CREATE INDEX IF NOT EXISTS idx_exp_deleted ON expenses(deleted);
                CREATE INDEX IF NOT EXISTS idx_exp_payer ON expenses(payer);
                CREATE INDEX IF NOT EXISTS idx_admin_audit_created ON admin_audit(created_at);
                """
            )
            con.execute(
                "INSERT OR IGNORE INTO schema_migrations(version, name) VALUES (?, ?)",
                (self.SCHEMA_VERSION, "admin_service_baseline"),
            )
        self._ready = True
        return prepared

    def migrate_with_preflight(self, preflight: Callable[[], object]) -> object | None:
        """Run a preflight after acquiring the same writer lock used by migration."""

        return self.migrate(before_migrate=preflight)

    @classmethod
    def _schema_ready_on_connection(cls, con: sqlite3.Connection) -> bool:
        required = {"notes", "expenses", "admin_audit", "schema_migrations"}
        names = {
            row["name"]
            for row in con.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            )
            if row["name"] in required
        }
        if names != required:
            return False
        version = con.execute(
            "SELECT 1 FROM schema_migrations WHERE version=?",
            (cls.SCHEMA_VERSION,),
        ).fetchone()
        note_columns = {row["name"] for row in con.execute("PRAGMA table_info(notes)")}
        expense_columns = {row["name"] for row in con.execute("PRAGMA table_info(expenses)")}
        return bool(version) and set(NOTE_COLUMN_MIGRATIONS) <= note_columns and set(
            EXPENSE_COLUMN_MIGRATIONS
        ) <= expense_columns

    def schema_ready(self) -> bool:
        """Return schema readiness without creating or migrating the database."""
        if not self.db_path.is_file():
            return False
        uri = self.db_path.resolve().as_uri() + "?mode=ro"
        con: sqlite3.Connection | None = None
        try:
            con = sqlite3.connect(
                uri,
                uri=True,
                timeout=self.busy_timeout_ms / 1000,
                isolation_level=None,
            )
            con.row_factory = sqlite3.Row
            con.execute(f"PRAGMA busy_timeout={self.busy_timeout_ms}")
            return self._schema_ready_on_connection(con)
        except sqlite3.OperationalError as exc:
            if self._is_busy_error(exc):
                raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
            if not self.db_path.exists():
                return False
            raise
        finally:
            if con is not None:
                con.close()

    def _ensure_ready(self) -> None:
        if self._ready:
            return
        if not self.db_path.exists():
            self.migrate()
            return
        schema_ready = False
        try:
            with self._connect() as con:
                names = {
                    row["name"]
                    for row in con.execute(
                        """
                        SELECT name FROM sqlite_master
                        WHERE type='table'
                          AND name IN ('notes','expenses','admin_audit','schema_migrations')
                        """
                    )
                }
                if names == {"notes", "expenses", "admin_audit", "schema_migrations"}:
                    version = con.execute(
                        "SELECT 1 FROM schema_migrations WHERE version=?",
                        (self.SCHEMA_VERSION,),
                    ).fetchone()
                    note_columns = {
                        row["name"] for row in con.execute("PRAGMA table_info(notes)")
                    }
                    expense_columns = {
                        row["name"] for row in con.execute("PRAGMA table_info(expenses)")
                    }
                    schema_ready = bool(version) and set(NOTE_COLUMN_MIGRATIONS) <= note_columns and set(
                        EXPENSE_COLUMN_MIGRATIONS
                    ) <= expense_columns
        except sqlite3.OperationalError as exc:
            if self._is_busy_error(exc):
                raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
            raise
        if schema_ready:
            self._ready = True
        else:
            self.migrate()

    @staticmethod
    def _clean_text(value: object, field: str, max_length: int, *, required: bool = False) -> str:
        text = str(value or "").strip()
        if required and not text:
            raise ValidationError(f"{field} 不可空白")
        if len(text) > max_length:
            raise ValidationError(f"{field} 超過長度限制")
        return text

    @staticmethod
    def _normalize_date(value: str | None, fallback: str) -> str:
        text = value or fallback
        try:
            return datetime.strptime(text, "%Y-%m-%d").strftime("%Y-%m-%d")
        except (TypeError, ValueError) as exc:
            raise ValidationError("日期格式必須是 YYYY-MM-DD") from exc

    @staticmethod
    def _normalize_time(value: str | None, fallback: str) -> str:
        text = value or fallback
        for fmt in ("%H:%M:%S", "%H:%M"):
            try:
                return datetime.strptime(text, fmt).strftime("%H:%M:%S")
            except (TypeError, ValueError):
                continue
        raise ValidationError("時間格式必須是 HH:MM 或 HH:MM:SS")

    @classmethod
    def _normalize_amount(cls, value: Decimal | str | float | int) -> Decimal:
        try:
            amount = Decimal(str(value))
        except (InvalidOperation, ValueError) as exc:
            raise ValidationError("金額格式不正確") from exc
        if not amount.is_finite() or amount <= 0 or amount > cls.MAX_AMOUNT:
            raise ValidationError("金額必須大於 0 且在允許範圍內")
        return amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)

    @staticmethod
    def _split_values(value: str) -> tuple[str, ...]:
        return tuple(part.strip() for part in (value or "").split(",") if part.strip())

    @staticmethod
    def _parse_attachments(value: str) -> list[dict[str, object]]:
        if not value:
            return []
        try:
            parsed = json.loads(value)
        except (TypeError, json.JSONDecodeError):
            return []
        return [item for item in parsed if isinstance(item, dict)] if isinstance(parsed, list) else []

    @staticmethod
    def _compose_search_text(*values: object) -> str:
        return " ".join(str(value).strip() for value in values if value and str(value).strip())

    def _insert_audit(
        self,
        con: sqlite3.Connection,
        action: str,
        kind: RecordKind,
        record_id: int,
        result: str,
    ) -> None:
        con.execute(
            "INSERT INTO admin_audit(action, record_type, record_id, result) VALUES (?, ?, ?, ?)",
            (action, kind, record_id, result),
        )

    @staticmethod
    def _validate_kind(kind: str) -> RecordKind:
        if kind not in {"note", "expense"}:
            raise ValidationError("record kind 不正確")
        return kind  # type: ignore[return-value]

    @staticmethod
    def _validate_id(record_id: int) -> int:
        try:
            value = int(record_id)
        except (TypeError, ValueError) as exc:
            raise ValidationError("record ID 必須是正整數") from exc
        if value <= 0:
            raise ValidationError("record ID 必須是正整數")
        return value

    def create_note(self, request: NoteCreate) -> RecordView:
        self._ensure_ready()
        now = self._now_fn()
        content = self._clean_text(request.content, "內容", self.MAX_NOTE_LENGTH, required=True)
        raw_text = self._clean_text(request.raw_text or content, "原始文字", self.MAX_NOTE_LENGTH)
        category = self._clean_text(request.category or "其他", "類別", 100)
        tags = self._clean_text(request.tags, "標籤", self.MAX_TEXT_LENGTH)
        source = self._clean_text(request.source or "admin", "來源", 32, required=True)
        attachment_summary = self._clean_text(
            request.attachment_summary,
            "附件摘要",
            self.MAX_TEXT_LENGTH,
        )
        try:
            priority = int(request.priority)
        except (TypeError, ValueError) as exc:
            raise ValidationError("優先度必須是 0、1 或 2") from exc
        if priority not in {0, 1, 2}:
            raise ValidationError("優先度必須是 0、1 或 2")
        note_date = self._normalize_date(request.note_date, now.strftime("%Y-%m-%d"))
        note_time = self._normalize_time(request.note_time, now.strftime("%H:%M:%S"))
        search_text = self._compose_search_text(
            content,
            raw_text,
            category,
            tags,
            attachment_summary,
        )

        with self._transaction() as con:
            cursor = con.execute(
                """
                INSERT INTO notes (
                    content, raw_text, category, tags, source, priority, pinned,
                    note_date, note_time, has_attachment, attachments,
                    attachment_summary, search_text, created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, '', ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
                """,
                (
                    content,
                    raw_text,
                    category,
                    tags,
                    source,
                    priority,
                    1 if request.pinned else 0,
                    note_date,
                    note_time,
                    attachment_summary,
                    search_text,
                ),
            )
            note_id = int(cursor.lastrowid)
            self._insert_audit(con, "create", "note", note_id, "ok")
        return self.get_record("note", note_id)

    def create_expense(self, request: ExpenseCreate) -> RecordView:
        self._ensure_ready()
        now = self._now_fn()
        amount = self._normalize_amount(request.amount)
        if request.expense_type not in {"expense", "income"}:
            raise ValidationError("記帳類型必須是 expense 或 income")
        category = self._clean_text(request.category or "其他", "類別", 100)
        content = self._clean_text(request.content, "說明", self.MAX_NOTE_LENGTH)
        raw_text = self._clean_text(request.raw_text or content, "原始文字", self.MAX_NOTE_LENGTH)
        payer = self._clean_text(request.payer, "先付者", 200)
        people = self._clean_text(request.people, "參與者", self.MAX_TEXT_LENGTH)
        source = self._clean_text(request.source or "admin", "來源", 32, required=True)
        try:
            split_with = int(request.split_with or 0)
        except (TypeError, ValueError) as exc:
            raise ValidationError("分攤人數必須是整數") from exc
        if split_with < 0 or split_with > 100:
            raise ValidationError("分攤人數必須介於 0 到 100")
        split_amount = (
            (amount / split_with).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
            if split_with > 0
            else Decimal("0.00")
        )
        exp_date = self._normalize_date(request.exp_date, now.strftime("%Y-%m-%d"))
        exp_time = self._normalize_time(request.exp_time, now.strftime("%H:%M:%S"))

        with self._transaction() as con:
            cursor = con.execute(
                """
                INSERT INTO expenses (
                    amount, expense_type, category, content, raw_text, payer,
                    people, split_with, split_amount, source, exp_date, exp_time,
                    created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
                """,
                (
                    float(amount),
                    request.expense_type,
                    category,
                    content,
                    raw_text,
                    payer,
                    people,
                    split_with,
                    float(split_amount),
                    source,
                    exp_date,
                    exp_time,
                ),
            )
            expense_id = int(cursor.lastrowid)
            self._insert_audit(con, "create", "expense", expense_id, "ok")
        return self.get_record("expense", expense_id)

    @staticmethod
    def _note_select(pin_sort: bool = False) -> str:
        pin_expression = "pinned" if pin_sort else "0"
        return f"""
            SELECT
                'note' AS kind,
                id,
                {pin_expression} AS pin_sort,
                note_date || ' ' || note_time AS display_at,
                content AS title,
                CASE
                    WHEN attachment_summary <> '' THEN attachment_summary
                    WHEN raw_text <> content THEN raw_text
                    ELSE ''
                END AS summary,
                category,
                tags,
                NULL AS amount,
                '' AS direction,
                0 AS split_with,
                0 AS split_amount,
                '' AS payer,
                '' AS people,
                has_attachment,
                attachments,
                deleted,
                pinned,
                source,
                raw_text,
                note_date AS record_date,
                note_time AS record_time,
                priority,
                attachment_summary
            FROM notes
        """

    @staticmethod
    def _expense_select() -> str:
        return """
            SELECT
                'expense' AS kind,
                id,
                0 AS pin_sort,
                exp_date || ' ' || exp_time AS display_at,
                CASE
                    WHEN content <> '' THEN content
                    WHEN raw_text <> '' THEN raw_text
                    ELSE category
                END AS title,
                CASE WHEN raw_text <> content THEN raw_text ELSE '' END AS summary,
                category,
                '' AS tags,
                amount,
                expense_type AS direction,
                split_with,
                split_amount,
                payer,
                people,
                0 AS has_attachment,
                '' AS attachments,
                deleted,
                0 AS pinned,
                source,
                raw_text,
                exp_date AS record_date,
                exp_time AS record_time,
                0 AS priority,
                '' AS attachment_summary
            FROM expenses
        """

    def _row_to_record(self, row: sqlite3.Row) -> RecordView:
        kind = self._validate_kind(str(row["kind"]))
        attachment_items = self._parse_attachments(str(row["attachments"] or ""))
        amount = None
        if row["amount"] is not None:
            amount = Decimal(str(row["amount"])).quantize(Decimal("0.01"))
        return RecordView(
            kind=kind,
            id=int(row["id"]),
            display_at=str(row["display_at"]),
            title=str(row["title"] or ""),
            summary=str(row["summary"] or ""),
            category=str(row["category"] or ""),
            tags=self._split_values(str(row["tags"] or "")),
            amount=amount,
            direction=str(row["direction"] or ""),
            split_with=int(row["split_with"] or 0),
            split_amount=Decimal(str(row["split_amount"] or 0)).quantize(Decimal("0.01")),
            payer=str(row["payer"] or ""),
            people=self._split_values(str(row["people"] or "")),
            has_attachment=bool(row["has_attachment"]),
            attachment_count=len(attachment_items),
            deleted=bool(row["deleted"]),
            pinned=bool(row["pinned"]),
            source=str(row["source"] or ""),
            raw_text=str(row["raw_text"] or ""),
            record_date=str(row["record_date"] or ""),
            record_time=str(row["record_time"] or ""),
            priority=int(row["priority"] or 0),
            attachment_summary=str(row["attachment_summary"] or ""),
        )

    def get_record(
        self,
        kind: str,
        record_id: int,
        *,
        include_deleted: bool = False,
    ) -> RecordView:
        self._ensure_ready()
        valid_kind = self._validate_kind(kind)
        valid_id = self._validate_id(record_id)
        select = self._note_select() if valid_kind == "note" else self._expense_select()
        deleted_clause = "" if include_deleted else " AND deleted=0"
        try:
            with self._connect() as con:
                row = con.execute(
                    f"SELECT * FROM ({select}) WHERE id=?{deleted_clause}",
                    (valid_id,),
                ).fetchone()
        except sqlite3.OperationalError as exc:
            if self._is_busy_error(exc):
                raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
            raise
        if row is None:
            raise NotFoundError("找不到記錄")
        return self._row_to_record(row)

    @staticmethod
    def _escape_like(value: str) -> str:
        return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

    def search_suggestions(
        self,
        prefix: str,
        *,
        kind: RecordKind | None = None,
        scope: RecordScope = "active",
        limit: int = 8,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> tuple[str, ...]:
        """Return a small, authenticated-UI-friendly set of matching values."""
        self._ensure_ready()
        if kind is not None:
            self._validate_kind(kind)
        if scope not in {"active", "deleted"}:
            raise ValidationError("scope 必須是 active 或 deleted")
        if type(limit) is not int or not 1 <= limit <= 20:
            raise ValidationError("suggestion limit 必須介於 1 到 20")
        value = self._clean_text(prefix, "搜尋字串", 100).strip()
        if len(value) < 2:
            return ()

        deleted_value = 1 if scope == "deleted" else 0
        normalized_start = self._normalize_date(start_date, start_date) if start_date else None
        normalized_end = self._normalize_date(end_date, end_date) if end_date else None
        if normalized_start and normalized_end and normalized_start > normalized_end:
            raise ValidationError("日期範圍不正確")
        like = f"%{self._escape_like(value)}%"
        selects: list[str] = []
        params: list[object] = []

        def add_select(table: str, date_column: str, column: str) -> None:
            clauses = ["deleted=?", column + " LIKE ? ESCAPE ?"]
            local_params: list[object] = [deleted_value, like, "\\"]
            if normalized_start:
                clauses.append(date_column + ">=?")
                local_params.append(normalized_start)
            if normalized_end:
                clauses.append(date_column + "<=?")
                local_params.append(normalized_end)
            selects.append(
                "SELECT " + column + " AS value FROM " + table + " WHERE " + " AND ".join(clauses)
            )
            params.extend(local_params)

        if kind in {None, "note"}:
            for column in ("content", "category", "tags", "attachment_summary"):
                add_select("notes", "note_date", column)
        if kind in {None, "expense"}:
            for column in ("content", "category", "payer", "people"):
                add_select("expenses", "exp_date", column)
        if not selects:
            return ()

        sql = f"""
            SELECT value
            FROM ({' UNION ALL '.join(selects)})
            WHERE value IS NOT NULL AND TRIM(value) <> ''
            GROUP BY value
            ORDER BY value COLLATE NOCASE
            LIMIT ?
        """
        params.append(limit)
        try:
            with self._read_connection() as con:
                rows = con.execute(sql, params).fetchall()
        except sqlite3.OperationalError as exc:
            if self._is_busy_error(exc):
                raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
            raise
        return tuple(str(row["value"])[:160] for row in rows)

    @staticmethod
    def _query_fingerprint(query: RecordQuery) -> str:
        payload = {
            "kind": query.kind,
            "scope": query.scope,
            "search": query.search,
            "category": query.category,
            "direction": query.direction,
            "start_date": query.start_date,
            "end_date": query.end_date,
            "has_attachment": query.has_attachment,
        }
        raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:20]

    @staticmethod
    def _encode_cursor(row: sqlite3.Row, fingerprint: str) -> str:
        payload = {
            "v": 1,
            "f": fingerprint,
            "p": int(row["pin_sort"] or 0),
            "d": str(row["display_at"]),
            "k": str(row["kind"]),
            "i": int(row["id"]),
        }
        raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")

    @staticmethod
    def _decode_cursor(cursor: str, fingerprint: str) -> tuple[int, str, str, int]:
        try:
            padded = cursor + "=" * (-len(cursor) % 4)
            payload = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8"))
            raw_pin_sort = payload["p"]
            raw_record_id = payload["i"]
            if type(raw_pin_sort) is not int or type(raw_record_id) is not int:
                raise ValueError("cursor field types invalid")
            pin_sort = raw_pin_sort
            display_at = str(payload["d"])
            kind = str(payload["k"])
            record_id = raw_record_id
            if payload.get("v") != 1 or payload.get("f") != fingerprint:
                raise ValueError("cursor query mismatch")
            if kind not in {"note", "expense"} or record_id <= 0 or pin_sort not in {0, 1}:
                raise ValueError("cursor fields invalid")
            datetime.strptime(display_at, "%Y-%m-%d %H:%M:%S")
            return pin_sort, display_at, kind, record_id
        except Exception as exc:
            if isinstance(exc, InvalidCursorError):
                raise
            raise InvalidCursorError("無效或不相容的 cursor") from exc

    def _validated_query(self, query: RecordQuery) -> RecordQuery:
        if query.kind is not None:
            self._validate_kind(query.kind)
        if query.scope not in {"active", "deleted"}:
            raise ValidationError("scope 必須是 active 或 deleted")
        if query.direction not in {"", "expense", "income"}:
            raise ValidationError("direction 不正確")
        if query.has_attachment is not None and type(query.has_attachment) is not bool:
            raise ValidationError("附件篩選必須是布林值或空值")
        if type(query.limit) is not int:
            raise ValidationError("limit 必須是 1 到 100 的整數")
        if not 1 <= query.limit <= 100:
            raise ValidationError("limit 必須介於 1 到 100")
        if query.start_date:
            self._normalize_date(query.start_date, query.start_date)
        if query.end_date:
            self._normalize_date(query.end_date, query.end_date)
        if query.start_date and query.end_date and query.start_date > query.end_date:
            raise ValidationError("日期範圍不正確")
        self._clean_text(query.search, "搜尋字串", 300)
        self._clean_text(query.category, "類別", 100)
        return query

    def list_records(self, query: RecordQuery | None = None) -> RecordPage:
        self._ensure_ready()
        query = self._validated_query(query or RecordQuery())
        deleted_value = 1 if query.scope == "deleted" else 0
        search = self._clean_text(query.search, "搜尋字串", 300)
        category = self._clean_text(query.category, "類別", 100)
        selects: list[str] = []
        all_params: list[object] = []

        include_notes = query.kind in {None, "note"} and not query.direction
        include_expenses = query.kind in {None, "expense"} and not query.has_attachment

        if include_notes:
            clauses = ["deleted=?"]
            params: list[object] = [deleted_value]
            if search:
                like = f"%{self._escape_like(search)}%"
                clauses.append(
                    "(content LIKE ? ESCAPE '\\' OR raw_text LIKE ? ESCAPE '\\' "
                    "OR tags LIKE ? ESCAPE '\\' OR category LIKE ? ESCAPE '\\' "
                    "OR attachment_summary LIKE ? ESCAPE '\\' OR search_text LIKE ? ESCAPE '\\')"
                )
                params.extend([like] * 6)
            if category:
                clauses.append("category=?")
                params.append(category)
            if query.start_date:
                clauses.append("note_date>=?")
                params.append(query.start_date)
            if query.end_date:
                clauses.append("note_date<=?")
                params.append(query.end_date)
            if query.has_attachment is not None:
                clauses.append("has_attachment=?")
                params.append(1 if query.has_attachment else 0)
            pin_sort = query.kind == "note" and query.scope == "active"
            selects.append(self._note_select(pin_sort) + " WHERE " + " AND ".join(clauses))
            all_params.extend(params)

        if include_expenses:
            clauses = ["deleted=?"]
            params = [deleted_value]
            if search:
                like = f"%{self._escape_like(search)}%"
                clauses.append(
                    "(content LIKE ? ESCAPE '\\' OR raw_text LIKE ? ESCAPE '\\' "
                    "OR category LIKE ? ESCAPE '\\' OR payer LIKE ? ESCAPE '\\' "
                    "OR people LIKE ? ESCAPE '\\')"
                )
                params.extend([like] * 5)
            if category:
                clauses.append("category=?")
                params.append(category)
            if query.direction:
                clauses.append("expense_type=?")
                params.append(query.direction)
            if query.start_date:
                clauses.append("exp_date>=?")
                params.append(query.start_date)
            if query.end_date:
                clauses.append("exp_date<=?")
                params.append(query.end_date)
            selects.append(self._expense_select() + " WHERE " + " AND ".join(clauses))
            all_params.extend(params)

        if not selects:
            return RecordPage(items=(), next_cursor=None)

        fingerprint = self._query_fingerprint(query)
        cursor_clause = ""
        if query.cursor:
            pin_sort, display_at, kind, record_id = self._decode_cursor(query.cursor, fingerprint)
            cursor_clause = """
                WHERE
                    pin_sort < ?
                    OR (pin_sort = ? AND display_at < ?)
                    OR (pin_sort = ? AND display_at = ? AND kind > ?)
                    OR (pin_sort = ? AND display_at = ? AND kind = ? AND id < ?)
            """
            all_params.extend(
                [
                    pin_sort,
                    pin_sort,
                    display_at,
                    pin_sort,
                    display_at,
                    kind,
                    pin_sort,
                    display_at,
                    kind,
                    record_id,
                ]
            )

        sql = f"""
            WITH records AS (
                {' UNION ALL '.join(selects)}
            )
            SELECT * FROM records
            {cursor_clause}
            ORDER BY pin_sort DESC, display_at DESC, kind ASC, id DESC
            LIMIT ?
        """
        all_params.append(int(query.limit) + 1)
        try:
            with self._connect() as con:
                rows = con.execute(sql, all_params).fetchall()
        except sqlite3.OperationalError as exc:
            if self._is_busy_error(exc):
                raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
            raise

        page_rows = rows[: int(query.limit)]
        next_cursor = self._encode_cursor(page_rows[-1], fingerprint) if len(rows) > query.limit else None
        return RecordPage(
            items=tuple(self._row_to_record(row) for row in page_rows),
            next_cursor=next_cursor,
        )

    def available_months(self, *, scope: RecordScope = "active") -> tuple[str, ...]:
        """Return months that contain at least one note or expense.

        The dashboard uses this read-only index to move through real history
        instead of guessing from the current month and the immediately prior
        calendar month.
        """
        self._ensure_ready()
        if scope not in {"active", "deleted"}:
            raise ValidationError("scope 必須是 active 或 deleted")
        deleted_value = 1 if scope == "deleted" else 0
        sql = """
            SELECT record_month
            FROM (
                SELECT substr(note_date, 1, 7) AS record_month
                FROM notes
                WHERE deleted=? AND note_date IS NOT NULL AND note_date <> ''
                UNION
                SELECT substr(exp_date, 1, 7) AS record_month
                FROM expenses
                WHERE deleted=? AND exp_date IS NOT NULL AND exp_date <> ''
            )
            WHERE record_month GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]'
            ORDER BY record_month DESC
        """
        try:
            with self._read_connection() as con:
                rows = con.execute(sql, (deleted_value, deleted_value)).fetchall()
        except sqlite3.OperationalError as exc:
            if self._is_busy_error(exc):
                raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
            raise
        return tuple(str(row["record_month"]) for row in rows)

    def dashboard_summary(
        self,
        *,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> DashboardSummary:
        self._ensure_ready()
        if start_date:
            start_date = self._normalize_date(start_date, start_date)
        if end_date:
            end_date = self._normalize_date(end_date, end_date)
        if start_date and end_date and start_date > end_date:
            raise ValidationError("日期範圍不正確")

        note_clauses = ["deleted=0"]
        note_params: list[object] = []
        expense_clauses = ["deleted=0"]
        expense_params: list[object] = []
        if start_date:
            note_clauses.append("note_date>=?")
            note_params.append(start_date)
            expense_clauses.append("exp_date>=?")
            expense_params.append(start_date)
        if end_date:
            note_clauses.append("note_date<=?")
            note_params.append(end_date)
            expense_clauses.append("exp_date<=?")
            expense_params.append(end_date)

        with self._read_connection() as con:
            note_row = con.execute(
                f"""
                SELECT COUNT(*) AS note_count,
                       COALESCE(SUM(CASE WHEN has_attachment=1 THEN 1 ELSE 0 END), 0) AS attachment_count
                FROM notes WHERE {' AND '.join(note_clauses)}
                """,
                note_params,
            ).fetchone()
            expense_rows = con.execute(
                f"""
                SELECT expense_type, COALESCE(SUM(amount),0) AS total
                FROM expenses WHERE {' AND '.join(expense_clauses)}
                GROUP BY expense_type
                """,
                expense_params,
            ).fetchall()
        totals = {str(row["expense_type"]): Decimal(str(row["total"])).quantize(Decimal("0.01")) for row in expense_rows}
        return DashboardSummary(
            note_count=int(note_row["note_count"]),
            expense_total=totals.get("expense", Decimal("0.00")),
            income_total=totals.get("income", Decimal("0.00")),
            attachment_count=int(note_row["attachment_count"]),
        )

    def calendar_days(
        self,
        start_date: str,
        end_date: str,
        *,
        scope: RecordScope = "active",
    ) -> tuple[CalendarDaySummary, ...]:
        self._ensure_ready()
        if scope not in {"active", "deleted"}:
            raise ValidationError("scope 必須是 active 或 deleted")
        normalized_start = self._normalize_date(start_date, start_date)
        normalized_end = self._normalize_date(end_date, end_date)
        if normalized_start > normalized_end:
            raise ValidationError("日期範圍不正確")

        deleted_value = 1 if scope == "deleted" else 0
        sql = """
            WITH dated_records AS (
                SELECT note_date AS record_date, 1 AS note_count, 0 AS expense_count
                FROM notes
                WHERE deleted=? AND note_date>=? AND note_date<=?
                UNION ALL
                SELECT exp_date AS record_date, 0 AS note_count, 1 AS expense_count
                FROM expenses
                WHERE deleted=? AND exp_date>=? AND exp_date<=?
            )
            SELECT record_date,
                   SUM(note_count) AS note_count,
                   SUM(expense_count) AS expense_count
            FROM dated_records
            GROUP BY record_date
            ORDER BY record_date ASC
        """
        try:
            with self._read_connection() as con:
                rows = con.execute(
                    sql,
                    (
                        deleted_value,
                        normalized_start,
                        normalized_end,
                        deleted_value,
                        normalized_start,
                        normalized_end,
                    ),
                ).fetchall()
        except sqlite3.OperationalError as exc:
            if self._is_busy_error(exc):
                raise DatabaseBusyError("資料庫忙碌，請稍後重試") from exc
            raise
        return tuple(
            CalendarDaySummary(
                date=str(row["record_date"]),
                note_count=int(row["note_count"]),
                expense_count=int(row["expense_count"]),
            )
            for row in rows
        )

    def soft_delete(self, kind: str, record_id: int) -> MutationResult:
        self._ensure_ready()
        valid_kind = self._validate_kind(kind)
        valid_id = self._validate_id(record_id)
        table = "notes" if valid_kind == "note" else "expenses"
        with self._transaction() as con:
            cursor = con.execute(
                f"""
                UPDATE {table}
                SET deleted=1, deleted_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP
                WHERE id=? AND deleted=0
                """,
                (valid_id,),
            )
            changed = cursor.rowcount > 0
            self._insert_audit(con, "delete", valid_kind, valid_id, "ok" if changed else "noop")
        return MutationResult(valid_kind, valid_id, changed)

    def restore(self, kind: str, record_id: int) -> MutationResult:
        self._ensure_ready()
        valid_kind = self._validate_kind(kind)
        valid_id = self._validate_id(record_id)
        table = "notes" if valid_kind == "note" else "expenses"
        with self._transaction() as con:
            cursor = con.execute(
                f"""
                UPDATE {table}
                SET deleted=0, deleted_at=NULL, updated_at=CURRENT_TIMESTAMP
                WHERE id=? AND deleted=1
                """,
                (valid_id,),
            )
            changed = cursor.rowcount > 0
            self._insert_audit(con, "restore", valid_kind, valid_id, "ok" if changed else "noop")
        return MutationResult(valid_kind, valid_id, changed)

    def list_audit_events(self, *, limit: int = 200) -> list[AuditEvent]:
        self._ensure_ready()
        if type(limit) is not int or not 1 <= limit <= 1000:
            raise ValidationError("audit limit 不正確")
        with self._read_connection() as con:
            rows = con.execute(
                """
                SELECT id, action, record_type, record_id, result, created_at
                FROM admin_audit ORDER BY id ASC LIMIT ?
                """,
                (limit,),
            ).fetchall()
        return [
            AuditEvent(
                id=int(row["id"]),
                action=str(row["action"]),
                record_type=self._validate_kind(str(row["record_type"])),
                record_id=int(row["record_id"]),
                result=str(row["result"]),
                created_at=str(row["created_at"]),
            )
            for row in rows
        ]

    def get_attachment(
        self,
        note_id: int,
        index: int,
        *,
        include_deleted: bool = False,
    ) -> AttachmentRef:
        self._ensure_ready()
        valid_id = self._validate_id(note_id)
        self.get_record("note", valid_id, include_deleted=include_deleted)
        try:
            valid_index = int(index)
        except (TypeError, ValueError) as exc:
            raise NotFoundError("找不到附件") from exc
        if valid_index < 0:
            raise NotFoundError("找不到附件")
        deleted_clause = "" if include_deleted else " AND deleted=0"
        with self._read_connection() as con:
            row = con.execute(
                f"SELECT attachments FROM notes WHERE id=?{deleted_clause}",
                (valid_id,),
            ).fetchone()
        items = self._parse_attachments(str(row["attachments"] or "")) if row else []
        if valid_index >= len(items):
            raise NotFoundError("找不到附件")
        item = items[valid_index]
        raw_path = str(item.get("path") or "")
        if not raw_path:
            raise NotFoundError("找不到附件")
        candidate = Path(raw_path)
        if not candidate.is_absolute():
            candidate = self.root / candidate
        resolved = candidate.resolve()
        attachment_root = self.attachments_dir.resolve()
        try:
            resolved.relative_to(attachment_root)
        except ValueError as exc:
            raise AttachmentAccessError("附件路徑不在允許範圍") from exc
        if not resolved.exists() or not resolved.is_file():
            raise NotFoundError("附件檔案不存在")

        requested_name = str(item.get("filename") or resolved.name)
        download_name = Path(requested_name.replace("\\", "/")).name
        download_name = re.sub(r"[\x00-\x1f\x7f]", "", download_name).strip() or "attachment"
        mime_type = mimetypes.guess_type(resolved.name)[0] or "application/octet-stream"
        return AttachmentRef(
            path=resolved,
            download_name=download_name,
            mime_type=mime_type,
            inline=mime_type in self.SAFE_INLINE_MIME_TYPES,
        )