最近天气多变暴雨和高温交替出现很多经常户外活动的朋友已经开始感受到蚊虫的威力了。尤其是喜欢爬山、露营、钓鱼的朋友几乎人手一瓶驱蚊液还是不够用。作为一个既要写代码又经常往外跑的“老猎人”我就在想能不能搞一个自动化的防蚊提醒服务自动读取天气数据判断蚊子活跃程度在需要的时候推送到手机甚至联动智能驱蚊设备。这篇文章就完整分享一下这个系统的设计与实现思路全部代码都可以直接复用。1. 为什么天气会影响蚊虫滋生1.1 蚊虫繁育的气象条件蚊虫的整个生命周期都依赖温度和水分。蚊子从卵、幼虫、蛹到成虫幼虫阶段必须在水中完成。当天气持续降雨、排水不畅时各类小型积水就成了蚊子幼虫的天然温床。而温度又是蚊虫发育速度的关键变量气温较低时蚊虫发育缓慢甚至停止气温在25℃到30℃之间湿度又较高的环境下蚊子繁殖速度会明显加快。一个典型的恶劣天气场景是这样的连续几天降雨气温维持在28℃上下空气湿度接近80%。这种条件下积水面积扩大蚊子产卵量增加幼虫孵化周期缩短最终表现为户外成蚊密度骤增。因此如果我们能提前获取天气数据并结合温度、湿度、降水情况生成一个“蚊虫活跃指数”就可以在出门前得到防蚊提醒。1.2 防蚊提醒系统需要解决什么问题在日常开发中这类“采集数据 → 逻辑判断 → 主动通知”的流程很常见。防蚊提醒系统的核心价值是把零散的气象信息和用户的出行场景结合起来自动判断当天以及未来几小时是否需要加强防蚊措施。当判断结果为高风险时通过多种渠道推送提醒。支持与智能家居设备联动比如自动打开电蚊香液或超声波驱蚊器。这样下来用户不需要每天手动查湿度、看温度也不需要记忆复杂的蚊子活跃规律系统会替用户完成判断。1.3 技术选型思路本文的系统定位为轻量级个人/团队工具因此选择了以下技术组合模块选型说明开发语言Python 3生态丰富适合快速开发和定时任务天气数据和风天气 API提供实时天气、湿度、温度、降水等数据定时调度APScheduler支持按分钟、小时、每天定时触发检查通知推送钉钉机器人 Bark分别覆盖团队协作和企业用户、iPhone 个人用户设备联动MQTT Paho可连接智能插座、驱蚊器等 IoT 设备如果你不想引入重量级框架这个组合足够轻量并且每个环节都可以独立替换成你熟悉的服务。2. 环境准备与项目结构设计2.1 本地开发环境本文示例代码以 Python 3.10 为基准环境操作系统可以是 Windows、macOS 或 Linux。先用以下命令创建项目目录和虚拟环境mkdir mosquito-alert cd mosquito-alert python3 -m venv venv source venv/bin/activate # Windows 下可执行 venv\Scripts\activate然后安装依赖pip install requests apscheduler paho-mqtt这里对你的 Python 版本没有严格要求Python 3.8 以上基本都能运行。如果你使用的是更早的版本建议先把 Python 升级到 3.8避免部分语法不兼容。2.2 项目文件结构为了便于后续扩展我建议按下面这种方式组织项目mosquito-alert/ ├── config.py # 配置文件存放 API Key、目标城市、Webhook 地址 ├── weather.py # 天气数据获取与解析模块 ├── mosquito_index.py # 蚊虫活跃度计算模块 ├── notifier.py # 钉钉、Bark 消息推送模块 ├── iot_actions.py # MQTT 智能设备联动模块 ├── main.py # 主程序入口定时调度与任务编排 ├── requirements.txt # 依赖清单 └── logs/ # 运行日志目录拆分模块的好处是以后你想更换天气数据源只需要修改weather.py想新增飞书通知只需要在notifier.py中增加一个方法。层次清晰也方便测试。2.3 系统数据流整个系统的运行流程可以概括为定时触发每天指定时间 ↓ 调用天气 API 获取当前城市的气温、湿度、天气现象 ↓ 根据蚊虫活跃度规则计算威胁等级 ↓ 如果等级符合提醒条件则推送到钉钉、Bark ↓ 如果启用了设备联动则通过 MQTT 控制驱蚊设备这个流程不涉及复杂的人工交互一旦跑起来就完全自动化。3. 天气数据获取模块实现3.1 天气数据源选择防蚊判断最需要的数据是温度、相对湿度、降水情况。市面上的天气 API 有很多本文以和风天气为例因为它提供了开发友好的 JSON 接口免费版额度足够个人项目使用。使用前需要到和风天气控制台注册账号创建一个项目并获取 API Key。由于各平台接口会有调整本文的 URL 参数以官方文档为准主要演示请求和解析的思路。需要注意API Key 属于敏感信息千万不要提交到公开代码仓库。建议写入本地配置文件并添加.gitignore规则将其忽略。3.2 编写天气获取代码接下来在weather.py中写入以下代码# 文件路径mosquito-alert/weather.py import requests from config import WEATHER_API_KEY, CITY_ID class WeatherClient: 天气数据客户端用于获取指定城市的实时天气信息。 def __init__(self, api_key: str, city_id: str): self.api_key api_key self.city_id city_id self.base_url https://devapi.qweather.com/v7/weather/now def get_current_weather(self) - dict: 获取实时天气数据。 返回字段包括温度、相对湿度、天气现象编码等。 params { key: self.api_key, location: self.city_id, } try: resp requests.get(self.base_url, paramsparams, timeout10) resp.raise_for_status() data resp.json() if data.get(code) ! 200: raise RuntimeError(f天气接口返回错误码: {data.get(code)}) now data.get(now, {}) return { temperature: float(now.get(temp, 0)), humidity: float(now.get(humidity, 0)), text: now.get(text, ), wind_dir: now.get(windDir, ), } except requests.exceptions.Timeout: raise RuntimeError(天气接口请求超时) except requests.exceptions.RequestException as exc: raise RuntimeError(f天气接口请求失败: {exc}) if __name__ __main__: client WeatherClient(WEATHER_API_KEY, CITY_ID) print(client.get_current_weather())3.3 参数说明与异常处理上面的代码有几个细节值得说明location参数可以使用城市 ID 或者经纬度。使用城市 ID 时可以预先从和风天气城市列表接口获取使用经纬度则更灵活适合移动端定位场景。timeout10设置了请求超时时间避免因为网络波动导致程序长时间卡住。code字段是接口返回的状态码只有200才表示成功。实际开发中还应该针对401、402等错误码做单独日志记录便于排查配额和鉴权问题。另外要注意天气 API 的业务字段可能会随版本调整。比如temp字段在不同历史版本中可能是字符串或数字建议在解析时统一用float()转换避免类型导致的计算错误。4. 蚊虫活跃度计算模块4.1 活跃度等级划分蚊虫活跃度没有标准官方定义我们可以根据常见生物学知识和经验给出一个可解释的评分规则。本文采用“基础分 修正分”的方式计算威胁等级判断维度条件得分温度25℃ ~ 30℃40 分温度20℃ ~ 25℃ 或 30℃ ~ 35℃20 分相对湿度大于 70%30 分相对湿度55% ~ 70%15 分天气现象雨、暴雨、雷阵雨30 分天气现象阴、多云10 分根据总分将防蚊等级划分为低风险0 ~ 29 分常规出行无需额外防护。中风险30 ~ 59 分建议穿长袖长裤携带驱蚊液。高风险60 ~ 79 分强烈建议使用驱蚊液避免在草丛、水边长时间逗留。极高风险80 分及以上非必要不出门户外狩猎/露营活动需要做好全套防护。这里要额外强调一下以上评分规则是启发式规则并不是医学或昆虫学标准。它的作用是为自动化提醒提供可编程的判定逻辑实际项目中你可以根据自己的经验调整阈值。4.2 实现蚊虫指数计算在mosquito_index.py中写入以下代码# 文件路径mosquito-alert/mosquito_index.py from weather import WeatherClient class MosquitoIndex: def __init__(self): self.temperature_score_rules [ (25, 30, 40), (20, 25, 20), (30, 35, 20), ] self.humidity_score_rules [ (70, 100, 30), (55, 70, 15), ] self.rain_weather_texts [雨, 雷, 暴雨, 阵雨, 毛毛雨] def _temperature_score(self, temp: float) - int: for low, high, score in self.temperature_score_rules: if low temp high: return score return 0 def _humidity_score(self, humidity: float) - int: for low, high, score in self.humidity_score_rules: if low humidity high: return score return 0 def _weather_score(self, weather_text: str) - int: for keyword in self.rain_weather_texts: if keyword in weather_text: return 30 if 阴 in weather_text or 多云 in weather_text: return 10 return 0 def level_text(self, score: int) - str: if score 80: return 极高风险 if score 60: return 高风险 if score 30: return 中风险 return 低风险 def evaluate(self, weather: dict) - dict: 根据天气数据计算蚊虫活跃指数。 score 0 score self._temperature_score(weather[temperature]) score self._humidity_score(weather[humidity]) score self._weather_score(weather[text]) level self.level_text(score) return {score: score, level: level} if __name__ __main__: # 模拟一组天气数据进行测试 sample {temperature: 28.0, humidity: 80.0, text: 雷阵雨} index MosquitoIndex() print(index.evaluate(sample))运行这段测试代码可以看到输出{score: 100, level: 极高风险}这个例子中温度 28℃、湿度 80%、天气为雷阵雨三个条件叠加得分是 100 分属于极高风险系统会触发最高级别的防蚊提醒。4.3 提醒条件的判定逻辑提醒逻辑不宜过于频繁。比如每天定时提醒两次分别对应早晨出门前和傍晚户外活动高发时段。具体判定条件可以写在主程序中等级为“高风险”或“极高风险”时进行推送。等级为“中风险”时只在傍晚的第二次检查中推送。等级为“低风险”时不推送避免通知疲劳。这样的策略既保证了实用性也不会让用户在低风险天气中收到大量无用消息。5. 通知推送模块实现5.1 使用钉钉机器人推送钉钉自定义机器人是比较常用的团队通知渠道。创建方式为进入钉钉群 → 群设置 → 智能群助手 → 添加机器人 → 自定义机器人。创建后会得到 Webhook 地址如果你开启了“加签”安全设置还需要一个密钥。钉钉机器人签名计算方式是固定的流程当前时间戳 换行符 密钥再使用 HMAC-SHA256 算法计算签名。关键实现如下# 文件路径mosquito-alert/notifier.py import base64 import hashlib import hmac import time import urllib.parse import requests from config import DINGTALK_WEBHOOK, DINGTALK_SECRET def _generate_sign(timestamp: int) - str: 生成钉钉机器人签名。 string_to_sign f{timestamp}\n{DINGTALK_SECRET} hmac_code hmac.new( DINGTALK_SECRET.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code)) return sign def send_dingtalk_message(title: str, text: str) - bool: 向钉钉群推送 Markdown 消息。 注意需要在 webhook 中拼接签名参数。 timestamp str(round(time.time() * 1000)) sign _generate_sign(timestamp) webhook f{DINGTALK_WEBHOOK}timestamp{timestamp}sign{sign} payload { msgtype: markdown, markdown: { title: title, text: text, }, } resp requests.post(webhook, jsonpayload, timeout10) result resp.json() if result.get(errcode) 0: return True print(钉钉推送失败:, result) return False如果你的机器人不需要加签就可以直接使用原始 Webhook不拼timestamp和sign参数。安全方面建议生产环境务必开启加签同时不要把 Webhook 地址粘贴到公共仓库。5.2 使用 Bark 推送 iPhoneBark 是一个开源的个人推送工具适合 iPhone 用户。你只需要在 App 中注册一个推送地址格式通常是https://api.day.app/your_device_keyBark 的推送请求非常简单def send_bark_message(title: str, body: str) - bool: 通过 Bark 推送消息到 iPhone。 bark_url fhttps://api.day.app/{BARK_DEVICE_KEY} params { title: title, body: body, group: mosquito-alert, level: active, } resp requests.get(bark_url, paramsparams, timeout10) return resp.status_code 200这里使用 GET 请求即可完成推送适合在服务器或树莓派上运行。如果你想推送更复杂的内容比如携带图片或者跳转链接可以参考 Bark 官方文档调整参数。5.3 定时调度与主程序整合在main.py中我们将天气获取、指数计算、消息推送和定时调度整合起来# 文件路径mosquito-alert/main.py import logging from datetime import datetime from apscheduler.schedulers.blocking import BlockingScheduler from config import CITY_ID, ENABLE_BARK, ENABLE_DINGTALK, WEATHER_API_KEY from mosquito_index import MosquitoIndex from notifier import send_bark_message, send_dingtalk_message from weather import WeatherClient logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, ) logger logging.getLogger(__name__) weather_client WeatherClient(WEATHER_API_KEY, CITY_ID) mosquito_index MosquitoIndex() def check_and_notify(is_evening_check: bool False): 执行一次完整的防蚊指数检查并根据等级决定是否通知。 try: weather weather_client.get_current_weather() result mosquito_index.evaluate(weather) score result[score] level result[level] logger.info(f当前得分: {score}, 等级: {level}, 天气: {weather}) # 这里可以按需要调整提醒策略 if level 高风险 or level 极高风险: _send_alert(level, score, weather) elif level 中风险 and is_evening_check: _send_alert(level, score, weather) except Exception as exc: logger.exception(检查蚊虫指数时出现异常: %s, exc) def _send_alert(level: str, score: int, weather: dict): title f防蚊提醒{level}当前指数 {score} content ( f### 防蚊提醒\n\n f- 当前等级**{level}**\n f- 当前温度{weather[temperature]}℃\n f- 相对湿度{weather[humidity]}%\n f- 天气现象{weather[text]}\n\n f出门建议做好防蚊准备避免长时间在草丛、积水附近活动。 ) if ENABLE_DINGTALK: send_dingtalk_message(title, content) if ENABLE_BARK: send_bark_message(title, content) if __name__ __main__: scheduler BlockingScheduler(timezoneAsia/Shanghai) # 每天 8:00 和 18:00 检查天气数据 scheduler.add_job( check_and_notify, cron, hour8, minute0, args[False], idmorning_mosquito_check, ) scheduler.add_job( check_and_notify, cron, hour18, minute0, args[True], idevening_mosquito_check, ) logger.info(防蚊提醒服务启动...) scheduler.start()BlockingScheduler适合直接运行在云服务器命令行中。如果你的项目已经在使用其他 Web 框架也可以改用BackgroundScheduler放入进程内运行。5.4 运行效果启动服务后如果当天天气满足高风险条件钉钉群会收到一条 Markdown 格式的防蚊提醒手机端的 Bark 也会同步收到一条推送。推送内容示例### 防蚊提醒 - 当前等级高风险 - 当前温度29℃ - 相对湿度85% - 天气现象阵雨 出门建议做好防蚊准备避免长时间在草丛、积水附近活动。这种方式的优势在于消息内容直观用户可以快速看到为什么会被提醒。6. 扩展联动智能驱蚊设备6.1 使用 MQTT 控制设备如果家里已经有智能插座、智能驱蚊夜灯可以通过 MQTT 协议实现联动。比如当判断结果为高风险时自动打开智能插座上的电蚊香液设备。使用 Paho MQTT 客户端推送控制指令# 文件路径mosquito-alert/iot_actions.py import paho.mqtt.client as mqtt from config import MQTT_HOST, MQTT_PORT, MQTT_USERNAME, MQTT_PASSWORD, MQTT_TOPIC class SmartDeviceController: def __init__(self): self.client mqtt.Client() if MQTT_USERNAME: self.client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD) self.client.connect(MQTT_HOST, MQTT_PORT, keepalive60) def switch_device(self, action: str): 向智能插座发送控制指令。 action 可以传入 ON 或 OFF。 self.client.publish(MQTT_TOPIC, action) print(f已发送设备指令: {action}) def disconnect(self): self.client.disconnect()在主程序的_send_alert方法中可以根据等级调用switch_device(ON)当等级回落到低风险后再发送 OFF 指令。这样就从“被动接收提醒”升级成了“主动环境干预”体验会更好。6.2 IoT 联动的安全注意智能设备联动涉及真实电器使用时一定要遵守安全规范确保智能插座承载功率大于电蚊香器功率。不要将加热类设备长时间无人看管运行。MQTT 的账号密码不能使用默认弱口令。远程控制指令应当走加密连接或至少使用强认证。如果你只是做实验不一定需要真实设备可以使用 MQTT 客户端订阅测试主题验证消息是否正常发送。7. 部署到服务器与常见问题排查7.1 使用 systemd 部署到云服务器如果你的脚本需要长期运行推荐使用 systemd 托管进程。创建文件/etc/systemd/system/mosquito-alert.service[Unit] DescriptionMosquito Alert Service Afternetwork.target [Service] Userubuntu WorkingDirectory/home/ubuntu/mosquito-alert ExecStart/home/ubuntu/mosquito-alert/venv/bin/python /home/ubuntu/mosquito-alert/main.py Restartalways RestartSec10 [Install] WantedBymulti-user.target然后执行sudo systemctl daemon-reload sudo systemctl enable mosquito-alert sudo systemctl start mosquito-alert查看日志journalctl -u mosquito-alert -f使用 systemd 的好处是进程异常退出后会自动重启且每次开机都自动运行。如果你不想引入 systemd也可以用最简单的方式后台运行nohup python3 main.py logs/mosquito.log 21 不过这种方式重启策略较弱只适合临时使用。另外还要考虑定时任务的幂等性重复提醒场景下最好加入“当天已推送过同等级消息”的记录防止服务重启后重复刷屏。7.2 常见问题与排查表问题现象常见原因解决思路天气接口返回错误码API Key 错误、额度不足、参数格式不对检查 Key 是否有效确认城市 ID 或经纬度格式是否正确请求超时本地网络不通或接口不可达先使用curl测试接口再检查代码中的timeout设置钉钉推送失败Webhook 地址错误、签名计算失败确认加签参数是否齐全注意 URL 编码Bark 推送收不到设备 Key 错误、网络问题在浏览器直接访问推送 URL 测试定时任务不执行时区配置错误、任务被异常中断检查系统时区查看日志中的异常堆栈MQTT 控制无反应主题错误、设备离线使用 MQTT 客户端订阅主题验证消息是否到达7.3 工程最佳实践最后总结一些实际工程中值得注意的实践点配置文件尽量使用环境变量或本地配置文件不要硬编码在代码里。API Key、Webhook 密钥、MQTT 密码建议存放在独立环境变量文件或密钥管理服务中。所有外部请求都要有超时和异常兜底不能让单个接口异常导致整个调度任务崩溃。推送频率要有限制避免用户收到过多重复消息后关闭通知。日志需要有级别区分日常运行输出 INFO异常堆栈输出 ERROR。如果要暴露天气查询接口给其他人使用必须增加限流与鉴权防止接口被恶意刷取。这个系统的核心价值不是“把天气数据搬到手机上”而是把“什么时候需要防蚊”这个判断规则沉淀成可复用、可调整的自动化服务。你可以继续扩展的方向很多比如接入未来 24 小时降雨预测、增加用户位置信息、加入历史数据训练更精准的蚊子指数模型或者把提醒功能做成微信小程序。如果你只是想要一个能跑的脚本本文给出的代码已经够用如果你想往智能家居方向继续深入可以把 MQTT 联动模块和温湿度传感器结合起来做成真正的“室内外蚊子防控系统”。建议先把自己的城市天气跑通再逐步增加通知渠道和硬件联动这样可以避免一开始就把系统做复杂。希望这篇文章对你有用动手试一试吧。