diff --git a/weread-sync/sync.py b/weread-sync/sync.py new file mode 100755 index 0000000..7cfa015 --- /dev/null +++ b/weread-sync/sync.py @@ -0,0 +1,280 @@ +#!/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()