feat: 阅读周报 + 工作日志(日报/周报) 报告生成脚本

analyzers/daily_work_report.py

@@ -0,0 +1,155 @@
+"""Daily work report generator - extracts work-related Memos and generates a daily work log."""
+
+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 get_output_dir
+from tools.memos_client import MemosClient
+from tools.config import load_secrets
+
+logger = logging.getLogger(__name__)
+
+TZ_BEIJING = timezone(timedelta(hours=8))
+
+# Tags that indicate work-related memos
+WORK_TAGS = {"工作", "会议", "项目", "出差", "培训", "汇报", "材料", "接待", "调研"}
+
+
+def is_work_memo(tags):
+    """Check if a memo's tags indicate work content."""
+    work_tags = set(tags) & WORK_TAGS
+    return len(work_tags) > 0
+
+
+def generate_daily_work_report(target_date=None):
+    """Generate a daily work report from Memos.
+
+    Args:
+        target_date: datetime.date or None (defaults to today in Beijing)
+    """
+    if target_date is None:
+        target_date = datetime.now(TZ_BEIJING).date()
+    else:
+        if hasattr(target_date, 'date'):
+            target_date = target_date.date()
+    
+    date_str = target_date.strftime("%Y-%m-%d")
+    logger.info(f"Generating daily work report for {date_str}")
+
+    # Connect to Memos
+    secrets = load_secrets()
+    client = MemosClient(secrets["memos_url"], secrets["memos_token"])
+
+    # Fetch today's memos
+    start_dt = datetime.combine(target_date, datetime.min.time(), tzinfo=TZ_BEIJING)
+    end_dt = datetime.combine(target_date, datetime.max.time(), tzinfo=TZ_BEIJING)
+    memos = client.list_all_memos_from_range(start_dt, end_dt)
+
+    # Filter work-related
+    work_memos = [m for m in memos if is_work_memo(m.get("tags", []))]
+
+    logger.info(f"Total memos: {len(memos)}, work-related: {len(work_memos)}")
+
+    if not work_memos:
+        return _write_empty_report(date_str)
+
+    # Build report
+    report = _build_report(date_str, work_memos)
+    _write_report(date_str, report)
+
+    return date_str
+
+
+def _build_report(date_str, work_memos):
+    """Build markdown report from work memos."""
+    lines = [
+        f"# 工作日志 · {date_str}",
+        "",
+        f"> 共 {len(work_memos)} 条工作记录 · 数据来源：Memos",
+        "",
+    ]
+
+    # Group by tag
+    tag_groups = {}
+    for m in work_memos:
+        for tag in m.get("tags", []):
+            if tag in WORK_TAGS:
+                if tag not in tag_groups:
+                    tag_groups[tag] = []
+                tag_groups[tag].append(m)
+
+    # Summary
+    lines.append("## 概览")
+    lines.append("")
+    for tag, items in sorted(tag_groups.items(), key=lambda x: -len(x[1])):
+        lines.append(f"- **{tag}**：{len(items)} 条")
+    lines.append("")
+
+    # Detailed entries
+    lines.append("## 详情")
+    lines.append("")
+
+    for m in work_memos:
+        content = m.get("content", "").strip()
+        tags = m.get("tags", [])
+        work_tags = [t for t in tags if t in WORK_TAGS]
+        tag_str = " · ".join(f"#{t}" for t in work_tags)
+
+        # Remove tag markers from display content for cleaner reading
+        # Keep the raw content but show tags separately
+        display_content = content
+
+        created = m.get("created_at", "")
+        try:
+            dt = datetime.fromisoformat(created.replace("Z", "+00:00"))
+            time_str = dt.astimezone(TZ_BEIJING).strftime("%H:%M")
+        except (ValueError, AttributeError):
+            time_str = ""
+
+        lines.append(f"### {time_str}  {tag_str}")
+        lines.append("")
+        lines.append(display_content)
+        lines.append("")
+
+    return "\n".join(lines)
+
+
+def _write_report(date_str, report_text):
+    """Write report to ai-insights/daily/YYYY-MM-DD/work/"""
+    output_root = get_output_dir("daily")
+    work_dir = os.path.join(output_root, date_str, "work")
+    os.makedirs(work_dir, exist_ok=True)
+
+    filepath = os.path.join(work_dir, f"工作日志_{date_str}.md")
+    with open(filepath, "w", encoding="utf-8") as f:
+        f.write(report_text)
+
+    logger.info(f"Work report saved to {filepath}")
+
+
+def _write_empty_report(date_str):
+    """Write a minimal report when no work memos found."""
+    report = f"# 工作日志 · {date_str}\n\n> 今日无工作相关 Memos\n"
+    _write_report(date_str, report)
+    return date_str
+
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--date", type=str, help="Target date YYYY-MM-DD, defaults to today")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    if args.date:
+        target = datetime.strptime(args.date, "%Y-%m-%d").date()
+    else:
+        target = None
+
+    generate_daily_work_report(target)
+    print("DONE")