开始之前
请参考 企业微信 Wecom 推送服务教程 创建应用并添加可信域
简单上手
进入域名目录下的 public_html 子目录，执行下述命令打开文件修改配置
wget https://github.com/easychen/wecomchan/blob/main/index.php && vi index.php
| 名称 | 描述 |
|---|---|
| SENDKEY | 自定义 sendkey |
| WECOM_CID | 企业 ID |
| WECOM_AID | AgentId |
| WECOM_SECRET | Secret |
| WECOM_TOUID | 接收人，@all 表示发送给所有人 |
| REDIS_STAT | 是否启用 redis，ON-启用 OFF 或空-不启用 |
| REDIS_ADDR | Redis 服务器地址，如不启用 redis 缓存可不设置 |
| REDIS_PORT | Redis 端口号 |
| REDIS_EXPIRED | Redis 过期时间 |
| REDIS_PASSWORD | Redis 的连接密码，如不启用 redis 缓存可不设置 |
GET示例
curl "http://username.serv00.net/?sendkey=aaaaa&msg=Hello+WeCom"
POST示例
curl -X POST -d "sendkey=aaaaa&msg=Hello+WeCom" http://username.serv00.net/
应用实例
下面在 【serv00系列教程】部署始皇NewAPI 自动更新的基础上添加了消息通知
update.py
#!/usr/local/bin/python
import os
import time
import requests
import subprocess
from datetime import datetime, timedelta
import logging
# 用户配置项
USERNAME = '你的用户名'
DOMAIN = '你的域名'
WORK_DIR = f'/home/{USERNAME}/domains/{DOMAIN}/public_html'
LOG_FILE = f'/home/{USERNAME}/domains/update.log'
GITHUB_PROJECT = 'eggacheb/new-api-freebsd'
# 推送配置项
SENDKEY = "aaaaaaa"
NOTIFICATION_URL = "http://username.serv00.net"
logging.basicConfig(filename=LOG_FILE, level=logging.INFO, format='%(asctime)s %(levelname)s:%(message)s')
def send_notification(message):
try:
url = f"{NOTIFICATION_URL}?sendkey={SENDKEY}&msg={message}"
response = requests.get(url)
response.raise_for_status()
logging.info("Notification sent successfully")
except requests.RequestException as e:
logging.error(f"Failed to send notification: {e}")
def get_current_version():
version_file = f"{WORK_DIR}/read.me"
if os.path.exists(version_file):
with open(version_file, "r") as file:
content = file.read().strip()
if content.startswith("Version:"):
return content.split(":", 1)[1].strip()
return None
def update_service():
try:
current_version = get_current_version()
# 获取最新版本信息
url = f"https://api.github.com/repos/{GITHUB_PROJECT}/releases/latest"
response = requests.get(url)
response.raise_for_status()
release = response.json()
latest_version = release['tag_name']
if current_version == latest_version:
logging.info(f"Already at the latest version: {current_version}")
send_notification(f"无需更新: 当前版本 {current_version} 已是最新")
return
asset_url = next((asset['browser_download_url'] for asset in release.get('assets', []) if asset['name'] != 'source code'), None)
if not asset_url:
logging.error("No asset URL found")
send_notification("更新失败: 未找到资源URL")
return
# 下载并准备新版本
subprocess.run(["curl", "-L", "-o", f"{WORK_DIR}/newapi_new", asset_url], check=True)
subprocess.run(["chmod", "+x", f"{WORK_DIR}/newapi_new"], check=True)
# 停止服务，更新，然后重启
subprocess.run([f"killall", "-u", f"{USERNAME}"], check=True)
subprocess.run(["mv", f"{WORK_DIR}/newapi_new", f"{WORK_DIR}/newapi"], check=True)
subprocess.run([f"/home/{USERNAME}/.npm-global/bin/pm2", "resurrect"], check=True)
# 更新版本信息
with open(f"{WORK_DIR}/read.me", "w") as file:
file.write(f"Version: {latest_version}\n")
logging.info(f"Updated from version {current_version} to {latest_version}")
send_notification(f"更新成功: 从版本 {current_version} 更新到 {latest_version}")
except Exception as e:
error_message = f"更新过程中出错: {str(e)}"
logging.error(error_message)
send_notification(error_message)
def main():
while True:
now = datetime.now()
next_run = now.replace(hour=0, minute=0, second=0, microsecond=0)
if now >= next_run:
next_run += timedelta(days=1)
time.sleep((next_run - now).total_seconds())
update_service()
if __name__ == "__main__":
main()
