diff --git a/analyzers/daily_digest.py b/analyzers/daily_digest.py index ee3fc3a..e511ed0 100644 --- a/analyzers/daily_digest.py +++ b/analyzers/daily_digest.py @@ -5,6 +5,7 @@ import logging import os import sys import subprocess +from collections import Counter from datetime import datetime, timezone, timedelta sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -33,7 +34,6 @@ def utc_to_beijing(ts_str): 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")) @@ -45,10 +45,8 @@ def read_previous_digest(date): 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() @@ -60,7 +58,6 @@ def read_previous_digest(date): ) 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") @@ -71,19 +68,28 @@ def read_previous_digest(date): # ============================================================ -# System prompt for daily analysis +# System prompt for daily analysis (v2 - tag-aware) # ============================================================ DAILY_SYSTEM_PROMPT = """你是一个私人思考伙伴。你的任务是认真阅读用户今天的每一条灵感记录，结合之前的分析，写一篇有深度的分析文章。 注意：所有时间戳都是**北京时间**（UTC+8），不是 UTC。用户在中国，记录和活动时间均以北京时间为准。 +## 关于标签 + +用户的每条记录可能带有标签（如 #培训 #交易 #生活 等）。标签是用户自己对记录的分类和上下文标注，对理解记录非常重要： + +1. **标签揭示场景**：同一条灵感，标注 #培训 和标注 #交易 的解读角度完全不同。标签告诉你用户在什么情境下产生的这个想法。 +2. **标签揭示延续性**：如果多天都有 #90后四级副职培训班 标签，说明这是一个持续性的活动/项目，分析时应体现跨日的连续思考。 +3. **同标签分组**：当多条记录共享同一标签时，它们通常是同一主题下的不同侧面，应该融会贯通地分析，而不是孤立看待。 +4. **不要忽略无标签记录**：没有标签的记录可能是随想、零散灵感，但也有其价值。 + 核心原则：不做缩写，不做分类表，不写空话。 具体要求： 1. **引用原文**：分析每一条灵感时，必须先引用（或精炼复述）用户的原文，让用户一眼就知道"哦我在说这个"。引用要压缩但不失原意，不能断章取义。 -2. **融会贯通，而非逐条罗列**：不要"第一条...第二条...第三条..."机械堆砌。要把所有灵感当作一个整体来思考——哪几条在说同一个主题？哪几条看似无关实则互补？把它们串起来写。文章是流淌的整体，不是并列的零件。 +2. **融会贯通，而非逐条罗列**：不要"第一条...第二条...第三条..."机械堆砌。要把所有灵感当作一个整体来思考——哪几条在说同一个主题？哪几条看似无关实则互补？把它们串起来写。文章是流淌的整体，不是并列的零件。**注意利用标签来识别主题聚类**。 3. **承接延续之前的分析**：你会看到上一次的分析内容。今天的分析不能只写"今天的事"，要把今天的灵感和昨天的分析结合起来——昨天讨论了什么？今天有什么进展？哪些问题有了答案？哪些问题还在延续？要让文章有跨日的时间纵深感。 @@ -110,13 +116,61 @@ DAILY_SYSTEM_PROMPT = """你是一个私人思考伙伴。你的任务是认真 --- -注意：即便只有1条灵感，也要写出深度。宁可写长，不可简略。""" +注意：即便只有1条灵感，也要写出深度。篇幅不设上限，内容完整即可。""" + + +def build_user_prompt(memos, date): + """Build user prompt for AI, with tag-aware formatting and grouping. + + Memos are grouped by shared tags so the AI can see thematic clusters. + Each memo includes its tags in the text block. + """ + # Step 1: Collect all tags across today's memos + tag_freq = MemosClient.aggregate_tags(memos) + groups = MemosClient.group_memos_by_tag(memos) + + prompt_parts = [] + + # Header: date + tag overview + date_str = date.strftime("%Y-%m-%d") + prompt_parts.append(f"以下是我 {date_str} 的灵感记录（共 {len(memos)} 条）。") + + if tag_freq: + tag_summary = "、".join(f"#{t}（{c}条）" for t, c in tag_freq.items()) + prompt_parts.append(f"\n今天的标签分布：{tag_summary}") + prompt_parts.append("\n标签说明：用户通过标签自行分类记录。同标签的记录通常属于同一主题或场景，分析时应关注标签揭示的上下文。") + + # Step 2: Group memos by tags for thematic presentation + prompt_parts.append("\n---\n") + + # Priority: present tagged memos first (grouped by tag), then untagged + sorted_tags = sorted(groups.keys(), key=lambda t: (t == "无标签", -len(groups[t]))) + + for tag in sorted_tags: + group_memos = groups[tag] + if tag == "无标签": + prompt_parts.append(f"\n## 无标签记录（{len(group_memos)}条）") + else: + prompt_parts.append(f"\n## 标签 #{tag}（{len(group_memos)}条）") + + for m in group_memos: + time_bj = utc_to_beijing(m["created_at"]) + content = m["content"].strip() + memo_tags = m.get("tags", []) + + # Show all tags on this memo (for cross-tagged memos) + tag_line = "" + if memo_tags: + tag_line = " [" + ", ".join("#" + t for t in memo_tags) + "]" + + prompt_parts.append(f"\n- **[{time_bj}]**{tag_line}\n{content}") + + return "\n".join(prompt_parts) 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, @@ -126,7 +180,6 @@ def git_push(digest_file): 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, @@ -146,7 +199,6 @@ def git_push(digest_file): logger.warning("git commit failed: %s", result.stderr.strip()) return False - # git push result = subprocess.run( ["git", "push", "origin", "main"], cwd=PROJECT_DIR, @@ -156,7 +208,6 @@ def git_push(digest_file): 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()) @@ -192,17 +243,11 @@ def run(memos_client, llm_client, date=None): 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 2: Build tag-aware user prompt + user_prompt = build_user_prompt(memos, date) # Step 3: Include previous analysis for continuity prev_date, prev_analysis = read_previous_digest(date) @@ -215,17 +260,58 @@ def run(memos_client, llm_client, date=None): + prev_analysis ) + # Step 4: Check for cross-day tag continuity + # If today has tags that appeared in previous digest, note this + tag_freq = MemosClient.aggregate_tags(memos) + today_tags = set(tag_freq.keys()) + if prev_date and today_tags: + try: + prev_tags_line = None + # Extract tags from previous digest frontmatter + daily_root = get_output_dir("daily") + prev_dir = os.path.join(daily_root, prev_date.strftime("%Y-%m-%d")) + if os.path.isdir(prev_dir): + prev_files = sorted([f for f in os.listdir(prev_dir) if f.endswith(".md")], reverse=True) + if prev_files: + with open(os.path.join(prev_dir, prev_files[0]), "r", encoding="utf-8") as f: + prev_content = f.read(500) + if "tags:" in prev_content: + prev_tags_line = prev_content.split("tags:")[1].split("\n")[0] + + if prev_tags_line: + import re + prev_tags = set(re.findall(r"'([^']+)'", prev_tags_line)) + cross_tags = today_tags & prev_tags + if cross_tags: + user_prompt += ( + "\n\n---\n\n" + "跨标签延续提示：以下标签与上一次分析（" + + prev_date.strftime("%Y-%m-%d") + + "）共享：" + + "、".join("#" + t for t in sorted(cross_tags)) + + "。这些标签代表持续性的主题，请特别关注跨日的思维连续性。" + ) + except Exception as e: + logger.debug("Cross-tag check failed: %s", e) + user_prompt += "\n\n---\n\n请基于以上所有素材，写一篇有深度的分析文章。" - # Step 4: Call DeepSeek API + # Step 5: 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") + # Step 6: Build frontmatter tags (default tags + today's user tags) + default_tags = ["灵感收集器", "每日总结", "AI分析"] + user_tags = sorted(tag_freq.keys()) + all_tags = default_tags + user_tags + + content = format_daily_digest( + date, ai_body=raw_response, + tags=all_tags, doc_type="daily-digest" + ) date_str = date.strftime("%Y-%m-%d") daily_dir = os.path.join(get_output_dir("daily"), date_str) @@ -238,11 +324,11 @@ def run(memos_client, llm_client, date=None): f.write(content) logger.info( - "Daily digest written to %s | %d memos | %d chars analysis", - filepath, len(memos), len(raw_response) + "Daily digest written to %s | %d memos | %d chars analysis | tags: %s", + filepath, len(memos), len(raw_response), ", ".join(user_tags) if user_tags else "(none)" ) - # Step 6: Auto-push to Gitea + # Step 7: Auto-push to Gitea git_push(filepath) return filepath, len(memos) diff --git a/tools/memos_client.py b/tools/memos_client.py index dff7093..b632482 100644 --- a/tools/memos_client.py +++ b/tools/memos_client.py @@ -1,12 +1,14 @@ """Memos API client - fetch memos using Connect RPC protocol.""" import logging +from collections import Counter from datetime import datetime, timezone, timedelta - -TZ_BEIJING = timezone(timedelta(hours=8)) +from re import findall import requests +TZ_BEIJING = timezone(timedelta(hours=8)) + logger = logging.getLogger(__name__) @@ -45,13 +47,11 @@ class MemosClient: def list_memos(self, days=1, page_size=100): """Fetch memos from the last N days (Beijing time reference) via Connect RPC.""" user = self.get_user_id() - # Use Beijing time as reference for "last N days" now_bj = datetime.now(TZ_BEIJING) since_bj = now_bj - timedelta(days=days) since_utc = since_bj.astimezone(timezone.utc) since = since_utc.isoformat() - # Connect RPC uses POST with JSON body payload = { "pageSize": page_size, "filter": f"creator == '{user}'", @@ -68,7 +68,6 @@ class MemosClient: memos = data.get("memos", []) logger.info("Memos API OK | fetched %d memos", len(memos)) - # Filter by time client-side (Beijing time reference) results = [] for m in memos: created = m.get("createTime", "") @@ -88,9 +87,7 @@ class MemosClient: "visibility": m.get("visibility", ""), }) - logger.info( - "Filtered to %d memos since %s", len(results), since[:10] - ) + logger.info("Filtered to %d memos since %s", len(results), since[:10]) return results def list_all_memos_from_range(self, start_date, end_date, page_size=200): @@ -110,7 +107,6 @@ class MemosClient: memos = data.get("memos", []) logger.info("Memos API OK | fetched %d memos total", len(memos)) - # Client-side date range filtering results = [] for m in memos: created = m.get("createTime", "") @@ -136,8 +132,76 @@ class MemosClient: ) return results + def list_memos_by_tag(self, tag, page_size=200): + """Fetch all memos containing a specific tag. + + Useful for retrieving memos across multiple days related to a specific + topic (e.g. a training course that spans a full week). + """ + payload = {"pageSize": page_size} + + resp = self.session.post( + f"{self.base_url}/memos.api.v1.MemoService/ListMemos", + json=payload, + timeout=15 + ) + resp.raise_for_status() + data = resp.json() + memos = data.get("memos", []) + + results = [] + for m in memos: + content_text = m.get("content", "") + tags = self._extract_tags(content_text) + if tag not in tags: + continue + + results.append({ + "id": m.get("name", "").split("/")[-1], + "content": content_text, + "created_at": m.get("createTime", ""), + "tags": tags, + "visibility": m.get("visibility", ""), + }) + + logger.info("Found %d memos with tag #%s", len(results), tag) + return results + + @staticmethod + def aggregate_tags(memos): + """Count tag frequency across a list of memos. + + Returns dict of {tag: count}, sorted by frequency descending. + """ + counter = Counter() + for m in memos: + for tag in m.get("tags", []): + counter[tag] += 1 + return dict(counter.most_common()) + + @staticmethod + def group_memos_by_tag(memos): + """Group memos by their tags. + + Returns dict of {tag: [memo_list]}, with "无标签" key for memos + without any tags. A memo can appear under multiple tags. + """ + groups = {} + untagged = [] + for m in memos: + tags = m.get("tags", []) + if not tags: + untagged.append(m) + continue + for tag in tags: + if tag not in groups: + groups[tag] = [] + groups[tag].append(m) + if untagged: + groups["无标签"] = untagged + return groups + @staticmethod def _extract_tags(content): """Extract #tags from memo content.""" - import re - return re.findall(r"#(\w[\w\-]*)", content) + return findall(r"#(\w[\w\-]*)", content)