init: inspiration collector v1.0

analyzers/weekly_trend.py

@@ -0,0 +1,159 @@
+"""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()