aiagent/scripts/tools/daily_work_summary.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
===========================================
📋 每日工作总结自动生成器
功能：每天早上9点 → 生成今日工作总结 → 推送到指定渠道
===========================================
"""

import json
import os
import sys
import datetime
import smtplib
import requests
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from pathlib import Path

# ===========================================
# 📝 第一部分：配置区（请按需修改）
# ===========================================

class Config:
    """所有配置集中管理，方便修改"""
    
    # --- 数据来源方式 ---
    # mode = "manual"  → 手动输入今日事项
    # mode = "file"    → 从文件读取今日事项
    DATA_MODE = "manual"
    
    # 当 DATA_MODE = "file" 时，读取此文件
    DATA_FILE = "today_work_items.txt"
    
    # --- 推送方式（可多选）---
    # 支持：wecom（企业微信）、dingtalk（钉钉）、email（邮件）
    PUSH_CHANNELS = ["wecom"]  # 可改为 ["dingtalk"] 或 ["email"]
    
    # ----- 企业微信机器人 -----
    WECOM_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY_HERE"
    
    # ----- 钉钉机器人 -----
    DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN_HERE"
    
    # ----- 邮件配置 -----
    SMTP_SERVER = "smtp.qq.com"
    SMTP_PORT = 465
    SMTP_USER = "your_email@qq.com"
    SMTP_PASSWORD = "your_auth_code"  # 邮件授权码，非密码
    MAIL_TO = ["boss@company.com", "yourself@company.com"]
    
    # --- 总结模板 ---
    SUMMARY_TEMPLATE = """
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
📅 今日工作总结 · {date}
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

👤 姓名：{name}
🏢 部门：{department}

📌 今日工作事项：
{work_items}

📊 工作总结：
{summary}

