在日常工作和生活中，及时了解天气预报和灾害预警信息非常重要。本文将介绍如何在 Ubuntu 22.04 系统中配置一个天气机器人脚本实现开机自启动，让它自动推送天气预报和灾害预警信息到企业微信群。
脚本功能介绍
这个天气机器人脚本具备以下核心功能：
- 定时推送全国主要城市的今明两天天气预报，包括白天和夜间的天气状况、温度范围、风向风力等信息
- 实时监测天气灾害预警，一旦有新的预警信息会及时推送，支持不同预警等级（蓝色、黄色、橙色、红色）的区分显示
- 避免重复推送相同的预警信息，自动清理过期预警记录
- 详细的日志记录功能，方便排查问题
配置开机自启动的步骤
一、准备工作
脚本存放与权限设置
首先，我们需要将脚本放置在一个固定的目录，建议放在/opt/weather_robot/目录下，并设置合适的权限：
创建目录
mkdir -p /opt/weather_robot
编写脚本（假设脚本名为weather_robot.py）
import requests
import json
import time
import schedule
import logging
from datetime import datetime, timedelta
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.FileHandler('weather_robot.log'), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)
# 配置信息
class Config:
# 企业微信群机器人Webhook地址，需替换为自己的
WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=76344cf4-f542-xxxx-xxxx-2680
12720aac"
# 要查询的城市列表及对应的LocationID（手动指定）
CITY_LOCATION_IDS = {
"上海": "101020100",
"北京": "101010100",
"广州": "101280101",
"深圳": "101280601",
"重庆": "101040100",
"天津": "101030100",
"成都": "101270101",
"青岛": "101120201",
"三亚": "101310201",
"西安": "101110101",
"昆明": "101290101",
"大连": "101070201",
"哈尔滨": "101050101",
"贵阳": "101260101",
"长沙": "101250101",
"福州": "101230101"
}
# 定时任务执行时间（24小时制）
SEND_TIME = "08:00" # 每天发送一次预报
WARNING_CHECK_INTERVAL = 60 # 灾害预警检查间隔(分钟)
# 每个消息最多包含的城市数量
CITIES_PER_MESSAGE = 8
# 预警等级颜色映射
WARNING_LEVEL_COLOR = {
"蓝色": "#1E90FF",
"黄色": "#FFFF00",
"橙色": "#FFA500",
"红色": "#FF0000"
}
# 天气API调用类
class WeatherAPI:
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like
Gecko) Chrome/91.0.4472.124 Safari/537.36'
})
# 存储已发送的预警信息，避免重复推送
self.sent_warnings = {} # 格式: {城市: {预警类型: 过期时间}}
def get_weather_forecast(self, city):
"""获取指定城市的今明两天天气预报"""
try:
# 使用和风天气API（需要申请KEY）
api_key = "00c6f0d7585e46bd8cd46736e09f9746" # 替换为自己申请的和风天气api_key
api_url = "https://n25u9va4vc.re.qweatherapi.com/v7/weather/7d" # 替换为自己申请的和风天气api_url
# 查询城市ID（从配置中获取）
location_id = Config.CITY_LOCATION_IDS.get(city)
if not location_id:
logger.error(f"未找到城市ID: {city}")
return None
params = {
"key": api_key,
"location": location_id
}
response = self.session.get(api_url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data.get("code") == "200":
return self._parse_weather_data(data, city)
else:
logger.error(f"API返回错误: {data.get('code')} - {data.get('message')}")
return None
except Exception as e:
logger.exception(f"获取{city}天气预报失败")
return None
def get_disaster_warnings(self, city):
"""获取指定城市的天气灾害预警信息"""
try:
api_key = "00c6xxxxxxxxxxxxxxxxx9746" # 替换为自己申请的和风天气api_key
api_url = "https://n2xxxx4vc.re.qweatherapi.com/v7/warning/now" # 替换为自己申请的和风天气api_url
location_id = Config.CITY_LOCATION_IDS.get(city)
if not location_id:
logger.error(f"未找到城市ID: {city}")
return None
params = {
"key": api_key,
"location": location_id
}
response = self.session.get(api_url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data.get("code") == "200":
return self._parse_warning_data(data, city)
else:
logger.error(f"预警API返回错误: {data.get('code')} - {data.get('message')}")
return None
except Exception as e:
logger.exception(f"获取{city}灾害预警失败")
return None
def _parse_weather_data(self, data, city):
"""解析今明两天天气数据"""
daily_forecasts = data.get("daily", [])[:2]
if len(daily_forecasts) < 2:
return None
today = daily_forecasts[0]
tomorrow = daily_forecasts[1]
return {
"city": city,
"today": {
"date": datetime.now().strftime('%m-%d'),
"weather_day": today.get("textDay", "未知"),
"weather_night": today.get("textNight", "未知"),
"temp_min": today.get("tempMin", "未知"),
"temp_max": today.get("tempMax", "未知"),
"wind_dir_day": today.get("windDirDay", "未知"),
"wind_scale_day": today.get("windScaleDay", "未知"),
"wind_dir_night": today.get("windDirNight", "未知"),
"wind_scale_night": today.get("windScaleNight", "未知")
},
"tomorrow": {
"date": (datetime.now() + timedelta(days=1)).strftime('%m-%d'),
"weather_day": tomorrow.get("textDay", "未知"),
"weather_night": tomorrow.get("textNight", "未知"),
"temp_min": tomorrow.get("tempMin", "未知"),
"temp_max": tomorrow.get("tempMax", "未知"),
"wind_dir_day": tomorrow.get("windDirDay", "未知"),
"wind_scale_day": tomorrow.get("windScaleDay", "未知"),
"wind_dir_night": tomorrow.get("windDirNight", "未知"),
"wind_scale_night": tomorrow.get("windScaleNight", "未知")
},
"update_time": data.get("updateTime", "")
}
def _parse_warning_data(self, data, city):
"""解析灾害预警数据"""
warnings = data.get("warning", [])
if not warnings:
return None
parsed_warnings = []
for warning in warnings:
# 计算预警过期时间（假设有效期3小时）
expire_time = datetime.now() + timedelta(hours=3)
parsed_warnings.append({
"city": city,
"type": warning.get("typeName", "未知预警"),
"level": warning.get("level", "未知等级"),
"text": warning.get("text", ""),
"issue_time": warning.get("pubTime", ""),
"expire_time": expire_time
})
return parsed_warnings
def check_and_send_new_warnings(self, robot):
"""检查并发送新的灾害预警"""
logger.info("开始检查天气灾害预警")
for city in Config.CITY_LOCATION_IDS.keys():
warnings = self.get_disaster_warnings(city)
if not warnings:
continue
# 清理已过期的预警记录
self._clean_expired_warnings()
for warning in warnings:
warning_key = f"{warning['type']}_{warning['level']}"
current_time = datetime.now()
# 检查是否已发送且未过期
if (city in self.sent_warnings and
warning_key in self.sent_warnings[city] and
self.sent_warnings[city][warning_key] > current_time):
continue
# 发送新预警
if self._send_warning_message(robot, warning):
# 记录已发送的预警
if city not in self.sent_warnings:
self.sent_warnings[city] = {}
self.sent_warnings[city][warning_key] = warning['expire_time']
time.sleep(1) # 避免API调用过于频繁
def _send_warning_message(self, robot, warning):
"""发送预警消息"""
level = warning['level']
color = Config.WARNING_LEVEL_COLOR.get(level, "#000000")
message = f"⚠️ **{warning['city']}发布{level}预警** ⚠️\n\n"
message += f"**预警类型**：{warning['type']}\n\n"
message += f"**预警内容**：\n{warning['text']}\n\n"
message += f"**发布时间**：{warning['issue_time']}\n"
message += f"<font color=\"{color}\">⚠️ 请相关地区人员注意防范！</font>"
return robot.send_markdown(message)
def _clean_expired_warnings(self):
"""清理已过期的预警记录"""
current_time = datetime.now()
cities_to_remove = []
for city, warnings in self.sent_warnings.items():
warnings_to_remove = [key for key, expire_time in warnings.items() if expire_time < curr
ent_time]
for key in warnings_to_remove:
del warnings[key]
if not warnings:
cities_to_remove.append(city)
for city in cities_to_remove:
del self.sent_warnings[city]
# 企业微信机器人类
class WeChatRobot:
def __init__(self, webhook_url):
self.webhook_url = webhook_url
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json'
})
def send_text(self, content):
"""发送文本消息"""
data = {
"msgtype": "text",
"text": {
"content": content
}
}
return self._send_message(data)
def send_markdown(self, content):
"""发送Markdown消息"""
data = {
"msgtype": "markdown",
"markdown": {
"content": content
}
}
return self._send_message(data)
def _send_message(self, data):
try:
response = self.session.post(self.webhook_url, json=data, timeout=10)
response.raise_for_status()
result = response.json()
if result.get("errcode") == 0:
logger.info("消息发送成功")
return True
else:
logger.error(f"消息发送失败: {result.get('errmsg')}")
return False
except Exception as e:
logger.exception("消息发送异常")
return False
# 主程序
def main():
config = Config()
weather_api = WeatherAPI()
robot = WeChatRobot(config.WEBHOOK_URL)
def send_weather_forecast():
"""发送今明两天天气预报"""
logger.info("开始发送今明两天天气预报")
cities = list(config.CITY_LOCATION_IDS.keys())
total_cities = len(cities)
city_chunks = [cities[i:i + config.CITIES_PER_MESSAGE] for i in
range(0, total_cities, config.CITIES_PER_MESSAGE)]
for chunk_index, city_chunk in enumerate(city_chunks):
message = f"📢 **全国主要城市“24H/48H”天气预报** (第{chunk_index + 1}/{len(city_chunks)}
部分)\n\n"
for city in city_chunk:
weather_data = weather_api.get_weather_forecast(city)
if weather_data:
message += f"### 🏙{weather_data['city']}\n"
today = weather_data['today']
tomorrow = weather_data['tomorrow']
message += f"**24H**（{today['date']}）：{today['weather_day']}转{today['weather
_night']}，{today['temp_min']}°C ~ {today['temp_max']}°C\n"
message += f"🌬️ 风力风向：白天 {today['wind_dir_day']}{today['wind_scale_day']}级
，夜间 {today['wind_dir_night']}{today['wind_scale_night']}级\n\n"
message += f"**48H**（{tomorrow['date']}）：{tomorrow['weather_day']}转{tomorrow
['weather_night']}，{tomorrow['temp_min']}°C ~ {tomorrow['temp_max']}°C\n"
message += f"🌬️ 风力风向：白天 {tomorrow['wind_dir_day']}{tomorrow['wind_scale_da
y']}级，夜间 {tomorrow['wind_dir_night']}{tomorrow['wind_scale_night']}级\n\n"
message += f"📅 更新时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
message += "数据来源：和风天气"
# 检查消息长度是否超过限制
if len(message) > 4096:
logger.warning(f"第{chunk_index + 1}部分消息长度{len(message)}超过4096，将尝试发送")
result = robot.send_markdown(message)
if not result:
logger.error(f"第{chunk_index + 1}部分消息发送失败，将尝试拆分为文本消息")
# 尝试作为文本消息发送
robot.send_text(message[:2048]) # 文本消息限制为2048字节
# 每条消息之间间隔1秒，避免频率限制
time.sleep(1)
# 设置定时任务
schedule.every().day.at(config.SEND_TIME).do(send_weather_forecast)
schedule.every(config.WARNING_CHECK_INTERVAL).minutes.do(
weather_api.check_and_send_new_warnings,
robot=robot
)
# 立即执行一次天气预报和预警检查
logger.info("程序启动，立即执行一次测试...")
send_weather_forecast()
weather_api.check_and_send_new_warnings(robot)
# 运行定时任务
logger.info(f"定时任务已设置：每天{config.SEND_TIME}发送今明两天天气预报")
logger.info(f"天气灾害预警检查间隔：每{config.WARNING_CHECK_INTERVAL}分钟")
while True:
schedule.run_pending()
time.sleep(60)
if __name__ == "__main__":
main()
赋予执行权限
chmod +x /opt/weather_robot/weather_robot.py
确保系统已安装 Python 及脚本所需的依赖库
安装Python3和pip
apt update && sudo apt install -y python3 python3-pip
安装脚本依赖
pip3 install requests schedule
创建 systemd 服务文件
执行以下命令创建weather-robot.service服务文件：
vim /etc/systemd/system/weather-robot.service
写入服务配置
在文件中添加以下内容，注意保持语法正确，不要在行内添加注释：
[Unit]
Description=Weather Robot Service (天气预报及灾害预警推送)
After=network.target
Wants=network.target
[Service]
User=root
WorkingDirectory=/opt/weather_robot
ExecStart=/usr/bin/python3 /opt/weather_robot/weather_robot.py
Restart=always
RestartSec=5
StandardOutput=append:/var/log/weather_robot/service.log
StandardError=append:/var/log/weather_robot/error.log
[Install]
WantedBy=multi-user.target
启动服务并设置开机自启
# 重载 systemd 配置，使新创建的服务文件生效：
systemctl daemon-reload
# 启动服务
systemctl start weather-robot.service
# 设置开机自启
systemctl enable weather-robot.service
服务管理常用命令
查看服务状态
systemctl status weather-robot.service -l
重启服务
systemctl restart weather-robot.service
停止服务
systemctl stop weather-robot.service
关闭开机自启
systemctl disable weather-robot.service
查看服务日志（实时）
tail -f /var/log/weather_robot/service.log
查看错误日志
cat /var/log/weather_robot/error.log
总结
通过本文介绍的步骤，我们可以在 Ubuntu 22.04 系统中成功配置天气机器人脚本的开机自启动。关键在于正确设置服务文件、确保相关路径存在且有权限、安装必要的依赖库。当遇到问题时，要善于查看日志文件，根据错误提示逐步排查和解决问题。
配置完成后，这个天气机器人将在系统启动时自动运行，持续为我们提供及时的天气预报和灾害预警信息，为工作和生活带来便利。