从检测到告警：实时手机检测镜像对接企业微信/钉钉消息推送教程-开发者社区

从检测到告警：实时手机检测镜像对接企业微信/钉钉消息推送教程

1. 项目概述

1.1 系统简介

这是一个基于DAMO-YOLO和TinyNAS技术的实时手机检测系统，专门针对手机端低算力、低功耗场景优化设计。系统采用"小、快、省"的技术路线，能够在资源受限的环境中实现高效准确的手机检测。

核心特性：

检测准确率：88.8% (AP@0.5)
处理速度：约3.83毫秒/张图片
模型大小：仅125MB
支持实时检测和告警推送

1.2 应用场景扩展

除了基础的手机检测功能外，本教程重点介绍如何将检测结果实时推送到企业微信或钉钉，实现从检测到告警的完整闭环：

考场监控：检测到违规使用手机立即告警
会议管理：实时监控会议纪律
生产安全：在禁止使用手机的区域内进行监控
驾驶安全：检测驾驶员是否违规使用手机

2. 环境准备与部署

2.1 系统要求

在开始对接消息推送前，请确保手机检测系统已正确部署：

# 检查服务状态 supervisorctl status phone-detection # 正常输出示例 phone-detection RUNNING pid 12345, uptime 1:23:45

2.2 安装必要的依赖包

为了实现消息推送功能，需要安装额外的Python依赖：

# 进入项目目录 cd /root/phone-detection # 安装消息推送相关依赖 pip install requests python-dotenv # 或者添加到requirements.txt echo "requests" >> requirements.txt echo "python-dotenv" >> requirements.txt pip install -r requirements.txt

3. 企业微信消息推送配置

3.1 创建企业微信应用

首先需要在企业微信后台创建自建应用：

登录企业微信管理后台
进入「应用管理」→「自建应用」→「创建应用」
填写应用名称（如：手机检测告警）
记录以下关键信息：
- AgentId：应用ID
- Secret：应用密钥
- 企业ID：我的企业→企业信息→企业ID

3.2 获取访问令牌

在企业微信中，需要先获取access_token才能发送消息：

import requests import json def get_wecom_token(corp_id, corp_secret): """ 获取企业微信访问令牌 """ url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corp_id}&corpsecret={corp_secret}" response = requests.get(url) result = response.json() if result['errcode'] == 0: return result['access_token'] else: raise Exception(f"获取token失败: {result['errmsg']}") # 使用示例 corp_id = "你的企业ID" corp_secret = "你的应用Secret" access_token = get_wecom_token(corp_id, corp_secret)

3.3 发送检测告警消息

修改手机检测系统的代码，添加消息推送功能：

def send_wecom_message(access_token, user_id, detection_result): """ 发送企业微信消息 """ url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}" # 构建消息内容 message = { "touser": user_id, "msgtype": "text", "agentid": 你的AgentId, "text": { "content": f" 手机检测告警\n\n" f"检测时间: {detection_result['timestamp']}\n" f"检测数量: {detection_result['count']}部手机\n" f"置信度: {detection_result['confidence']}%\n" f"处理建议: 请及时查看核实" } } response = requests.post(url, json=message) result = response.json() if result['errcode'] == 0: print("消息发送成功") else: print(f"消息发送失败: {result['errmsg']}") # 在检测到手机后调用 def on_phone_detected(detection_data): # 获取access_token token = get_wecom_token(corp_id, corp_secret) # 发送告警消息 send_wecom_message(token, "@all", detection_data)

4. 钉钉消息推送配置

4.1 创建钉钉机器人

在钉钉群中点击「群设置」→「智能群助手」→「添加机器人」
选择「自定义」机器人
设置机器人名称（如：手机检测机器人）
选择「加签」安全设置，记录密钥
获取webhook地址

4.2 钉钉消息发送实现

import time import hmac import hashlib import base64 import urllib.parse def get_dingtalk_signature(secret): """ 生成钉钉签名 """ timestamp = str(round(time.time() * 1000)) secret_enc = secret.encode('utf-8') string_to_sign = '{}\n{}'.format(timestamp, secret) string_to_sign_enc = string_to_sign.encode('utf-8') hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) return timestamp, sign def send_dingtalk_message(webhook, secret, detection_result): """ 发送钉钉消息 """ timestamp, sign = get_dingtalk_signature(secret) message = { "msgtype": "markdown", "markdown": { "title": "手机检测告警", "text": f"## 手机检测告警 \n\n" f"**检测时间**: {detection_result['timestamp']} \n\n" f"**检测结果**: 发现 {detection_result['count']} 部手机 \n\n" f"**置信度**: {detection_result['confidence']}% \n\n" f"**处理建议**: 请及时查看核实 \n\n" f"---\n" f"![检测截图]({detection_result['image_url']})" } } # 构建带签名的webhook地址 url = f"{webhook}&timestamp={timestamp}&sign={sign}" response = requests.post(url, json=message) result = response.json() if result['errcode'] == 0: print("钉钉消息发送成功") else: print(f"钉钉消息发送失败: {result['errmsg']}") # 使用示例 webhook = "你的钉钉webhook地址" secret = "你的钉钉机器人密钥" send_dingtalk_message(webhook, secret, detection_data)

5. 完整集成方案

5.1 配置文件设置

创建配置文件.env来管理敏感信息：

