AI视觉识别+钉钉推送：构建智能水质监测系统

前言

之前我实现了用AI 识别视频截图内容，现在进一步进行优化，将模型分析结果推动到钉钉群，实现 AI 的业务应用。

背景与挑战

随着环境保护意识的不断提升，水质监测已成为城市管理的重要组成部分。传统的水质监测主要依赖人工巡检和定点采样，这种方式存在效率低、成本高、实时性差等问题。特别是在河流、湖泊等大面积水域环境中，如何实现高效、准确、实时的水质监测成为了一个重要挑战。

现代城市拥有大量的视频监控设备，这些设备如果能够结合 AI 技术进行智能分析，就能实现对水质状况的自动化监测。本文将介绍如何构建一个基于 AI 视觉识别和钉钉推送的智能水质监测系统，利用现有视频资源和开源技术，实现低成本、高效率的智能监控。

系统架构与核心技术

整体架构设计

我们的智能水质监测系统采用模块化设计，主要包括以下几个核心组件，协同工作以实现水质的自动化监测与告警：

视频流获取模块：负责从海康威视等监控设备获取实时视频流。
图像处理模块：从视频流中截取关键帧，并进行必要的图像预处理。
AI 分析模块：利用先进的大模型对截取的图像进行智能分析，判断水质状况。
数据存储与图床模块：将分析结果和视频截图保存到本地，并利用本地部署的开源 MinIO 作为图床，为钉钉消息提供可公开访问的图片链接。
消息推送模块：通过钉钉机器人将 AI 分析结果和相关截图实时推送到指定的管理群，实现即时告警。

核心技术实现

本系统通过以下关键技术实现智能水质监测的完整流程：

1. 视频流获取与图像截取

系统首先通过 API 接口获取指定监控设备的实时视频流地址。随后，利用视频处理技术连接视频流，并从中截取高质量的视频帧。为确保获取到最新的画面，系统会尝试读取多帧并选取最新的一帧作为分析对象，同时设置超时机制以应对网络不稳定情况，提高系统健壮性。

2. AI 视觉分析技术

系统采用如 THUDM/GLM-4.1 V-9 B-Thinking 等高性能视觉大模型进行图像分析。通过精心设计的提示词模板，将 AI 模型转化为专业的流域水质监测专家。提示词明确定义了正常水体状态和各类异常问题（如水体浑浊但未发黑、疑似污染源汇入、排污口污水、水面漂浮物、藻类爆发、无明显问题等），并要求 AI 模型描述画面现象、判断问题类型并给出初步处理建议。这种方式极大地提高了 AI 分析的准确性和实用性。

3. 本地图床与数据存储

为了实现成本节约和数据自主可控，系统采用本地部署的开源对象存储服务 MinIO 作为图床。截取到的视频截图在本地保存的同时，也会被上传至 MinIO。MinIO 提供了一个可公开访问的 URL，使得钉钉消息能够直接引用这些图片，实现图文并茂的告警。此外，AI 分析的文本结果也会与图片一同保存到本地，便于后续审计和数据管理。

minio

4. 钉钉消息推送技术

系统利用钉钉机器人 Webhook 接口，将 AI 分析结果实时推送到预设的管理群。推送消息采用钉钉支持的 Markdown 格式，能够将 AI 生成的文本分析报告与 MinIO 提供的视频截图 URL 结合，形成内容丰富、直观易懂的告警信息。这种方式确保了管理人员能够第一时间通过手机或电脑接收到详细的水质异常报告，包括问题描述、AI 判断和现场截图。

添加机器人

系统优势与创新点

1. 实时性与自动化

支持定时任务执行，可根据需求（如每天 8 点、12 点、16 点）自动启动监测。
实时获取视频流并进行 AI 分析，一旦发现问题立即通过钉钉推送告警。
系统实现全天候自动化运行，无需人工干预，显著提升监测效率。

2. 智能化分析与丰富内容

利用先进的大模型视觉理解能力，准确识别多种水质异常情况，如水体浑浊、漂浮物、藻类爆发、疑似污染源等。
AI 模型不仅识别问题，还能提供专业的分析建议和初步处理方案。
通过钉钉 Markdown 消息，将 AI 分析文本与 MinIO 图床上的视频截图结合，提供图文并茂的告警报告，信息更直观、更全面。

3. 成本节约与便捷管理

充分利用现有视频监控设备，无需额外硬件投入。
采用本地部署的 MinIO 作为图床，避免了第三方云存储的费用，实现“零成本”或极低成本的图片存储与分发。
通过钉钉推送，管理人员可以随时随地通过移动设备掌握水质状况，极大提升了管理的便捷性和响应速度。
支持多站点同时监测，集中管理，降低运维成本。

实际应用效果

在实际部署中，该系统已经成功监测了多个重要水域点位，系统能够准确识别水体浑浊、漂浮物、藻类爆发等常见问题，并及时推送告警信息，大大提高了水质监测的效率和准确性。通过 MinIO 和钉钉 Markdown 的结合，告警信息更加生动直观，提升了决策效率。

关键代码

1. 获取视频流并截取帧

系统通过调用海康威视平台的开放 API 获取实时视频流地址，并使用 OpenCV 进行帧提取。为提升稳定性，加入了重试机制和超时控制。

import requests
import cv2
import time

