【青龙面板】和风天气预警脚本

1. #!/usr/bin/env python3
   
2. # -*- coding: utf-8 -*-
   
3. """
   
4. 天气预警监控脚本（API Host 适配版）
   
5. 和风天气 JWT + 企业微信
   
6. """
   
7. 
   
8. import requests
   
9. import json
   
10. import sqlite3
    
11. import os
    
12. import time
    
13. import jwt
    
14. from datetime import datetime, timedelta
    
15. 
    
16. # ==================== 配置读取（修复版） ====================
    
17. 
    
18. jwt_key = os.getenv("QWEATHER_PRIVATE_KEY", "").replace('\\n', '\n')
    
19. kid = os.getenv("QWEATHER_KID")
    
20. pid = os.getenv("QWEATHER_PROJECT_ID")
    
21. api_host = os.getenv("QWEATHER_HOST", "").strip()  # 专属API Host，必填
    
22. lat = os.getenv("LAT")
    
23. lon = os.getenv("LON")
    
24. wechat_key = os.getenv("WECHAT_KEY")
    
25. mention = os.getenv("WECHAT_MENTION_ALL", "false").lower() == "true"
    
26. db_path = "/ql/data/db/weather_warnings.db"
    
27. 
    
28. # 验证配置（新增 API_HOST 强制检查）
    
29. if not api_host:
    
30.     print("❌ 缺少 QWEATHER_HOST（专属API域名）")
    
31.     print("   请前往和风天气控制台 → 设置 → 查看你的API Host")
    
32.     print("   格式类似：abc1234xyz.def.qweatherapi.com")
    
33.     exit(1)
    
34. 
    
35. if not all([jwt_key, kid, pid, lat, lon, wechat_key]):
    
36.     print("❌ 缺少必要环境变量")
    
37.     exit(1)
    
38. 
    
39. # ==================== 数据库（防重复推送） ====================
    
40. 
    
41. def init_db():
    
42.     """初始化数据库并清理过期记录"""
    
43.     os.makedirs(os.path.dirname(db_path), exist_ok=True)
    
44.     conn = sqlite3.connect(db_path)
    
45.     conn.execute('''CREATE TABLE IF NOT EXISTS warnings (
    
46.         id TEXT PRIMARY KEY, level TEXT, expire TIMESTAMP)''')
    
47.     cur = conn.execute("DELETE FROM warnings WHERE expire < ?", (datetime.now(),))
    
48.     if cur.rowcount > 0:
    
49.         print(f"🧹 清理 {cur.rowcount} 条过期记录")
    
50.     conn.commit()
    
51.     conn.close()
    
52. 
    
53. def is_sent(wid):
    
54.     """检查是否已推送且未过期"""
    
55.     conn = sqlite3.connect(db_path)
    
56.     res = conn.execute("SELECT 1 FROM warnings WHERE id=? AND expire > ?",
    
57.                        (wid, datetime.now())).fetchone()
    
58.     conn.close()
    
59.     return res is not None
    
60. 
    
61. def save(wid, level, hours=48):
    
62.     """保存预警"""
    
63.     expire = datetime.now() + timedelta(hours=hours)
    
64.     conn = sqlite3.connect(db_path)
    
65.     conn.execute("INSERT OR REPLACE INTO warnings VALUES (?,?,?)", (wid, level, expire))
    
66.     conn.commit()
    
67.     conn.close()
    
68. 
    
69. # ==================== JWT 生成 ====================
    
70. 
    
71. def get_token():
    
72.     """生成 Ed25519 JWT Token"""
    
73.     try:
    
74.         now = int(time.time())
    
75.         headers = {"alg": "EdDSA", "kid": kid, "typ": "JWT"}
    
76.         payload = {"sub": pid, "iat": now-30, "exp": now+900}
    
77.         token = jwt.encode(payload, jwt_key, algorithm='EdDSA', headers=headers)
    
78.         print("✅ JWT 生成成功")
    
79.         return token
    
80.     except Exception as e:
    
81.         print(f"❌ JWT 生成失败: {e}")
    
82.         return None
    
83. 
    
84. # ==================== 和风天气 API（修复版） ====================
    
85. 
    
86. def get_alerts(token):
    
87.     """获取实时天气预警（使用专属API Host）"""
    
88.     url = f"https://{api_host}/weatheralert/v1/current/{lat}/{lon}"
    
89.     print(f"🌐 API Host: {api_host}")
    
90.    
    
91.     try:
    
92.         r = requests.get(url,
    
93.                         headers={'Authorization': f'Bearer {token}', 'Accept': 'application/json'},
    
94.                         timeout=30)
    
95.         
    
96.         if r.status_code == 200:
    
97.             data = r.json()
    
98.             if data.get("metadata", {}).get("zeroResult"):
    
99.                 return []
    
100.             return data.get("alerts", [])
     
101.         elif r.status_code == 403:
     
102.             print("❌ 403 权限错误：API Host 不正确或 JWT 认证失败")
     
103.             print(f"   当前使用的 API Host: {api_host}")
     
104.             print("   请确认这是控制台显示的专属API Host")
     
105.         elif r.status_code == 400:
     
106.             print("❌ 坐标不支持，请更换大城市坐标")
     
107.         elif r.status_code == 401:
     
108.             print("❌ JWT认证失败，检查KID和Project ID")
     
109.         else:
     
110.             print(f"❌ API错误: {r.status_code}")
     
111.             print(f"   响应: {r.text[:200]}")
     
112.     except Exception as e:
     