# 企业微信配置 WECOM_CORP_ID=你的企业ID WECOM_AGENT_ID=你的应用ID WECOM_SECRET=你的应用Secret # 钉钉配置 DINGTALK_WEBHOOK=你的钉钉webhook DINGTALK_SECRET=你的钉钉密钥 # 消息推送设置 NOTIFICATION_ENABLED=true NOTIFICATION_THRESHOLD=0.8 # 置信度阈值

5.2 消息推送服务集成

在手机检测主程序中集成消息推送功能：

import os from dotenv import load_dotenv # 加载配置 load_dotenv() class NotificationService: def __init__(self): self.enabled = os.getenv('NOTIFICATION_ENABLED', 'true').lower() == 'true' self.threshold = float(os.getenv('NOTIFICATION_THRESHOLD', 0.8)) def send_notification(self, detection_data): """ 发送通知消息 """ if not self.enabled: return if detection_data['confidence'] < self.threshold: print("置信度低于阈值，不发送通知") return # 同时发送到企业微信和钉钉 try: # 企业微信通知 if all([os.getenv('WECOM_CORP_ID'), os.getenv('WECOM_SECRET')]): token = get_wecom_token( os.getenv('WECOM_CORP_ID'), os.getenv('WECOM_SECRET') ) send_wecom_message(token, "@all", detection_data) # 钉钉通知 if all([os.getenv('DINGTALK_WEBHOOK'), os.getenv('DINGTALK_SECRET')]): send_dingtalk_message( os.getenv('DINGTALK_WEBHOOK'), os.getenv('DINGTALK_SECRET'), detection_data ) except Exception as e: print(f"消息发送失败: {str(e)}") # 在检测回调中使用 notification_service = NotificationService() def detection_callback(image_path, results): """ 检测结果回调函数 """ detection_data = { 'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'), 'count': len(results), 'confidence': sum([r['confidence'] for r in results]) / len(results) if results else 0, 'image_url': f"file://{image_path}" # 或者上传到图床获取URL } # 发送通知 notification_service.send_notification(detection_data)

6. 高级功能与优化

6.1 消息频率控制

避免频繁发送消息，添加频率限制：

from collections import defaultdict import time class RateLimiter: def __init__(self, max_calls, period): self.max_calls = max_calls self.period = period self.calls = defaultdict(list) def allow(self, key): now = time.time() calls = self.calls[key] # 移除过期的记录 calls = [call for call in calls if now - call < self.period] self.calls[key] = calls if len(calls) < self.max_calls: calls.append(now) return True return False # 使用示例：每分钟最多发送2条消息 rate_limiter = RateLimiter(2, 60) if rate_limiter.allow('phone_detection'): notification_service.send_notification(detection_data) else: print("消息频率限制，暂不发送")

6.2 消息模板定制

支持多种消息模板，根据不同的检测场景发送不同类型的消息：

MESSAGE_TEMPLATES = { 'default': { 'wecom': " 检测到手机设备\n时间: {timestamp}\n数量: {count}部", 'dingtalk': "## 手机检测提醒\n\n检测时间: {timestamp}\n发现数量: {count}部手机" }, 'urgent': { 'wecom': "🚨 紧急告警！检测到多部手机\n时间: {timestamp}\n数量: {count}部\n请立即处理！", 'dingtalk': "## 🚨 紧急告警！\n\n检测到多部手机设备\n时间: {timestamp}\n数量: {count}部\n请立即核实处理！" } } def get_message_template(count): """根据检测数量选择消息模板""" if count >= 3: # 检测到3部以上手机使用紧急模板 return 'urgent' return 'default'

7. 故障排除与监控

7.1 消息发送状态监控

添加消息发送状态监控和重试机制：

def send_message_with_retry(message_func, *args, max_retries=3, **kwargs): """ 带重试的消息发送 """ for attempt in range(max_retries): try: message_func(*args, **kwargs) return True except Exception as e: print(f"消息发送失败 (尝试 {attempt + 1}/{max_retries}): {str(e)}") time.sleep(2 ** attempt) # 指数退避 return False # 使用带重试的消息发送 send_message_with_retry( send_wecom_message, access_token, "@all", detection_data, max_retries=3 )

7.2 日志记录与监控

完善日志记录，便于故障排查：

import logging # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/root/phone-detection/logs/notification.log'), logging.StreamHandler() ] ) logger = logging.getLogger('notification') # 在消息发送处添加日志 try: send_wecom_message(token, user_id, detection_data) logger.info(f"消息发送成功: {detection_data}") except Exception as e: logger.error(f"消息发送失败: {str(e)}")

8. 总结

通过本教程，你已经学会了如何将实时手机检测系统与企业微信和钉钉消息推送功能进行集成。这种集成方案可以帮助你：

实现的核心功能：

实时手机检测结果自动推送
多平台消息通知（企业微信+钉钉）
智能消息频率控制
完整的错误处理和重试机制

最佳实践建议：

在生产环境中使用配置文件和环境变量管理敏感信息
设置合理的消息发送频率，避免打扰用户
添加完整的日志记录，便于故障排查
根据实际业务场景定制消息模板
定期测试消息推送功能，确保可用性

下一步改进方向：

添加消息推送的开关配置
支持更多消息平台（飞书、短信等）
实现消息模板的自定义配置
添加消息发送的数据统计和报表

现在你的手机检测系统已经具备了从检测到告警的完整能力，可以在各种监控场景中发挥重要作用。

获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

从检测到告警：实时手机检测镜像对接企业微信/钉钉消息推送教程