Python 3.9 · Flask 火灾监测识别系统 与 钉钉告警机器人 集成说明文档

在上一份《YOLOv11 火灾识别监测系统教程》的 Flask + MySQL 后端基础上，接入钉钉群机器人，
当模型检出 fire/smoke 时，实时向钉钉群推送告警消息（支持加签安全校验与推送冷却去重）。

---

【接上一篇文章, 系列】
YOLOv11 python3.9 Flask Mysql8火灾识别监测系统 全流程实战教程源码

Python 3.9 · Flask 火灾监测识别系统 与 钉钉告警机器人 集成说明文档

【其他系列，2026持续更新中】

VisionForge SDK使用教程06-AI安全帽检测API调用说明文档

VisionForge SDK使用教程05-AI人员打架检测API调用教程

VisionForge SDK使用教程04-AI打电话检测API调用说明文档

VisionForge SDK使用教程03- opencv和FaceDetector类实现人脸检测API调用

VisionForge SDK使用教程02- yolov8实现垃圾检测调用

VisionForge SDK使用教程01- yolov8实现火灾检测调用

目录

1. 集成架构
2. 环境准备（Python 3.9）
3. 创建钉钉群机器人
4. 钉钉推送模块 dingtalk.py
5. Flask 后端集成改造
6. 告警去重与冷却控制
7. 启动与测试
8. 常见问题
9. 进阶：图片告警与定时汇总

---

一、集成架构

┌──────────┐    ┌───────────────────────────────┐    ┌──────────┐
│ 摄像头/    │ → │ Flask 后端（Python 3.9）        │ → │ MySQL    │
│ 图片/视频  │   │  ├─ YOLO11 推理检出 fire/smoke  │ → │ 告警记录  │
└──────────┘   │  └─ 检出 → 写库 + 钉钉推送        │    └──────────┘
               └───────────────┬───────────────┘
                               ↓
                        ┌──────────────┐
                        │ 钉钉群机器人   │
                        │  (Webhook)   │
                        └──────────────┘

触发链路：图片/视频上传 → YOLO11 检出 fire/smoke → ① 写入 MySQL alert_record → ② 调用钉钉 Webhook 推送告警（受冷却时间控制，防止刷屏）。

前置条件：已完成上一份教程的模型训练（best.pt）与 Flask 后端（app.py / model_service.py / db.py）。

---

二、环境准备（Python 3.9）

2.1 创建 Python 3.9 虚拟环境

# 若已用 conda：
conda create -n fire39 python=3.9 -y
conda activate fire39

# 或用 venv（Windows）：
python3.9 -m venv fire39_env
fire39_env\Scripts\activate

2.2 安装依赖

pip install -U pip
pip install flask==3.0.0 pymysql==1.1.0 requests==2.31.0
pip install ultralytics opencv-python numpy

说明：

• requests 用于调用钉钉 Webhook（新增，钉钉推送核心依赖）。
• Flask/PyMySQL 版本与主教程保持一致（Flask 3.x、PyMySQL 1.x）。
• 若 GPU 环境未就绪，先 pip install torch torchvision --index-url https://download.pytorch.org/whl/cpu。

验证环境：

python -c "import flask, pymysql, requests; print('env ok, flask', flask.__version__)"

---

三、创建钉钉群机器人

3.1 创建步骤

1. 打开钉钉，进入目标群 → 右上角「设置」→「机器人」→「添加机器人」→ 选择「自定义」。
2. 机器人名称随意，如「火灾告警」。
3. 复制 Webhook 地址，形如：

   https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxxxxxxxxx
   

3.2 安全设置（三选一，推荐「加签」）

安全方式	说明	适用场景
自定义关键字	消息内容必须包含关键字（如「火灾告警」），否则发送失败	简单，不推荐用于告警（内容易被截断）
加签（Secret）	请求带 timestamp + sign 签名，最安全	推荐
IP 白名单	限制来源 IP	适合固定服务器

若勾选「加签」，会生成一个 Secret，务必保存到配置中。

3.3 测试机器人连通性

先发一条测试消息验证 Webhook 可用（注意：若选「加签」则此方式不通过，直接用第 4 章代码测）：

