前言

这几天天气无常变化快，写了一个青龙脚本监控天气，当有雨时推送消息到手机上提醒带雨伞。配置方法很简单。现把方法和代码分享出来。

教程如下：

1、点此输入城市名获取经纬度备用。如下图：

2、进青龙面板，新建一个脚本命名weather_alert.py，把下面的代码粘贴进去。修改上面获取到的经纬度。如下图：

3、安装青龙依赖。requests

4、创建定时任务，如下图：

5、部署后手动创建记录文件，进Docker容器运行bash命令。

touch /ql/data/scripts/last_push.txt
chmod 666 /ql/data/scripts/last_push.txt

6、最后在环境变量中配置pashdeer推送信息就可以使用了

• PUSHDEER_PUSHKEY=“秘钥”

备注：代码写死了首次推送通知后，再次推送间隔须2小时，并且运行时间为每天6点到20点，每30分钟运行一次。其它时间不运行。

最后附上脚本代码：

#!/usr/bin/env python3
# 青龙依赖：requests
import os
import requests
from datetime import datetime, timedelta
# 配置参数
PUSHDEER_PUSHKEY = os.environ.get("PUSHDEER_PUSHKEY")
HANCHUAN_COORDS = (30.6500, 113.8300)
LAST_PUSH_FILE = "/ql/data/scripts/last_push.txt"  # 注意路径匹配青龙目录
PUSH_INTERVAL = timedelta(hours=2)
def get_weather_forecast():
    """获取未来2小时降水预报"""
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": HANCHUAN_COORDS[0],
        "longitude": HANCHUAN_COORDS[1],
        "hourly": "weathercode,precipitation_probability",
        "forecast_hours": 2,
        "timezone": "Asia/Shanghai"
    }

    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        data = response.json()

        if "hourly" in data:
            return {
                "time": data["hourly"]["time"][:2],
                "weathercode": data["hourly"]["weathercode"][:2],
                "precip_prob": data["hourly"]["precipitation_probability"][:2]
            }
        return None
    except Exception as e:
        print(f"[错误] 获取天气数据失败: {str(e)}")
        return None
def should_alert(forecast):
    """判断是否需要提醒带伞"""
    RAIN_CODES = {61, 63, 65, 66, 67, 80, 81, 82, 95, 96, 99}

    for i in range(2):
        if forecast["weathercode"][i] in RAIN_CODES:
            return True
        if forecast["precip_prob"][i] >= 40:
            return True
    return False
def check_push_cooldown():
    """检查推送冷却时间"""
    try:
        if not os.path.exists(LAST_PUSH_FILE):
            return True

        with open(LAST_PUSH_FILE, "r") as f:
            last_push_str = f.read().strip()

        last_push_time = datetime.fromisoformat(last_push_str)
        return datetime.now() - last_push_time >= PUSH_INTERVAL
    except Exception as e:
        print(f"[警告] 读取推送记录失败: {str(e)}，允许推送")
        return True
def record_push_time():
    """记录推送时间"""
    try:
        with open(LAST_PUSH_FILE, "w") as f:
            f.write(datetime.now().isoformat())
    except Exception as e:
        print(f"[错误] 保存推送时间失败: {str(e)}")
def send_umbrella_alert():
    """发送带伞提醒"""
    if not check_push_cooldown():
        print("未达到推送间隔要求，跳过发送")
        return False
    url = "https://api2.pushdeer.com/message/push"
    params = {
        "pushkey": PUSHDEER_PUSHKEY,
        "text": "🌧️ 武汉即将下雨，记得带伞！",
        "type": "markdown",
        "desp": "### 天气预警\n\n建议携带雨具出门，注意出行安全！"
    }
    try:
        response = requests.get(url, params=params)
        if response.json().get("code") == 0:
            record_push_time()
            return True
        return False
    except Exception as e:
        print(f"[错误] 推送失败: {str(e)}")
        return False
def main():
    print("开始执行天气检查任务...")

    if not PUSHDEER_PUSHKEY:
        raise ValueError("未设置 PUSHDEER_PUSHKEY 环境变量")

    forecast = get_weather_forecast()
    if not forecast:
        print("未获取到有效天气数据，任务终止")
        return

    if should_alert(forecast):
        time_str = datetime.now().strftime('%Y-%m-%d %H:%M')
        print(f"{time_str} 检测到降水可能")
        if send_umbrella_alert():
            print("推送成功")
        else:
            print("推送尝试失败")
    else:
        print("当前无降水风险，无需推送")
if __name__ == "__main__":
    main()