210 lines
7.2 KiB
Python
210 lines
7.2 KiB
Python
"""Weekly work report synthesizer - reads daily work logs and generates weekly analysis."""
|
||
|
||
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
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
TZ_BEIJING = timezone(timedelta(hours=8))
|
||
PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
WORK_WEEKLY_SYSTEM_PROMPT = """你是一个私人工作日志分析师。你的任务是阅读用户过去一周的每日工作日志，生成一篇有深度的周度工作分析。
|
||
|
||
## 核心原则
|
||
|
||
1. **概述先行**：给出本周工作的整体面貌——主要精力投向了哪些领域、总体节奏如何。
|
||
|
||
2. **按主题组织，不按时间罗列**：不要"周一做了X、周二做了Y"。要提炼跨日的工作主线——哪些项目贯穿了整周？哪些任务是单日完成的？
|
||
|
||
3. **识别模式和趋势**：本周的工作节奏有什么特点？是否出现了新的工作类型？哪些工作在重复发生？是否有一些事在全周的工作日志中反复出现但没有被解决？
|
||
|
||
4. **产出导向**：本周有什么可量化的产出？写了什么材料？参加了什么会议？完成了什么节点？
|
||
|
||
5. **前瞻建议**：基于本周工作日志，下周需要关注什么？有什么待办被遗漏了？
|
||
|
||
## 输出格式
|
||
|
||
纯 Markdown，不要代码块包裹。
|
||
|
||
### 1. 本周工作概览
|
||
|
||
一段话概括本周工作的底色——主要在忙什么、节奏如何、核心成果。
|
||
|
||
### 2. 工作统计
|
||
|
||
| 维度 | 数值 |
|
||
|:---|:---|
|
||
| 有工作记录的天数 | X/7 |
|
||
| 工作 Memos 总数 | X 条 |
|
||
| 涉及标签 | #工作 #会议 ... |
|
||
| 单日最高工作量 | 周X · X条 |
|
||
|
||
### 3. 深入分析
|
||
|
||
按主题组织：
|
||
- 持续性的工作主线（跨日出现的项目/主题）
|
||
- 本周重要节点（会议、截止日期、关键决策）
|
||
- 工作节奏评价（是否有某天特别忙？是否有连续高压？）
|
||
- 新出现的工作类型或关注点
|
||
- 被遗漏或被推迟的事项
|
||
|
||
### 4. 下周关注
|
||
|
||
基于本周工作轨迹，列出下周需要重点关注的事项。具体到事项+时间节点。
|
||
|
||
---
|
||
|
||
注意：
|
||
- 如果某天没有工作日志，在统计中注明，但分析中跳过。
|
||
- 分析要有观点——不只是"做了什么"，更要回答"这意味着什么"和"接下来该做什么"。
|
||
- 篇幅不设上限。"""
|
||
|
||
|
||
def read_week_work_logs(week_start, week_end):
|
||
"""Read all daily work logs from week_start to week_end."""
|
||
all_texts = []
|
||
daily_root = get_output_dir("daily")
|
||
|
||
current = week_start
|
||
while current <= week_end:
|
||
date_str = current.strftime("%Y-%m-%d")
|
||
work_file = os.path.join(daily_root, date_str, "work", f"工作日志_{date_str}.md")
|
||
|
||
if os.path.exists(work_file):
|
||
with open(work_file, "r", encoding="utf-8") as f:
|
||
text = f.read()
|
||
all_texts.append((date_str, text))
|
||
logger.info(f" ✓ {date_str}: {len(text)} chars")
|
||
else:
|
||
logger.info(f" ✗ {date_str}: no work log")
|
||
|
||
current += timedelta(days=1)
|
||
|
||
return all_texts
|
||
|
||
|
||
def extract_weekly_work_stats(all_texts):
|
||
"""Extract statistics from daily work logs."""
|
||
total_memos = 0
|
||
all_tags = set()
|
||
highest_day = ("", 0)
|
||
days_with_data = 0
|
||
|
||
for date_str, text in all_texts:
|
||
if "今日无工作" in text:
|
||
continue
|
||
days_with_data += 1
|
||
|
||
# Count memos (each ### header with content)
|
||
import re
|
||
memo_count = len(re.findall(r'^###\s+\d{2}:\d{2}', text, re.MULTILINE))
|
||
total_memos += memo_count
|
||
|
||
if memo_count > highest_day[1]:
|
||
highest_day = (date_str, memo_count)
|
||
|
||
# Extract tags
|
||
tag_matches = re.findall(r'#(\w+)', text)
|
||
work_tags = {"工作", "会议", "项目", "出差", "培训", "汇报", "材料", "接待", "调研"}
|
||
for tag in tag_matches:
|
||
if tag in work_tags:
|
||
all_tags.add(tag)
|
||
|
||
return {
|
||
"days_with_data": days_with_data,
|
||
"total_memos": total_memos,
|
||
"top_day": highest_day,
|
||
"tags": sorted(all_tags),
|
||
}
|
||
|
||
|
||
def generate_weekly_work_report():
|
||
"""Main entry: generate weekly work report."""
|
||
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 daily work logs
|
||
all_texts = read_week_work_logs(week_start, week_end)
|
||
|
||
if not all_texts:
|
||
logger.warning("No work logs found this week")
|
||
return None
|
||
|
||
# Step 2: Extract stats
|
||
stats = extract_weekly_work_stats(all_texts)
|
||
|
||
# Step 3: Build context
|
||
context_parts = []
|
||
context_parts.append(f"## 本周工作统计\n")
|
||
context_parts.append(f"- 有工作记录的天数：{stats['days_with_data']}/7")
|
||
context_parts.append(f"- 工作 Memos 总数：{stats['total_memos']} 条")
|
||
context_parts.append(f"- 涉及标签：{' '.join('#' + t for t in stats['tags'])}")
|
||
if stats['top_day'][0]:
|
||
context_parts.append(f"- 单日最高：{stats['top_day'][0]} · {stats['top_day'][1]}条")
|
||
context_parts.append("")
|
||
|
||
context_parts.append("## 每日工作日志原文\n")
|
||
for date_str, text in all_texts:
|
||
context_parts.append(f"### {date_str}\n")
|
||
context_parts.append(text[:3500]) # Trim each day
|
||
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"])
|
||
|
||
logger.info(f"Sending to DeepSeek... ({len(full_context)} chars)")
|
||
result = client.chat(
|
||
system_prompt=WORK_WEEKLY_SYSTEM_PROMPT,
|
||
user_message=full_context,
|
||
temperature=0.5,
|
||
)
|
||
|
||
# Step 5: Assemble and write
|
||
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')}
|
||
|
||
> 基于 {stats['days_with_data']}/7 天工作日志生成 · 依托 DeepSeek API 分析
|
||
|
||
"""
|
||
|
||
final_report = header + result
|
||
|
||
output_dir = os.path.join(get_output_dir("weekly"), "work")
|
||
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}")
|
||
return output_path
|
||
|
||
|
||
if __name__ == "__main__":
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
path = generate_weekly_work_report()
|
||
if path:
|
||
print(f"DONE: {path}")
|
||
else:
|
||
print("FAILED: No report generated")
|
||
sys.exit(1)
|