curl -X POST "https://oapi.dingtalk.com/robot/send?access_token=xxx" \
     -H "Content-Type: application/json" \
     -d '{"msgtype":"text","text":{"content":"【测试】钉钉机器人已就绪"}}'

正常响应：{"errcode":0,"errmsg":"ok"}。

---

四、钉钉推送模块 dingtalk.py

在 backend/ 下新建 dingtalk.py，封装钉钉机器人（支持加签与无加签两种模式）：

# -*- coding: utf-8 -*-
"""
钉钉群机器人告警推送
支持：
  - 加签（Secret）安全校验：timestamp + HMAC-SHA256 签名
  - 文本 / Markdown 两种消息类型
  - 发送失败重试（2 次，间隔 1 秒）
"""
import base64
import hashlib
import hmac
import time
import urllib.parse

import requests

RETRY_TIMES = 2        # 发送失败重试次数
RETRY_INTERVAL = 1     # 重试间隔（秒）


class DingTalkNotifier:
    def __init__(self, webhook, secret=None):
        """
        :param webhook: 钉钉机器人 Webhook 完整地址
        :param secret:  加签密钥（未启用加签则传 None）
        """
        self.webhook = webhook
        self.secret = secret

    def _build_url(self):
        """拼接带签名的请求地址"""
        if not self.secret:
            return self.webhook
        timestamp = str(round(time.time() * 1000))
        string_to_sign = '{}\n{}'.format(timestamp, self.secret)
        hmac_code = hmac.new(
            self.secret.encode('utf-8'),
            string_to_sign.encode('utf-8'),
            digestmod=hashlib.sha256,
        ).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return '{}&timestamp={}&sign={}'.format(self.webhook, timestamp, sign)

    def _post(self, payload):
        """发送并重试，返回 (success: bool, errmsg: str)"""
        url = self._build_url()
        for attempt in range(RETRY_TIMES + 1):
            try:
                resp = requests.post(url, json=payload, timeout=5)
                data = resp.json()
                if data.get('errcode') == 0:
                    return True, data.get('errmsg', 'ok')
                last_err = data.get('errmsg', 'unknown')
            except Exception as exc:
                last_err = str(exc)
            if attempt < RETRY_TIMES:
                time.sleep(RETRY_INTERVAL)
        return False, last_err

    def send_text(self, content):
        """发送纯文本消息"""
        return self._post({'msgtype': 'text', 'text': {'content': content}})

    def send_markdown(self, title, text):
        """发送 Markdown 消息（title 为会话列表标题，text 为正文）"""
        return self._post({
            'msgtype': 'markdown',
            'markdown': {'title': title, 'text': text},
        })


if __name__ == '__main__':
    # 单元自测：单独运行本文件测试连通性
    import os
    bot = DingTalkNotifier(
        webhook=os.environ.get('DINGTALK_WEBHOOK', ''),
        secret=os.environ.get('DINGTALK_SECRET', ''),
    )
    ok, msg = bot.send_text('【测试】钉钉告警机器人连接成功')
    print('发送结果:', '成功' if ok else '失败', msg)

验证签名逻辑（加签模式）：直接运行

python dingtalk.py

控制台输出 发送结果: 成功 ok 即代表 Webhook + 签名均正确。

---

五、Flask 后端集成改造

5.1 配置项 backend/config.py 增加钉钉相关配置

# ---------- 钉钉机器人配置 ----------
DINGTALK_WEBHOOK = 'https://oapi.dingtalk.com/robot/send?access_token=你的token'
DINGTALK_SECRET = '你的加签密钥（无加签留空）'
ALERT_COOLDOWN_SECONDS = 60      # 同一告警推送冷却间隔（秒），防止刷屏
DINGTALK_ENABLED = True          # 总开关

5.2 告警推送封装 backend/notify_service.py

把"检出结果 → 钉钉推送"逻辑独立出来，便于复用与测试：

# -*- coding: utf-8 -*-
"""告警推送服务：检出火情时推送钉钉并做冷却控制"""
import time
from datetime import datetime

from config import (DINGTALK_WEBHOOK, DINGTALK_SECRET,
                    ALERT_COOLDOWN_SECONDS, DINGTALK_ENABLED)
from dingtalk import DingTalkNotifier

_bot = None
_last_push_time = {}


