从预警到行动：手把手教你用Python搭建个人暴雨预警与监控系统
简介
近期，贵州、重庆、湖南、湖北、江西等多地迎来强降雨，水利部、自然资源部、中央气象台等多部门联动，密集发布暴雨、山洪、地质灾害等高级别预警。对于普通公众，这些预警是重要的避险指南。而对于我们技术爱好者，尤其是Python开发者而言，这正是一个将“被动接收信息”转化为“主动监控与自动化响应”的绝佳实战场景。
本文将带你从零开始，利用Python搭建一套个人化的暴雨预警信息监控与处理系统。我们将模拟一个简化版的“预警数据中台”，学习如何获取、解析、可视化预警数据，并最终实现自动化消息推送。即使你只是初中级开发者，也能跟着本教程完成一个属于自己的、具有实用价值的预警工具。
前置准备
在开始之前，请确保你的环境满足以下条件：
- Python 环境：安装 Python 3.7 或更高版本。
- 代码编辑器：推荐使用 VSCode 或 PyCharm 等，能极大提升编码效率。
- 基础知识：了解Python基础语法、HTTP请求、JSON数据格式。
- 必要的库：我们将使用
requests（用于网络请求）、pandas（用于数据处理）、folium（用于地图可视化）。可以通过 pip 安装：
bash
pip install requests pandas folium - 硬件建议：进行地图渲染和数据处理时，一台性能不错的笔记本电脑会让体验更流畅。对于需要7×24小时运行的监控任务，可以考虑使用低功耗的 树莓派 作为服务器。
分步骤教程
## 第一步：数据获取 – 模拟接入预警数据源
真实的国家级预警系统接口通常不直接对外开放。为了教学，我们将模拟一个结构清晰的预警数据源。在实际应用中，你可以尝试从一些提供公开数据的气象服务或爬取（需注意合规性）官方发布的文本信息，并自行解析。
我们创建一个本地的 mock_data.json 文件，模拟接收到的多部门预警信息：
[
{
"id": "WARN20230621180001",
"department": "中央气象台",
"type": "暴雨红色预警",
"level": "一级",
"region": ["湖南省北部", "湖北省东部"],
"description": "预计6月21日20时至22日20时，湖南北部、湖北东部等地部分地区有大暴雨...",
"issue_time": "2023-06-21 18:00:00",
"expire_time": "2023-06-22 20:00:00",
"coordinates": {
"center_lat": 29.5,
"center_lon": 113.3,
"radius_km": 200
}
},
{
"id": "WARN20230621180002",
"department": "水利部",
"type": "山洪灾害气象预警",
"level": "二级",
"region": ["贵州省北部", "重庆市东南部"],
"description": "预计6月21日20时至22日20时，贵州北部、重庆东南部等地可能发生山洪...",
"issue_time": "2023-06-21 18:00:00",
"expire_time": "2023-06-22 20:00:00",
"coordinates": {
"center_lat": 28.8,
"center_lon": 107.5,
"radius_km": 250
}
}
]
现在，我们用Python代码来读取并初步解析这份“预警数据”：
import json
import pandas as pd
# 1. 加载模拟数据
with open('mock_data.json', 'r', encoding='utf-8') as f:
warning_data = json.load(f)
# 2. 使用pandas转换为DataFrame，便于后续处理
df = pd.DataFrame(warning_data)
# 3. 简单查看数据
print("预警信息概览：")
print(df[['id', 'department', 'type', 'level', 'region']])
print("\n最新预警详情：")
print(f"部门: {df.iloc[0]['department']}")
print(f"预警类型: {df.iloc[0]['type']} ({df.iloc[0]['level']})")
print(f"影响区域: {', '.join(df.iloc[0]['region'])}")
print(f"生效时间: {df.iloc[0]['issue_time']}")
## 第二步：数据可视化 – 在地图上标记预警区域
文字信息不够直观，让我们将预警区域在地图上可视化出来。这里我们使用 folium 库来创建一个交互式的地图。
import folium
# 创建一个以中国为中心的地图
m = folium.Map(location=[35.0, 105.0], zoom_start=5, tiles='https://mt{s}.google.com/vt/lyrs=m&x={x}&y={y}&z={z}', attr='Google')
# 根据预警等级设置颜色
def get_color(level):
if level == '一级':
return 'red'
elif level == '二级':
return 'orange'
else:
return 'blue'
# 遍历每条预警，在地图上添加标记和圆圈
for index, row in df.iterrows():
coords = row['coordinates']
popup_content = f"""
<b>{row['type']}</b><br>
<b>发布部门:</b> {row['department']}<br>
<b>影响区域:</b> {', '.join(row['region'])}<br>
<b>有效时间:</b> {row['issue_time']} 至 {row['expire_time']}<br>
<b>描述:</b> {row['description']}
"""
# 添加圆形标记区域
folium.Circle(
location=[coords['center_lat'], coords['center_lon']],
radius=coords['radius_km'] * 1000, # 将千米转换为米
color=get_color(row['level']),
fill=True,
fill_opacity=0.2,
popup=folium.Popup(popup_content, max_width=300),
tooltip=f"{row['type']} - {row['department']}"
).add_to(m)
# 保存地图为HTML文件
m.save('warning_map.html')
print("预警地图已生成，请用浏览器打开 warning_map.html 查看。")
运行这段代码后，打开生成的 warning_map.html 文件，你将看到一个标记着两个预警区域（红色和橙色圆圈）的交互式地图，点击圆圈可以查看详细信息。
## 第三步：自动化通知 – 建立你的预警哨兵
监控的核心在于“自动”。我们可以编写一个脚本，定期检查数据源（在实际应用中是真实API），当检测到符合我们关心条件（例如，影响到我们所在城市、预警级别高）时，自动发送通知。
这里以发送到企业微信机器人为例（你需要自行替换webhook_url）：
import requests
import time
import hashlib
from datetime import datetime
# 配置部分
WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY" # 替换为你的真实Webhook地址
MY_CITY = "长沙" # 设置你关心的城市
CHECK_INTERVAL = 600 # 每10分钟检查一次，单位秒
def send_wechat_message(content):
"""发送企业微信机器人消息"""
headers = {'Content-Type': 'application/json'}
data = {
"msgtype": "markdown",
"markdown": {
"content": content
}
}
try:
resp = requests.post(WEBHOOK_URL, json=data, headers=headers, timeout=5)
if resp.json().get('errcode') == 0:
print(f"[{datetime.now()}] 消息发送成功")
else:
print(f"[{datetime.now()}] 消息发送失败: {resp.json()}")
except Exception as e:
print(f"[{datetime.now()}] 发送异常: {e}")
def check_for_new_warnings(last_check_id):
"""模拟检查新预警，并返回新预警列表和最新的检查ID"""
# 在实际应用中，这里会调用真实API
# 当前模拟数据是静态的，我们通过ID来模拟“新”数据
# 假设初始last_check_id为空，第一次运行会处理所有数据
new_warnings = []
current_latest_id = last_check_id
for index, row in df.iterrows():
# 简单判断：如果当前行ID不等于上次检查的最新ID，且影响区域包含我们关心的城市
if row['id'] != last_check_id and MY_CITY in ','.join(row['region']):
new_warnings.append(row)
current_latest_id = max(current_latest_id, row['id']) # 简单取最大ID
return new_warnings, current_latest_id if new_warnings else last_check_id
def format_warning_message(warning):
"""将预警数据格式化为Markdown消息"""
message = f"""**🌧 暴雨预警自动监控提醒**
> **预警类型**: <font color="warning">{warning['type']}</font> ({warning['level']})
> **发布部门**: {warning['department']}
> **影响区域**: {', '.join(warning['region'])}
> **生效时间**: {warning['issue_time']}
> **失效时间**: {warning['expire_time']}
> **预警详情**: {warning['description']}
> **请密切关注天气变化，做好防范措施！**"""
return message
# 主监控循环（示例）
last_check_id = ""
print("预警监控服务启动，正在监听...")
while True:
try:
new_warnings, last_check_id = check_for_new_warnings(last_check_id)
if new_warnings:
print(f"[{datetime.now()}] 检测到 {len(new_warnings)} 条新预警！")
for warn in new_warnings:
msg = format_warning_message(warn)
send_wechat_message(msg)
# 这里可以添加更多动作，如记录日志、发送邮件等
else:
print(f"[{datetime.now()}] 暂无影响{MY_CITY}的新预警。")
except Exception as e:
print(f"[{datetime.now()}] 监控循环异常: {e}")
time.sleep(CHECK_INTERVAL) # 等待下一次检查
你可以将这个脚本放在一台常开的电脑或树莓派上运行。在移动端，你也可以使用智能手机上的Pythonista等APP来运行简化版。如果你更喜欢桌面提醒，可以将通知方式改为系统弹窗（如使用 win10toast 库）。
相关工具推荐
除了代码开发，善用现有工具能让你的生活更便捷。在应对极端天气时，这些产品能提升你的体验：
- 机械键盘：长时间编写和调试监控代码，一把手感舒适的机械键盘能有效提升效率，减少疲劳。
- 笔记本电脑：一台续航好、屏幕素质高的笔记本是移动开发和随时处理预警通知的得力助手。
- 摄像头：如果你关心家附近的道路积水或河流水位情况，安装一个室外防水摄像头，结合简单的图像分析脚本，可以构建更直观的监控。
- 智能手机：接收各类官方预警App（如“国家预警发布”）的推送，是第一道防线。确保其通知权限已开启。
- 空气炸锅：在因暴雨不便外出时，用它快速准备一些简单食物，保持体力。
常见问题
Q1: 我获取的真实预警数据源在哪里？
A: 中国气象局、水利部等部门的官网会发布文本信息，但通常没有标准的API接口。一些第三方商业气象服务（如和风天气、彩云天气）提供付费或免费的预警API。请务必遵守数据使用协议。
Q2: 地图中的圆圈位置不准确怎么办？
A: 本教程中的模拟数据坐标是随意给的。要获得精确坐标，你需要一个将“地名文本”转换为经纬度的地理编码服务（Geocoding），如高德地图、百度地图的API。这需要额外申请开发者密钥。
Q3: 监控脚本如何实现开机自启和后台运行？
A: 在Windows上可以使用“任务计划程序”；在Linux/macOS上，可以编写一个systemd service文件或使用 nohup 命令配合 & 让脚本在后台运行。
Q4: 除了企业微信，还能推送到哪里？
A: 绝大多数即时通讯软件都提供机器人Webhook（如钉钉、飞书）。对于个人，也可以发送邮件（使用 smtplib）、Telegram消息，甚至通过手机厂商的推送服务（如小米、华为的Push）来发送到手机。
总结
通过本教程，我们完成了一次从数据到行动的技术实践：从模拟接入预警数据，到利用 pandas 进行分析，用 folium 实现空间可视化，最后用自动化脚本搭建起预警通知系统。整个过程涵盖了数据处理、API调用、可视化、自动化等多个开发核心技能。
更重要的是，这个项目具有极强的扩展性和实用性。你可以：
– 接入真实数据，监控你所在省份或城市。
– 扩展预警类型，加入地震、台风、高温等。
– 增加复杂逻辑，如结合历史数据分析降雨趋势，或与智能家居联动（下雨自动关窗）。
– 包装成小型服务，为家人朋友提供订阅式预警推送。
技术服务于生活。希望这个小项目不仅能提升你的编程能力，也能在风雨来临之际，为你和关心的人提供一份多一层的安全保障。现在，就打开你的编辑器，开始构建你的“私人气象预警中心”吧！