399 lines
13 KiB
Python
Executable File
399 lines
13 KiB
Python
Executable File
#!/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),
|
||
"noteCount": b.get("noteCount", 0),
|
||
"readingProgress": b.get("readingProgress", 0),
|
||
})
|
||
return result
|
||
|
||
|
||
def get_book_info(book_id):
|
||
"""获取书籍详细信息"""
|
||
data = call_weread("/book/info", bookId=book_id)
|
||
if not data:
|
||
return {}
|
||
return {
|
||
"cover": data.get("cover", ""),
|
||
"translator": data.get("translator", ""),
|
||
"publisher": data.get("publisher", ""),
|
||
"isbn": data.get("isbn", ""),
|
||
"category": data.get("category", ""),
|
||
"intro": data.get("intro", ""),
|
||
"publishTime": data.get("publishTime", ""),
|
||
}
|
||
|
||
|
||
def get_book_progress(book_id):
|
||
"""获取阅读进度"""
|
||
data = call_weread("/book/getprogress", bookId=book_id)
|
||
if not data:
|
||
return {}
|
||
book = data.get("book", {})
|
||
return {
|
||
"progress": book.get("progress", 0),
|
||
"readingTime": book.get("readingTime", 0),
|
||
"startReadingTime": book.get("startReadingTime", 0),
|
||
"updateTime": book.get("updateTime", 0),
|
||
"chapterIdx": book.get("chapterIdx", 0),
|
||
}
|
||
|
||
|
||
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 format_duration(minutes):
|
||
"""分钟 → 可读时长"""
|
||
if not minutes:
|
||
return "—"
|
||
h = minutes // 60
|
||
m = minutes % 60
|
||
if h > 0:
|
||
return f"{h}小时{m}分钟"
|
||
return f"{m}分钟"
|
||
|
||
|
||
def generate_header(book, book_info, progress):
|
||
"""生成书籍信息头部"""
|
||
lines = []
|
||
lines.append(f"# {book['title']}")
|
||
lines.append("")
|
||
|
||
# 封面图（居中显示，宽350px）
|
||
cover = book_info.get("cover", "")
|
||
if cover:
|
||
lines.append('<p align="center">')
|
||
lines.append(f' <img src="{cover}" width="350">')
|
||
lines.append('</p>')
|
||
lines.append("")
|
||
|
||
# 基本信息表（居中）
|
||
lines.append('<div align="center">')
|
||
lines.append("")
|
||
lines.append("| | |")
|
||
lines.append("|:---|:---|")
|
||
lines.append(f"| **作者** | {book.get('author', '—')} |")
|
||
if book_info.get("translator"):
|
||
lines.append(f"| **译者** | {book_info['translator']} |")
|
||
if book_info.get("publisher"):
|
||
lines.append(f"| **出版社** | {book_info['publisher']} |")
|
||
if book_info.get("category"):
|
||
lines.append(f"| **分类** | {book_info['category']} |")
|
||
if book_info.get("isbn"):
|
||
lines.append(f"| **ISBN** | {book_info['isbn']} |")
|
||
if book_info.get("publishTime"):
|
||
lines.append(f"| **出版时间** | {book_info['publishTime']} |")
|
||
|
||
# 阅读进度
|
||
pct = progress.get("progress", 0)
|
||
read_minutes = progress.get("readingTime", 0)
|
||
start_ts = progress.get("startReadingTime", 0)
|
||
update_ts = progress.get("updateTime", 0)
|
||
|
||
lines.append(f"| **阅读进度** | {pct}% |")
|
||
if read_minutes:
|
||
lines.append(f"| **阅读时长** | {format_duration(read_minutes)} |")
|
||
if start_ts:
|
||
lines.append(f"| **开始阅读** | {format_time(start_ts)} |")
|
||
if update_ts:
|
||
lines.append(f"| **最近阅读** | {format_time(update_ts)} |")
|
||
|
||
lines.append(f"| **划线数量** | {book.get('noteCount', '—')} 条 |")
|
||
lines.append("")
|
||
lines.append('</div>')
|
||
lines.append("")
|
||
|
||
# 简介
|
||
intro = book_info.get("intro", "")
|
||
if intro:
|
||
# 取前 200 字作为摘要
|
||
short_intro = intro[:200] + ("..." if len(intro) > 200 else "")
|
||
lines.append("> " + short_intro)
|
||
lines.append("")
|
||
|
||
return lines
|
||
|
||
|
||
def generate_markdown(book, bookmarks, reviews, book_info, progress):
|
||
"""生成一本书的笔记 Markdown"""
|
||
lines = []
|
||
|
||
# 书籍信息头部
|
||
lines.extend(generate_header(book, book_info, progress))
|
||
|
||
# 划线
|
||
if bookmarks:
|
||
lines.append("---")
|
||
lines.append("## 划线笔记")
|
||
lines.append("")
|
||
current_chapter = None
|
||
first = True
|
||
for bm in bookmarks:
|
||
if not first:
|
||
lines.append("---")
|
||
first = False
|
||
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
|
||
first = True
|
||
for r in reviews:
|
||
if not first:
|
||
lines.append("---")
|
||
first = False
|
||
# 按章节分组（跳过空章节名）
|
||
chapter = r.get("chapter", "").strip()
|
||
if chapter and chapter != current_chapter:
|
||
lines.append(f"### {chapter}")
|
||
lines.append("")
|
||
current_chapter = chapter
|
||
elif not chapter:
|
||
current_chapter = None
|
||
# 原文在引用框内，想法在引用框外（加粗）
|
||
if r.get("abstract"):
|
||
lines.append(f"> {r['abstract']}")
|
||
if r["time"]:
|
||
lines.append(f"> 🕐 {format_time(r['time'])}")
|
||
lines.append("")
|
||
lines.append(f"💬 **{r['content']}**")
|
||
if r["time"] and not r.get("abstract"):
|
||
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=" ")
|
||
sys.stdout.flush()
|
||
|
||
# 获取书籍信息
|
||
book_info = get_book_info(bid)
|
||
progress = get_book_progress(bid)
|
||
|
||
# 获取划线
|
||
bookmarks, _ = get_bookmarks(bid)
|
||
# 获取想法
|
||
reviews = get_reviews(bid)
|
||
|
||
if not bookmarks and not reviews:
|
||
print("无笔记")
|
||
continue
|
||
|
||
# 生成 Markdown
|
||
safe_title = title.replace("/", "_").replace("\\", "_").replace(":", "_")
|
||
safe_title = safe_title.replace("(", "（").replace(")", "）").replace("[", "【").replace("]", "】")
|
||
md = generate_markdown(book, bookmarks, reviews, book_info, progress)
|
||
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(':', '_')
|
||
safe_title = safe_title.replace('(', '（').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('noteCount', 0)} | {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()
|