web/WeReadBoard
1
0 Fork 0
mirror of https://github.com/KingHWJ/WeReadBoard.git synced 2026-11-03 18:24:58 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
7bc02075bd639c77d712b34431824ea8e0e47bd9
WeReadBoard/app.py
T
admin 7bc02075bd 刷新按钮支持数据同步：调用 /api/sync 拉取最新数据 
- app.py 新增 /api/sync POST 接口：运行 fetch_data.py + migrate_to_mysql.py
- refreshShelf() 先调用 /api/sync 同步数据，再重新加载书架
- 按钮文案三阶段：同步中... → 加载中... → 刷新
- CSS 增加 min-width 防止按钮宽度抖动
- launchd plist 和 start_server.sh 均配置 WEREAD_API_KEY 环境变量
2026-07-17 13:22:19 +08:00

1275 lines
44 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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 subprocess
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): 分类筛选（精确匹配完整分类名，如 "历史-中国古代"）
parent_category (str): 父分类筛选（如 "历史" 匹配所有 "历史-*" 子分类）
finished (str): "1" 表示只看已读完
sort (str): "category" 表示按分类排序，其他默认按置顶+阅读时间
Returns:
JSON: {books: [...], total, page, per_page, total_pages, categories: [...], category_tree: {...}}
"""
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()
parent_category = request.args.get("parent_category", "").strip()
finished_only = request.args.get("finished", "0") == "1"
sort_mode = request.args.get("sort", "").strip()
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 parent_category:
where_clauses.append("(category = %s OR category LIKE %s)")
params.extend([parent_category, f"{parent_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"]
# 排序
if sort_mode == "category":
order_sql = " ORDER BY category ASC, is_top DESC, read_update_time DESC"
else:
order_sql = " ORDER BY is_top DESC, read_update_time DESC"
# 查询当前页数据
cursor.execute(
f"SELECT * FROM books{where_sql}{order_sql} LIMIT %s OFFSET %s",
params + [per_page, offset],
)
rows = cursor.fetchall()
# 获取所有分类列表及层级树
cursor.execute(
"SELECT category, COUNT(*) AS cnt FROM books "
"WHERE category IS NOT NULL AND category != '' "
"GROUP BY category ORDER BY category"
)
cat_rows = cursor.fetchall()
# 构建层级树: { "历史": {"count": 114, "subs": {"世界史": 5, "中国古代": 10, ...}}, ... }
category_tree = {}
for r in cat_rows:
parts = r["category"].split("-", 1)
parent = parts[0]
sub = parts[1] if len(parts) > 1 else ""
if parent not in category_tree:
category_tree[parent] = {"count": 0, "subs": {}}
category_tree[parent]["count"] += r["cnt"]
if sub:
category_tree[parent]["subs"][sub] = r["cnt"]
categories = [r["category"] for r in cat_rows]
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,
"category_tree": category_tree,
})
@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),
})
# ======================================================================
# 数据同步接口 — 从微信读书 API 拉取最新数据并迁移到 MySQL
# ======================================================================
@app.route("/api/sync", methods=["POST"])
def sync_data():
"""运行 fetch_data.py + migrate_to_mysql.py，全量同步数据"""
project_dir = os.path.dirname(os.path.abspath(__file__))
python_bin = os.path.join(
os.path.expanduser("~"),
".workbuddy/binaries/python/envs/default/bin/python",
)
env = os.environ.copy()
env["WEREAD_API_KEY"] = env.get("WEREAD_API_KEY", "")
if not env["WEREAD_API_KEY"]:
return jsonify({"ok": False, "error": "WEREAD_API_KEY 未设置"}), 500
results = {"steps": []}
try:
# Step 1: 采集数据 → SQLite
r1 = subprocess.run(
[python_bin, "fetch_data.py"],
cwd=project_dir,
env=env,
capture_output=True,
text=True,
timeout=120,
)
results["steps"].append({
"name": "fetch_data",
"ok": r1.returncode == 0,
"stdout": r1.stdout[-500:] if r1.stdout else "",
"stderr": r1.stderr[-500:] if r1.stderr else "",
})
if r1.returncode != 0:
results["ok"] = False
results["error"] = "fetch_data.py 执行失败"
return jsonify(results), 500
# Step 2: SQLite → MySQL
r2 = subprocess.run(
[python_bin, "migrate_to_mysql.py"],
cwd=project_dir,
env=env,
capture_output=True,
text=True,
timeout=60,
)
results["steps"].append({
"name": "migrate_to_mysql",
"ok": r2.returncode == 0,
"stdout": r2.stdout[-500:] if r2.stdout else "",
"stderr": r2.stderr[-500:] if r2.stderr else "",
})
if r2.returncode != 0:
results["ok"] = False
results["error"] = "migrate_to_mysql.py 执行失败"
return jsonify(results), 500
results["ok"] = True
results["message"] = "数据同步完成"
return jsonify(results)
except subprocess.TimeoutExpired:
results["ok"] = False
results["error"] = "同步超时（超过120秒）"
return jsonify(results), 500
except Exception as e:
results["ok"] = False
results["error"] = str(e)
return jsonify(results), 500
# ======================================================================
# 启动入口
# ======================================================================
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)
Reference in new issue View Git Blame Copy Permalink