hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/core/storage.py

"""本地存储服务 —— JSON/SQLite 文件持久化与数据备份."""

from __future__ import annotations

import json
import logging
import os
import re
import sqlite3
from collections import defaultdict
from contextlib import contextmanager, suppress
from typing import Any

from .author import Author
from .paper import Paper
from .paper_paths import LIBRARY_PDF_ROOT_DIR, category_slug_for_pdf_dir
from ..settings import get_settings

logger = logging.getLogger(__name__)

class PaperDatabase:
    """SQLite 论文数据库 —— 论文 CRUD、全文搜索(FTS)、分类与标签管理."""

    def __init__(self, db_path: str | None = None) -> None:
        if db_path is None:
            db_path = os.path.join(os.path.abspath(get_settings().data_dir), "papers.db")

        self.db_path = db_path
        self._library_fts_ready = False
        self._ensure_directory()
        self._init_database()
        self._library_fts_ready = self._detect_fts_table()

    def _data_root(self) -> str:
        return os.path.dirname(os.path.abspath(self.db_path))

    def _abs_local_pdf(self, relpath: str | None) -> str | None:
        if not relpath or not str(relpath).strip():
            return None
        return os.path.normpath(os.path.join(self._data_root(), str(relpath).strip()))

    def _ensure_directory(self) -> None:
        os.makedirs(os.path.dirname(self.db_path), exist_ok=True)

    def _detect_fts_table(self) -> bool:
        try:
            return self._query(
                "SELECT 1 FROM sqlite_master WHERE type='table' AND name='papers_fts' LIMIT 1",
                fetch='one'
            ) is not None
        except Exception:
            return False

    @contextmanager
    def _get_connection(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception as e:
            conn.rollback()
            logger.error("Database transaction failed: %s", e)
            raise
        finally:
            conn.close()

    def _query(self, sql, params=(), fetch='all'):
        with self._get_connection() as conn:
            cur = conn.cursor()
            cur.execute(sql, params)
            if fetch == 'one':
                return cur.fetchone()
            if fetch == 'all':
                return cur.fetchall()
            return None

    def _ensure_column(self, conn: sqlite3.Connection, col_name: str, col_type: str = "TEXT") -> None:
        cur = conn.cursor()
        cur.execute("PRAGMA table_info(papers)")
        cols = [r[1] for r in cur.fetchall()]
        if col_name not in cols:
            cur.execute(f"ALTER TABLE papers ADD COLUMN {col_name} {col_type}")

    def _init_database(self) -> None:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("PRAGMA user_version")
            db_version = int(cursor.fetchone()[0])

            if db_version < 1:
                cursor.execute(
                    """
                    CREATE TABLE IF NOT EXISTS papers (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        title TEXT NOT NULL,
                        abstract TEXT,
                        doi TEXT UNIQUE,
                        pmid TEXT UNIQUE,
                        arxiv_id TEXT UNIQUE,
                        pmc_id TEXT UNIQUE,
                        journal TEXT,
                        year INTEGER,
                        volume TEXT,
                        issue TEXT,
                        pages TEXT,
                        publisher TEXT,
                        pdf_url TEXT,
                        source_url TEXT,
                        local_pdf_path TEXT,
                        keywords TEXT,
                        mesh_terms TEXT,
                        "references" TEXT,
                        citations INTEGER DEFAULT 0,
                        source TEXT DEFAULT 'unknown',
                        notes TEXT,
                        tags TEXT,
                        category TEXT,
                        rating INTEGER,
                        read_status TEXT DEFAULT 'unread',
                        importance TEXT DEFAULT 'normal',
                        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                    )
                    """
                )
                self._ensure_column(conn, "local_pdf_path")
                self._ensure_column(conn, "category")
                self._ensure_column(conn, "venue_type")

                cursor.execute(
                    """
                    CREATE TABLE IF NOT EXISTS authors (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        name TEXT NOT NULL,
                        affiliation TEXT,
                        email TEXT,
                        orcid TEXT UNIQUE
                    )
                    """
                )
                cursor.execute(
                    """
                    CREATE TABLE IF NOT EXISTS paper_authors (
                        paper_id INTEGER NOT NULL,
                        author_id INTEGER NOT NULL,
                        author_order INTEGER DEFAULT 0,
                        PRIMARY KEY (paper_id, author_id),
                        FOREIGN KEY (paper_id) REFERENCES papers(id) ON DELETE CASCADE,
                        FOREIGN KEY (author_id) REFERENCES authors(id) ON DELETE CASCADE
                    )
                    """
                )
                cursor.execute("PRAGMA user_version = 1")
                db_version = 1

            if db_version < 2:
                cursor.executescript(
                    """
                    CREATE INDEX IF NOT EXISTS idx_papers_category ON papers(category);
                    CREATE INDEX IF NOT EXISTS idx_papers_year ON papers(year);
                    CREATE INDEX IF NOT EXISTS idx_papers_read_status ON papers(read_status);
                    CREATE INDEX IF NOT EXISTS idx_papers_created_at ON papers(created_at);
                    CREATE INDEX IF NOT EXISTS idx_category_year ON papers(category, year);
                    """
                )
                try:
                    cursor.executescript(
                        """
                        CREATE VIRTUAL TABLE IF NOT EXISTS papers_fts USING fts5(
                            title, abstract,
                            content='papers', content_rowid='id'
                        );

                        CREATE TRIGGER IF NOT EXISTS papers_ai AFTER INSERT ON papers BEGIN
                            INSERT INTO papers_fts(rowid, title, abstract)
                            VALUES (new.id, new.title, new.abstract);
                        END;

                        CREATE TRIGGER IF NOT EXISTS papers_ad AFTER DELETE ON papers BEGIN
                            INSERT INTO papers_fts(papers_fts, rowid, title, abstract)
                            VALUES ('delete', old.id, old.title, old.abstract);
                        END;

                        CREATE TRIGGER IF NOT EXISTS papers_au AFTER UPDATE ON papers BEGIN
                            INSERT INTO papers_fts(papers_fts, rowid, title, abstract)
                            VALUES ('delete', old.id, old.title, old.abstract);
                            INSERT INTO papers_fts(rowid, title, abstract)
                            VALUES (new.id, new.title, new.abstract);
                        END;
                        """
                    )
                    cursor.execute("INSERT INTO papers_fts(papers_fts) VALUES('rebuild')")
                except sqlite3.OperationalError as e:
                    logger.warning("FTS5 不可用或未启用，跳过全文索引: %s", e)
                cursor.execute("PRAGMA user_version = 2")

    @staticmethod
    def _norm_id_field(val: str | None) -> str | None:
        s = (val or "").strip()
        return s if s else None

    def _sync_saved_meta(self, cursor: sqlite3.Cursor, paper_id: int, paper: Paper) -> None:
        cat = getattr(paper, "category", None)
        doi = self._norm_id_field(paper.doi)
        arxiv_id = self._norm_id_field(paper.arxiv_id)
        abs_new = (paper.abstract or "").strip() or None
        title_new = (paper.title or "").strip() or None
        cursor.execute(
            """UPDATE papers SET category = ?, tags = ?, pdf_url = ?, source_url = ?,
               doi = COALESCE(?, doi),
               arxiv_id = COALESCE(?, arxiv_id),
               abstract = COALESCE(?, abstract),
               title = COALESCE(?, title),
               venue_type = COALESCE(?, venue_type),
               updated_at = CURRENT_TIMESTAMP WHERE id = ?""",
            (
                cat,
                json.dumps(paper.tags or [], ensure_ascii=False),
                paper.pdf_url,
                paper.source_url,
                doi,
                arxiv_id,
                abs_new,
                title_new,
                getattr(paper, "venue_type", None),
                paper_id,
            ),
        )

    def _add_paper_internal(self, conn: sqlite3.Connection, paper: Paper) -> tuple[int, bool]:
        cursor = conn.cursor()
        doi = self._norm_id_field(paper.doi)
        arxiv_id = self._norm_id_field(paper.arxiv_id)
        pmid = self._norm_id_field(paper.pmid)
        pmc_id = self._norm_id_field(paper.pmc_id)

        for field, val in (("doi", doi), ("arxiv_id", arxiv_id), ("pmid", pmid), ("pmc_id", pmc_id)):
            if val:
                cursor.execute(f"SELECT id FROM papers WHERE {field} = ?", (val,))
                existing = cursor.fetchone()
                if existing:
                    eid = int(existing[0])
                    self._sync_saved_meta(cursor, eid, paper)
                    return eid, False

        cursor.execute(
            """
            INSERT INTO papers (
                title, abstract, doi, pmid, arxiv_id, pmc_id,
                journal, year, volume, issue, pages, publisher,
                pdf_url, source_url, local_pdf_path, keywords, mesh_terms, "references",
                citations, source, notes, tags, category, venue_type, rating, read_status, importance
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                paper.title,
                paper.abstract,
                doi,
                pmid,
                arxiv_id,
                pmc_id,
                paper.journal,
                paper.year,
                paper.volume,
                paper.issue,
                paper.pages,
                paper.publisher,
                paper.pdf_url,
                paper.source_url,
                getattr(paper, "local_pdf_path", None),
                json.dumps(paper.keywords, ensure_ascii=False),
                json.dumps(paper.mesh_terms, ensure_ascii=False),
                json.dumps(paper.references, ensure_ascii=False),
                paper.citations,
                paper.source,
                paper.notes,
                json.dumps(paper.tags, ensure_ascii=False),
                getattr(paper, "category", None),
                getattr(paper, "venue_type", None),
                paper.rating,
                paper.read_status,
                paper.importance,
            ),
        )

        paper_id = cursor.lastrowid
        self._add_authors(conn, paper_id, paper.authors)
        return int(paper_id), True

    def add_paper(self, paper: Paper) -> tuple[int, bool]:
        with self._get_connection() as conn:
            return self._add_paper_internal(conn, paper)

    def add_papers(self, papers: list[Paper]) -> tuple[list[int], int, int]:
        ids: list[int] = []
        added = 0
        updated = 0
        with self._get_connection() as conn:
            for paper in papers:
                try:
                    paper_id, is_new = self._add_paper_internal(conn, paper)
                    ids.append(int(paper_id))
                    if is_new:
                        added += 1
                    else:
                        updated += 1
                except Exception as e:
                    logger.error("批量添加文献时出错 '%s': %s", getattr(paper, "title", ""), e)
                    ids.append(-1)
        return ids, added, updated

    def _add_authors(self, conn: sqlite3.Connection, paper_id: int, authors: list[Author]) -> None:
        cursor = conn.cursor()
        for order, author in enumerate(authors):
            if author.orcid:
                cursor.execute("SELECT id FROM authors WHERE orcid = ?", (author.orcid,))
            else:
                cursor.execute("SELECT id FROM authors WHERE name = ?", (author.name,))

            result = cursor.fetchone()
            if result:
                author_id = result[0]
            else:
                cursor.execute(
                    "INSERT INTO authors (name, affiliation, email, orcid) VALUES (?, ?, ?, ?)",
                    (author.name, author.affiliation, author.email, author.orcid),
                )
                author_id = cursor.lastrowid

            with suppress(sqlite3.IntegrityError):
                cursor.execute(
                    "INSERT INTO paper_authors (paper_id, author_id, author_order) VALUES (?, ?, ?)",
                    (paper_id, author_id, order),
                )

    def _fetch_authors_for_papers(
        self, conn: sqlite3.Connection, paper_ids: list[int]
    ) -> dict[int, list[Author]]:
        if not paper_ids:
            return {}
        cursor = conn.cursor()
        placeholders = ",".join("?" * len(paper_ids))
        cursor.execute(
            f"""
            SELECT pa.paper_id, a.* FROM authors a
            JOIN paper_authors pa ON a.id = pa.author_id
            WHERE pa.paper_id IN ({placeholders})
            ORDER BY pa.paper_id, pa.author_order
            """,
            paper_ids,
        )
        authors_by_paper: dict[int, list[Author]] = defaultdict(list)
        for row in cursor.fetchall():
            authors_by_paper[int(row["paper_id"])].append(
                Author(
                    name=row["name"],
                    affiliation=row["affiliation"],
                    email=row["email"],
                    orcid=row["orcid"],
                    db_id=int(row["id"]) if row["id"] is not None else None,
                )
            )
        return dict(authors_by_paper)

    def _row_to_paper_fast(self, row: sqlite3.Row, authors: list[Author]) -> Paper:
        keys = row.keys()
        return Paper(
            id=row["id"],
            title=row["title"],
            authors=authors,
            abstract=row["abstract"],
            doi=row["doi"],
            pmid=row["pmid"],
            arxiv_id=row["arxiv_id"],
            pmc_id=row["pmc_id"],
            journal=row["journal"],
            year=row["year"],
            volume=row["volume"],
            issue=row["issue"],
            pages=row["pages"],
            publisher=row["publisher"],
            pdf_url=row["pdf_url"],
            source_url=row["source_url"],
            local_pdf_path=row["local_pdf_path"] if "local_pdf_path" in keys else None,
            keywords=json.loads(row["keywords"] or "[]"),
            mesh_terms=json.loads(row["mesh_terms"] or "[]"),
            references=json.loads(row["references"] or "[]"),
            citations=row["citations"] or 0,
            source=row["source"] or "unknown",
            notes=row["notes"],
            tags=json.loads(row["tags"] or "[]"),
            category=row["category"] if "category" in keys else None,
            venue_type=row["venue_type"] if "venue_type" in keys else None,
            rating=row["rating"],
            read_status=row["read_status"] or "unread",
            importance=row["importance"] or "normal",
        )

    def count_papers(self) -> int:
        return self._query("SELECT COUNT(*) FROM papers", fetch='one')[0]

    def get_all_papers(self, limit: int | None = None, offset: int = 0, order_by: str = "created_at DESC") -> list[Paper]:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            query = f"SELECT * FROM papers ORDER BY {order_by}"
            if limit:
                query += f" LIMIT {int(limit)}"
            if offset:
                query += f" OFFSET {int(offset)}"
            cursor.execute(query)
            rows = cursor.fetchall()
            if not rows:
                return []
            paper_ids = [int(r["id"]) for r in rows]
            authors_map = self._fetch_authors_for_papers(conn, paper_ids)
            return [
                self._row_to_paper_fast(row, authors_map.get(int(row["id"]), []))
                for row in rows
            ]

    def get_paper_by_id(self, paper_id: int) -> Paper | None:
        row = self._query("SELECT * FROM papers WHERE id = ?", (paper_id,), fetch='one')
        if not row:
            return None
        with self._get_connection() as conn:
            authors_map = self._fetch_authors_for_papers(conn, [paper_id])
        return self._row_to_paper_fast(row, authors_map.get(paper_id, []))

    def search_library(
        self,
        query: str | None = None,
        tags: list[str] | None = None,
        year_from: int | None = None,
        year_to: int | None = None,
        read_status: str | None = None,
        category: str | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[Paper]:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            clauses: list[str] = ["1=1"]
            params: list[Any] = []
            use_fts = False
            match_expr = ""
            clean_query = ""

            if query and str(query).strip():
                clean_query = re.sub(r'["\'*^]', " ", str(query)).strip()
                if clean_query and self._library_fts_ready:
                    parts = [w for w in clean_query.split() if w.strip()]
                    if parts:
                        match_expr = " AND ".join(f'"{w}"' for w in parts)
                        use_fts = True

            if use_fts:
                base_from = "papers p"
                clauses.append(
                    "(p.id IN (SELECT rowid FROM papers_fts WHERE papers_fts MATCH ?)"
                    " OR p.id IN (SELECT pa.paper_id FROM paper_authors pa JOIN authors a ON pa.author_id = a.id WHERE a.name LIKE ?))"
                )
                params.append(match_expr)
                like_author = f"%{clean_query}%"
                params.append(like_author)
            elif query and str(query).strip():
                clauses.append("(p.title LIKE ? OR p.abstract LIKE ? OR p.id IN (SELECT pa.paper_id FROM paper_authors pa JOIN authors a ON pa.author_id = a.id WHERE a.name LIKE ?))")
                like = f"%{str(query).strip()}%"
                params.extend([like, like, like])
                base_from = "papers p"
            else:
                base_from = "papers p"

            if category:
                cat = category.strip()
                if cat.endswith("/*"):
                    prefix = cat[:-2].strip()
                    if prefix == "未分类":
                        clauses.append(
                            "(p.category IS NULL OR TRIM(COALESCE(p.category, '')) IN ('', '未分类') "
                            "OR TRIM(COALESCE(p.category, '')) LIKE '未分类/%')"
                        )
                    elif prefix:
                        clauses.append(
                            "(TRIM(COALESCE(p.category, '')) = ? OR TRIM(COALESCE(p.category, '')) LIKE ?)"
                        )
                        params.extend([prefix, prefix + "/%"])
                elif cat == "未分类":
                    clauses.append(
                        "(p.category IS NULL OR TRIM(COALESCE(p.category, '')) IN ('', '未分类'))"
                    )
                else:
                    clauses.append("TRIM(COALESCE(p.category, '')) = ?")
                    params.append(cat)

            if year_from is not None:
                clauses.append("(p.year IS NOT NULL AND p.year >= ?)")
                params.append(year_from)
            if year_to is not None:
                clauses.append("(p.year IS NOT NULL AND p.year <= ?)")
                params.append(year_to)
            if read_status:
                clauses.append("p.read_status = ?")
                params.append(read_status)

            order_clause = "ORDER BY p.created_at DESC"

            sql = f"SELECT p.* FROM {base_from} WHERE {' AND '.join(clauses)} {order_clause} LIMIT ?"
            params.append(int(limit))
            if offset:
                sql += " OFFSET ?"
                params.append(int(offset))
            cursor.execute(sql, params)
            rows = cursor.fetchall()
            if not rows:
                return []
            paper_ids = [int(r["id"]) for r in rows]
            authors_map = self._fetch_authors_for_papers(conn, paper_ids)
            papers = [self._row_to_paper_fast(row, authors_map.get(int(row["id"]), [])) for row in rows]
            if tags:
                tag_set = set(tags)
                papers = [p for p in papers if tag_set.intersection(set(p.tags))]
            return papers

    def update_paper(self, paper_id: int, **fields) -> bool:
        allowed = {"notes", "tags", "rating", "read_status", "importance", "category", "abstract"}
        updates = {k: v for k, v in fields.items() if k in allowed and v is not None}
        if not updates:
            return False
        if "tags" in updates and isinstance(updates["tags"], list):
            updates["tags"] = json.dumps(updates["tags"], ensure_ascii=False)
        set_parts = [f"{k} = ?" for k in updates]
        values = list(updates.values()) + [paper_id]
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                f"UPDATE papers SET {', '.join(set_parts)}, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
                values,
            )
            return cursor.rowcount > 0

    def set_local_pdf_path(self, paper_id: int, relative_path: str | None) -> bool:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "UPDATE papers SET local_pdf_path = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
                (relative_path, paper_id),
            )
            return cursor.rowcount > 0

    def delete_paper(self, paper_id: int) -> bool:
        with self._get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT local_pdf_path FROM papers WHERE id = ?", (paper_id,))
            row = cursor.fetchone()
            if row and row[0]:
                abspath = self._abs_local_pdf(row[0])
                if abspath and os.path.isfile(abspath):
                    with suppress(OSError):
                        os.remove(abspath)
            cursor.execute("DELETE FROM paper_authors WHERE paper_id = ?", (paper_id,))
            cursor.execute("DELETE FROM papers WHERE id = ?", (paper_id,))
            return cursor.rowcount > 0

    def repair_library_local_pdf_paths_batch(self, paper_ids: list[int]) -> dict[int, str]:
        want = {int(x) for x in paper_ids if x is not None and int(x) >= 0}
        if not want:
            return {}
        data_root = self._data_root()
        lib_root = os.path.join(data_root, LIBRARY_PDF_ROOT_DIR)
        if not os.path.isdir(lib_root):
            return {}
        candidates: dict[int, list[tuple[float, str]]] = {k: [] for k in want}
        name_pat = re.compile(r"^(\d+)\.pdf$")
        for dirpath, _, filenames in os.walk(lib_root):
            for fn in filenames:
                m = name_pat.match(fn)
                if not m:
                    continue
                pid = int(m.group(1))
                if pid not in want:
                    continue
                full = os.path.join(dirpath, fn)
                try:
                    mt = os.path.getmtime(full)
                except OSError:
                    continue
                rel = os.path.relpath(full, data_root).replace("\\", "/")
                if rel.startswith(".."):
                    continue
                candidates[pid].append((mt, rel))
        out: dict[int, str] = {}
        for pid, rows in candidates.items():
            if not rows:
                continue
            rows.sort(key=lambda x: -x[0])
            best_rel = rows[0][1]
            if self.set_local_pdf_path(pid, best_rel):
                out[pid] = best_rel
        return out

    def get_library_pdf_abspath(self, paper_id: int) -> str | None:
        p = self.get_paper_by_id(paper_id)
        if not p or not (getattr(p, "local_pdf_path", None) or "").strip():
            return None
        rel = (p.local_pdf_path or "").strip()
        candidates = [self._abs_local_pdf(rel)]
        if rel.startswith(f"{LIBRARY_PDF_ROOT_DIR}/"):
            candidates.append(
                self._abs_local_pdf("pdfs/" + rel[len(LIBRARY_PDF_ROOT_DIR) + 1 :])
            )
        elif rel.startswith("pdfs/"):
            candidates.append(
                self._abs_local_pdf(f"{LIBRARY_PDF_ROOT_DIR}/" + rel[len("pdfs/") :])
            )
        root = os.path.realpath(self._data_root())
        for abspath in candidates:
            if not abspath or not os.path.isfile(abspath):
                continue
            real_f = os.path.realpath(abspath)
            if real_f != root and not real_f.startswith(root + os.sep):
                continue
            return real_f
        return None

    def list_library_category_folders(self) -> list[dict[str, Any]]:
        rows = self._query(
            """
            SELECT COALESCE(NULLIF(TRIM(category), ''), '未分类') AS c, COUNT(*) AS n
            FROM papers
            GROUP BY c
            ORDER BY n DESC, c ASC
            """
        )

        standalone: dict[str, int] = {}
        by_parent: dict[str, list[dict[str, Any]]] = defaultdict(list)

        for row in rows:
            c = row["c"] or "未分类"
            n = int(row["n"])
            if "/" not in c:
                standalone[c] = standalone.get(c, 0) + n
                continue
            parts = [p.strip() for p in c.split("/") if p.strip()]
            if len(parts) < 2:
                standalone[c] = standalone.get(c, 0) + n
                continue
            parent = parts[0]
            label = "/".join(parts[1:])
            by_parent[parent].append(
                {
                    "category": c,
                    "label": label,
                    "folder": category_slug_for_pdf_dir(c),
                    "count": n,
                }
            )

        consumed_standalone: set[str] = set()
        out: list[dict[str, Any]] = []

        for parent in sorted(
            by_parent.keys(),
            key=lambda p: (-sum(x["count"] for x in by_parent[p]), p),
        ):
            ch = sorted(by_parent[parent], key=lambda x: (-x["count"], x["label"]))
            extra = standalone.get(parent, 0)
            total = sum(x["count"] for x in ch) + extra
            children: list[dict[str, Any]] = []
            if extra > 0:
                children.append(
                    {
                        "category": parent,
                        "label": "未分子类",
                        "folder": category_slug_for_pdf_dir(parent),
                        "count": extra,
                    }
                )
                consumed_standalone.add(parent)
            children.extend(ch)
            out.append(
                {
                    "category": parent,
                    "folder": category_slug_for_pdf_dir(parent),
                    "count": total,
                    "children": children,
                }
            )

        for cat, n in standalone.items():
            if cat in consumed_standalone:
                continue
            out.append(
                {
                    "category": cat,
                    "folder": category_slug_for_pdf_dir(cat),
                    "count": n,
                    "children": [],
                }
            )

        out.sort(key=lambda x: (-x["count"], x["category"]))
        return out

    def list_library_categories_by_count(self, limit: int = 80) -> list[str]:
        limit = int(limit or 0)
        if limit <= 0:
            limit = 80
        rows = self._query(
            """
            SELECT COALESCE(NULLIF(TRIM(category), ''), '未分类') AS c, COUNT(*) AS n
            FROM papers
            GROUP BY c
            ORDER BY n DESC, c ASC
            LIMIT ?
            """,
            (limit,),
        )
        out: list[str] = []
        for r in rows:
            c = (r["c"] or "").strip() or "未分类"
            if c not in out:
                out.append(c)
        return out