#!/usr/bin/env python3 """ 微信读书数据采集脚本 从微信读书 API 拉取数据并存入本地 SQLite 数据库 """ import json import os import sqlite3 import time import urllib.request API_URL = "https://i.weread.qq.com/api/agent/gateway" API_KEY = os.environ.get("WEREAD_API_KEY", "") SKILL_VERSION = "1.0.3" DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "weread.db") PROJECT_DIR = os.path.dirname(os.path.abspath(__file__)) def api_call(api_name, **kwargs): """调用微信读书 API""" body = {"api_name": api_name, "skill_version": SKILL_VERSION} body.update(kwargs) req = urllib.request.Request( 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) as resp: return json.loads(resp.read().decode("utf-8")) def init_db(db_path): """ 初始化 SQLite 数据库表结构。 创建所有数据表（如果不存在），包括： - 基础数据表：书架、阅读统计、笔记本等（采集时写入） - 按需数据表：书籍详情、划线、想法等（用户点击时写入） """ conn = sqlite3.connect(db_path) c = conn.cursor() # ===== 基础数据表（采集脚本写入） ===== # 书架书籍表 - 存储用户书架中的所有书籍信息 c.execute(""" CREATE TABLE IF NOT EXISTS books ( book_id TEXT PRIMARY KEY, -- 书籍唯一标识 title TEXT, -- 书名 author TEXT, -- 作者 cover TEXT, -- 封面图 URL category TEXT, -- 分类（如 "文学-古代诗词"） finish_reading INTEGER DEFAULT 0, -- 是否读完 (1=读完, 0=未读完) read_update_time INTEGER, -- 最近阅读时间 (Unix 时间戳) secret INTEGER DEFAULT 0, -- 是否私密书籍 (1=私密) is_top INTEGER DEFAULT 0, -- 是否置顶 (1=置顶) update_time INTEGER, -- 书籍信息更新时间 (Unix 时间戳) deep_link TEXT -- 微信读书 App 深度链接 ) """) # 阅读统计概要表 - 存储总体/年度阅读统计数据 c.execute(""" CREATE TABLE IF NOT EXISTS reading_stats ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度: overall(总体)/annually(年度) period TEXT, -- 周期标识: "overall" 或 "2026" total_read_time INTEGER, -- 总阅读时长 (秒) read_days INTEGER, -- 有效阅读天数 day_average_read_time REAL, -- 日均阅读时长 (秒) regist_time INTEGER, -- 用户注册时间 (Unix 时间戳) raw_json TEXT -- 原始 API 返回 JSON (完整备份) ) """) # 阅读时长时间序列表 - 按年/月/日分桶的阅读时长数据 c.execute(""" CREATE TABLE IF NOT EXISTS read_times ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 (含 _daily 后缀表示每日数据) period TEXT, -- 周期标识 bucket_timestamp INTEGER, -- 分桶起始时间戳 (Unix 时间戳) duration INTEGER -- 该时段阅读时长 (秒) ) """) # 阅读时长排行表 - 阅读时间最久的书籍排名 c.execute(""" CREATE TABLE IF NOT EXISTS read_longest ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 book_id TEXT, -- 书籍 ID title TEXT, -- 书名 author TEXT, -- 作者 cover TEXT, -- 封面图 URL read_time INTEGER, -- 阅读时长 (秒) tags TEXT -- 标签 (JSON 数组格式) ) """) # 偏好分类表 - 用户阅读的书籍分类统计 c.execute(""" CREATE TABLE IF NOT EXISTS prefer_categories ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 category_id INTEGER, -- 分类 ID category_title TEXT, -- 分类名称 (如 "文学") parent_category_title TEXT, -- 父分类名称 (如 "精品小说") val REAL, -- 偏好权重值 reading_time INTEGER, -- 该分类阅读时长 (秒) reading_count INTEGER -- 该分类阅读本数 ) """) # 偏好作者表 - 用户阅读最多的作者统计 c.execute(""" CREATE TABLE IF NOT EXISTS prefer_authors ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 author_id INTEGER, -- 作者 ID name TEXT, -- 作者名称 count INTEGER, -- 阅读该作者的书本数 read_time TEXT -- 阅读时长 (格式化字符串) ) """) # 24小时阅读时段分布表 - 一天中每个小时的阅读时长 c.execute(""" CREATE TABLE IF NOT EXISTS prefer_time ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 hour_index INTEGER, -- 小时 (0-23) duration INTEGER -- 该小时阅读时长 (秒) ) """) # 笔记本概览表 - 有笔记的书籍及其笔记统计 c.execute(""" CREATE TABLE IF NOT EXISTS notebooks ( book_id TEXT PRIMARY KEY, -- 书籍 ID title TEXT, -- 书名 author TEXT, -- 作者 cover TEXT, -- 封面图 URL review_count INTEGER DEFAULT 0, -- 想法/点评数 note_count INTEGER DEFAULT 0, -- 划线数 bookmark_count INTEGER DEFAULT 0, -- 书签数 reading_progress INTEGER DEFAULT 0, -- 阅读进度 (0-100) marked_status INTEGER DEFAULT 0, -- 标记状态 (1=读完) sort INTEGER -- 排序值 (微信读书返回的排序字段) ) """) # 阅读统计摘要项表 - App 首页展示的统计摘要 c.execute(""" CREATE TABLE IF NOT EXISTS read_stat_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 stat_name TEXT, -- 统计项名称 (读过/读完/阅读/笔记) counts TEXT, -- 统计值文案 (如 "404本") scheme TEXT -- App 跳转链接 ) """) # 书单/归档表 - 用户创建的书单 c.execute(""" CREATE TABLE IF NOT EXISTS archives ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, -- 书单名称 book_ids TEXT -- 包含的书籍 ID 列表 (JSON 数组) ) """) # 数据采集元信息表 - 记录采集时间等元数据 c.execute(""" CREATE TABLE IF NOT EXISTS fetch_meta ( key TEXT PRIMARY KEY, -- 元信息键名 (如 "last_fetch_time") value TEXT -- 元信息值 ) """) # ===== 按需数据表（用户点击时实时获取并写入） ===== # 书籍详细信息表 - 来自微信读书 /book/info 接口 c.execute(""" CREATE TABLE IF NOT EXISTS book_details ( book_id TEXT PRIMARY KEY, -- 书籍 ID title TEXT, -- 书名 author TEXT, -- 作者 translator TEXT, -- 译者 cover TEXT, -- 封面图 URL intro TEXT, -- 书籍简介 category TEXT, -- 分类 publisher TEXT, -- 出版社 publish_time TEXT, -- 出版时间 isbn TEXT, -- ISBN 号 word_count INTEGER, -- 总字数 new_rating REAL, -- 评分 (百分制) new_rating_count INTEGER, -- 评分人数 fetched_at INTEGER -- 数据获取时间 (Unix 时间戳) ) """) # 阅读进度表 - 来自微信读书 /book/getprogress 接口 c.execute(""" CREATE TABLE IF NOT EXISTS book_progress ( book_id TEXT PRIMARY KEY, -- 书籍 ID progress INTEGER, -- 阅读进度百分比 (0-100) chapter_uid INTEGER, -- 当前阅读章节 UID chapter_offset INTEGER, -- 当前章节内偏移量 update_time INTEGER, -- 最后阅读时间 (Unix 时间戳) record_reading_time INTEGER, -- 累计阅读时长 (秒) finish_time INTEGER, -- 读完时间 (Unix 时间戳) is_start_reading INTEGER, -- 是否已开始阅读 (1=是) fetched_at INTEGER -- 数据获取时间 (Unix 时间戳) ) """) # 划线内容表 - 来自微信读书 /book/bookmarklist 接口 c.execute(""" CREATE TABLE IF NOT EXISTS book_highlights ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT, -- 书籍 ID bookmark_id TEXT, -- 划线唯一 ID chapter_uid INTEGER, -- 所在章节 UID mark_text TEXT, -- 划线原文内容 create_time INTEGER, -- 创建时间 (Unix 时间戳) type INTEGER, -- 划线类型 range_text TEXT, -- 位置范围 color_style INTEGER, -- 划线颜色样式 fetched_at INTEGER -- 数据获取时间 (Unix 时间戳) ) """) # 章节信息表 - 来自微信读书 /book/bookmarklist 接口的 chapters 字段 c.execute(""" CREATE TABLE IF NOT EXISTS book_chapters ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT, -- 书籍 ID chapter_uid INTEGER, -- 章节 UID chapter_idx INTEGER, -- 章节序号 title TEXT, -- 章节标题 UNIQUE(book_id, chapter_uid) -- 联合唯一约束 ) """) # 个人想法/点评表 - 来自微信读书 /review/list/mine 接口 c.execute(""" CREATE TABLE IF NOT EXISTS book_reviews ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT, -- 书籍 ID review_id TEXT, -- 想法唯一 ID content TEXT, -- 想法内容文本 create_time INTEGER, -- 创建时间 (Unix 时间戳) star INTEGER, -- 评分 (0-5, -1=无评分) chapter_name TEXT, -- 所在章节名称 is_finish INTEGER, -- 是否书评 (1=书评) fetched_at INTEGER -- 数据获取时间 (Unix 时间戳) ) """) # 推荐书籍表 - 来自微信读书 /book/recommend 接口 c.execute(""" CREATE TABLE IF NOT EXISTS recommended_books ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT, -- 书籍 ID title TEXT, -- 书名 author TEXT, -- 作者 cover TEXT, -- 封面图 URL intro TEXT, -- 书籍简介 category TEXT, -- 分类 reason TEXT, -- 推荐理由 reading_count INTEGER, -- 在读人数 new_rating REAL, -- 评分 (百分制) new_rating_count INTEGER, -- 评分人数 rating_title TEXT, -- 评分标签 (神作/力荐等) fetched_at INTEGER -- 数据获取时间 (Unix 时间戳) ) """) # 搜索结果表 - 来自微信读书 /store/search 接口 c.execute(""" CREATE TABLE IF NOT EXISTS search_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT, -- 搜索关键词 book_id TEXT, -- 书籍 ID title TEXT, -- 书名 author TEXT, -- 作者 cover TEXT, -- 封面图 URL intro TEXT, -- 书籍简介 publisher TEXT, -- 出版社 category TEXT, -- 分类 reading_count INTEGER, -- 在读人数 new_rating REAL, -- 评分 (百分制) new_rating_count INTEGER, -- 评分人数 rating_title TEXT, -- 评分标签 search_idx INTEGER, -- 搜索序号 fetched_at INTEGER -- 数据获取时间 (Unix 时间戳) ) """) conn.commit() return conn def fetch_shelf(conn): """拉取书架数据""" print(" 拉取书架数据...") data = api_call("/shelf/sync") c = conn.cursor() books = data.get("books", []) for book in books: c.execute( """INSERT OR REPLACE INTO books (book_id, title, author, cover, category, finish_reading, read_update_time, secret, is_top, update_time, deep_link) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( book.get("bookId"), book.get("title", ""), book.get("author", ""), book.get("cover", ""), book.get("category", ""), book.get("finishReading", 0), book.get("readUpdateTime", 0), book.get("secret", 0), book.get("isTop", 0), book.get("updateTime", 0), book.get("deepLink", ""), ), ) # 保存书单 archives = data.get("archive", []) c.execute("DELETE FROM archives") for arch in archives: c.execute( "INSERT INTO archives (name, book_ids) VALUES (?, ?)", (arch.get("name", ""), json.dumps(arch.get("bookIds", []))), ) conn.commit() mp_flag = "yes" if data.get("mp") else "no" print(f" 书架: {len(books)} 本电子书, {len(data.get('albums', []))} 个专辑, MP={mp_flag}, {len(archives)} 个书单") def fetch_readdata(conn, mode, period_label): """拉取阅读统计数据""" print(f" 拉取阅读统计 [{mode}]...") params = {"mode": mode} if mode != "overall": params["baseTime"] = 0 data = api_call("/readdata/detail", **params) c = conn.cursor() total_read_time = data.get("totalReadTime", 0) read_days = data.get("readDays", 0) day_avg = data.get("dayAverageReadTime", 0) regist_time = data.get("registTime", 0) c.execute( """INSERT INTO reading_stats (mode, period, total_read_time, read_days, day_average_read_time, regist_time, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?)""", ( mode, period_label, total_read_time, read_days, day_avg, regist_time, json.dumps(data, ensure_ascii=False), ), ) # 保存 readTimes 分桶数据 read_times = data.get("readTimes", {}) for ts, duration in read_times.items(): c.execute( "INSERT INTO read_times (mode, period, bucket_timestamp, duration) VALUES (?, ?, ?, ?)", (mode, period_label, int(ts), duration), ) # 保存 dailyReadTimes（如果有） daily_times = data.get("dailyReadTimes", {}) for ts, duration in daily_times.items(): c.execute( "INSERT INTO read_times (mode, period, bucket_timestamp, duration) VALUES (?, ?, ?, ?)", (mode + "_daily", period_label, int(ts), duration), ) # 保存阅读最久的书籍 for item in data.get("readLongest", []): book = item.get("book", {}) album = item.get("albumInfo", {}) c.execute( """INSERT INTO read_longest (mode, period, book_id, title, author, cover, read_time, tags) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( mode, period_label, book.get("bookId") or album.get("albumId", ""), book.get("title") or album.get("name", ""), book.get("author") or album.get("authorName", ""), book.get("cover") or album.get("cover", ""), item.get("readTime", 0), json.dumps(item.get("tags", []), ensure_ascii=False), ), ) # 保存偏好分类 for cat in data.get("preferCategory", []): c.execute( """INSERT INTO prefer_categories (mode, period, category_id, category_title, parent_category_title, val, reading_time, reading_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( mode, period_label, cat.get("categoryId", 0), cat.get("categoryTitle", ""), cat.get("parentCategoryTitle", ""), cat.get("val", 0), cat.get("readingTime", 0), cat.get("readingCount", 0), ), ) # 保存偏好作者 for author in data.get("preferAuthor", []): c.execute( """INSERT INTO prefer_authors (mode, period, author_id, name, count, read_time) VALUES (?, ?, ?, ?, ?, ?)""", ( mode, period_label, author.get("authorId", 0), author.get("name", ""), author.get("count", 0), author.get("readTime", ""), ), ) # 保存24小时时段分布 prefer_time = data.get("preferTime", []) # preferTime 从6点开始到次日5点，映射到 hour_index for i, duration in enumerate(prefer_time): hour = (6 + i) % 24 c.execute( "INSERT INTO prefer_time (mode, period, hour_index, duration) VALUES (?, ?, ?, ?)", (mode, period_label, hour, duration), ) # 保存统计摘要项 for stat in data.get("readStat", []): c.execute( """INSERT INTO read_stat_items (mode, period, stat_name, counts, scheme) VALUES (?, ?, ?, ?, ?)""", ( mode, period_label, stat.get("stat", ""), stat.get("counts", ""), stat.get("scheme", ""), ), ) conn.commit() print(f" [{mode}] 总时长: {total_read_time}秒 ({total_read_time//3600}小时), 阅读天数: {read_days}") def fetch_notebooks(conn): """拉取笔记本概览（分页拉取全部）""" print(" 拉取笔记本数据...") c = conn.cursor() c.execute("DELETE FROM notebooks") count = 100 last_sort = None total = 0 while True: params = {"count": count} if last_sort is not None: params["lastSort"] = last_sort data = api_call("/user/notebooks", **params) books = data.get("books", []) for item in books: book = item.get("book", {}) c.execute( """INSERT OR REPLACE INTO notebooks (book_id, title, author, cover, review_count, note_count, bookmark_count, reading_progress, marked_status, sort) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( item.get("bookId", ""), book.get("title", ""), book.get("author", ""), book.get("cover", ""), item.get("reviewCount", 0), item.get("noteCount", 0), item.get("bookmarkCount", 0), item.get("readingProgress", 0), item.get("markedStatus", 0), item.get("sort", 0), ), ) total += 1 if data.get("hasMore", 0) == 1 and books: last_sort = books[-1].get("sort") else: break time.sleep(0.3) conn.commit() print(f" 笔记本: {total} 本有笔记的书") def main(): if not API_KEY: print("错误: 请先设置 WEREAD_API_KEY 环境变量") print(" export WEREAD_API_KEY=wrk-xxxxxxxx") return print("=" * 50) print("微信读书数据采集") print("=" * 50) # 初始化数据库 print(f"\n[1/4] 初始化数据库: {DB_PATH}") conn = init_db(DB_PATH) # 清空旧数据（保留表结构） for table in [ "reading_stats", "read_times", "read_longest", "prefer_categories", "prefer_authors", "prefer_time", "read_stat_items", ]: conn.execute(f"DELETE FROM {table}") conn.commit() # 拉取书架 print("\n[2/4] 拉取书架数据") fetch_shelf(conn) # 拉取阅读统计（overall + 当前年） print("\n[3/4] 拉取阅读统计数据") fetch_readdata(conn, "overall", "overall") current_year = time.localtime().tm_year fetch_readdata(conn, "annually", str(current_year)) # 拉取笔记本 print("\n[4/4] 拉取笔记本数据") fetch_notebooks(conn) # 记录采集时间 c = conn.cursor() c.execute( "INSERT OR REPLACE INTO fetch_meta (key, value) VALUES (?, ?)", ("last_fetch_time", time.strftime("%Y-%m-%d %H:%M:%S")), ) conn.commit() # 统计 c.execute("SELECT COUNT(*) FROM books") book_count = c.fetchone()[0] c.execute("SELECT COUNT(*) FROM notebooks") notebook_count = c.fetchone()[0] c.execute("SELECT SUM(total_read_time) FROM reading_stats WHERE mode='overall'") total_time = c.fetchone()[0] or 0 print("\n" + "=" * 50) print("采集完成!") print(f" 书架书籍: {book_count} 本") print(f" 笔记本: {notebook_count} 本") print(f" 总阅读时长: {total_time // 3600} 小时 {(total_time % 3600) // 60} 分钟") print(f" 数据库: {DB_PATH}") print("=" * 50) conn.close() if __name__ == "__main__": main()