🚗 Tesla 数据接入全流程（三）：✉️ 实时 & 规则发送消通过微信进行集成-比贝网

📨 基于 TeslaMate 的企业微信通知与自动化实战

承接前两篇（OAuth2/Fleet-Owner 入门、TeslaMate+Grafana 可视化），本篇把重点放在**“事件驱动的消息通知与自动化”：
我们将用 TeslaMate 的 *MQTT + PostgreSQL* 能力，构建一个企业微信通知系统**，实现充电/行程/温度/更新提醒等即时推送，并补充日报/周报/月报统计。

为什么用企业微信？-> 当然是免费呀，这个逻辑也能带入公众号/小程序，但是每年需要300年费，国外用户则可以是用 Telegram bot替代我这个方案

📚 目录

📝 前言
🏗 一、整体架构与数据流
🔌 二、关键能力清单
⚙ 三、快速部署（Docker Compose）
🧩 四、功能详解：MQTT 即时推送
📊 五、功能详解：数据库统计报告
🖼️ 六、部分效果展示
🛡 七、安全与稳健性清单
🧪 八、排障建议
📋 九、核心代码展示
📌 十、总结与下一步

📝 前言

在第二篇里我们部署了 TeslaMate 并用 Grafana 看到了漂亮的仪表盘。现实使用中，我们更需要在关键节点收到即时提醒：充满电了、温度异常、行程结束出报告、系统有更新等。本篇带你把这些“动态数据”转化为企业微信通知，让信息流主动找你。

🏗 一、整体架构与数据流

Tesla API →   TeslaMate
           ↙            ↘
       MQTT Broker     PostgreSQL
           ↓               ↓
    MQTT Listener     Stats Queries
           ↘            ↙
        WeChat Client (企业微信Bot)

MQTT：秒级事件（充电开始/结束、状态变更、温度等）
PostgreSQL：聚合统计（日/周/月行驶与充电）
WeChat：统一通知出口（频控+去重，减少骚扰）

🔌 二、关键能力清单

🔔 即时推送：充电开始/结束、行程结束、温度异常、哨兵模式、门窗状态等
📅 统计推送：行程/充电日报、周报、月报
🧠 智能节流：同类消息短时间不重复、重复内容不再次推送
🧰 易扩展：消息模板、目标人群、阈值、定时都可配置化

⚙ 三、快速部署（整合到上一篇 Compose 中）

⚠ 注意：

请将以下服务段直接合并到上一篇 TeslaMate + Grafana 的 docker-compose.yml 文件中，而不是单独启动。

所有示例中的凭据（企业微信、数据库密码等）均需替换为你自己的真实值，避免明文泄露。

建议使用 .env 文件或 Docker secrets 注入敏感参数。

tesla-wechat:
  container_name: "teslamate-wechat"
  image: registry.cn-hangzhou.aliyuncs.com/bigbey/smart-car-tesla-notifier:2.4
  restart: always
  volumes:
    - ./import/config.yaml:/app/config.yaml:ro
    - /etc/localtime:/etc/localtime:ro
  environment:
    - WECHAT_CORPID=wwxxxxxxxxxxxxxxxx
    - WECHAT_CORPSECRET=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
    - WECHAT_AGENTID=1000002
    - DB_HOST=database
    - DB_PORT=5432
    - DB_NAME=teslamate
    - DB_USER=teslamate
    - DB_PASSWORD=<你的数据库密码>
    - MQTT_HOST=mosquitto
    - MQTT_PORT=1883
    - MQTT_USER=
    - MQTT_PASSWORD=
    - LOG_LEVEL=INFO
    - TZ=Asia/Shanghai
  networks:
    - tesla-network
  depends_on:
    - teslamate
    - database
    - mosquitto

💡 整合要点：

tesla-network、teslamate、database、mosquitto 服务需与上一篇 Compose 保持一致。
WECHAT_CORPID / WECHAT_CORPSECRET / DB_PASSWORD / 代理地址等敏感信息不要硬编码到 Compose 文件，可放到 .env 文件中，例如：
```
WECHAT_CORPID=wwxxxxxxxxxxxxxxxx
WECHAT_CORPSECRET=xxxxxxxxxxxxxxxx
DB_PASSWORD=xxxxxxxx
```
然后在 Compose 文件里用 ${变量名} 方式引用。

🧩 四、功能详解：MQTT 即时推送

订阅话题（示例）：
teslamate/cars/+/state、.../charging_state、.../battery_level、.../inside_temp 等
事件→消息：收到状态变更/阈值触发事件 → 构造模板 → 调企业微信接口发送
防骚扰：每类消息设定 rate_limit，内容相同则去重

温度告警示例

# config.yaml 信息
temperature_alerts:
  inside_temp_max: 60.0
  outside_temp_min: -30.0
notifications:
  temperature_alert:
    enabled: true
    rate_limit_seconds: 300
    target_users: "@all"

