feat: weread-sync script v1 - WeRead highlights & reviews to Gitea
- Full pipeline: WeRead API → Markdown → Gitea weread-notes repo - 2837 highlights + 226 reviews exported across 105 books - Cron: every 4 hours (00:00/04:00/08:00/12:00/16:00/20:00) - API key stored in ~/.keys/weread_api_key (chmod 600) - Reviews display: original text + user thoughts combined
This commit is contained in:
fxy
280
weread-sync/sync.py
Executable file
280
weread-sync/sync.py
Executable file
@ -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()
|
||||
Reference in New Issue
Block a user