lt
251 lines
8.7 KiB
Python
251 lines
8.7 KiB
Python
#!/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 <notification_data_file>")
|
||
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() |