330 lines
12 KiB
Python
330 lines
12 KiB
Python
"""Weekly reading report synthesizer - reads daily reading reports and generates weekly analysis."""
|
||
|
||
import logging
|
||
import os
|
||
import re
|
||
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
|
||
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__)))
|
||
# Reading reports live in weread-notes repo, sibling to inspiration-collector
|
||
WEREAD_DIR = os.path.join(os.path.dirname(PROJECT_DIR), "weread-notes")
|
||
|
||
READING_WEEKLY_SYSTEM_PROMPT = """你是一个私人阅读分析师。你的任务是阅读用户过去一周的每日阅读报告，生成一篇有深度的周度阅读分析。
|
||
|
||
## 核心原则
|
||
|
||
1. **统计是基础，分析是核心**：先给出本周阅读的量化概览，然后深入分析阅读行为背后的认知模式。
|
||
|
||
2. **追踪阅读主题演变**：本周读了哪些书？它们之间有什么关联？阅读焦点在周初和周尾有什么变化？
|
||
|
||
3. **识别认知跃迁**：本周有没有哪本书/哪个概念让用户的思考发生了质变？在哪一天？是什么触发的？
|
||
|
||
4. **评估阅读质量**：不是读了多久、划了多少线，而是划线背后反映了什么思考模式？是精读还是泛读？是验证已有认知还是拓展新领域？
|
||
|
||
5. **建议下周方向**：基于本周的阅读轨迹，下周建议继续深耕哪本书？或者是否需要补充某个缺失的领域？
|
||
|
||
## 输出格式
|
||
|
||
纯 Markdown，不要代码块包裹。
|
||
|
||
### 1. 本周统计
|
||
|
||
用表格呈现：
|
||
| 维度 | 数值 |
|
||
|:---|:---|
|
||
| 阅读天数 | X/7 |
|
||
| 累计阅读时长 | X小时Y分钟 |
|
||
| 日均阅读时长 | X分钟 |
|
||
| 本周在读书籍 | X本 |
|
||
| 本周新增划线 | X条 |
|
||
| 本周新增想法 | X条 |
|
||
| 最投入的书 | 《XXX》· X小时 |
|
||
| 划线最多的书 | 《XXX》· X条 |
|
||
|
||
### 2. 阅读主题追踪
|
||
|
||
本周的阅读覆盖了哪些主题领域（如投资、哲学、历史、技术……）？每个主题在读哪些书？占了多少时间？哪些主题之间有交叉？
|
||
|
||
### 3. 深入分析
|
||
|
||
按主题组织，融会贯通地写：
|
||
- 本周阅读的主线是什么（不是逐日罗列，是提炼共同指向）
|
||
- 认知跃迁的关键时刻（哪天的什么划线让你意识到用户在想什么）
|
||
- 跨书关联（本周读的几本书之间有没有隐含的对话）
|
||
- 阅读节奏评价（是持续深耕一本书，还是多线并行？深度如何？）
|
||
- 本周最值得注意的一个阅读习惯（比如：划线密度突然变化、某个时段的阅读特别集中）
|
||
|
||
### 4. 本周金句
|
||
|
||
从本周的划线和批注中，选出 3-5 条最具代表性的原文或用户批注。每条注明来自哪本书、哪一天。
|
||
|
||
### 5. 下周阅读建议
|
||
|
||
基于本周轨迹，给出 1-3 条具体的阅读方向建议。要写到"继续读哪本书的哪个章节"或"补充哪类书"的颗粒度。
|
||
|
||
---
|
||
|
||
## 补充说明
|
||
|
||
- 如果某天没有阅读报告（用户当天没读书），在统计中注明，但分析中跳过。
|
||
- 统计数字必须从日报原文中提取，不要编造。
|
||
- 分析要有观点，不要求和稀泥。如果用户本周阅读质量明显下降，直接指出。
|
||
- 篇幅不设上限，深度优先。"""
|
||
|
||
|
||
def extract_stats_from_daily(daily_text, date_str):
|
||
"""Extract reading statistics from a daily report.
|
||
|
||
Returns dict with: date, books, highlights_count, notes_count,
|
||
reading_time_minutes, book_list
|
||
"""
|
||
stats = {
|
||
"date": date_str,
|
||
"has_data": False,
|
||
"reading_time_minutes": 0,
|
||
"highlights_count": 0,
|
||
"notes_count": 0,
|
||
"books": [],
|
||
}
|
||
|
||
if not daily_text or len(daily_text.strip()) < 50:
|
||
return stats
|
||
|
||
# Check if there was any reading today
|
||
if "今日没有阅读记录" in daily_text or "暂无阅读" in daily_text:
|
||
return stats
|
||
|
||
stats["has_data"] = True
|
||
|
||
# Extract reading time - look for patterns like "X小时Y分钟", "X分钟"
|
||
time_patterns = [
|
||
r'(\d+)\s*小时\s*(\d+)\s*分钟',
|
||
r'(\d+)\s*小时',
|
||
r'(\d+)\s*分钟',
|
||
]
|
||
for pattern in time_patterns:
|
||
match = re.search(pattern, daily_text)
|
||
if match:
|
||
if len(match.groups()) == 2:
|
||
stats["reading_time_minutes"] = int(match.group(1)) * 60 + int(match.group(2))
|
||
elif "小时" in pattern:
|
||
stats["reading_time_minutes"] = int(match.group(1)) * 60
|
||
else:
|
||
stats["reading_time_minutes"] = int(match.group(1))
|
||
break
|
||
|
||
# Extract highlight/note counts - look for patterns
|
||
highlight_match = re.search(r'划线.*?(\d+)\s*条', daily_text)
|
||
if highlight_match:
|
||
stats["highlights_count"] = int(highlight_match.group(1))
|
||
|
||
notes_match = re.search(r'想法.*?(\d+)\s*条', daily_text)
|
||
if notes_match:
|
||
stats["notes_count"] = int(notes_match.group(1))
|
||
|
||
# Extract book titles - look for 《书名》 patterns
|
||
book_titles = re.findall(r'《([^》]+)》', daily_text)
|
||
# De-duplicate while preserving order
|
||
seen = set()
|
||
unique_books = []
|
||
for title in book_titles:
|
||
if title not in seen and len(title) > 1:
|
||
seen.add(title)
|
||
unique_books.append(title)
|
||
stats["books"] = unique_books[:10] # Top 10 to avoid noise
|
||
|
||
return stats
|
||
|
||
|
||
def read_week_daily_reports(week_start, week_end):
|
||
"""Read all daily reading reports from week_start to week_end.
|
||
|
||
Args:
|
||
week_start, week_end: datetime objects in Beijing time
|
||
|
||
Returns:
|
||
tuple of (accumulated_stats, full_texts)
|
||
"""
|
||
daily_dir = os.path.join(WEREAD_DIR, "daily")
|
||
all_stats = []
|
||
all_texts = []
|
||
|
||
current = week_start
|
||
while current <= week_end:
|
||
date_str = current.strftime("%Y-%m-%d")
|
||
# Daily reports are named: 每日阅读_YYYY-MM-DD.md
|
||
filename = f"每日阅读_{date_str}.md"
|
||
parts = date_str.split("-")
|
||
filepath = os.path.join(daily_dir, parts[0], parts[1], parts[2], filename)
|
||
|
||
if os.path.exists(filepath):
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
text = f.read()
|
||
stats = extract_stats_from_daily(text, date_str)
|
||
all_stats.append(stats)
|
||
all_texts.append((date_str, text))
|
||
logger.info(f" ✓ {date_str}: {len(text)} chars, {len(stats['books'])} books")
|
||
else:
|
||
all_stats.append({"date": date_str, "has_data": False, "books": []})
|
||
logger.info(f" ✗ {date_str}: no report")
|
||
|
||
current += timedelta(days=1)
|
||
|
||
return all_stats, all_texts
|
||
|
||
|
||
def aggregate_weekly_stats(all_stats):
|
||
"""Aggregate daily stats into weekly totals."""
|
||
total = {
|
||
"reading_days": 0,
|
||
"total_minutes": 0,
|
||
"total_highlights": 0,
|
||
"total_notes": 0,
|
||
"all_books": [],
|
||
"daily_detail": [],
|
||
}
|
||
|
||
seen_books = set()
|
||
for s in all_stats:
|
||
if s.get("has_data"):
|
||
total["reading_days"] += 1
|
||
total["total_minutes"] += s.get("reading_time_minutes", 0)
|
||
total["total_highlights"] += s.get("highlights_count", 0)
|
||
total["total_notes"] += s.get("notes_count", 0)
|
||
for book in s.get("books", []):
|
||
if book not in seen_books:
|
||
seen_books.add(book)
|
||
total["all_books"].append(book)
|
||
|
||
total["daily_detail"].append({
|
||
"date": s["date"],
|
||
"has_data": s.get("has_data", False),
|
||
"minutes": s.get("reading_time_minutes", 0),
|
||
"books": s.get("books", []),
|
||
})
|
||
|
||
return total
|
||
|
||
|
||
def build_stats_table(weekly_stats):
|
||
"""Build the statistics markdown table."""
|
||
avg_minutes = weekly_stats["total_minutes"] // max(weekly_stats["reading_days"], 1)
|
||
hours = weekly_stats["total_minutes"] // 60
|
||
minutes = weekly_stats["total_minutes"] % 60
|
||
avg_h = avg_minutes // 60
|
||
avg_m = avg_minutes % 60
|
||
|
||
# Find most engaged book by looking at daily detail
|
||
book_mentions = {}
|
||
for d in weekly_stats["daily_detail"]:
|
||
for book in d.get("books", []):
|
||
book_mentions[book] = book_mentions.get(book, 0) + 1
|
||
top_book = max(book_mentions, key=book_mentions.get) if book_mentions else "—"
|
||
|
||
return f"""| 维度 | 数值 |
|
||
|:---|:---|
|
||
| 阅读天数 | {weekly_stats['reading_days']}/7 |
|
||
| 累计阅读时长 | {hours}小时{minutes}分钟 |
|
||
| 日均阅读时长 | {avg_h}小时{avg_m}分钟 |
|
||
| 本周在读书籍 | {len(weekly_stats['all_books'])}本 |
|
||
| 本周新增划线 | {weekly_stats['total_highlights']}条 |
|
||
| 本周新增想法 | {weekly_stats['total_notes']}条 |
|
||
| 最常出现的书 | 《{top_book}》|"""
|
||
|
||
|
||
def generate_weekly_report():
|
||
"""Main entry: generate weekly reading report."""
|
||
# Determine week range (past 7 days, ending yesterday)
|
||
now = datetime.now(TZ_BEIJING)
|
||
yesterday = now - timedelta(days=1)
|
||
week_end = yesterday.replace(hour=23, minute=59, second=59)
|
||
week_start = (week_end - timedelta(days=6)).replace(hour=0, minute=0, second=0)
|
||
|
||
logger.info(f"Week range: {week_start.strftime('%Y-%m-%d')} → {week_end.strftime('%Y-%m-%d')}")
|
||
|
||
# Step 1: Read all daily reports
|
||
all_stats, all_texts = read_week_daily_reports(week_start, week_end)
|
||
|
||
if not all_texts:
|
||
logger.error("No daily reading reports found this week")
|
||
return None
|
||
|
||
# Step 2: Aggregate statistics
|
||
weekly_stats = aggregate_weekly_stats(all_stats)
|
||
stats_md = build_stats_table(weekly_stats)
|
||
|
||
# Step 3: Build context for LLM
|
||
context_parts = []
|
||
context_parts.append(f"## 本周阅读统计\n\n{stats_md}\n")
|
||
|
||
context_parts.append("## 每日阅读报告原文\n")
|
||
for date_str, text in all_texts:
|
||
# Trim each daily report - keep the key analysis sections, skip repetitive formatting
|
||
context_parts.append(f"### {date_str}\n")
|
||
# Limit each day to ~3000 chars to avoid token overflow
|
||
context_parts.append(text[:3500])
|
||
context_parts.append("\n---\n")
|
||
|
||
full_context = "\n".join(context_parts)
|
||
|
||
# Step 4: Call DeepSeek API
|
||
secrets = load_secrets()
|
||
client = DeepSeekClient(secrets["deepseek_api_key"])
|
||
|
||
prompt = READING_WEEKLY_SYSTEM_PROMPT
|
||
logger.info(f"Sending to DeepSeek... ({len(full_context)} chars)")
|
||
|
||
result = client.chat(
|
||
system_prompt=prompt,
|
||
user_message=full_context,
|
||
temperature=0.5,
|
||
)
|
||
|
||
# Step 5: Assemble final report
|
||
year = week_start.strftime("%Y")
|
||
week_num = week_start.isocalendar()[1]
|
||
week_start_str = week_start.strftime("%m%d")
|
||
week_end_str = week_end.strftime("%m%d")
|
||
|
||
filename = f"每周阅读_W{week_num}_{week_start_str}-{week_end_str}.md"
|
||
|
||
header = f"""# 每周阅读报告 · {week_start.strftime('%Y.%m.%d')} — {week_end.strftime('%Y.%m.%d')}
|
||
|
||
> 本文基于 {weekly_stats['reading_days']} 天阅读数据生成 · 依托 DeepSeek API 分析
|
||
|
||
"""
|
||
|
||
final_report = header + result
|
||
|
||
# Step 6: Write and return
|
||
output_dir = os.path.join(WEREAD_DIR, "weekly")
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
output_path = os.path.join(output_dir, filename)
|
||
|
||
with open(output_path, "w", encoding="utf-8") as f:
|
||
f.write(final_report)
|
||
|
||
logger.info(f"Report saved to {output_path}")
|
||
logger.info(f"Stats: {weekly_stats['reading_days']} days, {weekly_stats['total_minutes']}min, {len(weekly_stats['all_books'])} books")
|
||
|
||
return output_path
|
||
|
||
|
||
if __name__ == "__main__":
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
path = generate_weekly_report()
|
||
if path:
|
||
print(f"DONE: {path}")
|
||
else:
|
||
print("FAILED: No report generated")
|
||
sys.exit(1)
|