#!/usr/bin/env python3 """ 微信读书数据采集脚本 从微信读书 API 拉取数据并存入本地 SQLite 数据库。 覆盖全部 8 份官方 API 文档： - shelf.md: 书架（电子书 + 有声书 + 书单） - readdata.md: 阅读统计（时长、排行、偏好分类/作者/出版社/版权方等） - notes.md: 笔记（笔记本概览、划线、个人想法） - book.md: 书籍详情、章节目录、阅读进度 - discover.md: 推荐、相似书 - search.md: 搜索 - review.md: 公开点评 - profile.md: 用户概况（组合接口） 设计规范： 1. 所有整型时间戳字段伴随一个 _dt 文本列（ISO 格式时间） 2. 每张表维护 created_at 和 updated_at 3. 表结构与 MySQL 版本（schema_mysql.sql）完全对齐 """ import json import os import sqlite3 import time import urllib.request from datetime import datetime, timezone, timedelta from config_utils import get_env_value # ===== 配置 ===== API_URL = "https://i.weread.qq.com/api/agent/gateway" SKILL_VERSION = "1.0.4" DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "weread.db") PROJECT_DIR = os.path.dirname(os.path.abspath(__file__)) # 中国时区 UTC+8 CST = timezone(timedelta(hours=8)) def ts_to_dt(ts): """将 Unix 整型时间戳转为 ISO 格式字符串 (中国时区)""" if not ts or ts == 0: return None try: return datetime.fromtimestamp(ts, tz=CST).strftime("%Y-%m-%d %H:%M:%S") except (ValueError, OSError): return None def now_str(): """当前时间字符串""" return datetime.now(tz=CST).strftime("%Y-%m-%d %H:%M:%S") def api_call(api_name, **kwargs): """调用微信读书 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": 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 数据库表结构。 表结构与 MySQL 版本（schema_mysql.sql）完全对齐。 """ conn = sqlite3.connect(db_path) c = conn.cursor() # ===== 第一组：书架数据 ===== # 1. 书架电子书表 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=读完) read_update_time INTEGER, -- 最近阅读时间 (Unix 时间戳) read_update_time_dt TEXT, -- 最近阅读时间 (ISO 格式) secret INTEGER DEFAULT 0, -- 是否私密 (1=私密) is_top INTEGER DEFAULT 0, -- 是否置顶 (1=置顶) update_time INTEGER, -- 书籍更新时间 (Unix 时间戳) update_time_dt TEXT, -- 书籍更新时间 (ISO 格式) deep_link TEXT, -- 微信读书 App 深度链接 created_at TEXT DEFAULT (datetime('now', 'localtime')), -- 记录创建时间 updated_at TEXT DEFAULT (datetime('now', 'localtime')) -- 记录更新时间 ) """) # 2. 书架有声书/专辑表 c.execute(""" CREATE TABLE IF NOT EXISTS shelf_albums ( album_id TEXT PRIMARY KEY, -- 专辑唯一标识 name TEXT, -- 专辑名称 author_name TEXT, -- 演播/作者 cover TEXT, -- 封面图 URL track_count INTEGER DEFAULT 0, -- 音频集数 finish_status TEXT, -- 完结状态文案 finish INTEGER DEFAULT 0, -- 是否完结 (1=完结) pay_type INTEGER, -- 付费类型 intro TEXT, -- 专辑简介 update_time INTEGER, -- 更新时间 (Unix 时间戳) update_time_dt TEXT, -- 更新时间 (ISO 格式) secret INTEGER DEFAULT 0, -- 是否私密 lecture_paid INTEGER DEFAULT 0, -- 是否已购买 lecture_read_update_time INTEGER, -- 最近收听时间 (Unix 时间戳) lecture_read_update_time_dt TEXT, -- 最近收听时间 (ISO 格式) is_top INTEGER DEFAULT 0, -- 是否置顶 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 3. 书架元信息表 c.execute(""" CREATE TABLE IF NOT EXISTS shelf_meta ( id INTEGER PRIMARY KEY AUTOINCREMENT, mp_enabled INTEGER DEFAULT 0, -- 是否有文章收藏入口 book_count INTEGER DEFAULT 0, -- 可枚举电子书数量 album_count INTEGER DEFAULT 0, -- 有声书数量 total_count INTEGER DEFAULT 0, -- 书架总条目数 raw_json TEXT, -- 原始 API 返回 JSON created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 4. 书单/归档表 c.execute(""" CREATE TABLE IF NOT EXISTS archives ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, -- 书单名称 book_ids TEXT, -- bookId 列表 (JSON 数组) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # ===== 第二组：阅读统计数据 ===== # 5. 阅读统计概要表 c.execute(""" CREATE TABLE IF NOT EXISTS reading_stats ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度: overall/annually/monthly/weekly period TEXT, -- 周期标识 base_time INTEGER, -- 统计周期基准时间戳 base_time_dt TEXT, -- 基准时间 (ISO 格式) total_read_time INTEGER DEFAULT 0, -- 总阅读时长 (秒) read_days INTEGER DEFAULT 0, -- 有效阅读天数 day_average_read_time REAL DEFAULT 0, -- 日均阅读时长 (秒) compare REAL, -- 与上周期对比比例 regist_time INTEGER, -- 用户注册时间 (Unix 时间戳) regist_time_dt TEXT, -- 注册时间 (ISO 格式) record_reading_time INTEGER DEFAULT 0, -- 朗读/记录类阅读时长 (秒) read_rate REAL, -- 文字阅读占比百分比 wr_read_time INTEGER, -- 文字阅读时长 (秒) wr_listen_time INTEGER, -- 听书时长 (秒) author_count INTEGER, -- 作者总数 prefer_category_word TEXT, -- 偏好分类文案 prefer_time_word TEXT, -- 偏好时段文案 read_records_word TEXT, -- 书籍分布模块标题 read_distribution_word TEXT, -- 点评分布模块标题 read_time_gears TEXT, -- 阅读时长档位 (JSON) style_type TEXT, -- 样式类型 raw_json TEXT, -- 原始 API 返回 JSON created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 6. 阅读时长时间序列表 c.execute(""" CREATE TABLE IF NOT EXISTS read_times ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 (含 _daily 后缀) period TEXT, -- 周期标识 bucket_timestamp INTEGER, -- 分桶起始时间戳 bucket_timestamp_dt TEXT, -- 分桶起始时间 (ISO 格式) duration INTEGER DEFAULT 0, -- 阅读时长 (秒) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 7. 阅读时长排行表 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 DEFAULT 0, -- 阅读时长 (秒) record_reading_time INTEGER, -- 朗读/记录类阅读时长 (秒) tags TEXT, -- 标签 (JSON 数组) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 8. 阅读统计摘要项表 c.execute(""" CREATE TABLE IF NOT EXISTS read_stat_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 stat_name TEXT, -- 统计项名称 (读过/读完/阅读/笔记) counts TEXT, -- 统计值文案 scheme TEXT, -- App 跳转链接 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 9. 偏好分类表 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_id INTEGER, -- 父分类 ID parent_category_title TEXT, -- 父分类名称 category_type INTEGER DEFAULT 0, -- 分类类型标记 val REAL DEFAULT 0, -- 偏好权重 reading_time INTEGER DEFAULT 0, -- 阅读时长 (秒) reading_count INTEGER DEFAULT 0, -- 阅读本数 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 10. 偏好作者表 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 DEFAULT 0, -- 阅读书本数 read_time TEXT, -- 阅读时长 (格式化字符串) user_json TEXT, -- 作者关联用户信息 (JSON) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 11. 偏好出版社表 c.execute(""" CREATE TABLE IF NOT EXISTS prefer_publishers ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 name TEXT, -- 出版社名 count INTEGER DEFAULT 0, -- 阅读本数 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 12. 偏好版权方表 c.execute(""" CREATE TABLE IF NOT EXISTS prefer_cps ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 count INTEGER DEFAULT 0, -- 阅读本数 cp_name TEXT, -- 版权方名称 cp_user_vid TEXT, -- 版权方用户 VID cp_avatar TEXT, -- 版权方头像 URL cp_role TEXT, -- 版权方角色 copyright_json TEXT, -- 版权方完整信息 (JSON) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 13. 偏好书籍卡片表 c.execute(""" CREATE TABLE IF NOT EXISTS prefer_books ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 book_id TEXT, -- 书籍 ID title TEXT, -- 书名 author TEXT, -- 作者 cover TEXT, -- 封面图 URL reason TEXT, -- 推荐理由 prefer_type TEXT, -- 偏好类型 book_json TEXT, -- 书籍完整信息 (JSON) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 14. 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 DEFAULT 0, -- 阅读时长 (秒) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 15. 好友阅读排行表 c.execute(""" CREATE TABLE IF NOT EXISTS read_rank ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 rank_text TEXT, -- 排行文案 scheme TEXT, -- 排行跳转链接 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 16. 勋章表 c.execute(""" CREATE TABLE IF NOT EXISTS medals ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 medal_json TEXT, -- 勋章信息 (JSON) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 17. 年度报告表 c.execute(""" CREATE TABLE IF NOT EXISTS year_reports ( id INTEGER PRIMARY KEY AUTOINCREMENT, mode TEXT, -- 统计维度 period TEXT, -- 周期标识 report_year TEXT, -- 报告年份 report_json TEXT, -- 报告入口信息 (JSON) monthly_times TEXT, -- 12个月阅读时长 (JSON 数组) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # ===== 第三组：笔记数据 ===== # 18. 笔记本概览表 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, -- 书签数 total_note_count INTEGER DEFAULT 0, -- 笔记总数 reading_progress INTEGER DEFAULT 0, -- 阅读进度 (0-100) marked_status INTEGER DEFAULT 0, -- 标记状态 (1=读完) sort INTEGER, -- 排序值 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 19. 划线内容表 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 时间戳) create_time_dt TEXT, -- 创建时间 (ISO 格式) type INTEGER, -- 类型 (1=划线) range_text TEXT, -- 位置范围 color_style INTEGER, -- 划线颜色样式 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 20. 章节信息表 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, -- 章节标题 word_count INTEGER, -- 章节字数 level INTEGER, -- 目录层级 (1=一级) update_time INTEGER, -- 章节更新时间 (Unix 时间戳) update_time_dt TEXT, -- 章节更新时间 (ISO 格式) price INTEGER DEFAULT 0, -- 章节价格 (0=免费) paid INTEGER DEFAULT 0, -- 是否已购买 is_mp_chapter INTEGER DEFAULT 0, -- 是否公众号章节 anchors TEXT, -- 章节内锚点 (JSON) synckey INTEGER, -- 同步 key chapter_update_time INTEGER, -- 章节列表最后更新时间 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')), UNIQUE(book_id, chapter_uid) ) """) # 21. 个人想法/点评表 c.execute(""" CREATE TABLE IF NOT EXISTS book_reviews ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT, -- 书籍 ID review_id TEXT, -- 想法唯一 ID content TEXT, -- 想法内容文本 abstract TEXT, -- 想法对应的划线原文 range_text TEXT, -- 划线原文位置范围 chapter_uid INTEGER, -- 章节 UID chapter_idx INTEGER, -- 章节序号 chapter_name TEXT, -- 所在章节名 create_time INTEGER, -- 创建时间 (Unix 时间戳) create_time_dt TEXT, -- 创建时间 (ISO 格式) star INTEGER DEFAULT -1, -- 评分 (0-5, -1=无评分) is_finish INTEGER DEFAULT 0, -- 是否书评 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 22. 热门划线表 c.execute(""" CREATE TABLE IF NOT EXISTS book_best_bookmarks ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT, -- 书籍 ID bookmark_id TEXT, -- 划线唯一 ID chapter_uid INTEGER, -- 所在章节 UID range_text TEXT, -- 划线位置范围 mark_text TEXT, -- 划线原文文本 total_count INTEGER DEFAULT 0, -- 划线人数 user_vid TEXT, -- 代表用户 VID simplified_range TEXT, -- 简体 range traditional_range TEXT, -- 繁体 range synckey INTEGER, -- 同步 key created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 23. 划线热度统计表 c.execute(""" CREATE TABLE IF NOT EXISTS book_underlines ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT, -- 书籍 ID chapter_uid INTEGER, -- 章节 UID range_text TEXT, -- 划线位置范围 count INTEGER DEFAULT 0, -- 划线人数 score REAL DEFAULT 0, -- 热度分数 type INTEGER, -- 划线类型 synckey INTEGER, -- 同步 key created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 24. 公开点评表 c.execute(""" CREATE TABLE IF NOT EXISTS public_reviews ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT, -- 书籍 ID review_id TEXT, -- 点评唯一 ID content TEXT, -- 点评文本内容 html_content TEXT, -- 点评 HTML 富文本 star INTEGER DEFAULT -1, -- 评分 (20-100) is_finish INTEGER DEFAULT 0, -- 是否读完 create_time INTEGER, -- 创建时间 (Unix 时间戳) create_time_dt TEXT, -- 创建时间 (ISO 格式) chapter_name TEXT, -- 所在章节名 author_vid TEXT, -- 评论者 VID author_name TEXT, -- 评论者昵称 author_avatar TEXT, -- 评论者头像 URL review_list_type INTEGER DEFAULT 0, -- 筛选类型 idx INTEGER, -- 序号 (用于翻页) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # ===== 第四组：书籍详情 ===== # 25. 书籍详细信息表 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 DEFAULT 0, -- 总字数 new_rating REAL DEFAULT 0, -- 评分 (百分制) new_rating_count INTEGER DEFAULT 0, -- 评分人数 new_rating_detail TEXT, -- 评分分布详情 (JSON) deep_link TEXT, -- 跳转链接 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 26. 阅读进度表 c.execute(""" CREATE TABLE IF NOT EXISTS book_progress ( book_id TEXT PRIMARY KEY, -- 书籍 ID progress INTEGER DEFAULT 0, -- 阅读进度百分比 (0-100) chapter_uid INTEGER, -- 当前阅读章节 UID chapter_offset INTEGER, -- 当前章节内偏移量 update_time INTEGER, -- 最后阅读时间 (Unix 时间戳) update_time_dt TEXT, -- 最后阅读时间 (ISO 格式) record_reading_time INTEGER DEFAULT 0, -- 累计阅读时长 (秒) finish_time INTEGER, -- 读完时间 (Unix 时间戳) finish_time_dt TEXT, -- 读完时间 (ISO 格式) is_start_reading INTEGER DEFAULT 0, -- 是否已开始阅读 server_timestamp INTEGER, -- 服务端时间戳 server_timestamp_dt TEXT, -- 服务端时间 (ISO 格式) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # ===== 第五组：发现/推荐 ===== # 27. 推荐书籍表 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 DEFAULT 0, -- 在读人数 new_rating REAL DEFAULT 0, -- 评分 (百分制) new_rating_count INTEGER DEFAULT 0, -- 评分人数 rating_title TEXT, -- 评分标签 deep_link TEXT, -- 跳转链接 search_idx INTEGER DEFAULT 0, -- 结果序号 price INTEGER, -- 价格 (分) pay_type INTEGER, -- 付费类型 book_type INTEGER DEFAULT 0, -- 书籍类型 (0=电子书) batch_id TEXT, -- 推荐批次 ID created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 28. 相似书推荐表 c.execute(""" CREATE TABLE IF NOT EXISTS similar_books ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_book_id TEXT, -- 源书籍 ID book_id TEXT, -- 推荐书籍 ID title TEXT, -- 书名 author TEXT, -- 作者 cover TEXT, -- 封面图 URL deep_link TEXT, -- 跳转链接 idx INTEGER, -- 结果序号 session_id TEXT, -- 翻页会话 ID book_json TEXT, -- 书籍完整信息 (JSON) created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 29. 搜索会话表 c.execute(""" CREATE TABLE IF NOT EXISTS search_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, sid TEXT, -- 搜索会话 ID keyword TEXT, -- 搜索关键词 scope INTEGER, -- 搜索类型 has_more INTEGER DEFAULT 0, -- 是否有更多 max_idx INTEGER DEFAULT 0, -- 当前翻页偏移 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # 30. 搜索结果表 c.execute(""" CREATE TABLE IF NOT EXISTS search_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id INTEGER, -- 关联搜索会话 ID keyword TEXT, -- 搜索关键词 book_id TEXT, -- 书籍 ID title TEXT, -- 书名 author TEXT, -- 作者 cover TEXT, -- 封面图 URL intro TEXT, -- 书籍简介 publisher TEXT, -- 出版社 category TEXT, -- 分类 reading_count INTEGER DEFAULT 0, -- 在读人数 new_rating REAL DEFAULT 0, -- 评分 new_rating_count INTEGER DEFAULT 0, -- 评分人数 rating_title TEXT, -- 评分标签 deep_link TEXT, -- 跳转链接 pay_type INTEGER, -- 付费类型 price INTEGER, -- 价格 (分) soldout INTEGER DEFAULT 0, -- 是否下架 search_idx INTEGER DEFAULT 0, -- 搜索序号 result_scope INTEGER, -- 回包分组类型 result_title TEXT, -- 回包分组标题 scope_count INTEGER, -- 分组总结果数 current_count INTEGER, -- 本次返回数量 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) # ===== 第六组：系统表 ===== # 31. 数据采集元信息表 c.execute(""" CREATE TABLE IF NOT EXISTS fetch_meta ( key TEXT PRIMARY KEY, -- 元信息键名 value TEXT, -- 元信息值 created_at TEXT DEFAULT (datetime('now', 'localtime')), updated_at TEXT DEFAULT (datetime('now', 'localtime')) ) """) conn.commit() return conn # ===== 数据采集函数 ===== def fetch_shelf(conn): """拉取书架数据（电子书 + 有声书 + 书单）""" print(" 拉取书架数据...") data = api_call("/shelf/sync") c = conn.cursor() # 清空旧数据 for t in ["books", "shelf_albums", "shelf_meta", "archives"]: c.execute(f"DELETE FROM {t}") # 1. 电子书 books = data.get("books", []) for book in books: rut = book.get("readUpdateTime", 0) ut = book.get("updateTime", 0) c.execute( """INSERT OR REPLACE INTO books (book_id, title, author, cover, category, finish_reading, read_update_time, read_update_time_dt, secret, is_top, update_time, update_time_dt, deep_link) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( book.get("bookId"), book.get("title", ""), book.get("author", ""), book.get("cover", ""), book.get("category", ""), book.get("finishReading", 0), rut, ts_to_dt(rut), book.get("secret", 0), book.get("isTop", 0), ut, ts_to_dt(ut), book.get("deepLink", ""), ), ) # 2. 有声书/专辑 albums = data.get("albums", []) for album in albums: info = album.get("albumInfo", {}) extra = album.get("albumInfoExtra", {}) ut = info.get("updateTime", 0) lrut = extra.get("lectureReadUpdateTime", 0) c.execute( """INSERT OR REPLACE INTO shelf_albums (album_id, name, author_name, cover, track_count, finish_status, finish, pay_type, intro, update_time, update_time_dt, secret, lecture_paid, lecture_read_update_time, lecture_read_update_time_dt, is_top) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( info.get("albumId"), info.get("name", ""), info.get("authorName", ""), info.get("cover", ""), info.get("trackCount", 0), info.get("finishStatus", ""), info.get("finish", 0), info.get("payType"), info.get("intro", ""), ut, ts_to_dt(ut), extra.get("secret", 0), extra.get("lecturePaid", 0), lrut, ts_to_dt(lrut), extra.get("isTop", 0), ), ) # 3. 书架元信息 mp_flag = 1 if data.get("mp") else 0 book_count = data.get("bookCount", len(books)) album_count = len(albums) total = book_count + album_count + mp_flag c.execute( """INSERT INTO shelf_meta (mp_enabled, book_count, album_count, total_count, raw_json) VALUES (?, ?, ?, ?, ?)""", (mp_flag, book_count, album_count, total, json.dumps(data, ensure_ascii=False)), ) # 4. 书单/归档 archives = data.get("archive", []) for arch in archives: c.execute( "INSERT INTO archives (name, book_ids) VALUES (?, ?)", (arch.get("name", ""), json.dumps(arch.get("bookIds", []))), ) conn.commit() print(f" 书架: {len(books)} 本电子书, {len(albums)} 个有声书, " f"MP={'有' if mp_flag else '无'}, {len(archives)} 个书单, 总计 {total} 条目") 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() # 5. 阅读统计概要 base_time = data.get("baseTime") regist_time = data.get("registTime") gears = data.get("readTimeGears") c.execute( """INSERT INTO reading_stats (mode, period, base_time, base_time_dt, total_read_time, read_days, day_average_read_time, compare, regist_time, regist_time_dt, record_reading_time, read_rate, wr_read_time, wr_listen_time, author_count, prefer_category_word, prefer_time_word, read_records_word, read_distribution_word, read_time_gears, style_type, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( mode, period_label, base_time, ts_to_dt(base_time), data.get("totalReadTime", 0), data.get("readDays", 0), data.get("dayAverageReadTime", 0), data.get("compare"), regist_time, ts_to_dt(regist_time), data.get("recordReadingTime", 0), data.get("readRate"), data.get("wrReadTime"), data.get("wrListenTime"), data.get("authorCount"), data.get("preferCategoryWord"), data.get("preferTimeWord"), data.get("readRecordsWord"), data.get("readDistributionWord"), json.dumps(gears) if gears else None, data.get("styleType"), json.dumps(data, ensure_ascii=False), ), ) # 6. 阅读时长时间序列 read_times = data.get("readTimes", {}) for ts, duration in read_times.items(): c.execute( "INSERT INTO read_times (mode, period, bucket_timestamp, bucket_timestamp_dt, duration) VALUES (?, ?, ?, ?, ?)", (mode, period_label, int(ts), ts_to_dt(int(ts)), duration), ) daily_times = data.get("dailyReadTimes", {}) for ts, duration in daily_times.items(): c.execute( "INSERT INTO read_times (mode, period, bucket_timestamp, bucket_timestamp_dt, duration) VALUES (?, ?, ?, ?, ?)", (mode + "_daily", period_label, int(ts), ts_to_dt(int(ts)), duration), ) # 7. 阅读时长排行 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, record_reading_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), item.get("recordReadingTime"), json.dumps(item.get("tags", []), ensure_ascii=False), ), ) # 8. 统计摘要项 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", "")), ) # 9. 偏好分类 for cat in data.get("preferCategory", []): c.execute( """INSERT INTO prefer_categories (mode, period, category_id, category_title, parent_category_id, parent_category_title, category_type, val, reading_time, reading_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( mode, period_label, cat.get("categoryId", 0), cat.get("categoryTitle", ""), cat.get("parentCategoryId"), cat.get("parentCategoryTitle", ""), cat.get("categoryType", 0), cat.get("val", 0), cat.get("readingTime", 0), cat.get("readingCount", 0), ), ) # 10. 偏好作者 for author in data.get("preferAuthor", []): user = author.get("user") c.execute( """INSERT INTO prefer_authors (mode, period, author_id, name, count, read_time, user_json) VALUES (?, ?, ?, ?, ?, ?, ?)""", ( mode, period_label, author.get("authorId", 0), author.get("name", ""), author.get("count", 0), author.get("readTime", ""), json.dumps(user, ensure_ascii=False) if user else None, ), ) # 11. 偏好出版社 for pub in data.get("preferPublisher", []): c.execute( "INSERT INTO prefer_publishers (mode, period, name, count) VALUES (?, ?, ?, ?)", (mode, period_label, pub.get("name", ""), pub.get("count", 0)), ) # 12. 偏好版权方 for cp in data.get("preferCp", []): ci = cp.get("copyrightInfo", {}) c.execute( """INSERT INTO prefer_cps (mode, period, count, cp_name, cp_user_vid, cp_avatar, cp_role, copyright_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( mode, period_label, cp.get("count", 0), ci.get("name", ""), ci.get("userVid", ""), ci.get("avatar", ""), ci.get("role", ""), json.dumps(ci, ensure_ascii=False) if ci else None, ), ) # 13. 偏好书籍卡片 for pb in data.get("preferBooks", []): book = pb.get("book", {}) c.execute( """INSERT INTO prefer_books (mode, period, book_id, title, author, cover, reason, prefer_type, book_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( mode, period_label, book.get("bookId"), book.get("title", ""), book.get("author", ""), book.get("cover", ""), pb.get("reason", ""), pb.get("preferType", ""), json.dumps(pb, ensure_ascii=False), ), ) # 14. 24小时时段分布（从6点开始到次日5点） for i, duration in enumerate(data.get("preferTime", [])): hour = (6 + i) % 24 c.execute( "INSERT INTO prefer_time (mode, period, hour_index, duration) VALUES (?, ?, ?, ?)", (mode, period_label, hour, duration), ) # 15. 好友排行 rank = data.get("rank") if rank: c.execute( "INSERT INTO read_rank (mode, period, rank_text, scheme) VALUES (?, ?, ?, ?)", (mode, period_label, rank.get("text", ""), rank.get("scheme", "")), ) # 16. 勋章 for medal in data.get("medals", []): c.execute( "INSERT INTO medals (mode, period, medal_json) VALUES (?, ?, ?)", (mode, period_label, json.dumps(medal, ensure_ascii=False)), ) # 17. 年度报告 for yr in data.get("yearReport", []): times = yr.get("times") c.execute( """INSERT INTO year_reports (mode, period, report_year, report_json, monthly_times) VALUES (?, ?, ?, ?, ?)""", ( mode, period_label, str(yr.get("year", "")), json.dumps(yr, ensure_ascii=False), json.dumps(times) if times else None, ), ) conn.commit() total = data.get("totalReadTime", 0) print(f" [{mode}] 总时长: {total}秒 ({total // 3600}小时), " f"阅读天数: {data.get('readDays', 0)}, " f"排行: {len(data.get('readLongest', []))} 本, " f"偏好分类: {len(data.get('preferCategory', []))} 个, " f"偏好作者: {len(data.get('preferAuthor', []))} 个, " f"偏好出版社: {len(data.get('preferPublisher', []))} 个") def fetch_notebooks(conn): """拉取笔记本概览（分页拉取全部，含 total_note_count）""" 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", {}) rc = item.get("reviewCount", 0) nc = item.get("noteCount", 0) bc = item.get("bookmarkCount", 0) c.execute( """INSERT OR REPLACE INTO notebooks (book_id, title, author, cover, review_count, note_count, bookmark_count, total_note_count, reading_progress, marked_status, sort) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( item.get("bookId", ""), book.get("title", ""), book.get("author", ""), book.get("cover", ""), rc, nc, bc, rc + nc + bc, 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 get_env_value("WEREAD_API_KEY", ""): print("错误: 请先设置 WEREAD_API_KEY 环境变量") print(" 可配置在当前终端、项目 .env，或 ~/.zshrc") return print("=" * 60) print("微信读书数据采集") print("=" * 60) # 初始化数据库 print(f"\n[1/4] 初始化数据库: {DB_PATH}") conn = init_db(DB_PATH) # 清空旧的统计数据（每次采集全量覆盖） for table in [ "reading_stats", "read_times", "read_longest", "read_stat_items", "prefer_categories", "prefer_authors", "prefer_publishers", "prefer_cps", "prefer_books", "prefer_time", "read_rank", "medals", "year_reports", ]: conn.execute(f"DELETE FROM {table}") conn.commit() # 拉取书架 print("\n[2/4] 拉取书架数据") fetch_shelf(conn) # 拉取阅读统计 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", now_str()), ) c.execute( "INSERT OR REPLACE INTO fetch_meta (key, value) VALUES (?, ?)", ("schema_version", "2.0"), ) conn.commit() # 统计 c.execute("SELECT COUNT(*) FROM books") book_count = c.fetchone()[0] c.execute("SELECT COUNT(*) FROM shelf_albums") album_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 c.execute("SELECT COUNT(*) FROM prefer_publishers") pub_count = c.fetchone()[0] c.execute("SELECT COUNT(*) FROM prefer_cps") cp_count = c.fetchone()[0] print("\n" + "=" * 60) print("采集完成!") print(f" 书架电子书: {book_count} 本") print(f" 有声书/专辑: {album_count} 个") print(f" 笔记本: {notebook_count} 本") print(f" 总阅读时长: {total_time // 3600} 小时 {(total_time % 3600) // 60} 分钟") print(f" 偏好出版社: {pub_count} 个") print(f" 偏好版权方: {cp_count} 个") print(f" 数据库: {DB_PATH}") print(f" 表结构版本: 2.0 (31 张表)") print("=" * 60) conn.close() if __name__ == "__main__": main()