169 lines
5.0 KiB
Python
169 lines
5.0 KiB
Python
"""Daily digest analyzer - fetches today's memos and generates AI summary."""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
|
||
# Add parent dir to path for local imports
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from tools.config import load_secrets, get_output_dir
|
||
from tools.llm import DeepSeekClient
|
||
from tools.memos_client import MemosClient
|
||
from tools.formatter import format_daily_digest
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# System prompt for daily digest
|
||
DAILY_SYSTEM_PROMPT = """你是一个灵感整理助手。你的任务是将用户零散的灵感记录进行智能整理。
|
||
|
||
请严格按照要求整理，输出纯 JSON（不要代码块标记，不要额外说明）：
|
||
|
||
{
|
||
"summary": "一段简短的今日灵感总结，50字以内",
|
||
"categories": {
|
||
"分类名1": ["具体灵感项1", "具体灵感项2"],
|
||
"分类名2": ["具体灵感项3"]
|
||
},
|
||
"connections": ["跨主题关联发现1", "2-3条"],
|
||
"todos": ["待办事项1", "2-3条"]
|
||
}
|
||
|
||
分类原则：根据内容自然归类，分类名不超过4个字。
|
||
关联发现：识别不同灵感之间的关联、冲突或可串联的主题。
|
||
待办：从灵感中提取可执行的事项。"""
|
||
|
||
|
||
def parse_ai_response(text):
|
||
"""Parse AI response, handling both pure JSON and code-block wrapped JSON."""
|
||
text = text.strip()
|
||
# Remove markdown code block wrappers if present
|
||
if text.startswith("```"):
|
||
lines = text.split("\n")
|
||
# Remove first and last ``` lines
|
||
if lines[0].startswith("```"):
|
||
lines = lines[1:]
|
||
if lines and lines[-1].strip() == "```":
|
||
lines = lines[:-1]
|
||
text = "\n".join(lines).strip()
|
||
|
||
return json.loads(text)
|
||
|
||
|
||
def run(memos_client, llm_client, date=None):
|
||
"""Run daily digest analysis."""
|
||
date = date or datetime.now(timezone.utc)
|
||
|
||
logger.info("Daily digest started for %s", date.strftime("%Y-%m-%d"))
|
||
|
||
# Step 1: Fetch today's memos
|
||
memos = memos_client.list_memos(days=1)
|
||
|
||
if not memos:
|
||
logger.info("No memos today, skipping")
|
||
# Still generate a minimal file
|
||
content = format_daily_digest(
|
||
date=date,
|
||
categories={},
|
||
summary="今天没有记录灵感。",
|
||
connections=[],
|
||
todos=[]
|
||
)
|
||
output_dir = get_output_dir("daily")
|
||
filename = f"{date.strftime('%Y-%m-%d')}_digest.md"
|
||
filepath = os.path.join(output_dir, filename)
|
||
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
|
||
logger.info("Empty digest written to %s", filepath)
|
||
return filepath, 0
|
||
|
||
# Step 2: Prepare prompt
|
||
memo_texts = []
|
||
for m in memos:
|
||
time_str = m["created_at"][:16].replace("T", " ")
|
||
memo_texts.append(f"[{time_str}] {m['content']}")
|
||
|
||
user_prompt = f"请分析以下灵感记录：\n\n" + "\n".join(memo_texts)
|
||
|
||
# Step 3: Call DeepSeek API
|
||
raw_response = llm_client.ask(
|
||
system_prompt=DAILY_SYSTEM_PROMPT,
|
||
user_prompt=user_prompt,
|
||
temperature=0.3
|
||
)
|
||
|
||
# Step 4: Parse and format
|
||
try:
|
||
data = parse_ai_response(raw_response)
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
logger.error("Failed to parse AI response: %s", e)
|
||
logger.error("Raw response: %s", raw_response[:200])
|
||
data = {
|
||
"summary": f"AI 解析失败，请查看原始 Memos。",
|
||
"categories": {"未分类": [m["content"] for m in memos]},
|
||
"connections": [],
|
||
"todos": []
|
||
}
|
||
|
||
categories = data.get("categories", {})
|
||
connections = data.get("connections", [])
|
||
todos = data.get("todos", [])
|
||
summary = data.get("summary", "")
|
||
|
||
# Step 5: Format and write output
|
||
content = format_daily_digest(
|
||
date=date,
|
||
categories=categories,
|
||
summary=summary,
|
||
connections=connections,
|
||
todos=todos
|
||
)
|
||
|
||
output_dir = get_output_dir("daily")
|
||
filename = f"{date.strftime('%Y-%m-%d')}_digest.md"
|
||
filepath = os.path.join(output_dir, filename)
|
||
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
|
||
logger.info(
|
||
"Daily digest written to %s | %d memos | %d categories",
|
||
filepath, len(memos), len(categories)
|
||
)
|
||
return filepath, len(memos)
|
||
|
||
|
||
def main():
|
||
"""CLI entry point."""
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
|
||
)
|
||
|
||
try:
|
||
secrets = load_secrets()
|
||
except FileNotFoundError as e:
|
||
print(e)
|
||
sys.exit(1)
|
||
|
||
memos_client = MemosClient(
|
||
base_url=secrets.get("memos_url", "http://localhost:5230"),
|
||
access_token=secrets["memos_token"]
|
||
)
|
||
|
||
llm_client = DeepSeekClient(
|
||
api_key=secrets["deepseek_api_key"],
|
||
model=secrets.get("deepseek_model", "deepseek-chat")
|
||
)
|
||
|
||
filepath, count = run(memos_client, llm_client)
|
||
print(f"Done: {filepath} ({count} memos)")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|