def get_video_stream(device_id):
    """获取指定设备的实时视频流地址"""
    api_url = "http://119.84.149.201:18088/api/monitor/openapi/getVideoStreamByVideoPrintIdOrVideoDeviceCode"
    payload = [device_id]
    
    for attempt in range(3):
        try:
            response = requests.post(api_url, json=payload, timeout=15)
            if response.status_code == 200:
                data = response.json()
                if data.get("code") == 200 and data["data"]:
                    return data["data"][0]["videoStream"]
            time.sleep(5)
        except Exception as e:
            print(f"获取视频流失败 (第{attempt+1}次): {str(e)}")
            if attempt == 2:
                raise
    return None

def capture_frame(video_url):
    """从视频流中截取一帧"""
    cap = cv2.VideoCapture(video_url)
    cap.set(cv2.CAP_PROP_OPEN_TIMEOUT_MSEC, 10000)
    cap.set(cv2.CAP_PROP_READ_TIMEOUT_MSEC, 10000)

    frame = None
    for _ in range(5):  # 尝试读取多帧，取最后一帧
        ret, temp_frame = cap.read()
        if ret:
            frame = temp_frame
        time.sleep(0.5)
    
    cap.release()
    if frame is None:
        raise IOError("无法从视频流中读取有效帧")
    
    return cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)  # 转为RGB格式供PIL使用

2. 调用 AI 模型进行视觉分析

使用 requests 调用 SiliconFlow 提供的 GLM-4 V 大模型 API，通过 Base 64 编码图像并构造多模态请求体。

import base64
from PIL import Image
import io
import json

def analyze_image_with_ai(image: Image.Image, prompt: str, api_key: str):
    """调用AI模型分析图像内容"""
    # 图像转Base64
    buffered = io.BytesIO()
    image.save(buffered, format="JPEG")
    img_str = base64.b64encode(buffered.getvalue()).decode()

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    
    payload = {
        "model": "THUDM/GLM-4.1V-9B-Thinking",
        "messages": [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/jpeg;base64,{img_str}"}
                    }
                ]
            }
        ],
        "temperature": 0.7,
        "top_p": 0.9,
        "max_tokens": 512
    }

    response = requests.post(
        "https://api.siliconflow.cn/v1/chat/completions",
        json=payload,
        headers=headers,
        timeout=90
    )
    
    if response.status_code == 200:
        return response.json()["choices"][0]["message"]["content"]
    else:
        raise Exception(f"AI分析失败: {response.status_code}, {response.text}")

3. 上传图片至 MinIO 图床

使用 minio-py SDK 将截图上传至本地 MinIO 服务，生成可公开访问的 URL。

from minio import Minio
from minio.error import S3Error

minio_client = Minio(
    "172.21.92.37:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
    secure=False
)

def upload_to_minio(image_path, object_name):
    """上传图片到MinIO并返回公网URL"""
    try:
        if not minio_client.bucket_exists("hikvis"):
            minio_client.make_bucket("hikvis")
        
        minio_client.fput_object(
            "hikvis", object_name, image_path, content_type="image/jpeg"
        )
        return f"http://113.204.228.66:17999/files/hikvis/{object_name}"
    except S3Error as e:
        print(f"MinIO上传失败: {e}")
        return None

4. 钉钉机器人推送 Markdown 消息

通过钉钉自定义机器人发送图文告警，支持加签验证以提升安全性。

import hmac
import hashlib
import urllib.parse

def send_dingtalk_markdown(title, text, image_url=None):
    """发送Markdown格式的钉钉消息"""
    timestamp = str(round(time.time() * 1000))
    secret = "6feec97ee98b0f7e87a52e0fc34c1c6351f28b6253be995e065a49173f8546c4"
    string_to_sign = f"{timestamp}\n{secret}"
    hmac_code = hmac.new(
        secret.encode("utf-8"),
        string_to_sign.encode("utf-8"),
        digestmod=hashlib.sha256
    ).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))

    webhook = f"https://oapi.dingtalk.com/robot/send?access_token=6feec97ee98b0f7e87a52e0fc34c1c6351f28b6253be995e065a49173f8546c4&timestamp={timestamp}&sign={sign}"

    content = text
    if image_url:
        content += f"\n\n![现场截图]({image_url})"

    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": content
        }
    }

    requests.post(webhook, json=payload, timeout=10)

结语

AI 视觉识别结合钉钉推送的智能水质监测系统，为现代城市环境管理提供了一种高效、智能、低成本的解决方案。通过充分利用现有的视频监控资源，结合先进的人工智能技术和本地部署的 MinIO 图床，实现了对水质状况的自动化监测和管理。

随着技术的不断发展和完善，这类智能监测系统将在环境保护、智慧城市等领域发挥越来越重要的作用，为建设美丽中国贡献力量。

未来，我们将继续优化系统性能，扩展应用场景，让 AI 技术更好地服务于环境保护事业。

生态环境

#经验分享 #agent #AI应用

AI视觉识别+钉钉推送：构建智能水质监测系统

https://maoyu92.github.io/2025/09/02/02 生态环境/AI视觉识别+钉钉推送：构建智能水质监测系统/

作者

陈文茂

发布于

2025年9月2日

许可协议

n8n实战：从零搭建智能水质查询机器人上一篇

Jupyter 中实现交互式图表：ipywidgets 从入门到部署下一篇