"""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")