想给微信发通知,代码再复杂些

not image

以下是一个**更复杂、功能更丰富**的 Python 定时通知程序,支持通过**微信推送**(使用 PushPlus 第三方服务)发送通知,并包含**动态配置、日志记录、错误重试、多任务调度**等高级功能。 --- ## 🧩 功能概览 | 功能模块 | 说明 | |---------|------| | 定时任务 | 使用 `APScheduler` 实现灵活定时调度 | | 微信通知 | 通过 [PushPlus](https://www.pushplus.plus/) 发送微信消息 | | 动态配置 | 支持从 `config.yaml` 加载任务配置 | | 日志记录 | 使用 `logging` 模块记录运行日志 | | 错误重试 | 发送失败后自动重试 3 次 | | 多任务支持 | 支持多个不同时间、内容的通知任务 | --- ## 📦 依赖安装 ```bash pip install apscheduler requests pyyaml ``` --- ## 📁 项目结构 ``` wechat_notifier/ ├── config.yaml # 配置文件 ├── notifier.py # 主程序 ├── pushplus.py # PushPlus 接口封装 ├── logs/ # 日志文件目录 └── README.md # 使用说明 ``` --- ## 📄 示例配置文件 `config.yaml` ```yaml pushplus: token: your_pushplus_token # 请替换为你的 PushPlus Token tasks: - name: "每日健康提醒" time: "09:00" content: "记得喝水和做眼保健操哦!" enabled: true - name: "晚间总结" time: "21:30" content: "今天工作完成得怎么样?记得复盘总结!" enabled: true ``` --- ## 🧾 主程序 `notifier.py` ```python import logging import time import yaml from apscheduler.schedulers.blocking import BlockingScheduler from pushplus import send_wechat_message # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', filename='logs/notifier.log' ) # 读取配置 def load_config(): with open('config.yaml', 'r', encoding='utf-8') as f: return yaml.safe_load(f) # 任务执行函数 def scheduled_task(task_name, content): logging.info(f"正在执行任务:{task_name}") try: send_wechat_message(content) logging.info(f"任务 [{task_name}] 成功发送消息") except Exception as e: logging.error(f"任务 [{task_name}] 发送失败: {e}") # 启动定时器 def start_scheduler(): config = load_config() scheduler = BlockingScheduler() for task in config['tasks']: if task['enabled']: task_name = task['name'] send_time = task['time'] content = task['content'] scheduler.add_job( scheduled_task, 'cron', hour=send_time.split(':')[0], minute=send_time.split(':')[1], args=[task_name, content], misfire_grace_time=60, coalesce=True, max_instances=1 ) logging.info(f"任务 [{task_name}] 已注册,时间:{send_time}") try: logging.info("定时通知程序已启动,按 Ctrl+C 停止。") scheduler.start() except KeyboardInterrupt: logging.info("定时通知程序已停止。") if __name__ == '__main__': start_scheduler() ``` --- ## 📡 PushPlus 接口封装 `pushplus.py` ```python import requests import logging PUSHPLUS_URL = "http://www.pushplus.plus/send" def send_wechat_message(content, title="定时通知"): config = load_config() token = config['pushplus']['token'] data = { "token": token, "title": title, "content": content, "template": "markdown" } for i in range(3): # 最多重试 3 次 try: response = requests.post(PUSHPLUS_URL, json=data, timeout=10) if response.status_code == 200 and response.json().get("code") == 200: return True else: logging.warning(f"第 {i + 1} 次发送失败: {response.text}") time.sleep(5) except Exception as e: logging.warning(f"第 {i + 1} 次发送异常: {e}") time.sleep(5) return False def load_config(): with open('config.yaml', 'r', encoding='utf-8') as f: return yaml.safe_load(f) ``` --- ## 📋 README.md 使用说明 ```markdown # 微信定时通知程序 ## 安装依赖 ```bash pip install apscheduler requests pyyaml ``` ## 配置 PushPlus 1. 访问 [PushPlus 官网](https://www.pushplus.plus/) 2. 注册账号并获取 Token 3. 替换 `config.yaml` 中的 `token` 字段 ## 启动程序 ```bash python notifier.py ``` ## 后台运行(Linux/macOS) ```bash nohup python notifier.py > /dev/null 2>&1 & ``` ## 停止程序 按 `Ctrl+C` 或使用 `kill` 命令 ``` --- ## 🛠️ 扩展建议 - **支持企业微信**:替换 PushPlus 为 [企业微信 Webhook](https://work.weixin.qq.com/api/doc) 接口 - **支持模板消息**:将 `content` 改为 Markdown 或 HTML 格式 - **添加数据库支持**:使用 SQLite 或 MongoDB 存储任务和历史记录 - **添加 API 接口**:使用 Flask/FastAPI 提供任务管理接口 - **支持多用户推送**:为不同用户配置不同的 Token --- ## ✅ 总结 这个程序相比之前的版本更复杂,具备以下优势: - 更灵活的定时任务调度(APScheduler) - 支持微信推送(PushPlus) - 支持配置化任务管理 - 完善的日志记录和错误重试机制 - 可扩展性强,便于后续集成企业微信、多用户通知等 如果你希望进一步支持 **企业微信机器人、邮件通知、声音提醒、数据库记录** 等功能,也可以继续扩展。是否需要我为你添加这些功能?

评论区:

评论:

请输入内容: