feat: init weread data dashboard

- 数据采集脚本 fetch_data.py:从微信读书 API 拉取书架、阅读统计、笔记数据存入 SQLite
- Flask 后端 app.py:提供 8 个 API 接口(总览、年度趋势、读书排行、偏好分类、偏好作者、时段分布、书架、笔记)
- 前端页面:ECharts 图表(年度趋势、时段分布、分类饼图、作者排行)、书架网格、笔记概览
- 数据规模:404 本书架、339 小时阅读、304 天阅读、1709 条笔记
This commit is contained in:
admin committed 2026-07-17 09:59:37 +08:00
commit 1e9c537b86
9 files changed
+2361

No files matched your search

+375
View File
@@ -0,0 +1,375 @@
#!/usr/bin/env python3
"""
微信读书数据展示网站 - Flask 后端
提供 API 接口供前端查询本地 SQLite 数据库中的读书数据
"""
import json
import os
import sqlite3
import time
from datetime import datetime
from flask import Flask, jsonify, render_template, request
app = Flask(__name__, static_folder="static", template_folder="templates")
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "weread.db")
def get_db():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
def ts_to_date(ts):
"""时间戳转日期字符串"""
if not ts or ts == 0:
return ""
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d")
def ts_to_year(ts):
"""时间戳转年份"""
if not ts or ts == 0:
return ""
return datetime.fromtimestamp(ts).strftime("%Y")
def format_duration(seconds):
"""秒转可读时长"""
if not seconds:
return "0分钟"
hours = seconds // 3600
minutes = (seconds % 3600) // 60
if hours > 0:
return f"{hours}小时{minutes}分钟"
return f"{minutes}分钟"
@app.route("/")
def index():
return render_template("index.html")
@app.route("/api/overview")
def api_overview():
"""总览数据"""
conn = get_db()
c = conn.cursor()
# 总阅读统计
c.execute("SELECT * FROM reading_stats WHERE mode='overall' LIMIT 1")
overall = c.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
# 今年统计
current_year = str(time.localtime().tm_year)
c.execute("SELECT * FROM reading_stats WHERE mode='annually' LIMIT 1")
annual = c.fetchone()
year_read_time = annual["total_read_time"] if annual else 0
year_read_days = annual["read_days"] if annual else 0
# 书架数量
c.execute("SELECT COUNT(*) as cnt FROM books")
shelf_count = c.fetchone()["cnt"]
# 读完数量
c.execute("SELECT COUNT(*) as cnt FROM books WHERE finish_reading=1")
finished_count = c.fetchone()["cnt"]
# 笔记统计
c.execute("SELECT COUNT(*) as cnt FROM notebooks")
notebook_count = c.fetchone()["cnt"]
c.execute("SELECT SUM(review_count + note_count + bookmark_count) as cnt FROM notebooks")
total_notes = c.fetchone()["cnt"] or 0
# 最近阅读时间
c.execute("SELECT MAX(read_update_time) as max_ts FROM books")
last_read_ts = c.fetchone()["max_ts"] or 0
# 采集时间
c.execute("SELECT value FROM fetch_meta WHERE key='last_fetch_time'")
fetch_row = c.fetchone()
fetch_time = fetch_row["value"] if fetch_row else ""
conn.close()
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")
def api_yearly_trend():
"""年度阅读时长时间序列"""
conn = get_db()
c = conn.cursor()
c.execute(
"SELECT bucket_timestamp, duration FROM read_times WHERE mode='overall' ORDER BY bucket_timestamp"
)
rows = c.fetchall()
conn.close()
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")
def api_top_books():
"""阅读时长最久的书籍"""
conn = get_db()
c = conn.cursor()
c.execute(
"""SELECT * FROM read_longest WHERE mode='overall'
ORDER BY read_time DESC LIMIT 20"""
)
rows = c.fetchall()
conn.close()
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")
def api_categories():
"""偏好分类"""
conn = get_db()
c = conn.cursor()
c.execute(
"""SELECT * FROM prefer_categories WHERE mode='overall'
AND reading_time > 0
ORDER BY reading_time DESC"""
)
rows = c.fetchall()
conn.close()
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")
def api_authors():
"""偏好作者"""
conn = get_db()
c = conn.cursor()
c.execute(
"""SELECT * FROM prefer_authors WHERE mode='overall'
ORDER BY count DESC LIMIT 20"""
)
rows = c.fetchall()
conn.close()
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")
def api_time_distribution():
"""24小时阅读时段分布"""
conn = get_db()
c = conn.cursor()
c.execute(
"""SELECT hour_index, SUM(duration) as total
FROM prefer_time WHERE mode='overall'
GROUP BY hour_index ORDER BY hour_index"""
)
rows = c.fetchall()
conn.close()
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")
def api_shelf():
"""书架列表"""
page = int(request.args.get("page", 1))
per_page = int(request.args.get("per_page", 60))
category = request.args.get("category", "")
finished_only = request.args.get("finished", "0") == "1"
offset = (page - 1) * per_page
conn = get_db()
c = conn.cursor()
where_clauses = []
params = []
if category:
where_clauses.append("category LIKE ?")
params.append(f"%{category}%")
if finished_only:
where_clauses.append("finish_reading=1")
where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
c.execute(f"SELECT COUNT(*) as cnt FROM books{where_sql}", params)
total = c.fetchone()["cnt"]
c.execute(
f"SELECT * FROM books{where_sql} ORDER BY is_top DESC, read_update_time DESC LIMIT ? OFFSET ?",
params + [per_page, offset],
)
rows = c.fetchall()
# 获取分类列表
c.execute("SELECT DISTINCT category FROM books WHERE category != '' ORDER BY category")
categories = [r["category"] for r in c.fetchall()]
conn.close()
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")
def api_notebooks():
"""笔记本概览"""
conn = get_db()
c = conn.cursor()
c.execute(
"""SELECT * FROM notebooks
ORDER BY (review_count + note_count + bookmark_count) DESC"""
)
rows = c.fetchall()
conn.close()
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")
def api_read_stats():
"""阅读统计摘要项"""
conn = get_db()
c = conn.cursor()
c.execute("SELECT * FROM read_stat_items WHERE mode='overall'")
rows = c.fetchall()
conn.close()
stats = []
for row in rows:
stats.append({
"stat": row["stat_name"],
"counts": row["counts"],
"scheme": row["scheme"],
})
return jsonify(stats)
@app.route("/api/archives")
def api_archives():
"""书单"""
conn = get_db()
c = conn.cursor()
c.execute("SELECT * FROM archives")
rows = c.fetchall()
conn.close()
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)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080, debug=False)