基于Python构建本地天气自动监测与通知脚本
本方案旨在利用 Python 编写一个轻量级工具，通过当前网络 IP 自动识别所在城市，进而获取实时气象数据，并定期通过电子邮件将结果推送至指定终端。该系统主要包含地理位置解析、气象数据抓取、邮件发送服务及自动化调度四个核心部分。
一、实时地理位置检索模块
首先需要确定执行脚本机器的物理位置。我们将调用公共 IP 查询接口，将返回的 JSON 数据中的城市字段提取出来。为了增强稳定性，增加了重试机制。
"""
模块名：geo_locator.py
功能：通过IP地址解析当前所在的城市名称
"""
import requests
import time
API_ENDPOINT = "http://ip-api.com/json/"
def identify_current_city(lang="zh-CN"):
"""
发起请求并解析城市信息，包含自动重试逻辑
:return: 城市名称字符串或 None
"""
parameters = {"lang": lang}
max_retries = 2
for attempt in range(max_retries):
try:
# 使用GET请求替代POST以符合常规API规范
response = requests.get(API_ENDPOINT, params=parameters, timeout=8 + attempt * 5)
response.raise_for_status()
json_data = response.json()
if json_data.get('status') == 'success':
return json_data.get('city', '未知区域')
except (requests.RequestException, ValueError) as err:
print(f" [!] 第 {attempt+1} 次连接定位服务失败：{err}")
time.sleep(2)
print(" [!] 警告：无法获取有效的位置信息")
return None二、气象站点数据爬取策略
获得城市名称后，需要映射到具体的气象站编码，并访问气象网站获取详细预报。此处采用面向对象的方式封装 HTTP 请求与 HTML 解析过程，分离关注点。
"""
模块名：weather_engine.py
功能：获取城市天气详情并格式化输出
"""
import json
import re
import requests
from bs4 import BeautifulSoup
from geo_locator import identify_current_city
# 目标站点配置
SEARCH_API = "http://toy1.weather.com.cn/search"
FORECAST_URL_TMPL = "http://d1.weather.com.cn/weather_index/{cid}.html?_={ts}"
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
class WeatherScraper:
def __init__(self):
self.session = requests.Session()
self.session.headers.update({"User-Agent": USER_AGENT})
def resolve_station_id(self, city_name):
"""通过城市名查找气象局标准ID"""
try:
payload = f"cityname={city_name}"
resp = self.session.get(f"{SEARCH_API}?{payload}", timeout=5)
# 处理返回的JSONP格式数据
content = resp.text.strip()[1:-1]
data = json.loads(content)
if data:
return data[0]['ref'].split('~')[0]
except Exception as e:
print(f"[Error] 无法解析城市编码：{e}")
raise
return None
def parse_js_vars(self, html_content, var_names):
"""从JavaScript变量注入中提取特定数值"""
results = []
soup = BeautifulSoup(html_content, 'html.parser')
# 模拟遍历页面中的变量定义行
lines = str(soup).replace('\n', '').split(';')
for line in lines:
for v in var_names:
match = re.search(rf'{v}\s*=\s*(.*?)\s*,?', line)
if match:
val = match.group(1).strip('"')
results.append(val)
return results
def fetch_forecast_details(self, city_code):
"""抓取具体城市的天气报表"""
import time
ts = str(int(time.time() * 1000))
url = FORECAST_URL_TMPL.format(cid=city_code, ts=ts)
headers = {
"Referer": "http://www.weather.com.cn",
"User-Agent": USER_AGENT
}
try:
resp = self.session.get(url, headers=headers, timeout=10)
resp.encoding = 'ISO-8859-1' # 兼容网页编码
# 提取关键数据块
report_text = resp.text
# 此处简化正则逻辑用于匹配JS数据结构
temp_range = re.search(r'"temp":"(.*?)".*"tempn":"(.*)"', report_text)
current_cond = re.search(r'"weather":"(.*?)"', report_text)
aqi_val = re.search(r'"aqi":"(.*?)".*"aqi_pm25":"(.*)"', report_text)
if not all([temp_range, current_cond]):
return None
return {
"max_temp": temp_range.group(1),
"min_temp": temp_range.group(2),
"condition": current_cond.group(1),
"aqi": aqi_val.group(1) if aqi_val else "N/A",
"pm25": aqi_val.group(2) if aqi_val else "N/A"
}
except Exception as e:
print(f"[Fetch Error] {e}")
return None
def generate_report(self):
"""整合流程：定位 -> 查码 -> 抓数 -> 输出"""
city = identify_current_city()
if not city:
return ["定位服务不可用"]
station_code = self.resolve_station_id(city)
if not station_code:
return f"未找到城市[{city}]的气象编码"
data = self.fetch_forecast_details(station_code)
if not data:
return "未能拉取到实时天气数据"
lines = [
f"📍 监测地点：{city}",
f"🌤️ 实况天气：{data['condition']}",
f"🌡️ 气温范围：{data['min_temp']}℃ ~ {data['max_temp']}℃",
f"🧪 空气质量：AQI {data['aqi']} | PM2.5 {data['pm25']}"
]
return lines
if __name__ == "__main__":
bot = WeatherScraper()
print("\n".join(bot.generate_report()))三、邮件发送封装
为了方便运维人员随时查看，我们将最终生成的报告封装进 HTML 邮件，并通过 SMTP 协议投递。建议在代码中分离配置项以便部署管理。
"""
模块名：smtp_dispatcher.py
功能：构造邮件内容并发送至接收端
"""
import smtplib
from email.mime.text import MIMEText
from email.header import Header
from weather_engine import WeatherScraper
# 配置区域
MAIL_CONFIG = {
"host": "smtp.qq.com",
"port": 465,
"sender": "your_account@qq.com",
"password": "auth_code",
"recipients": ["admin@example.com", "ops@example.com"]
}
def build_email_body(report_lines):
"""将文本列表转换为HTML表格格式"""
header_rows = "<tr style='background:#f0f0f0'><th>指标</th><th>状态</th></tr>"
body_rows = ""
for item in report_lines:
parts = item.split(":", 1)
if len(parts) == 2:
label = parts[0]
value = parts[1].strip()
body_rows += f"<tr><td>{label}</td><td>{value}</td></tr>"
html_template = f"""
<html>
<body style="font-family: Arial, sans-serif;">
<h2>每日气象简报</h2>
<table border="1" cellpadding="5">
{header_rows}
{body_rows}
</table>
<p style="color:#888; font-size:12px;">由自动化监控脚本生成</p>
</body>
</html>
"""
return html_template
def send_alert():
main_scraper = WeatherScraper()
content = main_scraper.generate_report()
if isinstance(content, str):
print(content)
return False
msg = MIMEText(build_email_body(content), 'html', 'utf-8')
msg['Subject'] = Header("【环境监测】每日天气通报", 'utf-8')
msg['From'] = MAIL_CONFIG["sender"]
msg['To'] = ", ".join(MAIL_CONFIG["recipients"])
try:
server = smtplib.SMTP_SSL(MAIL_CONFIG["host"], MAIL_CONFIG["port"])
server.login(MAIL_CONFIG["sender"], MAIL_CONFIG["password"])
server.sendmail(MAIL_CONFIG["sender"], MAIL_CONFIG["recipients"], msg.as_string())
server.quit()
return True
except Exception as ex:
print(f"邮件投递异常：{ex}")
return False
if __name__ == "__main__":
if send_alert():
print("通知已发出")
else:
print("通知发送失败")四、自动化任务调度
为了实现无人值守运行，需将该脚本注册为操作系统的定时任务。以 Linux 系统为例，可以使用 Crontab 设置每天早上 8 点自动执行。
# 编辑crontab配置
crontab -e
# 添加以下行（假设脚本位于 /opt/tools/weather_check.py）
0 8 * * * /usr/bin/python3 /opt/tools/weather_check.py >> /var/log/weather.log 2>&1
```
上述指令表示每天 UTC 时间 8:00 执行一次 Python 脚本，并将标准输出和错误日志重定向到日志文件中，便于后续排查问题。
