#!/usr/bin/env python3 """ 每日日程推送 ============ 由自动化定时触发，生成当日日程报告。 """ import json import sys import os from datetime import date from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from scheduler import generate_daily_plan, generate_weekly_overview CATEGORY_EMOJI = { "outing": "📍", "personal": "👤", "other_work": "💼", "project": "📁", } CATEGORY_NAME = { "outing": "外出事务", "personal": "个人事务", "other_work": "其他工作", "project": "", } DASHBOARD_URL = os.environ.get("DASHBOARD_URL", "http://localhost:8765/dashboard.html") def format_time_badge(minutes): """格式化时长""" if minutes < 60: return f"{minutes}分钟" h = minutes // 60 m = minutes % 60 if m == 0: return f"{h}小时" return f"{h}小时{m}分钟" def push_daily(target_date=None): """生成每日推送 Markdown""" plan = generate_daily_plan(target_date) lines = [] lines.append(f"# 今日日程 · {plan['date']} ({plan['weekday']})") lines.append(f"工作时间：{plan['working_hours']} | 总任务：{plan['total_tasks']}项 | 预估：{format_time_badge(plan['total_estimated_minutes'])}") lines.append("") # 冲突警告 conflicts = plan.get("conflicts", {}) if conflicts.get("conflicts"): lines.append("## 冲突检测") for c in conflicts["conflicts"]: lines.append(f"- {c['date']}: 「{c['task_a_title']}」结束于 {c['task_a_end']}，与「{c['task_b_title']}」({c['task_b_start']}开始) 重叠 {c['overlap_minutes']}分钟") lines.append("") if conflicts.get("overload_days"): lines.append("## 超载警告") for d, info in conflicts["overload_days"]: lines.append(f"- {d}: 预估 {format_time_badge(info['total_minutes'])} / 上限 {format_time_badge(info['max_minutes'])}，超出 {format_time_badge(info['over_minutes'])}") lines.append("") if conflicts.get("warnings"): lines.append("## 提醒") for w in conflicts["warnings"]: if w["type"] == "non_working_day": lines.append(f"- {w['date']} 是{w['weekday']}（非工作日），有 {w['task_count']} 项任务") elif w["type"] == "deadline_before_schedule": lines.append(f"- 「{w['task_title']}」截止 {w['deadline']}，但安排在 {w['scheduled']}") lines.append("") # 任务列表 lines.append("## 任务列表") lines.append("") for i, t in enumerate(plan.get("tasks", []), 1): cat = t.get("category", "") emoji = CATEGORY_EMOJI.get(cat, "") cat_name = CATEGORY_NAME.get(cat, "") priority_map = {1: "🔴P1", 2: "🟠P2", 3: "🟡P3", 4: "⚪P4"} p_label = priority_map.get(t.get("priority", 99), "") deadline = f"截止{t['deadline']}" if t.get("deadline") else "" if cat == "project": project_label = f"{emoji} {t['project']}" else: project_label = f"{emoji} {cat_name}" lines.append(f"### {i}. [{t['time']}] {t['title']} `{p_label}`") lines.append(f"> {project_label} | 预估：{format_time_badge(t['estimated_minutes'])} {deadline}") if t.get("description"): lines.append(f"> {t['description']}") if t.get("tags"): tags = " · ".join(t["tags"]) lines.append(f"> 标签：{tags}") lines.append("") if not plan.get("tasks"): lines.append("今天没有安排任务。") lines.append("") lines.append("---") lines.append(f"*由日程管家自动生成 · {plan.get('generated_at', '')}*") lines.append("") lines.append(f"[查看全部日程]({DASHBOARD_URL})") return "\n".join(lines) def push_weekly(target_monday=None): """生成每周概览推送""" overview = generate_weekly_overview(target_monday) lines = [] lines.append(f"# 本周概览 · {overview['week_start']} ~ {overview['week_end']}") lines.append(f"总任务：{overview['total_tasks']}项 | 总预估：{overview['total_hours']}小时") lines.append("") # 项目分布 lines.append("## 项目分布") lines.append("") for pname, info in overview.get("project_breakdown", {}).items(): lines.append(f"- **{pname}**：{info['task_count']}项 · {format_time_badge(info['total_minutes'])}") lines.append("") # 每日概览 lines.append("## 每日安排") lines.append("") for day in overview.get("days", []): wd = day["weekday"] if day["task_count"] == 0: lines.append(f"### {wd} {day['date']} — 暂未安排") else: lines.append(f"### {wd} {day['date']} — {day['task_count']}项 · {day['hours']}小时") for t in day.get("tasks", []): cat = t.get("category", "") emoji = CATEGORY_EMOJI.get(cat, "") p_label = {1: "🔴P1", 2: "🟠P2", 3: "🟡P3", 4: "⚪P4"}.get(t.get("priority", 99), "") lines.append(f"- {emoji} [{t['time']}] {t['title']} `{p_label}`") lines.append("") lines.append("---") lines.append(f"*由日程管家自动生成 · {overview.get('generated_at', '')}*") lines.append("") lines.append(f"[查看全部日程]({DASHBOARD_URL})") return "\n".join(lines) if __name__ == "__main__": import sys mode = sys.argv[1] if len(sys.argv) > 1 else "daily" if mode == "weekly": print(push_weekly()) else: print(push_daily())