yumao
ce04d5a342
提取 collect/formatters/themes 等 pipeline 模块，新增 wecom 分条、RSS、delta、Agent 工作流测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
37 lines
1.0 KiB
Python
37 lines
1.0 KiB
Python
"""Skills 快照读写。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from daily.config import CACHE_DIR, SNAPSHOT_FILE
|
|
|
|
|
|
def skill_id(item: dict[str, Any]) -> str:
|
|
return str(item.get("id") or f"{item.get('source')}/{item.get('title')}")
|
|
|
|
|
|
def load_snapshot() -> set[str]:
|
|
if not SNAPSHOT_FILE.exists():
|
|
return set()
|
|
try:
|
|
data = json.loads(SNAPSHOT_FILE.read_text(encoding="utf-8"))
|
|
return set(str(x) for x in (data.get("skill_ids") or []))
|
|
except (OSError, json.JSONDecodeError):
|
|
return set()
|
|
|
|
|
|
def save_snapshot(feed: dict[str, Any], date_str: str) -> None:
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
ids: list[str] = []
|
|
for board in ("topTrending", "topHot"):
|
|
for item in feed.get(board, [])[:20]:
|
|
sid = skill_id(item)
|
|
if sid not in ids:
|
|
ids.append(sid)
|
|
SNAPSHOT_FILE.write_text(
|
|
json.dumps({"date": date_str, "skill_ids": ids}, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|