"""Daily digest analyzer - fetches today's memos and generates AI analysis.""" import json import logging import os import sys import subprocess 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_daily_digest logger = logging.getLogger(__name__) TZ_BEIJING = timezone(timedelta(hours=8)) PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) def utc_to_beijing(ts_str): """Convert UTC timestamp string (ISO 8601 with Z) to Beijing time string.""" try: dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) dt_bj = dt.astimezone(TZ_BEIJING) return dt_bj.strftime("%Y-%m-%d %H:%M") except (ValueError, AttributeError): return ts_str[:16].replace("T", " ") + " (UTC?)" def read_previous_digest(date): """Read the most recent previous daily digest (if exists) for continuity.""" daily_root = get_output_dir("daily") # Try yesterday first, then go back up to 7 days for days_back in range(1, 8): prev_date = date - timedelta(days=days_back) prev_dir = os.path.join(daily_root, prev_date.strftime("%Y-%m-%d")) if not os.path.isdir(prev_dir): continue files = sorted([f for f in os.listdir(prev_dir) if f.endswith(".md")], reverse=True) if not files: continue prev_file = os.path.join(prev_dir, files[0]) with open(prev_file, "r", encoding="utf-8") as f: content = f.read() # Extract just the AI analysis section, skip frontmatter ai_section_start = content.find("## AI 分析") if ai_section_start != -1: # Find the "我的批注" section boundary annotation_start = content.find("## 我的批注", ai_section_start) if annotation_start != -1: ai_section = content[ai_section_start:annotation_start].strip() else: ai_section = content[ai_section_start:].strip() logger.info( "Found previous digest from %s (%d chars)", prev_date.strftime("%Y-%m-%d"), len(ai_section) ) return prev_date, ai_section else: # No AI section found, return whole file logger.info( "Found previous digest from %s (no AI section found, using full)", prev_date.strftime("%Y-%m-%d") ) return prev_date, content logger.info("No previous digest found in the last 7 days") return None, None # ============================================================ # System prompt for daily analysis # ============================================================ DAILY_SYSTEM_PROMPT = """你是一个私人思考伙伴。你的任务是认真阅读用户今天的每一条灵感记录，结合之前的分析，写一篇有深度的分析文章。 注意：所有时间戳都是**北京时间**（UTC+8），不是 UTC。用户在中国，记录和活动时间均以北京时间为准。 核心原则：不做缩写，不做分类表，不写空话。 具体要求： 1. **引用原文**：分析每一条灵感时，必须先引用（或精炼复述）用户的原文，让用户一眼就知道"哦我在说这个"。引用要压缩但不失原意，不能断章取义。 2. **融会贯通，而非逐条罗列**：不要"第一条...第二条...第三条..."机械堆砌。要把所有灵感当作一个整体来思考——哪几条在说同一个主题？哪几条看似无关实则互补？把它们串起来写。文章是流淌的整体，不是并列的零件。 3. **承接延续之前的分析**：你会看到上一次的分析内容。今天的分析不能只写"今天的事"，要把今天的灵感和昨天的分析结合起来——昨天讨论了什么？今天有什么进展？哪些问题有了答案？哪些问题还在延续？要让文章有跨日的时间纵深感。 4. **逐渐深入，有节奏感**：从表面现象往下挖。先指出用户说了什么，再追问为什么，再展开你的洞察。每一段都要比上一段深一层。 5. **有实质内容**：不要写"这是一个有价值的思考"这种废话。要说清楚为什么有价值，值在哪里，用户可以从这个方向挖到什么。 6. **关联要真实**：如果多条灵感确实指向同一个主题，就写一段贯通的分析。不要硬凑关联。 7. **待办要锋利**：待办事项不是"调研X""优化Y"这种万金油。要写到"具体做什么、什么时机做、做到什么程度"的颗粒度。 输出格式：纯 Markdown，不要代码块包裹，不要 JSON。 文章结构参考（不是模板，不必严格遵守）： ### 1. 今日概览 一两句话点出今天思考的主旋律，并与之前的内容形成呼应。 ### 2. 深入分析 融会贯通地写。引用原文要自然嵌入行文。注意与上次分析的承接关系。 ### 3. 待办事项 具体、可执行的事项列表。 --- 注意：即便只有1条灵感，也要写出深度。宁可写长，不可简略。""" def git_push(digest_file): """Commit and push the digest file to Gitea.""" try: # git add result = subprocess.run( ["git", "add", digest_file], cwd=PROJECT_DIR, capture_output=True, text=True, timeout=15 ) if result.returncode != 0: logger.warning("git add failed: %s", result.stderr.strip()) return False # git commit (check for changes first) result = subprocess.run( ["git", "status", "--porcelain", "ai-insights/"], cwd=PROJECT_DIR, capture_output=True, text=True, timeout=10 ) if not result.stdout.strip(): logger.info("No changes to commit") return True date_str = datetime.now(TZ_BEIJING).strftime("%Y-%m-%d %H:%M") result = subprocess.run( ["git", "commit", "-m", "daily digest " + date_str], cwd=PROJECT_DIR, capture_output=True, text=True, timeout=15 ) if result.returncode != 0 and "nothing to commit" not in result.stdout: logger.warning("git commit failed: %s", result.stderr.strip()) return False # git push result = subprocess.run( ["git", "push", "origin", "main"], cwd=PROJECT_DIR, capture_output=True, text=True, timeout=30 ) if result.returncode != 0: logger.warning("git push failed: %s", result.stderr.strip()) return False # Extract remote result line for line in result.stdout.split("\n"): if "->" in line or "remote:" in line: logger.info("Gitea push: %s", line.strip()) logger.info("Pushed to Gitea successfully") return True except subprocess.TimeoutExpired: logger.error("git push timed out") return False except Exception as e: logger.error("git push error: %s", e) return False def run(memos_client, llm_client, date=None): """Run daily digest analysis.""" date = date or datetime.now(TZ_BEIJING) 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") content = format_daily_digest(date, ai_body="今天没有记录灵感。", tags=["灵感收集器", "每日总结", "静默"], doc_type="daily-digest") date_str = date.strftime("%Y-%m-%d") daily_dir = os.path.join(get_output_dir("daily"), date_str) os.makedirs(daily_dir, exist_ok=True) now_str = datetime.now(TZ_BEIJING).strftime("%H%M%S") filename = date_str + "_digest_" + now_str + ".md" filepath = os.path.join(daily_dir, filename) with open(filepath, "w", encoding="utf-8") as f: f.write(content) logger.info("Empty digest written to %s", filepath) # Still push git_push(filepath) return filepath, 0 # Step 2: Prepare user prompt memo_texts = [] for m in memos: time_bj = utc_to_beijing(m["created_at"]) memo_texts.append("- **[" + time_bj + "]** " + m["content"].strip()) user_prompt = "以下是我今天的灵感记录：\n\n" + "\n\n".join(memo_texts) # Step 3: Include previous analysis for continuity prev_date, prev_analysis = read_previous_digest(date) if prev_analysis: user_prompt += ( "\n\n---\n\n" "以下是我上一次的分析内容（" + prev_date.strftime("%Y-%m-%d") + "），" "请结合今天的灵感一起思考，保持连续性：\n\n" + prev_analysis ) user_prompt += "\n\n---\n\n请基于以上所有素材，写一篇有深度的分析文章。" # Step 4: Call DeepSeek API raw_response = llm_client.ask( system_prompt=DAILY_SYSTEM_PROMPT, user_prompt=user_prompt, temperature=0.5 ) # Step 5: Wrap with frontmatter and write content = format_daily_digest(date, ai_body=raw_response, tags=["灵感收集器", "每日总结", "AI分析"], doc_type="daily-digest") date_str = date.strftime("%Y-%m-%d") daily_dir = os.path.join(get_output_dir("daily"), date_str) os.makedirs(daily_dir, exist_ok=True) now_str = datetime.now(TZ_BEIJING).strftime("%H%M%S") filename = date_str + "_digest_" + now_str + ".md" filepath = os.path.join(daily_dir, filename) with open(filepath, "w", encoding="utf-8") as f: f.write(content) logger.info( "Daily digest written to %s | %d memos | %d chars analysis", filepath, len(memos), len(raw_response) ) # Step 6: Auto-push to Gitea git_push(filepath) return filepath, len(memos) def main(): 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("Done: " + filepath + " (" + str(count) + " memos)") if __name__ == "__main__": main()