368 lines
12 KiB
Python
368 lines
12 KiB
Python
"""Weekly digest synthesizer - reads all daily digests and generates a week-level analysis."""
|
||
|
||
import logging
|
||
import os
|
||
import subprocess
|
||
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
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
TZ_BEIJING = timezone(timedelta(hours=8))
|
||
PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
|
||
WEEKLY_SYSTEM_PROMPT = """你是一个私人思考伙伴。你的任务是认真阅读用户过去一周的7份日报，写一篇有深度的周度分析报告。
|
||
|
||
## 核心原则
|
||
|
||
你读的是日报——不是原始灵感记录，而是已经经过每日分析提炼的内容。你的工作是：
|
||
|
||
1. **提炼跨日主线**：7份日报各自都有"今日概览"——把它们串起来。这周的主旋律是什么？从周一到现在，思考的焦点如何演变？
|
||
|
||
2. **追踪延续与断裂**：日报中反复出现的标签（如 #浙江培训 #交易 #灵感收集器）代表持续性的主题。哪些主题从头贯穿到尾？哪些在中间消失了？为什么？
|
||
|
||
3. **识别本周的关键跃迁**：在7天的思考中，哪一天出现了认知上的质变？哪一个洞察是本周最核心的收获？
|
||
|
||
4. **不逐日罗列**：不要"周一...周二...周三..."。要融会贯通，把7天当作一个整体来写。按主题组织，不是按时间组织。
|
||
|
||
5. **深度优先**：每周分析应该比每日分析更深一层。如果日报是在追问"今天发生了什么"，周报应该回答"这一周意味着什么"。
|
||
|
||
## 输出格式
|
||
|
||
纯 Markdown，不要代码块包裹。
|
||
|
||
结构参考：
|
||
|
||
### 1. 本周概览
|
||
一两句话点出这周的底色，并指出最核心的1-2条暗线（可能不是最显眼的，但是对用户最具启发价值的）。
|
||
|
||
### 2. 深入分析
|
||
按主题组织，融会贯通地写。可以包含但不限于：
|
||
- 持续性主题的演变轨迹
|
||
- 认知跃迁的关键时刻
|
||
- 看似无关事件之间的深层关联
|
||
- 本周形成的新规则、新认知、新能力
|
||
- 与用户长期目标/身份认同的关系
|
||
|
||
引用日报内容要自然嵌入行文，让用户知道你在引用哪一天的观察。
|
||
|
||
### 3. 待办事项
|
||
基于本周分析，提出需要延续到下周的具体行动。要写到"什么时机做什么做到什么程度"的颗粒度。
|
||
|
||
---
|
||
|
||
注意：即便7份日报中有些比较简短，也要从整体中挖掘深度。篇幅不设上限。"""
|
||
|
||
|
||
def read_week_digests(week_start, week_end):
|
||
"""Read all daily digest files from week_start to week_end.
|
||
|
||
Returns list of (date_str, digest_text, tags) tuples.
|
||
"""
|
||
daily_root = get_output_dir("daily")
|
||
digests = []
|
||
|
||
current = week_start
|
||
while current <= week_end:
|
||
date_str = current.strftime("%Y-%m-%d")
|
||
daily_dir = os.path.join(daily_root, date_str)
|
||
|
||
if os.path.isdir(daily_dir):
|
||
# Find the latest digest file for this day
|
||
files = sorted([
|
||
f for f in os.listdir(daily_dir)
|
||
if f.endswith(".md") and "digest" in f and not f.endswith(".bak.md")
|
||
], reverse=True)
|
||
|
||
if files:
|
||
filepath = os.path.join(daily_dir, files[0])
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
|
||
# Extract AI analysis section
|
||
ai_start = content.find("## AI 分析")
|
||
annotation_start = content.find("## 我的批注", ai_start) if ai_start != -1 else -1
|
||
|
||
if ai_start != -1 and annotation_start != -1:
|
||
ai_body = content[ai_start:annotation_start].strip()
|
||
elif ai_start != -1:
|
||
ai_body = content[ai_start:].strip()
|
||
else:
|
||
ai_body = content
|
||
|
||
# Extract tags from frontmatter
|
||
tags = []
|
||
if "tags:" in content[:500]:
|
||
tag_line = content.split("tags:")[1].split("\n")[0]
|
||
import re
|
||
tags = re.findall(r"'([^']+)'", tag_line)
|
||
|
||
digests.append((date_str, ai_body, tags))
|
||
logger.info(
|
||
"Read digest %s: %d chars, tags=%s",
|
||
date_str, len(ai_body), ", ".join(tags[:5])
|
||
)
|
||
else:
|
||
logger.info("No digest file found for %s (skipped)", date_str)
|
||
else:
|
||
logger.info("No daily dir for %s (skipped)", date_str)
|
||
|
||
current += timedelta(days=1)
|
||
|
||
return digests
|
||
|
||
|
||
def build_weekly_prompt(digests):
|
||
"""Build the user prompt containing all 7 daily digest bodies."""
|
||
|
||
if not digests:
|
||
return None
|
||
|
||
# Extract week range
|
||
first_date = digests[0][0]
|
||
last_date = digests[-1][0]
|
||
|
||
prompt_parts = [
|
||
f"以下是我从 {first_date} 到 {last_date} 共7天的日报。",
|
||
f"共 {len(digests)} 份日报。",
|
||
"",
|
||
"请基于这些日报内容，写一篇周度分析报告。"
|
||
]
|
||
|
||
# Collect all unique tags across the week
|
||
all_tags = set()
|
||
for _, _, tags in digests:
|
||
all_tags.update(tags)
|
||
if all_tags:
|
||
tag_str = "、".join(f"#{t}" for t in sorted(all_tags))
|
||
prompt_parts.append(f"\n本周出现的标签：{tag_str}")
|
||
|
||
prompt_parts.append("\n---\n")
|
||
|
||
for date_str, body, tags in digests:
|
||
# Show a weekday indicator
|
||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||
weekday = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][dt.weekday()]
|
||
tag_hint = ""
|
||
if tags:
|
||
important_tags = [t for t in tags if t not in ("灵感收集器", "每日总结", "AI分析")]
|
||
if important_tags:
|
||
tag_hint = " | 标签：" + ", ".join(f"#{t}" for t in important_tags[:5])
|
||
|
||
prompt_parts.append(f"\n## {date_str}（{weekday}）{tag_hint}\n")
|
||
prompt_parts.append(body)
|
||
prompt_parts.append("\n")
|
||
|
||
prompt_parts.append("\n---\n")
|
||
prompt_parts.append("请基于以上所有日报内容，写一篇有深度的周度分析报告。")
|
||
|
||
return "\n".join(prompt_parts)
|
||
|
||
|
||
def format_weekly_digest(week_start, week_end, ai_body, tags=None):
|
||
"""Format weekly digest markdown with frontmatter, matching daily digest format."""
|
||
week_start_str = week_start.strftime("%Y-%m-%d")
|
||
week_end_str = week_end.strftime("%m/%d")
|
||
week_number = week_start.isocalendar()[1]
|
||
start_month = week_start.strftime("%m/%d")
|
||
|
||
char_count = len(ai_body.replace(" ", "").replace("\n", ""))
|
||
read_min = max(1, round(char_count / 300))
|
||
|
||
if tags is None:
|
||
tags = ["灵感收集器", "每周总结"]
|
||
tag_str = ", ".join(["'" + t + "'" for t in tags])
|
||
|
||
lines = []
|
||
lines.append("---")
|
||
lines.append(f"date: {week_start.strftime('%Y')}-W{week_number}")
|
||
lines.append("type: weekly-digest")
|
||
lines.append(f"tags: [{tag_str}]")
|
||
lines.append("---")
|
||
lines.append("")
|
||
lines.append(f"# 第{week_number}周 · 周度总结（{start_month} - {week_end_str}）")
|
||
lines.append("")
|
||
lines.append(f"> 本文共 **{char_count}** 字 · 预计阅读 **{read_min}** 分钟")
|
||
lines.append("")
|
||
lines.append("> 基于本周7份日报综合生成。")
|
||
lines.append("")
|
||
lines.append("## AI 分析（自动生成，请勿编辑）")
|
||
lines.append("")
|
||
lines.append(ai_body)
|
||
lines.append("")
|
||
lines.append("---")
|
||
lines.append("")
|
||
lines.append("## 我的批注")
|
||
lines.append("")
|
||
lines.append("> *在这里写下你的想法、质疑、补充*")
|
||
lines.append("")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def git_push_weekly(filepath):
|
||
"""Commit and push the weekly digest to Gitea."""
|
||
try:
|
||
result = subprocess.run(
|
||
["git", "add", filepath],
|
||
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
|
||
|
||
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
|
||
|
||
week_num = datetime.now(TZ_BEIJING).isocalendar()[1]
|
||
result = subprocess.run(
|
||
["git", "commit", "-m", f"weekly digest W{week_num}"],
|
||
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
|
||
|
||
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
|
||
|
||
for line in result.stdout.split("\n"):
|
||
if "->" in line or "remote:" in line:
|
||
logger.info("Gitea push: %s", line.strip())
|
||
logger.info("Pushed weekly digest to Gitea")
|
||
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(llm_client, date=None):
|
||
"""Run weekly digest synthesis.
|
||
|
||
Reads all daily digests from the past 7 days (Mon-Sun), synthesizes
|
||
a weekly report via DeepSeek API, and saves to ai-insights/weekly/.
|
||
"""
|
||
date = date or datetime.now(TZ_BEIJING)
|
||
|
||
# Calculate week range (Monday to Sunday)
|
||
weekday = date.weekday() # 0=Monday, 6=Sunday
|
||
week_start = date - timedelta(days=weekday) # This Monday
|
||
week_end = week_start + timedelta(days=6) # This Sunday
|
||
|
||
logger.info(
|
||
"Weekly digest started for W%d (%s - %s)",
|
||
week_start.isocalendar()[1],
|
||
week_start.strftime("%Y-%m-%d"),
|
||
week_end.strftime("%Y-%m-%d")
|
||
)
|
||
|
||
# Read all daily digests
|
||
digests = read_week_digests(week_start, week_end)
|
||
|
||
if not digests:
|
||
logger.info("No daily digests found for this week, skipping")
|
||
return None, 0
|
||
|
||
logger.info("Found %d daily digests to synthesize", len(digests))
|
||
|
||
# Build prompt
|
||
user_prompt = build_weekly_prompt(digests)
|
||
if user_prompt is None:
|
||
logger.error("Failed to build weekly prompt")
|
||
return None, 0
|
||
|
||
# Call DeepSeek API
|
||
raw_response = llm_client.ask(
|
||
system_prompt=WEEKLY_SYSTEM_PROMPT,
|
||
user_prompt=user_prompt,
|
||
temperature=0.5
|
||
)
|
||
|
||
if not raw_response:
|
||
logger.error("Empty response from DeepSeek API")
|
||
return None, 0
|
||
|
||
# Collect all unique tags from the week
|
||
all_tags = set()
|
||
for _, _, tags in digests:
|
||
all_tags.update(tags)
|
||
# Add default weekly tags
|
||
default_tags = ["灵感收集器", "每周总结"]
|
||
weekly_tags = default_tags + sorted(all_tags)
|
||
|
||
# Format output
|
||
content = format_weekly_digest(week_start, week_end, raw_response, weekly_tags)
|
||
|
||
# Save to weekly directory
|
||
output_dir = get_output_dir("weekly")
|
||
week_start_str = week_start.strftime("%Y-%m-%d")
|
||
filename = f"W{week_start_str}_digest.md"
|
||
filepath = os.path.join(output_dir, filename)
|
||
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
|
||
logger.info(
|
||
"Weekly digest written to %s | %d digests | %d chars analysis | %d tags",
|
||
filepath, len(digests), len(raw_response), len(weekly_tags)
|
||
)
|
||
|
||
# Auto-push to Gitea
|
||
git_push_weekly(filepath)
|
||
|
||
return filepath, len(digests)
|
||
|
||
|
||
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)
|
||
|
||
llm_client = DeepSeekClient(
|
||
api_key=secrets["deepseek_api_key"],
|
||
model=secrets.get("deepseek_model", "deepseek-chat")
|
||
)
|
||
|
||
filepath, count = run(llm_client)
|
||
if filepath:
|
||
print(f"Done: {filepath} ({count} digests synthesized)")
|
||
else:
|
||
print("No digests found, skipped")
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|