#!/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 ssl import subprocess import sys import time import urllib.request from datetime import datetime from functools import wraps import pymysql from config_utils import get_env_value try: import certifi _ssl_context = ssl.create_default_context(cafile=certifi.where()) except ImportError: _ssl_context = ssl.create_default_context() from flask import Flask, jsonify, render_template, request from api_v2 import v2_bp # ====================================================================== # Flask 应用实例 # ====================================================================== app = Flask(__name__, static_folder="static", template_folder="templates") # 注册 v2 数据接口（/api/v2/*，为「阅间」新版前端提供聚合数据） app.register_blueprint(v2_bp) # ====================================================================== # 数据库连接配置 # ====================================================================== 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_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 调用失败时抛出异常 """ api_key = get_env_value("WEREAD_API_KEY", "") if not api_key: raise RuntimeError("WEREAD_API_KEY 未设置，请检查当前终端环境、项目 .env 或 ~/.zshrc") 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 {api_key}", "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen(req, timeout=30, context=_ssl_context) 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 = sys.executable or "python3" env = os.environ.copy() env["WEREAD_API_KEY"] = get_env_value("WEREAD_API_KEY", "") if not env["WEREAD_API_KEY"]: return jsonify({ "ok": False, "error": "WEREAD_API_KEY 未设置，请检查当前终端环境、项目 .env 或 ~/.zshrc", }), 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}") # 监听端口：默认 8080，可用环境变量 PORT 覆盖（v2 前端与 v1 并存时可换端口） port = int(os.environ.get("PORT", 8080)) print(f"[信息] 打开 http://127.0.0.1:{port} 查看「阅间」看板") app.run(host="0.0.0.0", port=port, debug=False)