fxy/inspiration-collector
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cards: 备份 9 条 Memos 卡片

Browse Source
This commit is contained in:
Beast
2026-06-16 00:33:14 +08:00
parent 8d5ed081f8
commit 49fa1e8b08
10 changed files with 157 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
scripts/sync_cards.py
View File

@ -1,21 +1,21 @@
#!/usr/bin/env python3
"""
sync_cards.py — Memos 卡片备份脚本
sync_cards.py — Memos 卡片备份脚本（基于 uid 去重）
每30分钟由 cron 触发，将当前用户的所有 Memos 卡片备份到 cards/ 目录，
并推送到 Gitea inspiration-collector 仓库。
去重方式：按 memo.uid 比对，不重复写入。
"""
import os
import sys
import json
import re
import subprocess
from datetime import datetime, timezone, timedelta
import requests
# ── 配置 ──────────────────────────────────────────────
# ── 配置 ──
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CARDS_DIR = os.path.join(BASE_DIR, "cards")
GIT_DIR = BASE_DIR
@ -25,7 +25,8 @@ MEMOS_TOKEN = os.environ.get("MEMOS_ACCESS_TOKEN", "")
TZ_BEIJING = timezone(timedelta(hours=8))
# ── Memos API 客户端 ──────────────────────────────────
KNOWN_TAGS = {"#功能完善", "#90后四级副职培训班", "#灵感收集器"}
class MemosClient:
def __init__(self, base_url, token):
@ -37,7 +38,6 @@ class MemosClient:
})
def list_all_memos(self):
"""Fetch ALL memos via pagination."""
all_memos = []
page_token = ""
while True:
@ -46,13 +46,11 @@ class MemosClient:
params["pageToken"] = page_token
resp = self.session.post(
f"{self.base_url}/memos.api.v1.MemoService/ListMemos",
json=params,
timeout=15
json=params, timeout=15
)
resp.raise_for_status()
data = resp.json()
memos = data.get("memos", [])
all_memos.extend(memos)
all_memos.extend(data.get("memos", []))
next_token = data.get("nextPageToken", "")
if not next_token:
break
@ -61,33 +59,56 @@ class MemosClient:
def sanitize_filename(text):
"""清理文件名中的特殊字符"""
text = re.sub(r'[<>:"/\\|?*]', "", text)
text = re.sub(r'\s+', "_", text.strip())
return text[:80] or "memo"
def extract_tags(content):
"""提取内容中的 #标签 """
return [w for w in content.split() if w.startswith("#") and w in KNOWN_TAGS]
def load_existing_uids(cards_dir):
"""扫描现有卡片文件，收集所有 uid + 文件名作为去重依据"""
uids = set()
fnames = set()
if not os.path.isdir(cards_dir):
return uids, fnames
for fname in os.listdir(cards_dir):
if not fname.endswith(".md"):
continue
fnames.add(fname)
fpath = os.path.join(cards_dir, fname)
try:
with open(fpath, "r", encoding="utf-8") as f:
head = f.read(500)
# 匹配 uid: "xxx" 或 id: xxx 或 uid: xxx（无引号）
m = re.search(r'^(?:uid|id):\s*"?([^"\n]+)"?', head, re.MULTILINE)
if m:
val = m.group(1).strip()
if val:
uids.add(val)
except Exception:
continue
return uids, fnames
def memo_to_markdown(memo):
"""将 Memo API 返回的对象转为 Markdown 文件内容"""
uid = memo.get("uid", "")
content = memo.get("content", "")
create_time = memo.get("createTime", "")
# 解析时间
try:
dt = datetime.fromisoformat(create_time.replace("Z", "+00:00"))
dt_bj = dt.astimezone(TZ_BEIJING)
date_str = dt_bj.strftime("%Y-%m-%d")
time_str = dt_bj.strftime("%Y-%m-%dT%H:%M:%S+08:00")
file_ts = dt_bj.strftime("%Y%m%d%H%M%S")
except:
date_str = "unknown"
except Exception:
time_str = create_time
file_ts = uid[:14] if uid else "unknown"
# 提取标题（第一行）
first_line = content.strip().split("\n")[0][:60]
filename = f"{file_ts}_{sanitize_filename(first_line)}.md"
filepath = os.path.join(CARDS_DIR, filename)
@ -100,11 +121,10 @@ source: "memos"
{content}
"""
return filepath, md, date_str
return filepath, md
def git_push(directory, message):
"""git add + commit + push"""
try:
subprocess.run(["git", "add", "-A"], cwd=directory, check=True,
capture_output=True, timeout=30)
@ -113,7 +133,7 @@ def git_push(directory, message):
cwd=directory, capture_output=True, timeout=15
)
if result.returncode == 0:
return # 无变更
return
subprocess.run(
["git", "commit", "-m", message],
cwd=directory, check=True, capture_output=True, timeout=30
@ -134,34 +154,33 @@ def main():
print("[ERROR] MEMOS_ACCESS_TOKEN 未设置")
sys.exit(1)
client = MemosClient(MEMOS_URL, MEMOS_TOKEN)
print(f"[INFO] 正在拉取 Memos...")
try:
all_memos = client.list_all_memos()
except Exception as e:
print(f"[ERROR] 拉取 Memos 失败: {e}")
sys.exit(1)
print(f"[INFO] 共获取 {len(all_memos)} 条卡片")
os.makedirs(CARDS_DIR, exist_ok=True)
existing_files = set(os.listdir(CARDS_DIR))
new_count = 0
# 加载已有 uid + 文件名
existing_uids, existing_fnames = load_existing_uids(CARDS_DIR)
print(f"[INFO] 已有 {len(existing_fnames)} 个文件，{len(existing_uids)} 个 uid")
client = MemosClient(MEMOS_URL, MEMOS_TOKEN)
print(f"[INFO] 正在从 Memos 拉取...")
all_memos = client.list_all_memos()
print(f"[INFO] 共获取 {len(all_memos)} 条卡片")
new_count = 0
for memo in all_memos:
filepath, md, date_str = memo_to_markdown(memo)
filename = os.path.basename(filepath)
# 如果文件已存在则跳过
if filename in existing_files:
continue
uid = memo.get("uid", "")
if uid and uid in existing_uids:
continue # uid 已存在，跳过
filepath, md = memo_to_markdown(memo)
fname = os.path.basename(filepath)
if fname in existing_fnames:
continue # 文件名已存在，跳过（兼容旧版无 uid 的卡片）
with open(filepath, "w", encoding="utf-8") as f:
f.write(md)
new_count += 1
print(f" [+] {filename}")
print(f" [+] {fname}")
if uid:
existing_uids.add(uid)
existing_fnames.add(fname)
if new_count > 0:
print(f"[INFO] 新增 {new_count} 条卡片，推送至 Gitea...")