mirror of
https://github.com/KingHWJ/WeReadBoard.git
synced 2026-11-03 18:24:58 +08:00
admin
4c916eceb7
1. Fix hdCover() to also replace s_ prefix (search results use s_ not t6_) 2. Add persistent search bar below navbar, switches to discover tab on search 3. Shelf search now supports book title and author search (new keyword param) 4. Add category tag pills above book grid, click to filter by category 5. Update /api/shelf backend: keyword param for title/author, category for exact match 6. Mobile responsive styles for search bar and category tags
1169 lines
40 KiB
Python
1169 lines
40 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
微信读书数据展示网站 - Flask 后端服务
|
||
=====================================
|
||
功能说明：
|
||
1. 提供读书数据的 RESTful API 接口，数据源为本地 MySQL 数据库
|
||
2. 按需调用微信读书 API 获取书籍详情、笔记详情、推荐书籍、搜索结果，并入库缓存
|
||
3. 提供 /api/health 健康检查接口，前端据此判断数据库连接状态
|
||
|
||
技术栈：
|
||
- Flask：轻量级 Web 框架
|
||
- PyMySQL：MySQL 数据库驱动
|
||
- 微信读书 Agent Gateway API：按需数据获取
|
||
|
||
作者：WorkBuddy
|
||
创建时间：2026-07-17
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import time
|
||
import urllib.request
|
||
from datetime import datetime
|
||
from functools import wraps
|
||
|
||
import pymysql
|
||
from flask import Flask, jsonify, render_template, request
|
||
|
||
# ======================================================================
|
||
# Flask 应用实例
|
||
# ======================================================================
|
||
app = Flask(__name__, static_folder="static", template_folder="templates")
|
||
|
||
# ======================================================================
|
||
# 数据库连接配置
|
||
# ======================================================================
|
||
MYSQL_CONFIG = {
|
||
"host": "127.0.0.1", # MySQL 服务器地址
|
||
"port": 3306, # MySQL 端口
|
||
"user": "root", # 数据库用户名
|
||
"password": "258963", # 数据库密码
|
||
"database": "weread_dashboard", # 数据库名称
|
||
"charset": "utf8mb4", # 字符集（支持完整 Unicode 含 emoji）
|
||
"cursorclass": pymysql.cursors.DictCursor, # 使用字典游标，方便按字段名取值
|
||
}
|
||
|
||
# ======================================================================
|
||
# 微信读书 API 配置
|
||
# ======================================================================
|
||
WEREAD_API_URL = "https://i.weread.qq.com/api/agent/gateway"
|
||
WEREAD_API_KEY = os.environ.get("WEREAD_API_KEY", "")
|
||
WEREAD_SKILL_VERSION = "1.0.4"
|
||
|
||
|
||
# ======================================================================
|
||
# 数据库连接与健康检查
|
||
# ======================================================================
|
||
|
||
def get_db():
|
||
"""
|
||
获取 MySQL 数据库连接。
|
||
每次请求创建独立连接，请求结束后由调用方关闭。
|
||
使用 DictCursor，查询结果以字典形式返回。
|
||
|
||
Returns:
|
||
pymysql.Connection: MySQL 数据库连接对象
|
||
|
||
Raises:
|
||
pymysql.MySQLError: 当 MySQL 无法连接时抛出异常
|
||
"""
|
||
conn = pymysql.connect(**MYSQL_CONFIG)
|
||
return conn
|
||
|
||
|
||
def check_db_health():
|
||
"""
|
||
检查 MySQL 数据库连接是否正常。
|
||
执行一条简单的 SELECT 1 查询，成功则表示连接正常。
|
||
|
||
Returns:
|
||
tuple: (is_healthy: bool, error_message: str)
|
||
"""
|
||
try:
|
||
conn = pymysql.connect(**MYSQL_CONFIG)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT 1")
|
||
cursor.close()
|
||
conn.close()
|
||
return True, "MySQL 连接正常"
|
||
except pymysql.MySQLError as e:
|
||
return False, f"MySQL 连接失败: {e}"
|
||
except Exception as e:
|
||
return False, f"数据库连接异常: {e}"
|
||
|
||
|
||
def db_required(f):
|
||
"""
|
||
装饰器：自动处理数据库连接。
|
||
- 为被装饰的函数注入 conn（数据库连接）参数
|
||
- 如果 MySQL 连接失败，返回 503 错误和告警信息
|
||
- 请求结束后自动关闭连接
|
||
"""
|
||
@wraps(f)
|
||
def wrapper(*args, **kwargs):
|
||
try:
|
||
conn = get_db()
|
||
except pymysql.MySQLError as e:
|
||
# MySQL 连接失败，返回 503 Service Unavailable
|
||
return jsonify({
|
||
"error": "数据库连接失败",
|
||
"detail": str(e),
|
||
"alert": "MySQL 数据库无法连接，请检查 MySQL 服务是否已启动。"
|
||
}), 503
|
||
except Exception as e:
|
||
return jsonify({
|
||
"error": "数据库连接异常",
|
||
"detail": str(e),
|
||
"alert": "数据库连接异常，请联系管理员。"
|
||
}), 503
|
||
|
||
try:
|
||
result = f(conn=conn, *args, **kwargs)
|
||
return result
|
||
finally:
|
||
conn.close()
|
||
|
||
return wrapper
|
||
|
||
|
||
# ======================================================================
|
||
# 微信读书 API 调用
|
||
# ======================================================================
|
||
|
||
def weread_api_call(api_name, **kwargs):
|
||
"""
|
||
调用微信读书 Agent Gateway API。
|
||
用于按需获取书籍详情、笔记详情、推荐书籍、搜索结果等数据。
|
||
|
||
Args:
|
||
api_name (str): API 接口名称，如 "/book/info"
|
||
**kwargs: API 参数，将作为 JSON body 发送
|
||
|
||
Returns:
|
||
dict: API 返回的 JSON 数据
|
||
|
||
Raises:
|
||
Exception: API 调用失败时抛出异常
|
||
"""
|
||
body = {"api_name": api_name, "skill_version": WEREAD_SKILL_VERSION}
|
||
body.update(kwargs)
|
||
req = urllib.request.Request(
|
||
WEREAD_API_URL,
|
||
data=json.dumps(body).encode("utf-8"),
|
||
headers={
|
||
"Authorization": f"Bearer {WEREAD_API_KEY}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
method="POST",
|
||
)
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
|
||
|
||
# ======================================================================
|
||
# 工具函数
|
||
# ======================================================================
|
||
|
||
def ts_to_date(ts):
|
||
"""
|
||
Unix 时间戳转日期字符串。
|
||
Args:
|
||
ts (int): Unix 时间戳（秒）
|
||
Returns:
|
||
str: 格式化日期 "YYYY-MM-DD"，空值返回空字符串
|
||
"""
|
||
if not ts or ts == 0:
|
||
return ""
|
||
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d")
|
||
|
||
|
||
def ts_to_year(ts):
|
||
"""
|
||
Unix 时间戳转年份字符串。
|
||
Args:
|
||
ts (int): Unix 时间戳（秒）
|
||
Returns:
|
||
str: 年份 "YYYY"，空值返回空字符串
|
||
"""
|
||
if not ts or ts == 0:
|
||
return ""
|
||
return datetime.fromtimestamp(ts).strftime("%Y")
|
||
|
||
|
||
def format_duration(seconds):
|
||
"""
|
||
秒数转可读时长字符串。
|
||
Args:
|
||
seconds (int): 时长（秒）
|
||
Returns:
|
||
str: 如 "5小时30分钟"、"20分钟"
|
||
"""
|
||
if not seconds:
|
||
return "0分钟"
|
||
hours = seconds // 3600
|
||
minutes = (seconds % 3600) // 60
|
||
if hours > 0:
|
||
return f"{hours}小时{minutes}分钟"
|
||
return f"{minutes}分钟"
|
||
|
||
|
||
def ts_to_dt_str(ts):
|
||
"""
|
||
Unix 时间戳转完整日期时间字符串 (中国时区 UTC+8)。
|
||
用于写入数据库的 _dt 列。
|
||
|
||
Args:
|
||
ts (int): Unix 时间戳（秒）
|
||
Returns:
|
||
str: 格式化日期时间 "YYYY-MM-DD HH:MM:SS"，空值返回 None
|
||
"""
|
||
if not ts or ts == 0:
|
||
return None
|
||
from datetime import timezone, timedelta
|
||
cst = timezone(timedelta(hours=8))
|
||
return datetime.fromtimestamp(ts, tz=cst).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
# ======================================================================
|
||
# 健康检查接口
|
||
# ======================================================================
|
||
|
||
@app.route("/api/health")
|
||
def api_health():
|
||
"""
|
||
健康检查接口。
|
||
前端页面加载时调用此接口，判断 MySQL 数据库是否正常连接。
|
||
如果连接失败，前端会显示告警横幅。
|
||
|
||
Returns:
|
||
JSON: {"status": "ok"/"error", "database": "connected"/"disconnected", "message": "..."}
|
||
"""
|
||
is_healthy, message = check_db_health()
|
||
if is_healthy:
|
||
return jsonify({
|
||
"status": "ok",
|
||
"database": "connected",
|
||
"message": message
|
||
})
|
||
else:
|
||
return jsonify({
|
||
"status": "error",
|
||
"database": "disconnected",
|
||
"message": message
|
||
}), 503
|
||
|
||
|
||
# ======================================================================
|
||
# 页面路由
|
||
# ======================================================================
|
||
|
||
@app.route("/")
|
||
def index():
|
||
"""首页：渲染读书数据仪表盘页面"""
|
||
return render_template("index.html")
|
||
|
||
|
||
# ======================================================================
|
||
# API 接口 - 基础数据查询
|
||
# ======================================================================
|
||
|
||
@app.route("/api/overview")
|
||
@db_required
|
||
def api_overview(conn):
|
||
"""
|
||
总览数据接口。
|
||
返回用户的核心阅读指标：累计阅读时长、阅读天数、藏书量、笔记数等。
|
||
|
||
Returns:
|
||
JSON: 包含以下字段
|
||
- total_read_time: 总阅读时长（秒）
|
||
- total_read_time_str: 总阅读时长（格式化字符串）
|
||
- read_days: 阅读天数
|
||
- year_read_time: 今年阅读时长（秒）
|
||
- year_read_time_str: 今年阅读时长（格式化字符串）
|
||
- year_read_days: 今年阅读天数
|
||
- shelf_count: 书架书籍数
|
||
- finished_count: 读完书籍数
|
||
- notebook_count: 有笔记的书籍数
|
||
- total_notes: 笔记总数（划线+想法+书签）
|
||
- regist_time: 注册时间
|
||
- last_read_time: 最近阅读时间
|
||
- fetch_time: 数据采集时间
|
||
"""
|
||
cursor = conn.cursor()
|
||
|
||
# 查询总体阅读统计
|
||
cursor.execute("SELECT * FROM reading_stats WHERE mode='overall' LIMIT 1")
|
||
overall = cursor.fetchone()
|
||
total_read_time = overall["total_read_time"] if overall else 0
|
||
read_days = overall["read_days"] if overall else 0
|
||
regist_time = overall["regist_time"] if overall else 0
|
||
|
||
# 查询今年阅读统计
|
||
cursor.execute("SELECT * FROM reading_stats WHERE mode='annually' LIMIT 1")
|
||
annual = cursor.fetchone()
|
||
year_read_time = annual["total_read_time"] if annual else 0
|
||
year_read_days = annual["read_days"] if annual else 0
|
||
|
||
# 查询书架书籍总数
|
||
cursor.execute("SELECT COUNT(*) AS cnt FROM books")
|
||
shelf_count = cursor.fetchone()["cnt"]
|
||
|
||
# 查询已读完书籍数
|
||
cursor.execute("SELECT COUNT(*) AS cnt FROM books WHERE finish_reading=1")
|
||
finished_count = cursor.fetchone()["cnt"]
|
||
|
||
# 查询笔记本统计
|
||
cursor.execute("SELECT COUNT(*) AS cnt FROM notebooks")
|
||
notebook_count = cursor.fetchone()["cnt"]
|
||
cursor.execute("SELECT SUM(review_count + note_count + bookmark_count) AS cnt FROM notebooks")
|
||
total_notes = cursor.fetchone()["cnt"] or 0
|
||
|
||
# 查询最近阅读时间
|
||
cursor.execute("SELECT MAX(read_update_time) AS max_ts FROM books")
|
||
last_read_ts = cursor.fetchone()["max_ts"] or 0
|
||
|
||
# 查询数据采集时间
|
||
cursor.execute("SELECT value FROM fetch_meta WHERE `key`='last_fetch_time'")
|
||
fetch_row = cursor.fetchone()
|
||
fetch_time = fetch_row["value"] if fetch_row else ""
|
||
|
||
return jsonify({
|
||
"total_read_time": total_read_time,
|
||
"total_read_time_str": format_duration(total_read_time),
|
||
"read_days": read_days,
|
||
"year_read_time": year_read_time,
|
||
"year_read_time_str": format_duration(year_read_time),
|
||
"year_read_days": year_read_days,
|
||
"shelf_count": shelf_count,
|
||
"finished_count": finished_count,
|
||
"notebook_count": notebook_count,
|
||
"total_notes": total_notes,
|
||
"regist_time": ts_to_date(regist_time),
|
||
"last_read_time": ts_to_date(last_read_ts),
|
||
"fetch_time": fetch_time,
|
||
})
|
||
|
||
|
||
@app.route("/api/yearly-trend")
|
||
@db_required
|
||
def api_yearly_trend(conn):
|
||
"""
|
||
年度阅读时长时间序列接口。
|
||
返回各年度的阅读时长数据，用于前端柱状图展示。
|
||
|
||
Returns:
|
||
JSON数组: [{year: "2024", hours: 45.2, seconds: 162720}, ...]
|
||
"""
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
"SELECT bucket_timestamp, duration FROM read_times "
|
||
"WHERE mode='overall' ORDER BY bucket_timestamp"
|
||
)
|
||
rows = cursor.fetchall()
|
||
|
||
data = []
|
||
for row in rows:
|
||
year = ts_to_year(row["bucket_timestamp"])
|
||
hours = round(row["duration"] / 3600, 1)
|
||
data.append({"year": year, "hours": hours, "seconds": row["duration"]})
|
||
|
||
return jsonify(data)
|
||
|
||
|
||
@app.route("/api/top-books")
|
||
@db_required
|
||
def api_top_books(conn):
|
||
"""
|
||
阅读时长最久的书籍排行接口。
|
||
返回阅读时长排名前 20 的书籍信息，用于前端卡片展示。
|
||
|
||
Returns:
|
||
JSON数组: [{book_id, title, author, cover, read_time, read_time_str, tags}, ...]
|
||
"""
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
"SELECT * FROM read_longest WHERE mode='overall' "
|
||
"ORDER BY read_time DESC LIMIT 20"
|
||
)
|
||
rows = cursor.fetchall()
|
||
|
||
books = []
|
||
for row in rows:
|
||
tags = json.loads(row["tags"]) if row["tags"] else []
|
||
books.append({
|
||
"book_id": row["book_id"],
|
||
"title": row["title"],
|
||
"author": row["author"],
|
||
"cover": row["cover"],
|
||
"read_time": row["read_time"],
|
||
"read_time_str": format_duration(row["read_time"]),
|
||
"tags": tags,
|
||
})
|
||
|
||
return jsonify(books)
|
||
|
||
|
||
@app.route("/api/categories")
|
||
@db_required
|
||
def api_categories(conn):
|
||
"""
|
||
偏好分类接口。
|
||
返回用户阅读偏好分类数据，按阅读时长降序排列，用于前端饼图展示。
|
||
|
||
Returns:
|
||
JSON数组: [{category_title, parent_category_title, reading_time, reading_time_str, reading_count, val}, ...]
|
||
"""
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
"SELECT * FROM prefer_categories WHERE mode='overall' "
|
||
"AND reading_time > 0 ORDER BY reading_time DESC"
|
||
)
|
||
rows = cursor.fetchall()
|
||
|
||
cats = []
|
||
for row in rows:
|
||
cats.append({
|
||
"category_title": row["category_title"],
|
||
"parent_category_title": row["parent_category_title"],
|
||
"reading_time": row["reading_time"],
|
||
"reading_time_str": format_duration(row["reading_time"]),
|
||
"reading_count": row["reading_count"],
|
||
"val": row["val"],
|
||
})
|
||
|
||
return jsonify(cats)
|
||
|
||
|
||
@app.route("/api/authors")
|
||
@db_required
|
||
def api_authors(conn):
|
||
"""
|
||
偏好作者接口。
|
||
返回用户阅读最多的作者排行（前 20），用于前端横向条形图展示。
|
||
|
||
Returns:
|
||
JSON数组: [{name, count, read_time}, ...]
|
||
"""
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
"SELECT * FROM prefer_authors WHERE mode='overall' "
|
||
"ORDER BY count DESC LIMIT 20"
|
||
)
|
||
rows = cursor.fetchall()
|
||
|
||
authors = []
|
||
for row in rows:
|
||
authors.append({
|
||
"name": row["name"],
|
||
"count": row["count"],
|
||
"read_time": row["read_time"],
|
||
})
|
||
|
||
return jsonify(authors)
|
||
|
||
|
||
@app.route("/api/time-distribution")
|
||
@db_required
|
||
def api_time_distribution(conn):
|
||
"""
|
||
24 小时阅读时段分布接口。
|
||
返回一天中每个小时的阅读时长，用于前端折线图展示。
|
||
|
||
Returns:
|
||
JSON数组: [{hour: 0, duration: 3600, duration_min: 60.0}, ...]
|
||
"""
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
"SELECT hour_index, SUM(duration) AS total "
|
||
"FROM prefer_time WHERE mode='overall' "
|
||
"GROUP BY hour_index ORDER BY hour_index"
|
||
)
|
||
rows = cursor.fetchall()
|
||
|
||
dist = []
|
||
for row in rows:
|
||
dist.append({
|
||
"hour": row["hour_index"],
|
||
"duration": row["total"],
|
||
"duration_min": round(row["total"] / 60, 1),
|
||
})
|
||
|
||
return jsonify(dist)
|
||
|
||
|
||
@app.route("/api/shelf")
|
||
@db_required
|
||
def api_shelf(conn):
|
||
"""
|
||
书架列表接口（分页）。
|
||
返回书架中的书籍列表，支持按书名/作者搜索、按分类筛选和只看已读完。
|
||
|
||
Query Parameters:
|
||
page (int): 页码，默认 1
|
||
per_page (int): 每页数量，默认 60
|
||
keyword (str): 书名或作者搜索（模糊匹配）
|
||
category (str): 分类筛选（精确匹配分类名）
|
||
finished (str): "1" 表示只看已读完
|
||
|
||
Returns:
|
||
JSON: {books: [...], total, page, per_page, total_pages, categories: [...]}
|
||
"""
|
||
page = int(request.args.get("page", 1))
|
||
per_page = int(request.args.get("per_page", 60))
|
||
keyword = request.args.get("keyword", "").strip()
|
||
category = request.args.get("category", "").strip()
|
||
finished_only = request.args.get("finished", "0") == "1"
|
||
|
||
offset = (page - 1) * per_page
|
||
cursor = conn.cursor()
|
||
|
||
# 构建查询条件
|
||
where_clauses = []
|
||
params = []
|
||
if keyword:
|
||
where_clauses.append("(title LIKE %s OR author LIKE %s)")
|
||
params.extend([f"%{keyword}%", f"%{keyword}%"])
|
||
if category:
|
||
where_clauses.append("category = %s")
|
||
params.append(category)
|
||
if finished_only:
|
||
where_clauses.append("finish_reading=1")
|
||
|
||
where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
|
||
|
||
# 查询总数
|
||
cursor.execute(f"SELECT COUNT(*) AS cnt FROM books{where_sql}", params)
|
||
total = cursor.fetchone()["cnt"]
|
||
|
||
# 查询当前页数据（置顶优先，最近阅读时间倒序）
|
||
cursor.execute(
|
||
f"SELECT * FROM books{where_sql} ORDER BY is_top DESC, read_update_time DESC LIMIT %s OFFSET %s",
|
||
params + [per_page, offset],
|
||
)
|
||
rows = cursor.fetchall()
|
||
|
||
# 获取所有分类列表（用于前端筛选下拉框）
|
||
cursor.execute("SELECT DISTINCT category FROM books WHERE category != '' ORDER BY category")
|
||
categories = [r["category"] for r in cursor.fetchall()]
|
||
|
||
books = []
|
||
for row in rows:
|
||
books.append({
|
||
"book_id": row["book_id"],
|
||
"title": row["title"],
|
||
"author": row["author"],
|
||
"cover": row["cover"],
|
||
"category": row["category"],
|
||
"finish_reading": row["finish_reading"],
|
||
"read_update_time": ts_to_date(row["read_update_time"]),
|
||
"secret": row["secret"],
|
||
"is_top": row["is_top"],
|
||
"deep_link": row["deep_link"],
|
||
})
|
||
|
||
return jsonify({
|
||
"books": books,
|
||
"total": total,
|
||
"page": page,
|
||
"per_page": per_page,
|
||
"total_pages": (total + per_page - 1) // per_page,
|
||
"categories": categories,
|
||
})
|
||
|
||
|
||
@app.route("/api/notebooks")
|
||
@db_required
|
||
def api_notebooks(conn):
|
||
"""
|
||
笔记本概览接口。
|
||
返回所有有笔记的书籍列表，按笔记总数降序排列。
|
||
|
||
Returns:
|
||
JSON数组: [{book_id, title, author, cover, review_count, note_count, bookmark_count, total_notes, reading_progress, marked_status}, ...]
|
||
"""
|
||
cursor = conn.cursor()
|
||
cursor.execute(
|
||
"SELECT * FROM notebooks "
|
||
"ORDER BY (review_count + note_count + bookmark_count) DESC"
|
||
)
|
||
rows = cursor.fetchall()
|
||
|
||
notebooks = []
|
||
for row in rows:
|
||
total = row["review_count"] + row["note_count"] + row["bookmark_count"]
|
||
notebooks.append({
|
||
"book_id": row["book_id"],
|
||
"title": row["title"],
|
||
"author": row["author"],
|
||
"cover": row["cover"],
|
||
"review_count": row["review_count"],
|
||
"note_count": row["note_count"],
|
||
"bookmark_count": row["bookmark_count"],
|
||
"total_notes": total,
|
||
"reading_progress": row["reading_progress"],
|
||
"marked_status": row["marked_status"],
|
||
})
|
||
|
||
return jsonify(notebooks)
|
||
|
||
|
||
@app.route("/api/read-stats")
|
||
@db_required
|
||
def api_read_stats(conn):
|
||
"""
|
||
阅读统计摘要项接口。
|
||
返回微信读书 App 中展示的阅读统计摘要（读过/读完/阅读/笔记）。
|
||
|
||
Returns:
|
||
JSON数组: [{stat, counts, scheme}, ...]
|
||
"""
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT * FROM read_stat_items WHERE mode='overall'")
|
||
rows = cursor.fetchall()
|
||
|
||
stats = []
|
||
for row in rows:
|
||
stats.append({
|
||
"stat": row["stat_name"],
|
||
"counts": row["counts"],
|
||
"scheme": row["scheme"],
|
||
})
|
||
|
||
return jsonify(stats)
|
||
|
||
|
||
@app.route("/api/archives")
|
||
@db_required
|
||
def api_archives(conn):
|
||
"""
|
||
书单/归档接口。
|
||
返回用户创建的书单列表。
|
||
|
||
Returns:
|
||
JSON数组: [{name, book_count, book_ids: [...]}, ...]
|
||
"""
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT * FROM archives")
|
||
rows = cursor.fetchall()
|
||
|
||
archives = []
|
||
for row in rows:
|
||
book_ids = json.loads(row["book_ids"]) if row["book_ids"] else []
|
||
archives.append({
|
||
"name": row["name"],
|
||
"book_count": len(book_ids),
|
||
"book_ids": book_ids,
|
||
})
|
||
|
||
return jsonify(archives)
|
||
|
||
|
||
# ======================================================================
|
||
# API 接口 - 按需获取：书籍详情
|
||
# ======================================================================
|
||
|
||
@app.route("/api/book/detail")
|
||
@db_required
|
||
def api_book_detail(conn):
|
||
"""
|
||
书籍详情接口。
|
||
实时调用微信读书 /book/info 和 /book/getprogress API 获取书籍详情和阅读进度，
|
||
入库缓存后返回给前端。API 调用失败时回退到数据库缓存。
|
||
|
||
Query Parameters:
|
||
book_id (str): 书籍 ID（必填）
|
||
|
||
Returns:
|
||
JSON: 书籍详细信息 + 阅读进度
|
||
"""
|
||
book_id = request.args.get("book_id", "")
|
||
if not book_id:
|
||
return jsonify({"error": "book_id is required"}), 400
|
||
|
||
cursor = conn.cursor()
|
||
|
||
# 1. 获取书籍基本信息（/book/info）
|
||
try:
|
||
info = weread_api_call("/book/info", bookId=book_id)
|
||
rating_detail = info.get("newRatingDetail")
|
||
cursor.execute(
|
||
"REPLACE INTO book_details "
|
||
"(book_id, title, author, translator, cover, intro, category, "
|
||
"publisher, publish_time, isbn, word_count, new_rating, new_rating_count, "
|
||
"new_rating_detail, deep_link) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(
|
||
book_id,
|
||
info.get("title", ""),
|
||
info.get("author", ""),
|
||
info.get("translator", ""),
|
||
info.get("cover", ""),
|
||
info.get("intro", ""),
|
||
info.get("category", ""),
|
||
info.get("publisher", ""),
|
||
info.get("publishTime", ""),
|
||
info.get("isbn", ""),
|
||
info.get("wordCount", 0),
|
||
info.get("newRating", 0),
|
||
info.get("newRatingCount", 0),
|
||
json.dumps(rating_detail, ensure_ascii=False) if rating_detail else None,
|
||
info.get("deepLink", ""),
|
||
),
|
||
)
|
||
conn.commit()
|
||
except Exception as e:
|
||
# API 调用失败，尝试从数据库读取缓存
|
||
cursor.execute("SELECT * FROM book_details WHERE book_id=%s", (book_id,))
|
||
cached = cursor.fetchone()
|
||
if not cached:
|
||
return jsonify({"error": f"获取书籍信息失败: {str(e)}"}), 500
|
||
|
||
# 2. 获取阅读进度
|
||
progress_data = {}
|
||
try:
|
||
prog = weread_api_call("/book/getprogress", bookId=book_id)
|
||
book_prog = prog.get("book", {})
|
||
srv_ts = prog.get("timestamp")
|
||
ut = book_prog.get("updateTime", 0)
|
||
ft = book_prog.get("finishTime", 0)
|
||
cursor.execute(
|
||
"REPLACE INTO book_progress "
|
||
"(book_id, progress, chapter_uid, chapter_offset, update_time, update_time_dt, "
|
||
"record_reading_time, finish_time, finish_time_dt, is_start_reading, "
|
||
"server_timestamp, server_timestamp_dt) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(
|
||
book_id,
|
||
book_prog.get("progress", 0),
|
||
book_prog.get("chapterUid", 0),
|
||
book_prog.get("chapterOffset", 0),
|
||
ut, ts_to_dt_str(ut),
|
||
book_prog.get("recordReadingTime", 0),
|
||
ft, ts_to_dt_str(ft),
|
||
book_prog.get("isStartReading", 0),
|
||
srv_ts, ts_to_dt_str(srv_ts),
|
||
),
|
||
)
|
||
conn.commit()
|
||
progress_data = {
|
||
"progress": book_prog.get("progress", 0),
|
||
"record_reading_time": book_prog.get("recordReadingTime", 0),
|
||
"record_reading_time_str": format_duration(book_prog.get("recordReadingTime", 0)),
|
||
"update_time": ts_to_date(book_prog.get("updateTime", 0)),
|
||
"finish_time": ts_to_date(book_prog.get("finishTime", 0)),
|
||
"is_start_reading": book_prog.get("isStartReading", 0),
|
||
}
|
||
except Exception:
|
||
# API 调用失败，从数据库读取缓存的进度数据
|
||
cursor.execute("SELECT * FROM book_progress WHERE book_id=%s", (book_id,))
|
||
cached = cursor.fetchone()
|
||
if cached:
|
||
progress_data = {
|
||
"progress": cached["progress"],
|
||
"record_reading_time": cached["record_reading_time"],
|
||
"record_reading_time_str": format_duration(cached["record_reading_time"]),
|
||
"update_time": ts_to_date(cached["update_time"]),
|
||
"finish_time": ts_to_date(cached["finish_time"]),
|
||
"is_start_reading": cached["is_start_reading"],
|
||
}
|
||
|
||
# 3. 从数据库读取最终书籍详情并组装返回
|
||
cursor.execute("SELECT * FROM book_details WHERE book_id=%s", (book_id,))
|
||
detail = cursor.fetchone()
|
||
|
||
if not detail:
|
||
return jsonify({"error": "数据获取失败"}), 500
|
||
|
||
return jsonify({
|
||
"book_id": book_id,
|
||
"title": detail["title"],
|
||
"author": detail["author"],
|
||
"translator": detail["translator"],
|
||
"cover": detail["cover"],
|
||
"intro": detail["intro"],
|
||
"category": detail["category"],
|
||
"publisher": detail["publisher"],
|
||
"publish_time": detail["publish_time"],
|
||
"isbn": detail["isbn"],
|
||
"word_count": detail["word_count"],
|
||
"new_rating": float(detail["new_rating"]) if detail["new_rating"] else 0,
|
||
"new_rating_count": detail["new_rating_count"],
|
||
"progress": progress_data,
|
||
"deep_link": detail["deep_link"] or f"weread://reading?bId={book_id}",
|
||
})
|
||
|
||
|
||
# ======================================================================
|
||
# API 接口 - 按需获取：笔记详情
|
||
# ======================================================================
|
||
|
||
@app.route("/api/book/notes")
|
||
@db_required
|
||
def api_book_notes(conn):
|
||
"""
|
||
笔记详情接口。
|
||
实时调用微信读书 /book/bookmarklist 和 /review/list/mine API，
|
||
获取划线和想法/点评数据，入库缓存后返回。
|
||
划线按章节分组展示。API 调用失败时回退到数据库缓存。
|
||
|
||
Query Parameters:
|
||
book_id (str): 书籍 ID（必填）
|
||
|
||
Returns:
|
||
JSON: {book_id, highlights: [...], highlight_count, grouped_highlights: {...}, reviews: [...], review_count}
|
||
"""
|
||
book_id = request.args.get("book_id", "")
|
||
if not book_id:
|
||
return jsonify({"error": "book_id is required"}), 400
|
||
|
||
cursor = conn.cursor()
|
||
|
||
# 1. 获取划线数据（/book/bookmarklist）
|
||
highlights = []
|
||
chapters = {}
|
||
try:
|
||
data = weread_api_call("/book/bookmarklist", bookId=book_id)
|
||
|
||
# 保存章节信息（包含完整字段）
|
||
for ch in data.get("chapters", []):
|
||
cursor.execute(
|
||
"REPLACE INTO book_chapters "
|
||
"(book_id, chapter_uid, chapter_idx, title) "
|
||
"VALUES (%s,%s,%s,%s)",
|
||
(book_id, ch.get("chapterUid", 0), ch.get("chapterIdx", 0), ch.get("title", "")),
|
||
)
|
||
chapters[ch.get("chapterUid", 0)] = ch.get("title", "")
|
||
|
||
# 保存划线数据（先删除旧数据再插入）
|
||
cursor.execute("DELETE FROM book_highlights WHERE book_id=%s", (book_id,))
|
||
for bm in data.get("updated", []):
|
||
ct = bm.get("createTime", 0)
|
||
cursor.execute(
|
||
"INSERT INTO book_highlights "
|
||
"(book_id, bookmark_id, chapter_uid, mark_text, create_time, create_time_dt, "
|
||
"type, range_text, color_style) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(
|
||
book_id,
|
||
bm.get("bookmarkId", ""),
|
||
bm.get("chapterUid", 0),
|
||
bm.get("markText", ""),
|
||
ct, ts_to_dt_str(ct),
|
||
bm.get("type", 0),
|
||
bm.get("range", ""),
|
||
bm.get("colorStyle", 0),
|
||
),
|
||
)
|
||
highlights.append({
|
||
"bookmark_id": bm.get("bookmarkId", ""),
|
||
"chapter_uid": bm.get("chapterUid", 0),
|
||
"chapter_title": chapters.get(bm.get("chapterUid", 0), ""),
|
||
"mark_text": bm.get("markText", ""),
|
||
"create_time": ts_to_date(bm.get("createTime", 0)),
|
||
"range": bm.get("range", ""),
|
||
})
|
||
conn.commit()
|
||
except Exception:
|
||
# API 调用失败，从数据库读取缓存的划线数据
|
||
cursor.execute(
|
||
"SELECT * FROM book_highlights WHERE book_id=%s ORDER BY create_time", (book_id,)
|
||
)
|
||
for row in cursor.fetchall():
|
||
cursor.execute(
|
||
"SELECT title FROM book_chapters WHERE book_id=%s AND chapter_uid=%s",
|
||
(book_id, row["chapter_uid"])
|
||
)
|
||
ch_row = cursor.fetchone()
|
||
highlights.append({
|
||
"bookmark_id": row["bookmark_id"],
|
||
"chapter_uid": row["chapter_uid"],
|
||
"chapter_title": ch_row["title"] if ch_row else "",
|
||
"mark_text": row["mark_text"],
|
||
"create_time": ts_to_date(row["create_time"]),
|
||
"range": row["range_text"],
|
||
})
|
||
|
||
# 2. 获取想法/点评数据
|
||
reviews = []
|
||
try:
|
||
rev_data = weread_api_call("/review/list/mine", bookid=book_id, count=100)
|
||
cursor.execute("DELETE FROM book_reviews WHERE book_id=%s", (book_id,))
|
||
for item in rev_data.get("reviews", []):
|
||
rev = item.get("review", {})
|
||
ct = rev.get("createTime", 0)
|
||
cursor.execute(
|
||
"INSERT INTO book_reviews "
|
||
"(book_id, review_id, content, abstract, range_text, chapter_uid, chapter_idx, "
|
||
"chapter_name, create_time, create_time_dt, star, is_finish) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(
|
||
book_id,
|
||
rev.get("reviewId", ""),
|
||
rev.get("content", ""),
|
||
rev.get("abstract", ""),
|
||
rev.get("range", ""),
|
||
rev.get("chapterUid"),
|
||
rev.get("chapterIdx"),
|
||
rev.get("chapterName", ""),
|
||
ct, ts_to_dt_str(ct),
|
||
rev.get("star", -1),
|
||
rev.get("isFinish", 0),
|
||
),
|
||
)
|
||
reviews.append({
|
||
"review_id": rev.get("reviewId", ""),
|
||
"content": rev.get("content", ""),
|
||
"abstract": rev.get("abstract", ""),
|
||
"create_time": ts_to_date(rev.get("createTime", 0)),
|
||
"star": rev.get("star", -1),
|
||
"chapter_name": rev.get("chapterName", ""),
|
||
"is_finish": rev.get("isFinish", 0),
|
||
})
|
||
conn.commit()
|
||
except Exception:
|
||
# API 调用失败，从数据库读取缓存的想法数据
|
||
cursor.execute(
|
||
"SELECT * FROM book_reviews WHERE book_id=%s ORDER BY create_time DESC", (book_id,)
|
||
)
|
||
for row in cursor.fetchall():
|
||
reviews.append({
|
||
"review_id": row["review_id"],
|
||
"content": row["content"],
|
||
"create_time": ts_to_date(row["create_time"]),
|
||
"star": row["star"],
|
||
"chapter_name": row["chapter_name"],
|
||
"is_finish": row["is_finish"],
|
||
})
|
||
|
||
# 3. 按章节分组划线
|
||
grouped_highlights = {}
|
||
for h in highlights:
|
||
ch_title = h["chapter_title"] or f"章节{h['chapter_uid']}"
|
||
if ch_title not in grouped_highlights:
|
||
grouped_highlights[ch_title] = []
|
||
grouped_highlights[ch_title].append(h)
|
||
|
||
return jsonify({
|
||
"book_id": book_id,
|
||
"highlights": highlights,
|
||
"highlight_count": len(highlights),
|
||
"grouped_highlights": grouped_highlights,
|
||
"reviews": reviews,
|
||
"review_count": len(reviews),
|
||
})
|
||
|
||
|
||
# ======================================================================
|
||
# API 接口 - 推荐书籍
|
||
# ======================================================================
|
||
|
||
@app.route("/api/recommend")
|
||
@db_required
|
||
def api_recommend(conn):
|
||
"""
|
||
个性化推荐接口。
|
||
调用微信读书 /book/recommend API 获取推荐书籍列表，入库缓存后返回。
|
||
支持翻页（通过 maxIdx 参数）。
|
||
|
||
Query Parameters:
|
||
maxIdx (int): 翻页偏移量，默认 0
|
||
|
||
Returns:
|
||
JSON: {books: [...], has_more, last_search_idx}
|
||
"""
|
||
max_idx = int(request.args.get("maxIdx", 0))
|
||
cursor = conn.cursor()
|
||
batch_id = datetime.now().strftime("%Y%m%d%H%M%S")
|
||
|
||
try:
|
||
data = weread_api_call("/book/recommend", count=12, maxIdx=max_idx)
|
||
books = data.get("books", [])
|
||
|
||
# 将推荐书籍入库（完整字段）
|
||
for book in books:
|
||
rating_detail = book.get("newRatingDetail", {})
|
||
cursor.execute(
|
||
"REPLACE INTO recommended_books "
|
||
"(book_id, title, author, cover, intro, category, reason, "
|
||
"reading_count, new_rating, new_rating_count, rating_title, "
|
||
"deep_link, search_idx, price, pay_type, book_type, batch_id) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(
|
||
book.get("bookId", ""),
|
||
book.get("title", ""),
|
||
book.get("author", ""),
|
||
book.get("cover", ""),
|
||
book.get("intro", ""),
|
||
book.get("category", ""),
|
||
book.get("reason", ""),
|
||
book.get("readingCount", 0),
|
||
book.get("newRating", 0),
|
||
book.get("newRatingCount", 0),
|
||
rating_detail.get("title", "") if isinstance(rating_detail, dict) else "",
|
||
book.get("deepLink", ""),
|
||
book.get("searchIdx", 0),
|
||
book.get("price"),
|
||
book.get("payType"),
|
||
book.get("type", 0),
|
||
batch_id,
|
||
),
|
||
)
|
||
conn.commit()
|
||
except Exception as e:
|
||
return jsonify({"error": f"获取推荐失败: {str(e)}"}), 500
|
||
|
||
# 组装返回数据
|
||
result = []
|
||
for book in books:
|
||
rating_detail = book.get("newRatingDetail", {})
|
||
result.append({
|
||
"book_id": book.get("bookId", ""),
|
||
"title": book.get("title", ""),
|
||
"author": book.get("author", ""),
|
||
"cover": book.get("cover", ""),
|
||
"intro": book.get("intro", ""),
|
||
"category": book.get("category", ""),
|
||
"reason": book.get("reason", ""),
|
||
"reading_count": book.get("readingCount", 0),
|
||
"new_rating": book.get("newRating", 0),
|
||
"new_rating_count": book.get("newRatingCount", 0),
|
||
"rating_title": rating_detail.get("title", "") if isinstance(rating_detail, dict) else "",
|
||
})
|
||
|
||
return jsonify({
|
||
"books": result,
|
||
"has_more": data.get("hasMore", 0),
|
||
"last_search_idx": books[-1].get("searchIdx", 0) if books else 0,
|
||
})
|
||
|
||
|
||
# ======================================================================
|
||
# API 接口 - 搜索书籍
|
||
# ======================================================================
|
||
|
||
@app.route("/api/search")
|
||
@db_required
|
||
def api_search(conn):
|
||
"""
|
||
搜索书籍接口。
|
||
调用微信读书 /store/search API 按关键词搜索书籍，入库缓存后返回。
|
||
注意：Flask URL 参数中文编码需特殊处理（latin-1 -> utf-8 转码）。
|
||
|
||
Query Parameters:
|
||
keyword (str): 搜索关键词（必填）
|
||
maxIdx (int): 翻页偏移量，默认 0
|
||
|
||
Returns:
|
||
JSON: {keyword, books: [...], has_more}
|
||
"""
|
||
keyword = request.args.get("keyword", "").strip()
|
||
# 修复 Flask URL 参数中文编码问题
|
||
try:
|
||
keyword = keyword.encode("latin-1").decode("utf-8")
|
||
except (UnicodeDecodeError, UnicodeEncodeError):
|
||
pass
|
||
if not keyword:
|
||
return jsonify({"error": "keyword is required"}), 400
|
||
|
||
max_idx = int(request.args.get("maxIdx", 0))
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
data = weread_api_call("/store/search", keyword=keyword, scope=10, maxIdx=max_idx)
|
||
sid = data.get("sid", "")
|
||
has_more = data.get("hasMore", 0)
|
||
|
||
# 创建搜索会话记录
|
||
cursor.execute(
|
||
"INSERT INTO search_sessions (sid, keyword, scope, has_more, max_idx) "
|
||
"VALUES (%s,%s,%s,%s,%s)",
|
||
(sid, keyword, 10, has_more, max_idx),
|
||
)
|
||
session_id = cursor.lastrowid
|
||
|
||
results = data.get("results", [])
|
||
books = []
|
||
for group in results:
|
||
result_title = group.get("title", "")
|
||
result_scope = group.get("scope")
|
||
scope_count = group.get("scopeCount")
|
||
current_count = group.get("currentCount")
|
||
for item in group.get("books", []):
|
||
bi = item.get("bookInfo", {})
|
||
rating_detail = item.get("newRatingDetail", {})
|
||
cursor.execute(
|
||
"INSERT INTO search_results "
|
||
"(session_id, keyword, book_id, title, author, cover, intro, publisher, "
|
||
"category, reading_count, new_rating, new_rating_count, rating_title, "
|
||
"deep_link, pay_type, price, soldout, search_idx, "
|
||
"result_scope, result_title, scope_count, current_count) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(
|
||
session_id,
|
||
keyword,
|
||
bi.get("bookId", ""),
|
||
bi.get("title", ""),
|
||
bi.get("author", ""),
|
||
bi.get("cover", ""),
|
||
bi.get("intro", ""),
|
||
bi.get("publisher", ""),
|
||
bi.get("category", ""),
|
||
item.get("readingCount", 0),
|
||
item.get("newRating", 0),
|
||
item.get("newRatingCount", 0),
|
||
rating_detail.get("title", "") if isinstance(rating_detail, dict) else "",
|
||
bi.get("deepLink", ""),
|
||
bi.get("payType"),
|
||
bi.get("price"),
|
||
bi.get("soldout", 0),
|
||
item.get("searchIdx", 0),
|
||
result_scope,
|
||
result_title,
|
||
scope_count,
|
||
current_count,
|
||
),
|
||
)
|
||
books.append({
|
||
"book_id": bi.get("bookId", ""),
|
||
"title": bi.get("title", ""),
|
||
"author": bi.get("author", ""),
|
||
"cover": bi.get("cover", ""),
|
||
"intro": bi.get("intro", ""),
|
||
"publisher": bi.get("publisher", ""),
|
||
"category": bi.get("category", ""),
|
||
"reading_count": item.get("readingCount", 0),
|
||
"new_rating": item.get("newRating", 0),
|
||
"new_rating_count": item.get("newRatingCount", 0),
|
||
"rating_title": rating_detail.get("title", "") if isinstance(rating_detail, dict) else "",
|
||
"deep_link": bi.get("deepLink", ""),
|
||
"soldout": bi.get("soldout", 0),
|
||
})
|
||
conn.commit()
|
||
except Exception as e:
|
||
return jsonify({"error": f"搜索失败: {str(e)}"}), 500
|
||
|
||
return jsonify({
|
||
"keyword": keyword,
|
||
"books": books,
|
||
"has_more": data.get("hasMore", 0),
|
||
})
|
||
|
||
|
||
# ======================================================================
|
||
# 启动入口
|
||
# ======================================================================
|
||
|
||
if __name__ == "__main__":
|
||
# 启动前检查 MySQL 连接
|
||
is_healthy, msg = check_db_health()
|
||
if not is_healthy:
|
||
print(f"[警告] {msg}")
|
||
print("[提示] 服务仍会启动，但前端将显示数据库连接告警")
|
||
else:
|
||
print(f"[信息] {msg}")
|
||
|
||
# 启动 Flask 服务，监听 8080 端口
|
||
app.run(host="0.0.0.0", port=8080, debug=False)
|