#!/usr/bin/env python3 """ 企业微信通知脚本 根据收集器或创作器的结果，发送企业微信通知给用户 WangLiuTong """ import os import sys import json import logging from pathlib import Path import datetime import yaml # 项目根目录 PROJECT_ROOT = Path(__file__).parent.parent sys.path.insert(0, str(PROJECT_ROOT)) # 配置路径 CONFIG_DIR = PROJECT_ROOT / "config" LOGS_DIR = PROJECT_ROOT / "automation" / "logs" TODAY = datetime.datetime.now().strftime("%Y-%m-%d") # 日志配置 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(LOGS_DIR / f"notifier_{TODAY}.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class WeComNotifier: """企业微信通知器""" def __init__(self): self.load_config() def load_config(self): """加载配置文件""" try: with open(CONFIG_DIR / "wecom_config.yaml", "r", encoding='utf-8') as f: self.config = yaml.safe_load(f) except Exception as e: logger.warning(f"加载配置文件失败，使用默认配置: {e}") self.config = { "notification_channel": "wecom", "wecom": { "target_user": "WangLiuTong", "message_template": { "header": "【宇之然自动推送】", "footer": "详情请查看项目目录", "max_length": 2000 } }, "notification_templates": { "daily_summary": """【宇之然日报】{{DATE}} ✅ 今日完成： • 新增选题：{{TOPIC_COUNT}} 个 • 创作完成：{{CREATED_COUNT}} 篇 • 已发布：{{PUBLISHED_COUNT}} 篇 ({{PLATFORMS}}) 📁 发布包：{{PUBLISH_PATH}} {{FAILURES}} —————————— 全流程结束， awaiting tomorrow's run.""" } } def format_message(self, template_name: str, data: dict) -> str: """格式化消息""" templates = self.config["notification_templates"] template = templates.get(template_name, "") for key, value in data.items(): placeholder = f"{{{{{key}}}}}" template = template.replace(placeholder, str(value)) # 添加头部和尾部 header = self.config["wecom"]["message_template"]["header"] footer = self.config["wecom"]["message_template"]["footer"] message = f"{header}\n{template}\n{footer}" # 截断到最大长度 max_len = self.config["wecom"]["message_template"]["max_length"] if len(message) > max_len: message = message[:max_len-3] + "..." return message def send_via_openclaw(self, message: str, account: str = None) -> bool: """通过OpenClaw发送消息 Args: message: 要发送的消息 account: OpenClaw账户ID（对应openclaw.json中的channels.wecom.accounts key） 默认为None，自动根据项目选择：yu-zhi-ran项目用"yuzhiran"，main项目用"main" """ try: import subprocess # 确定账户 if account is None: # 根据项目根目录推断账户（PROJECT_ROOT已定义） cwd = str(PROJECT_ROOT) if 'yu-zhi-ran' in cwd: account = "yuzhiran" elif 'agent-lt' in cwd: account = "agent-lt" else: account = "main" target_user = self.config["wecom"]["target_user"] # 构建命令：使用openclaw message send channel = self.config.get("notification_channel", "wecom") cmd = [ "openclaw", "message", "send", "--channel", channel, "--account", account, "--target", target_user, "--message", message ] result = subprocess.run( cmd, capture_output=True, text=True, timeout=30 ) if result.returncode == 0: logger.info(f"通过OpenClaw发送成功") return True else: logger.error(f"OpenClaw发送失败: {result.stderr}") return False except FileNotFoundError: logger.warning("OpenClaw CLI未找到，使用备用方法") return self.send_via_stdout(message) except Exception as e: logger.error(f"发送失败: {e}") return self.send_via_stdout(message) def send_via_stdout(self, message: str) -> bool: """备用方法：输出到stdout""" print(f"企业微信通知（待发送给{self.config['wecom']['target_user']}）:") print("-" * 50) print(message) print("-" * 50) print("（实际发送需要配置企业微信通道）") return True def process_notification_file(self, data_file: Path): """处理通知数据文件""" if not data_file.exists(): logger.error(f"通知数据文件不存在: {data_file}") return False try: with open(data_file, 'r', encoding='utf-8') as f: data = json.load(f) task_type = data.get("task", "") time_str = data.get("time", datetime.datetime.now().strftime("%Y-%m-%d %H:%M")) if task_type == "sustainability_collection": message_data = { "TIME": time_str, "TOPIC_COUNT": data.get("topic_count", 0), "CASE_COUNT": data.get("case_count", 0), "SOURCE_COUNT": data.get("source_count", 0), "DETAILS_LINK": data.get("details_link", "") } message = self.format_message("sustainability_task_complete", message_data) elif task_type == "content_creation": message_data = { "TIME": time_str, "TOPIC_TITLE": data.get("topic_title", ""), "IMAGE_COUNT": data.get("image_count", 0), "OUTPUT_DIR": data.get("output_dir", ""), "STATUS": data.get("status", "") } message = self.format_message("content_creation_complete", message_data) elif task_type == "daily_summary": # 日报：使用 DATE 而非 TIME message_data = { "DATE": data.get("date", datetime.datetime.now().strftime("%Y-%m-%d")), "PUBLISHED_COUNT": data.get("published_count", 0), "PLATFORMS": ", ".join(data.get("platforms", [])), "PUBLISH_DIR": data.get("publish_dir", "") } message = self.format_message("daily_summary", message_data) else: message_data = { "TASK_NAME": task_type, "ERROR": data.get("error", "未知错误"), "TIME": time_str, "LOG_PATH": data.get("log_path", "") } message = self.format_message("system_error", message_data) # 发送消息 success = self.send_via_openclaw(message) if success: logger.info(f"通知发送成功: {task_type}") else: logger.warning(f"通知发送失败，已输出到stdout") return success except Exception as e: logger.error(f"处理通知文件失败: {e}") return False def main(): """主函数""" if len(sys.argv) < 2: print("Usage: python wecom_notifier.py ") sys.exit(1) data_file = Path(sys.argv[1]) if not data_file.exists(): print(f"Error: Data file not found: {data_file}") sys.exit(1) try: notifier = WeComNotifier() success = notifier.process_notification_file(data_file) if success: print("SUCCESS: Notification processed") sys.exit(0) else: print("WARNING: Notification failed") sys.exit(1) except Exception as e: logger.error(f"通知任务失败: {e}") print(f"ERROR: {e}") sys.exit(1) if __name__ == "__main__": main()