160 lines
4.9 KiB
Python
160 lines
4.9 KiB
Python
"""Weekly trend analyzer - identifies patterns across a week of memos."""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import sys
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
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_weekly_trend
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
WEEKLY_SYSTEM_PROMPT = """你是一个灵感趋势分析师。你的任务是对用户一周的灵感记录进行趋势分析。
|
||
|
||
请输出纯 JSON（不要代码块标记，不要额外说明）：
|
||
|
||
{
|
||
"daily_counts": {
|
||
"周一": 3,
|
||
"周二": 1
|
||
},
|
||
"categories": {
|
||
"分类名1": 5,
|
||
"分类名2": 3
|
||
},
|
||
"highlights": ["本周最有价值的灵感1", "2-3条"],
|
||
"insight": "一段关于用户本周思维模式的洞察，60字以内"
|
||
}
|
||
|
||
分类原则：根据内容自然归类，分类名不超过4个字。
|
||
亮点：选出本周最有启发性、行动性、或值得继续深挖的灵感。
|
||
洞察：识别用户本周的关注焦点变化、思维模式、或信息缺口。"""
|
||
|
||
|
||
def parse_ai_response(text):
|
||
"""Parse AI JSON response, handling code block wrappers."""
|
||
text = text.strip()
|
||
if text.startswith("```"):
|
||
lines = text.split("\n")
|
||
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 weekly trend analysis."""
|
||
date = date or datetime.now(timezone.utc)
|
||
# Calculate week start (Monday) and end (Sunday)
|
||
weekday = date.weekday() # 0=Monday
|
||
week_start = date - timedelta(days=weekday + 7) # Previous Monday
|
||
week_end = week_start + timedelta(days=6) # Sunday
|
||
|
||
logger.info(
|
||
"Weekly trend started for W%s (%s - %s)",
|
||
week_start.isocalendar()[1],
|
||
week_start.strftime("%Y-%m-%d"),
|
||
week_end.strftime("%Y-%m-%d")
|
||
)
|
||
|
||
# Fetch week's memos
|
||
memos = memos_client.list_all_memos_from_range(week_start, week_end)
|
||
|
||
if not memos:
|
||
logger.info("No memos this week, skipping")
|
||
stats = {"total": 0, "daily_counts": {}, "categories": {}}
|
||
content = format_weekly_trend(week_start, week_end, stats, [], "")
|
||
output_dir = get_output_dir("weekly")
|
||
filename = f"{week_start.strftime('W%Y-%m-%d')}_trend.md"
|
||
filepath = os.path.join(output_dir, filename)
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
return filepath, 0
|
||
|
||
# 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"分析以下本周（{week_start.strftime('%m/%d')} - {week_end.strftime('%m/%d')}）"
|
||
f"共 {len(memos)} 条灵感记录的趋势：\n\n"
|
||
+ "\n".join(memo_texts)
|
||
)
|
||
|
||
# Call API
|
||
raw_response = llm_client.ask(
|
||
system_prompt=WEEKLY_SYSTEM_PROMPT,
|
||
user_prompt=user_prompt,
|
||
temperature=0.3
|
||
)
|
||
|
||
# Parse
|
||
try:
|
||
data = parse_ai_response(raw_response)
|
||
except (json.JSONDecodeError, KeyError) as e:
|
||
logger.error("Failed to parse AI response: %s", e)
|
||
data = {"daily_counts": {}, "categories": {}, "highlights": [], "insight": ""}
|
||
|
||
stats = {
|
||
"total": len(memos),
|
||
"daily_counts": data.get("daily_counts", {}),
|
||
"categories": data.get("categories", {}),
|
||
}
|
||
highlights = data.get("highlights", [])
|
||
insight = data.get("insight", "")
|
||
|
||
# Format output
|
||
content = format_weekly_trend(week_start, week_end, stats, highlights, insight)
|
||
output_dir = get_output_dir("weekly")
|
||
filename = f"{week_start.strftime('W%Y-%m-%d')}_trend.md"
|
||
filepath = os.path.join(output_dir, filename)
|
||
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
|
||
logger.info(
|
||
"Weekly trend written to %s | %d memos | %d categories",
|
||
filepath, len(memos), len(stats["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()
|