在 Linux 中配置天气机器人脚本开机自启动的完整指南

在日常工作和生活中，及时了解天气预报和灾害预警信息非常重要。本文将介绍如何在 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, likeGecko) Chrome/91.0.4472.124 Safari/537.36'})# 存储已发送的预警信息，避免重复推送self.sent_warnings = {}  # 格式: {城市: {预警类型: 过期时间}}def get_weather_forecast(self, city):"""获取指定城市的今明两天天气预报"""try:# 使用和风天气API（需要申请KEY）api_key = "00c6f0d7585e46bd8cd46736e09f9746"   # 替换为自己申请的和风天气api_keyapi_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 Noneparams = {"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 Noneexcept Exception as e:logger.exception(f"获取{city}天气预报失败")return Nonedef get_disaster_warnings(self, city):"""获取指定城市的天气灾害预警信息"""try:api_key = "00c6xxxxxxxxxxxxxxxxx9746" # 替换为自己申请的和风天气api_keyapi_url = "https://n2xxxx4vc.re.qweatherapi.com/v7/warning/now"  # 替换为自己申请的和风天气api_urllocation_id = Config.CITY_LOCATION_IDS.get(city)if not location_id:logger.error(f"未找到城市ID: {city}")return Noneparams = {"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 Noneexcept Exception as e:logger.exception(f"获取{city}灾害预警失败")return Nonedef _parse_weather_data(self, data, city):"""解析今明两天天气数据"""daily_forecasts = data.get("daily", [])[:2]if len(daily_forecasts) < 2:return Nonetoday = 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 Noneparsed_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_warningsdef 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 andwarning_key in self.sent_warnings[city] andself.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_urlself.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 Trueelse:logger.error(f"消息发送失败: {result.get('errmsg')}")return Falseexcept 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 inrange(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 系统中成功配置天气机器人脚本的开机自启动。关键在于正确设置服务文件、确保相关路径存在且有权限、安装必要的依赖库。当遇到问题时，要善于查看日志文件，根据错误提示逐步排查和解决问题。
配置完成后，这个天气机器人将在系统启动时自动运行，持续为我们提供及时的天气预报和灾害预警信息，为工作和生活带来便利。