154 lines
4.8 KiB
Python
154 lines
4.8 KiB
Python
"""数据库操作层 — 封装所有 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)
|