✅ 明日计划：
{tomorrow_plan}

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
⏰ 生成时间：{generate_time}
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
"""


# ===========================================
# 🔧 第二部分：核心功能
# ===========================================

class WorkSummaryGenerator:
    """工作总结生成器"""
    
    def __init__(self, name="张三", department="技术部"):
        self.name = name
        self.department = department
        self.today = datetime.date.today()
        self.tomorrow = self.today + datetime.timedelta(days=1)
    
    def get_work_items_from_input(self):
        """方式一：手动输入今日工作事项"""
        print("\n📝 请输入你今天的工作事项（每行一条，输入空行结束）：")
        print("（例：完成了XX功能开发 / 修复了XXbug / 参加了XX会议）\n")
        items = []
        while True:
            line = input("  ➜ ").strip()
            if not line:
                break
            items.append(line)
        return items
    
    def get_work_items_from_file(self, filepath):
        """方式二：从文件读取今日工作事项"""
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                items = [line.strip() for line in f if line.strip()]
            return items
        except FileNotFoundError:
            print(f"⚠️ 文件 {filepath} 不存在，请先创建！")
            return []
    
    def generate_summary(self, work_items):
        """生成工作总结（基于规则+模板，你也可以接入AI API）"""
        if not work_items:
            return "今日暂无工作记录。"
        
        # 统计信息
        total_items = len(work_items)
        
        # 简单分类统计（关键词匹配）
        categories = {
            "开发": 0, "修复": 0, "会议": 0, "文档": 0,
            "测试": 0, "沟通": 0, "其他": 0
        }
        keywords_map = {
            "开发": ["开发", "编码", "编程", "实现", "搭建", "构建"],
            "修复": ["修复", "解决", "处理", "修改", "bug", "Bug"],
            "会议": ["会议", "评审", "讨论", "沟通会", "晨会", "周会"],
            "文档": ["文档", "文档", "方案", "设计", "PPT", "报告"],
            "测试": ["测试", "调试", "自测", "联调", "验证"],
            "沟通": ["沟通", "对齐", "同步", "协调", "对接"],
        }
        
        for item in work_items:
            matched = False
            for cat, keywords in keywords_map.items():
                if any(kw in item for kw in keywords):
                    categories[cat] += 1
                    matched = True
                    break
            if not matched:
                categories["其他"] += 1
        
        # 生成总结文本
        summary_parts = [f"今日共完成 {total_items} 项工作，具体如下：\n"]
        
        # 分类汇总
        active_cats = {k: v for k, v in categories.items() if v > 0}
        if active_cats:
            summary_parts.append("📊 工作类型分布：")
            for cat, count in active_cats.items():
                bar = "█" * count
                summary_parts.append(f"  {cat}：{bar} {count}项")
            summary_parts.append("")
        
        # 逐项列出
        summary_parts.append("📋 详细工作记录：")
        for i, item in enumerate(work_items, 1):
            summary_parts.append(f"  {i}. {item}")
        
        # 评价与建议
        summary_parts.append(f"\n💡 今日小结：今日工作效率{'较高' if total_items >= 5 else '正常'}，"
                           f"建议明日继续保持{'/优化时间管理' if total_items < 3 else ''}。")
        
        return "\n".join(summary_parts)
    
    def generate_tomorrow_plan(self):
        """生成明日计划（可扩展为从配置文件读取）"""
        plan_items = [
            "继续推进当前开发任务",
            "同步项目进度",
            "代码审查",
        ]
        return "\n".join(f"  {i+1}. {item}" for i, item in enumerate(plan_items))
    
    def build_summary_text(self, work_items, summary):
        """组装完整的总结文本"""
        tomorrow_plan = self.generate_tomorrow_plan()
        formatted_items = "\n".join(f"  • {item}" for item in work_items)
        
        now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        return Config.SUMMARY_TEMPLATE.format(
            date=self.today.strftime("%Y年%m月%d日"),
            name=self.name,
            department=self.department,
            work_items=formatted_items if formatted_items else "  （暂无记录）",
            summary=summary,
            tomorrow_plan=tomorrow_plan,
            generate_time=now_str
        )
    
    def run(self):
        """主运行流程"""
        print(f"\n{'='*50}")
        print(f"📅 今日工作总结生成器")
        print(f"📆 {self.today.strftime('%Y年%m月%d日')}")
        print(f"{'='*50}\n")
        
        # 1️⃣ 获取工作事项
        if Config.DATA_MODE == "file":
            items = self.get_work_items_from_file(Config.DATA_FILE)
        else:
            items = self.get_work_items_from_input()
        
        if not items:
            print("⚠️ 没有输入任何工作事项，将生成空总结。")
        
        # 2️⃣ 生成总结
        summary = self.generate_summary(items)
        final_text = self.build_summary_text(items, summary)
        
        # 3️⃣ 本地保存
        self.save_to_file(final_text)
        
        # 4️⃣ 推送
        pusher = MessagePusher()
        pusher.push(final_text)
        
        return final_text
    
    def save_to_file(self, content):
        """保存到本地文件"""
        filename = f"工作总结_{self.today.strftime('%Y%m%d')}.md"
        with open(filename, "w", encoding="utf-8") as f:
            f.write(content)
        print(f"\n✅ 已保存到本地文件：{os.path.abspath(filename)}")


# ===========================================
# 📨 第三部分：消息推送
# ===========================================

class MessagePusher:
    """消息推送器"""
    
    def push(self, content):
        """根据配置推送消息到各个渠道"""
        for channel in Config.PUSH_CHANNELS:
            if channel == "wecom":
                self.push_wecom(content)
            elif channel == "dingtalk":
                self.push_dingtalk(content)
            elif channel == "email":
                self.push_email(content)
            else:
                print(f"⚠️ 未知推送渠道: {channel}")
    
    def push_wecom(self, content):
        """推送到企业微信机器人"""
        url = Config.WECOM_WEBHOOK
        if "YOUR_KEY_HERE" in url:
            print("⚠️ 跳过企业微信推送：未配置Webhook地址")
            return
        
        data = {
            "msgtype": "markdown",
            "markdown": {"content": content}
        }
        try:
            resp = requests.post(url, json=data, timeout=10)
            result = resp.json()
            if result.get("errcode") == 0:
                print("✅ 企业微信推送成功！")
            else:
                print(f"⚠️ 企业微信推送失败: {result}")
        except Exception as e:
            print(f"⚠️ 企业微信推送异常: {e}")
    
    def push_dingtalk(self, content):
        """推送到钉钉机器人"""
        url = Config.DINGTALK_WEBHOOK
        if "YOUR_TOKEN_HERE" in url:
            print("⚠️ 跳过钉钉推送：未配置Webhook地址")
            return
        
        data = {
            "msgtype": "markdown",
            "markdown": {
                "title": "今日工作总结",
                "text": content
            }
        }
        try:
            resp = requests.post(url, json=data, timeout=10)
            result = resp.json()
            if result.get("errcode") == 0:
                print("✅ 钉钉推送成功！")
            else:
                print(f"⚠️ 钉钉推送失败: {result}")
        except Exception as e:
            print(f"⚠️ 钉钉推送异常: {e}")
    
    def push_email(self, content):
        """通过邮件推送"""
        if "your_email" in Config.SMTP_USER:
            print("⚠️ 跳过邮件推送：未配置邮箱信息")
            return
        
        try:
            msg = MIMEMultipart()
            msg["From"] = Config.SMTP_USER
            msg["To"] = ", ".join(Config.MAIL_TO)
            msg["Subject"] = f"今日工作总结 - {datetime.date.today().strftime('%Y-%m-%d')}"
            
            msg.attach(MIMEText(content, "plain", "utf-8"))
            
            with smtplib.SMTP_SSL(Config.SMTP_SERVER, Config.SMTP_PORT) as server:
                server.login(Config.SMTP_USER, Config.SMTP_PASSWORD)
                server.sendmail(Config.SMTP_USER, Config.MAIL_TO, msg.as_string())
            
            print("✅ 邮件推送成功！")
        except Exception as e:
            print(f"⚠️ 邮件推送异常: {e}")


# ===========================================
# 🚀 第四部分：程序入口
# ===========================================

def setup_windows_task():
    """生成Windows任务计划程序导入脚本"""
    script_path = os.path.abspath(sys.argv[0])
    python_path = sys.executable
    
    cmd = f'''@echo off
echo ========================================
echo   设置每日9点自动运行工作总结脚本
echo ========================================

REM 请以管理员身份运行此.bat文件

schtasks /create /tn "每日工作总结" /tr "{python_path} {script_path}" /sc daily /st 09:00 /f

if %errorlevel%==0 (
    echo ✅ 定时任务创建成功！每天早上9:00自动运行。
) else (
    echo ❌ 创建失败，请尝试以管理员身份运行。
)

pause
'''
    
    bat_path = "setup_windows_task.bat"
    with open(bat_path, "w", encoding="utf-8") as f:
        f.write(cmd)
    print(f"\n📁 Windows定时任务脚本已生成：{os.path.abspath(bat_path)}")
    print("   → 右键点击该文件，选择「以管理员身份运行」即可设置定时任务！")


def setup_linux_cron():
    """输出Linux crontab配置说明"""
    script_path = os.path.abspath(sys.argv[0])
    python_path = sys.executable
    
    print("\n" + "="*50)
    print("🐧 Linux/macOS 定时任务设置（crontab）")
    print("="*50)
    print(f"在终端中执行以下命令添加定时任务：")
    print(f"\n  crontab -e")
    print(f"\n然后添加一行：")
    print(f"\n  0 9 * * * {python_path} {script_path}")
    print(f"\n保存退出即可！📌")


if __name__ == "__main__":
    import sys
    
    # 检查是否有命令行参数
    if len(sys.argv) > 1:
        arg = sys.argv[1].lower()
        if arg == "--setup-win":
            setup_windows_task()
            sys.exit(0)
        elif arg == "--setup-linux":
            setup_linux_cron()
            sys.exit(0)
    
    print("""
╔══════════════════════════════════════════╗
║        🍊 每日工作总结生成器             ║
║                                          ║
║  用法：                                   ║
║    python daily_work_summary.py          ║
║    python daily_work_summary.py --setup-win  ║
║    python daily_work_summary.py --setup-linux ║
╚══════════════════════════════════════════╝
""")
    
    # 运行主程序
    generator = WorkSummaryGenerator(name="张三", department="技术部")
    result = generator.run()
    
    print("\n" + "="*50)
    print("🎉 工作总结生成完成！")
    print("="*50)
    
    # 生成本地定时任务设置脚本
    if sys.platform == "win32":
        setup_windows_task()
    else:
        setup_linux_cron()
    
    print("\n💡 小提示：修改 config.py 中的配置可自定义推送到不同渠道～")