157 lines
4.7 KiB
Python
157 lines
4.7 KiB
Python
"""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")
|
||
parts = date_str.split("-")
|
||
work_dir = os.path.join(output_root, parts[0], parts[1], parts[2], "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")
|