Python 實現微信模板消息推送完整腳本（含 Access Token 緩存與告警/恢復分發）

專案背景

適用於 Zabbix/監控系統通過微信公眾號/企業微信發送告警與恢復通知。核心功能：

• Access Token 本地文件緩存，自動過期刷新
• 支援告警（多關鍵字）與恢復兩種模板
• 按群組/超管分發，命令列參數驅動
• 詳細日誌記錄便於審計

核心類：WechatPush

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import urllib2, json, datetime, time, sys
from config import *  # 包含 appid, secret, file_name, alarm_id, recover_id, group_*, group_super

class WechatPush:
    def __init__(self, appid, secret, token_file):
        self.appid = appid
        self.secret = secret
        self.token_file = token_file

    def _calc_expire_time(self, seconds):
        """計算指定秒數後的時間字串"""
        return (datetime.datetime.now() + datetime.timedelta(seconds=seconds)).strftime('%Y-%m-%d %H:%M:%S')

    def _check_token(self):
        """檢查本地 token 是否過期，返回 (token, is_valid)"""
        try:
            with open(self.token_file, 'r') as f:
                line = f.read().strip()
            if not line:
                return "", False
            token, expire_str = line.split(',', 1)
            now = time.strftime('%Y-%m-%d %H:%M:%S')
            return (token, now < expire_str)
        except Exception:
            return "", False

    def _fetch_token(self):
        """從微信 API 獲取新 token 並寫入文件"""
        url = 'https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s' % (self.appid, self.secret)
        try:
            resp = urllib2.urlopen(url, timeout=10)
            data = json.loads(resp.read())
            token = data['access_token']
            expires_in = data['expires_in']
            # 提前 5 分鐘過期，留緩衝
            expire_time = self._calc_expire_time(expires_in - 300)
            with open(self.token_file, 'w') as f:
                f.write('%s,%s' % (token, expire_time))
            return token
        except Exception as e:
            print('Fetch token failed:', e)
            return None

    def _post(self, url, payload):
        req = urllib2.Request(url, json.dumps(payload), {'Content-Type': 'application/json'})
        try:
            return urllib2.urlopen(req, timeout=10).read()
        except Exception as e:
            print('POST failed:', e)
            return None

    def send_template(self, touser, template_id, url, topcolor, data):
        token, valid = self._check_token()
        if not valid:
            token = self._fetch_token()
            if not token:
                return False
        if not topcolor:
            topcolor = "#7B68EE"
        payload = {
            'touser': touser,
            'template_id': template_id,
            'url': url,
            'topcolor': topcolor,
            'data': data
        }
        resp = self._post('https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=' + token, payload)
        if resp:
            result = json.loads(resp)
            print('Send to %s: errcode=%s errmsg=%s' % (touser, result.get('errcode'), result.get('errmsg')))
            return result.get('errcode') == 0
        return False

模板數據構建

def build_alarm(title, hostname, timestamp, level, message, state, tail):
    color = "#FF0000"
    return {
        'first': {'value': title},
        'keyword1': {'value': hostname, 'color': color},
        'keyword2': {'value': timestamp, 'color': color},
        'keyword3': {'value': level, 'color': color},
        'keyword4': {'value': message, 'color': color},
        'keyword5': {'value': state, 'color': color},
        'remark': {'value': tail}
    }

def build_recover(title, message, alarm_time, recover_time, duration, tail):
    color = "#228B22"
    return {
        'first': {'value': title},
        'content': {'value': message, 'color': color},
        'occurtime': {'value': alarm_time, 'color': color},
        'recovertime': {'value': recover_time, 'color': color},
        'lasttime': {'value': duration, 'color': color},
        'remark': {'value': tail}
    }

主程式入口：命令列參數解析

if __name__ == '__main__':
    pusher = WechatPush(appid, secret, file_name)
    url = "https://monitor.example.com"

    # 告警模式: python push.py 告警 標題 主機 時間 級別 內容 狀態 尾巴 群組名
    # 恢復模式: python push.py 恢復 標題 內容 告警時間 恢復時間 持續時間 尾巴 群組名
    if len(sys.argv) == 9 and sys.argv[0].endswith('push.py'):  # 實際判斷參數個數
        # 簡化：假設第一個參數區分類型
        pass
    # 實際生產建議使用 argparse 解析子命令

配置文件範例：config.py

appid = "wx_your_appid"
secret = "your_secret"
file_name = "/tmp/wechat_token.txt"
alarm_id = "template_id_for_alarm"
recover_id = "template_id_for_recover"

# 群組定義：列表包含 openid
group_ops = ["oUser1", "oUser2"]
group_dev = ["oUser3"]
group_super = ["oAdmin1", "oAdmin2"]

部署建議

1. 將腳本放置於監控伺服器 /usr/local/bin/wechat_push.py，賦予執行權限。
2. Zabbix 動作腳本配置：{ALERT.SENDTO} 映射群組名，參數傳遞對應欄位。
3. Token 文件目錄需腳本執行用戶寫入權限。
4. 日誌建議輸出至 /var/log/wechat_push.log，配合 logrotate。

擴展方向

• 改用 requests 庫簡化 HTTP 處理。
• 支援企業微信應用消息（agentid + access_token 不同）。
• 加入重試機制與發送失敗告警（如發送失敗再發郵件）。
• 使用 argparse 規範化 CLI 介面。