113.         print(f"❌ 网络错误: {e}")
     
114.     return []
     
115. 
     
116. # ==================== 企业微信推送 ====================
     
117. 
     
118. def send(msg):
     
119.     """推送消息到企业微信"""
     
120.     url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={wechat_key}"
     
121.    
     
122.     if mention:
     
123.         # 纯文本模式，支持 @所有人
     
124.         text = msg.replace('**','').replace('#','').replace('<font color="info">','')\
     
125.                   .replace('<font color="warning">','').replace('</font>','').replace('`','')
     
126.         data = {"msgtype": "text", "text": {"content": text, "mentioned_list": ["@all"]}}
     
127.     else:
     
128.         # Markdown 模式，带颜色
     
129.         data = {"msgtype": "markdown", "markdown": {"content": msg}}
     
130.    
     
131.     try:
     
132.         r = requests.post(url, json=data, timeout=10)
     
133.         return r.json().get('errcode') == 0
     
134.     except Exception as e:
     
135.         print(f"   推送异常: {e}")
     
136.         return False
     
137. 
     
138. def format_msg(alert):
     
139.     """格式化预警消息"""
     
140.     colors = {
     
141.         "minor": ("🔵", "info", "蓝色"),
     
142.         "moderate": ("🟡", "warning", "黄色"),
     
143.         "severe": ("🟠", "warning", "橙色"),
     
144.         "extreme": ("🔴", "warning", "红色")
     
145.     }
     
146.     icon, color, name = colors.get(alert.get('severity'), ("⚪", "comment", "未知"))
     
147.    
     
148.     status_map = {"alert": "🆕 新发布", "update": "📝 更新", "cancel": "❌ 取消"}
     
149.     status = status_map.get(alert.get('messageType', {}).get('code', 'alert'), "未知")
     
150.    
     
151.     lines = [
     
152.         f"## {icon} 天气预警 - <font color='{color}'>{name}</font>",
     
153.         f"",
     
154.         f"**预警类型：** {alert.get('eventType', {}).get('name', '未知')}",
     
155.         f"**预警标题：** {alert.get('headline', '')}",
     
156.         f"**发布状态：** {status}",
     
157.         f"**发布机构：** {alert.get('senderName', '未知机构')}",
     
158.         f"**严重程度：** <font color='{color}'>{alert.get('severity', '未知')}</font>",
     
159.         f"",
     
160.         f"### 📝 预警详情",
     
161.         f"{alert.get('description', '暂无详情')}",
     
162.     ]
     
163.    
     
164.     if alert.get('instruction'):
     
165.         lines.extend([f"", f"### 🛡️ 防御指南", f"{alert['instruction']}"])
     
166.    
     
167.     lines.extend([
     
168.         f"",
     
169.         f"---",
     
170.         f"⏰ **发布时间：** `{alert.get('issuedTime', '未知')}`",
     
171.         f"⏳ **失效时间：** `{alert.get('expireTime', '未知')}`"
     
172.     ])
     
173.    
     
174.     return "\n".join(lines)
     
175. 
     
176. # ==================== 主程序 ====================
     
177. 
     
178. def main():
     
179.     print(f"🕐 当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
     
180.     print(f"📍 监控坐标：{lat}, {lon}")
     
181.     print(f"👥 @全体成员：{'开启' if mention else '关闭'}")
     
182.    
     
183.     init_db()
     
184.    
     
185.     token = get_token()
     
186.     if not token:
     
187.         return
     
188.    
     
189.     alerts = get_alerts(token)
     
190.     if not alerts:
     
191.         print("ℹ️ 当前无生效中的天气预警")
     
192.         return
     
193.    
     
194.     sent = 0
     
195.     for alert in alerts:
     
196.         wid = alert.get('id')
     
197.         if not wid:
     
198.             continue
     
199.         
     
200.         if is_sent(wid):
     
201.             print(f"⏭️ 已推送过: {alert.get('headline','')[:20]}...")
     
202.             continue
     
203.         
     
204.         if send(format_msg(alert)):
     
205.             # 计算过期时间
     
206.             hours = 48
     
207.             if alert.get('expireTime'):
     
208.                 try:
     
209.                     et = alert['expireTime'].replace('+08:00','').replace('Z','')
     
210.                     expire_dt = datetime.strptime(et, '%Y-%m-%dT%H:%M')
     
211.                     hours = max(1, (expire_dt - datetime.now()).total_seconds() / 3600)
     
212.                 except:
     
213.                     pass
     
214.             
     
215.             save(wid, alert.get('severity'), hours)
     
216.             sent += 1
     
217.             print(f"✅ 推送成功: {alert.get('headline','')[:30]}...")
     
218.         else:
     
219.             print(f"❌ 推送失败")
     
220.    
     
221.     print(f"🎉 完成：新推{sent}条")
     
222. 
     
223. if __name__ == "__main__":
     
224.     main()

复制代码

青龙非最新版本有安全风险，请各位务必拉取最新的镜像或者关注青龙官方的修复信息哈~

牛蛙！牛蛙！

厉害！！厉害！！

🐮🍺

潇潇雨歇 发表于 2026-2-9 09:44
🐮🍺

梦梦 发表于 2026-2-7 10:59
牛蛙！牛蛙！

你快写哦   都七级了

蓝小白 发表于 2026-2-9 10:28
你快写哦   都七级了

6

如果不推企微的话，怎么做

C6KoCl 发表于 2026-3-12 16:30
如果不推企微的话，怎么做

可以问ai