inspiration-collector/scripts/sync_cards.py

#!/usr/bin/env python3
"""
sync_cards.py — Memos 卡片备份脚本

每30分钟由 cron 触发，将当前用户的所有 Memos 卡片备份到 cards/ 目录，
并推送到 Gitea inspiration-collector 仓库。
"""

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

MEMOS_URL = os.environ.get("MEMOS_URL", "http://localhost:5230")
MEMOS_TOKEN = os.environ.get("MEMOS_ACCESS_TOKEN", "")

TZ_BEIJING = timezone(timedelta(hours=8))

# ── Memos API 客户端 ──────────────────────────────────

class MemosClient:
    def __init__(self, base_url, token):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        })

    def list_all_memos(self):
        """Fetch ALL memos via pagination."""
        all_memos = []
        page_token = ""
        while True:
            params = {"pageSize": 100}
            if page_token:
                params["pageToken"] = page_token
            resp = self.session.post(
                f"{self.base_url}/memos.api.v1.MemoService/ListMemos",
                json=params,
                timeout=15
            )
            resp.raise_for_status()
            data = resp.json()
            memos = data.get("memos", [])
            all_memos.extend(memos)
            next_token = data.get("nextPageToken", "")
            if not next_token:
                break
            page_token = next_token
        return all_memos


def sanitize_filename(text):
    """清理文件名中的特殊字符"""
    text = re.sub(r'[<>:"/\\|?*]', "", text)
    text = re.sub(r'\s+', "_", text.strip())
    return text[:80] or "memo"


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"
        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)

    md = f"""---
title: "{first_line}"
date: {time_str}
uid: "{uid}"
source: "memos"
---

{content}
"""
    return filepath, md, date_str


def git_push(directory, message):
    """git add + commit + push"""
    try:
        subprocess.run(["git", "add", "-A"], cwd=directory, check=True,
                       capture_output=True, timeout=30)
        result = subprocess.run(
            ["git", "diff", "--cached", "--quiet"],
            cwd=directory, capture_output=True, timeout=15
        )
        if result.returncode == 0:
            return  # 无变更
        subprocess.run(
            ["git", "commit", "-m", message],
            cwd=directory, check=True, capture_output=True, timeout=30
        )
        subprocess.run(
            ["git", "push", "origin", "main"],
            cwd=directory, check=True, capture_output=True, timeout=60
        )
        print(f"[OK] 已推送: {message}")
    except subprocess.CalledProcessError as e:
        print(f"[WARN] git 操作失败: {e.stderr.decode()[:200]}")
    except Exception as e:
        print(f"[WARN] git 异常: {e}")


def main():
    if not MEMOS_TOKEN:
        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

    for memo in all_memos:
        filepath, md, date_str = memo_to_markdown(memo)
        filename = os.path.basename(filepath)
        
        # 如果文件已存在则跳过
        if filename in existing_files:
            continue
        
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(md)
        new_count += 1
        print(f"  [+] {filename}")

    if new_count > 0:
        print(f"[INFO] 新增 {new_count} 条卡片，推送至 Gitea...")
        git_push(GIT_DIR, f"cards: 备份 {new_count} 条 Memos 卡片")
    else:
        print(f"[INFO] 无新增卡片")


if __name__ == "__main__":
    main()