#!/usr/bin/env python3 """ weread_sync.py — 微信读书笔记自动同步到 Gitea 流程： 微信读书 API → Markdown 文件 → git push → Gitea weread-notes 仓库 部署： 1. 将 API Key 写入 ~/.keys/weread_api_key 2. 将本脚本放到 ~/weread-sync/sync.py 3. chmod +x 并设置 cron Cron（每天6次，每4小时）： 0 */4 * * * cd ~/weread-sync && python3 sync.py >> sync.log 2>&1 """ import os import sys import json import time import subprocess from datetime import datetime # ─── 配置 ───────────────────────────────────────────── WEREAD_API = "https://i.weread.qq.com/api/agent/gateway" KEY_FILE = os.path.expanduser("~/.keys/weread_api_key") REPO_DIR = os.path.expanduser("~/weread-notes") SKILL_VERSION = "1.0.3" OUTPUT_DIR = os.path.join(REPO_DIR, "notes") # ─── API 调用 ───────────────────────────────────────── def call_weread(api_name, **params): """调用微信读书 API""" key_file = os.path.expanduser(KEY_FILE) if not os.path.exists(key_file): print(f"[ERROR] API Key not found: {key_file}") sys.exit(1) with open(key_file) as f: api_key = f.read().strip() body = {"api_name": api_name, "skill_version": SKILL_VERSION} body.update(params) import urllib.request req = urllib.request.Request( WEREAD_API, data=json.dumps(body).encode(), headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } ) try: resp = urllib.request.urlopen(req, timeout=30) return json.loads(resp.read()) except Exception as e: print(f"[ERROR] API call failed: {api_name} — {e}") return None def get_notebooks(): """获取书架（含笔记数量）""" data = call_weread("/user/notebooks", count=200) if not data: return [] books = data.get("books", []) result = [] for b in books: book = b.get("book", {}) or b.get("bookInfo", {}) result.append({ "bookId": book.get("bookId", b.get("bookId", "")), "title": book.get("title", "未知书名"), "author": book.get("author", ""), "reviewCount": b.get("reviewCount", 0), }) return result def get_bookmarks(book_id): """获取一本书的划线""" data = call_weread("/book/bookmarklist", bookId=book_id) if not data: return [], [] updated = data.get("updated", []) chapters = {c["chapterUid"]: c["title"] for c in data.get("chapters", [])} bookmarks = [] for u in updated: if u.get("type") == 1: # 划线 chap_title = chapters.get(u.get("chapterUid", 0), f"第{u.get('chapterIdx',0)}章") bookmarks.append({ "text": u.get("markText", ""), "chapter": chap_title, "time": u.get("createTime", 0), "range": u.get("range", ""), }) # 按章节排序 bookmarks.sort(key=lambda x: x.get("chapter", "")) return bookmarks, data.get("synckey", 0) def get_reviews(book_id): """获取一本书的笔记/想法""" data = call_weread("/review/list/mine", bookid=book_id, count=100) if not data: return [] reviews = data.get("reviews", []) result = [] for r in reviews: review_data = r.get("review", {}) if not review_data.get("content"): continue # 跳过空内容 result.append({ "content": review_data.get("content", ""), "time": review_data.get("createTime", 0), "chapter": review_data.get("chapterName", ""), "abstract": review_data.get("abstract", ""), }) return result # ─── Markdown 生成 ──────────────────────────────────── def format_time(ts): """时间戳 → 可读日期""" if not ts: return "" return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M") def generate_markdown(book, bookmarks, reviews): """生成一本书的笔记 Markdown""" lines = [] lines.append(f"# {book['title']}") if book.get("author"): lines.append(f"> 作者：{book['author']}") lines.append("") lines.append(f"> 同步时间：{format_time(time.time())}") lines.append("") # 划线 if bookmarks: lines.append("---") lines.append("## 划线笔记") lines.append("") current_chapter = None for bm in bookmarks: if bm["chapter"] != current_chapter: lines.append(f"### {bm['chapter']}") lines.append("") current_chapter = bm["chapter"] lines.append(f"> {bm['text']}") if bm["time"]: lines.append(f"> [{format_time(bm['time'])}]") lines.append("") # 想法 if reviews: lines.append("---") lines.append("## 笔记/想法") lines.append("") current_chapter = None for r in reviews: # 按章节分组（跳过空章节名） chapter = r.get("chapter", "").strip() if chapter and chapter != current_chapter: if current_chapter is not None: lines.append("") lines.append(f"### {chapter}") lines.append("") current_chapter = chapter elif not chapter: current_chapter = None # 原文 + 想法组合展示 if r.get("abstract"): lines.append(f"> {r['abstract']}") lines.append(">") lines.append(f"> {r['content']}") if r["time"]: lines.append(f"> [{format_time(r['time'])}]") lines.append("") return "\n".join(lines) # ─── Git 操作 ───────────────────────────────────────── def git_commit_push(repo_dir, message): """git commit + push""" try: subprocess.run(["git", "add", "-A"], cwd=repo_dir, check=True, capture_output=True, timeout=30) result = subprocess.run( ["git", "commit", "-m", message, "--allow-empty"], cwd=repo_dir, capture_output=True, timeout=30 ) if result.returncode == 0: subprocess.run(["git", "push"], cwd=repo_dir, check=True, capture_output=True, timeout=60) print(f"[GIT] Pushed: {message}") else: print(f"[GIT] No changes to commit") except subprocess.CalledProcessError as e: print(f"[GIT] Error: {e.stderr.decode()[:200]}") # ─── 主流程 ─────────────────────────────────────────── def main(): print(f"\n{'='*50}") print(f"微信读书同步 | {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"{'='*50}") # 1. 获取书架 print("\n[1/3] 获取书架...") books = get_notebooks() if not books: print("[WARN] 书架为空或 API 返回异常") return print(f" 找到 {len(books)} 本书") # 2. 逐本获取笔记 print("\n[2/3] 拉取笔记...") total_bookmarks = 0 total_reviews = 0 os.makedirs(OUTPUT_DIR, exist_ok=True) for i, book in enumerate(books): bid = book["bookId"] title = book["title"] print(f" [{i+1}/{len(books)}] {title}...", end=" ") # 获取划线 bookmarks, _ = get_bookmarks(bid) # 获取想法 reviews = get_reviews(bid) if not bookmarks and not reviews: print("无笔记") continue # 生成 Markdown safe_title = title.replace("/", "_").replace("\\", "_").replace(":", "_") md = generate_markdown(book, bookmarks, reviews) filepath = os.path.join(OUTPUT_DIR, f"{safe_title}.md") with open(filepath, "w", encoding="utf-8") as f: f.write(md) total_bookmarks += len(bookmarks) total_reviews += len(reviews) print(f"{len(bookmarks)}条划线, {len(reviews)}条想法") print(f"\n 共导出 {total_bookmarks} 条划线, {total_reviews} 条想法") # 3. 生成索引 print("\n[3/3] 生成索引...") index_lines = ["# 微信读书笔记索引", "", "| 书名 | 作者 | 划线 | 想法 | 最后同步 |", "|:---|:---|:---:|:---:|:---:|"] for book in books: safe_title = book['title'].replace('/','_').replace('\\','_').replace(':','_') md_file = os.path.join(OUTPUT_DIR, f"{safe_title}.md") if os.path.exists(md_file): index_lines.append(f"| [{book['title']}](notes/{os.path.basename(md_file)}) | {book.get('author','')} | {book.get('reviewCount',0)} | — | {format_time(time.time())} |") index_path = os.path.join(REPO_DIR, "README.md") with open(index_path, "w", encoding="utf-8") as f: f.write("\n".join(index_lines)) print(f" 索引已更新") # 4. Git 推送 print("\n 推送到 Gitea...") git_commit_push(REPO_DIR, f"sync: 微信读书笔记 {datetime.now().strftime('%Y-%m-%d %H:%M')}") print(f"\n{'='*50}") print(f"同步完成 ✅") print(f"{'='*50}") if __name__ == "__main__": main()