def get_bot():
    global _bot
    if _bot is None:
        _bot = DingTalkNotifier(webhook=DINGTALK_WEBHOOK, secret=DINGTALK_SECRET)
    return _bot


def _cooldown_ok(key):
    """冷却检查：同一设备/来源在冷却时间内只推一次"""
    now = time.time()
    if _last_push_time.get(key) and (now - _last_push_time[key]) < ALERT_COOLDOWN_SECONDS:
        return False
    _last_push_time[key] = now
    return True


def notify_fire(device_name, source_type, dets, alert_count=None):
    """
    检出火情后调用：推送钉钉告警
    :param dets: 检测结果列表 [{class_name, confidence, bbox}, ...]
    :return: (pushed: bool, message: str)
    """
    if not DINGTALK_ENABLED or not DINGTALK_WEBHOOK:
        return False, '钉钉推送未启用'

    key = f'{device_name}|{source_type}'
    if not _cooldown_ok(key):
        return False, '冷却期内，跳过本次推送'

    now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    fire_cnt = sum(1 for d in dets if d['class_name'] == 'fire')
    smoke_cnt = sum(1 for d in dets if d['class_name'] == 'smoke')

    # 拼装 Markdown 告警内容
    lines = [
        '## :fire: 【火灾告警】',
        f'**设备**：{device_name}',
        f'**来源**：{source_type}',
        f'**时间**：{now_str}',
        f'**检出目标**：fire × {fire_cnt}，smoke × {smoke_cnt}',
        '',
        '| 类别 | 置信度 | 位置(x,y,w,h) |',
        '| --- | --- | --- |',
    ]
    for d in dets[:10]:          # 最多展示 10 个目标，避免消息过长
        x, y, w, h = d['bbox']
        lines.append(f"| {d['class_name']} | {d['confidence'] * 100:.1f}% | {x:.0f},{y:.0f},{w:.0f},{h:.0f} |")
    lines.append('')
    lines.append('> 请及时处理！')

    ok, msg = get_bot().send_markdown(title='火灾告警', text='\n'.join(lines))
    return ok, msg

5.3 在 app.py 检测接口中接入推送

在「图片检测」与「视频检测」接口检出分支中调用推送：

# app.py 顶部新增导入
from notify_service import notify_fire

@app.route('/api/detect/image', methods=['POST'])
def api_detect_image():
    # ...（原有读取文件逻辑不变）...

    img_bytes = file.read()
    dets = detect_image_bytes(img_bytes)

    if dets:
        fname = save_upload(file)
        ids = insert_alert(device_name, 'image', dets, fname)

        # ★ 新增：钉钉告警推送
        pushed, push_msg = notify_fire(device_name, 'image', dets, len(ids))

        return jsonify({
            'success': True, 'fire_detected': True,
            'results': dets, 'alert_ids': ids,
            'image': fname,
            'dingtalk_pushed': pushed,        # 前端可展示推送状态
            'dingtalk_msg': push_msg,
        })

    return jsonify({'success': True, 'fire_detected': False, 'results': []})

@app.route('/api/detect/video', methods=['POST'])
def api_detect_video():
    # ...（原有逻辑不变）...
    dets = detect_video_file(fpath)
    if dets:
        ids = insert_alert(device_name, 'video', dets, fname)
        pushed, push_msg = notify_fire(device_name, 'video', dets, len(ids))   # ★ 新增
        return jsonify({'success': True, 'fire_detected': True, 'results': dets,
                        'alert_ids': ids, 'video': fname,
                        'dingtalk_pushed': pushed, 'dingtalk_msg': push_msg})
    return jsonify({'success': True, 'fire_detected': False, 'results': []})

关键点：推送放在 insert_alert 之后，保证"先落库、再推送"——即使钉钉发送失败，告警记录也不丢失。

---

六、告警去重与冷却控制

钉钉每个自定义机器人每分钟最多发送 20 条消息，火焰/烟雾视频检测极易刷屏，必须做冷却：

机制	实现	作用
冷却时间	notify_fire 内按 设备+来源 维度判断 ALERT_COOLDOWN_SECONDS	60 秒内同设备只推 1 条
失败重试	DingTalkNotifier._post 重试 2 次	网络抖动不丢告警
频控兜底	单条消息展示目标数上限 10	避免消息体过大触发限流

