lt
233 lines
7.7 KiB
Python
233 lines
7.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:
|
||
# 如果没有yaml，使用默认配置
|
||
self.config = {
|
||
"wecom": {
|
||
"target_user": "WangLiuTong",
|
||
"message_template": {
|
||
"header": "【宇之然自动推送】",
|
||
"footer": "详情请查看项目目录",
|
||
"max_length": 2000
|
||
}
|
||
},
|
||
"notification_templates": {
|
||
"sustainability_task_complete": """【可持续性内容收集完成】
|
||
时间: {{TIME}}
|
||
新增选题数: {{TOPIC_COUNT}}
|
||
新增案例数: {{CASE_COUNT}}
|
||
信息源: {{SOURCE_COUNT}}个
|
||
详情: {{DETAILS_LINK}}""",
|
||
"content_creation_complete": """【内容创作完成】
|
||
时间: {{TIME}}
|
||
选题: {{TOPIC_TITLE}}
|
||
平台版本: 知乎、公众号、小红书
|
||
图片数: {{IMAGE_COUNT}}
|
||
文件位置: {{OUTPUT_DIR}}
|
||
状态: {{STATUS}}""",
|
||
"system_error": """【定时任务异常】
|
||
任务: {{TASK_NAME}}
|
||
错误: {{ERROR}}
|
||
时间: {{TIME}}
|
||
请检查日志: {{LOG_PATH}}"""
|
||
}
|
||
}
|
||
|
||
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) -> bool:
|
||
"""通过OpenClaw发送消息"""
|
||
try:
|
||
import subprocess
|
||
|
||
# 尝试使用OpenClaw CLI发送消息
|
||
# 假设有企业微信通道配置
|
||
target_user = self.config["wecom"]["target_user"]
|
||
|
||
# 构建命令：使用openclaw message send
|
||
cmd = [
|
||
"openclaw", "message", "send",
|
||
"--channel", "wecom",
|
||
"--account", "default",
|
||
"--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)
|
||
|
||
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() |