Initial commit

db.py

@@ -0,0 +1,153 @@
+"""数据库操作层 — 封装所有 SQLite 操作。"""
+
+import sqlite3
+import os
+from datetime import datetime
+
+DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "books.db")
+
+
+def _connect():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA journal_mode=WAL")
+    return conn
+
+
+def init_db():
+    """初始化数据库和表结构。"""
+    conn = _connect()
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS books (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            title TEXT NOT NULL,
+            author TEXT,
+            translator TEXT,
+            publisher TEXT,
+            pub_date TEXT,
+            cover_url TEXT,
+            format TEXT DEFAULT 'paper',
+            status TEXT DEFAULT 'to-read',
+            rating INTEGER,
+            douban_score REAL,
+            goodreads_score REAL,
+            tags TEXT DEFAULT '',
+            notes TEXT,
+            start_date TEXT,
+            finish_date TEXT,
+            created_at TEXT DEFAULT (datetime('now', 'localtime')),
+            updated_at TEXT DEFAULT (datetime('now', 'localtime'))
+        )
+    """)
+    conn.commit()
+    conn.close()
+
+
+def add_book(title, *, author=None, translator=None, publisher=None,
+             pub_date=None, cover_url=None, format="paper", status="to-read",
+             rating=None, douban_score=None, goodreads_score=None,
+             tags="", notes=None, start_date=None, finish_date=None):
+    """添加一本书，返回新书 ID。"""
+    conn = _connect()
+    cur = conn.execute("""
+        INSERT INTO books (title, author, translator, publisher, pub_date,
+                           cover_url, format, status, rating, douban_score,
+                           goodreads_score, tags, notes, start_date, finish_date)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    """, (title, author, translator, publisher, pub_date, cover_url,
+          format, status, rating, douban_score, goodreads_score,
+          tags, notes, start_date, finish_date))
+    conn.commit()
+    book_id = cur.lastrowid
+    conn.close()
+    return book_id
+
+
+def update_book(book_id, **kwargs):
+    """更新书目字段。只更新传入的字段。"""
+    if not kwargs:
+        return
+    allowed = {
+        "title", "author", "translator", "publisher", "pub_date",
+        "cover_url", "format", "status", "rating", "douban_score",
+        "goodreads_score", "tags", "notes", "start_date", "finish_date",
+    }
+    fields = {k: v for k, v in kwargs.items() if k in allowed}
+    if not fields:
+        return
+    fields["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    set_clause = ", ".join(f"{k} = ?" for k in fields)
+    values = list(fields.values()) + [book_id]
+    conn = _connect()
+    conn.execute(f"UPDATE books SET {set_clause} WHERE id = ?", values)
+    conn.commit()
+    conn.close()
+
+
+def delete_book(book_id):
+    """删除一本书。"""
+    conn = _connect()
+    conn.execute("DELETE FROM books WHERE id = ?", (book_id,))
+    conn.commit()
+    conn.close()
+
+
+def get_book(book_id):
+    """获取单本书详情，返回 dict 或 None。"""
+    conn = _connect()
+    row = conn.execute("SELECT * FROM books WHERE id = ?", (book_id,)).fetchone()
+    conn.close()
+    return dict(row) if row else None
+
+
+def list_books(*, status=None, format=None, tag=None):
+    """查询书目列表，支持按状态/格式/标签过滤。"""
+    conn = _connect()
+    query = "SELECT * FROM books WHERE 1=1"
+    params = []
+    if status:
+        query += " AND status = ?"
+        params.append(status)
+    if format:
+        query += " AND format = ?"
+        params.append(format)
+    if tag:
+        # 逗号分隔的 tags 字段中模糊匹配
+        query += " AND (',' || tags || ',' LIKE ?)"
+        params.append(f"%,{tag},%")
+    query += " ORDER BY updated_at DESC"
+    rows = conn.execute(query, params).fetchall()
+    conn.close()
+    return [dict(r) for r in rows]
+
+
+def get_stats():
+    """获取统计数据。"""
+    conn = _connect()
+    total = conn.execute("SELECT COUNT(*) FROM books").fetchone()[0]
+    by_status = {}
+    for row in conn.execute("SELECT status, COUNT(*) as cnt FROM books GROUP BY status"):
+        by_status[row["status"]] = row["cnt"]
+    by_format = {}
+    for row in conn.execute("SELECT format, COUNT(*) as cnt FROM books GROUP BY format"):
+        by_format[row["format"]] = row["cnt"]
+    conn.close()
+    return {
+        "total": total,
+        "by_status": by_status,
+        "by_format": by_format,
+    }
+
+
+def get_all_tags():
+    """获取所有不重复的标签列表。"""
+    conn = _connect()
+    rows = conn.execute("SELECT tags FROM books WHERE tags != ''").fetchall()
+    conn.close()
+    tag_set = set()
+    for row in rows:
+        for t in row["tags"].split(","):
+            t = t.strip()
+            if t:
+                tag_set.add(t)
+    return sorted(tag_set)