若需要"持续火情持续告警"，可改为：冷却期后再次检出才推新告警，并把冷却时间调小（如 10~30 秒）。

---

七、启动与测试

7.1 启动后端

cd backend
python app.py

7.2 测试流程

步骤	操作	预期
1. 单测钉钉	python dingtalk.py	群内收到「测试」消息
2. 上传火图	打开 http://127.0.0.1:18032/detect 上传带火焰图片	钉钉群 1 秒内收到 Markdown 告警
3. 冷却验证	1 分钟内再次上传同设备火图	不重复推送（dingtalk_pushed: false）
4. 无火图片	上传无火图片	不推送、不写库
5. MySQL 核验	SELECT * FROM alert_record ORDER BY detect_time DESC;	告警已落库

7.3 curl 快速验证

# 图片检测（会自动触发钉钉推送）
curl -F "image=@fire.jpg" -F "device_name=车间1号" \
     http://127.0.0.1:18032/api/detect/image

响应中 dingtalk_pushed: true 即推送成功。

---

八、常见问题

问题	原因 / 解决
errcode: 310000 或 keywords not in content	安全设置选了「自定义关键字」，消息内容需包含该关键字，或在机器人设置中改选「加签」
errcode: 310000，sign not match	加签失败：检查 Secret 是否与机器人设置一致；服务器时间是否与钉钉服务器误差过大
errcode: 310000，robot is blocked	机器人被群管理员停用，需在群设置中重新启用
errcode: 40035	JSON 格式错误：检查 msgtype 与 payload 字段拼写
每分钟 20 条限流	调大 ALERT_COOLDOWN_SECONDS，或升级为 Stream 模式机器人
收不到消息但接口返回 ok	检查钉钉群是否有人工服务/全员禁言；机器人是否在目标群

调试技巧：把 _post 返回的原始 JSON 打印出来（data.get('errcode')、errmsg），比对钉钉官方错误码表即可快速定位。

---

九、进阶：图片告警与定时汇总

9.1 推送带告警截图

检出后把画框结果图保存为本地文件，调用钉钉「上传媒体文件」接口拿到 media_id，再用 actionCard 消息发送图片：

# 1) 保存画框图（用 ultralytics 结果对象）
#    result.save('alert_xxx.jpg')

# 2) 上传媒体（POST multipart 到 https://oapi.dingtalk.com/media/upload）
#    参数: type=image, media=<文件>，需带 access_token
# 3) 发送 actionCard 消息引用 media_id

注意：钉钉自定义机器人不能直接发图片 URL，必须走 media/upload 拿 media_id。

9.2 定时汇总告警

配合 APScheduler（项目已有该依赖），每天固定时间汇总当天告警：

from apscheduler.schedulers.background import BackgroundScheduler

def daily_summary():
    # 查询今天 alert_record 数量，推送一条 Markdown 汇总
    pass

scheduler = BackgroundScheduler()
scheduler.add_job(daily_summary, trigger='cron', hour=21, minute=0)
scheduler.start()

---

十、完整文件清单

backend/
├── app.py               # Flask 入口（已接入钉钉推送）
├── config.py            # 新增钉钉相关配置
├── dingtalk.py          # ★ 新增：钉钉机器人封装（含加签）
├── notify_service.py    # ★ 新增：告警推送 + 冷却控制
├── model_service.py     # YOLO 推理封装（原样）
├── db.py                # MySQL（原样）
├── templates/           # 前端页面（原样）
└── uploads/             # 上传文件

---

集成完成后的完整链路：火情出现 → 模型检出 → MySQL 落库 → 钉钉群实时告警 → 值班人员第一时间响应。

您的关注就是我们前进的动力！一起学习进步！

VisionForgeSDK: VisionForge SDK 为用户提供新一代人工智能解决方案，释放数据的真正潜力；
1、火灾监测识别系统：可用于森林、厂区等防火区域；
2、垃圾监测识别系统：支持常见垃圾监测；
3、人脸轨迹提取系统：根据视频画面提取人员的时间活动轨迹，追踪目标；
4、智慧工地监测系统：实时监控施工场景，保障工人安全，提高管理效率；
5、头盔监测识别系统：头盔佩戴等
项目地址： https://gitee.com/51diysoft/VisionForgeSDK