📊 五、功能详解：数据库统计报告

日志、周报、月报通过 PostgreSQL 查询聚合得到：
- 行驶总里程 / 次数 / 时长 / 平均效率
- 充电次数 / 总电量 / 平均功率 / 费用估算 / 家充占比
用 APScheduler 在固定时间推送到企业微信。
（可选）把统计结果也写入表中，便于审计与幂等。

🖼️ 六、部分效果展示

Tesla 数据接入全流程（三）-图片二.jpg

Tesla 数据接入全流程（三）-图片一.jpg

Tesla 数据接入全流程（三）-图片四.jpg

Tesla 数据接入全流程（三）-图片三.jpg

Tesla 数据接入全流程（三）-图片五.jpg

🛡 七、安全与稳健性清单

🔒 密钥管理：凭据改用环境变量/容器密钥；提交 config.example.yaml。
🧱 数据库连接池：减少重连与开销。
🔁 幂等：记录已发消息的哈希，重启不重复。
🚦 限流退避：遇 429/45009 指数退避再发。
🔐 MQTT/TLS：支持 CA 证书，qos=1、reconnect_delay_set。
📈 可观测性：/healthz + Prometheus /metrics。
🕒 时区统一：建议 Asia/Taipei 或按你实际需要统一配置。

🧪 八、排障建议

MQTT 无消息：检查订阅话题、QoS、broker 权限、TLS 配置。
DB 查询慢：给高频查询列加索引，或用物化视图；引入连接池。
企业微信不达：确认 agentid/corpid/corpsecret、应用已开通、限流码。
消息重复：确认频控与去重配置；实现幂等记录后可彻底解决。

📋 九、核心代码展示

1️⃣ MQTT 消息监听与事件触发

import paho.mqtt.client as mqtt
from wechat_client import send_wechat_message

def on_connect(client, userdata, flags, rc):
    print("MQTT connected:", rc)
    client.subscribe("teslamate/cars/+/state")
    client.subscribe("teslamate/cars/+/charging_state")

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    topic_parts = msg.topic.split("/")
    car_id = topic_parts[2]
    event_type = topic_parts[3]

    if event_type == "state" and payload == "asleep":
        send_wechat_message(f"🚗 车辆 {car_id} 已进入休眠模式")
    elif event_type == "charging_state" and payload == "Charging":
        send_wechat_message(f"🔋 车辆 {car_id} 开始充电")

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("mosquitto", 1883, 60)
client.loop_forever()

💡 订阅 teslamate 的相关主题，解码消息并根据事件类型触发不同的通知。

2️⃣ 企业微信消息发送

import requests, os

CORPID = os.getenv("WECHAT_CORPID")
CORPSECRET = os.getenv("WECHAT_CORPSECRET")
AGENTID = os.getenv("WECHAT_AGENTID")

def get_access_token():
    resp = requests.get(
        "https://qyapi.weixin.qq.com/cgi-bin/gettoken",
        params={"corpid": CORPID, "corpsecret": CORPSECRET}
    )
    return resp.json()["access_token"]

def send_wechat_message(content):
    token = get_access_token()
    data = {
        "touser": "@all",
        "msgtype": "text",
        "agentid": AGENTID,
        "text": {"content": content}
    }
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
    resp = requests.post(url, json=data)
    if resp.status_code != 200:
        print("企业微信发送失败:", resp.text)

💡 通过企业微信 API 推送文本消息，可替换 touser 为指定成员。

3️⃣ PostgreSQL 日报统计

import psycopg2, os
from datetime import date

def get_daily_stats():
    conn = psycopg2.connect(
        host=os.getenv("DB_HOST"),
        dbname=os.getenv("DB_NAME"),
        user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASSWORD")
    )
    cur = conn.cursor()
    cur.execute("""
        SELECT
          SUM(distance) AS total_km,
          SUM(energy_used) AS total_kwh
        FROM drives
        WHERE DATE(start_date) = %s
    """, (date.today(),))
    total_km, total_kwh = cur.fetchone()
    conn.close()
    return f"📅 今日行驶 {total_km:.1f} 公里，耗电 {total_kwh:.1f} kWh"

if __name__ == "__main__":
    print(get_daily_stats())

💡 直接从 teslamate 数据库聚合行驶距离和耗电量，用于日报推送。

📌 十、总结与下一步

通过 TeslaMate 的 MQTT + PostgreSQL，我们把“车辆事件/统计”转成了即时可用的信息流，以企业微信为统一入口触达。
下一步：

将通知系统的Prometheus 指标接入 Grafana，形成“看板+告警”的闭环；
打通 Home Assistant，基于 MQTT/通知触发自动化（例如夜间达到低电价且 SOC<50% 自动开启充电）。

目录CONTENT

🚗 Tesla 数据接入全流程（三）：✉️ 实时 & 规则发送消通过微信进行集成