使用Python 將 台中市36小時天氣資料輸出到Telegram
cutedu_leddht22_bot, [2025/6/6 下午 03:47]
--- 🗓️ 時段預報 ---
時間: 2025-06-06 12:00:00 ~ 2025-06-06 18:00:00
天氣: 多雲午後短暫雷陣雨
降雨機率: 30%
最低溫: 29°C
舒適度: 悶熱
最高溫: 31°C
cutedu_leddht22_bot, [2025/6/6 下午 03:47]
--- 🗓️ 時段預報 ---
時間: 2025-06-06 18:00:00 ~ 2025-06-07 06:00:00
天氣: 多雲短暫陣雨
降雨機率: 30%
最低溫: 25°C
舒適度: 舒適至悶熱
最高溫: 29°C
cutedu_leddht22_bot, [2025/6/6 下午 03:47]
--- 🗓️ 時段預報 ---
時間: 2025-06-07 06:00:00 ~ 2025-06-07 18:00:00
天氣: 多雲午後短暫雷陣雨
降雨機率: 30%
最低溫: 25°C
舒適度: 舒適至悶熱
最高溫: 32°C
Python程式
import requests
import json
import time
# --- 中央氣象署開放資料平台設定 ---
CWA_API_URL = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-C0032-001"
CWA_AUTHORIZATION_CODE = "CWB-40C25FFF-1224-4250-B9D9-3735AAE17DBF" # 您的CWB授權碼
LOCATION_NAME = "臺中市" # 要爬取資料的縣市
# --- Telegram Bot 設定 ---
TELEGRAM_BOT_TOKEN = "7738940254:AAHbrWu9ovb1BKPQyWsbNSjNxfCGCrEWU-o" # 請替換為您的Telegram Bot Token
TELEGRAM_CHAT_ID = "7965218469" # 請替換為您的Telegram Chat ID
def send_telegram_message(message):
"""
發送訊息到 Telegram。
"""
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
headers = {"Content-Type": "application/json"}
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message
}
try:
# Telegram API 允許 HTML 格式的訊息，這裡我們可以利用它讓訊息更美觀
# 但為了簡單，我們這裡使用純文字。如果要用 HTML，可以加上 "parse_mode": "HTML"
# payload["parse_mode"] = "HTML"
response = requests.post(url, headers=headers, data=json.dumps(payload))
response.raise_for_status() # 如果請求不成功 (例如 4xx 或 5xx)，會引發異常
print(f"Sending to Telegram: {json.dumps(payload)}")
print(f"Telegram API response code: {response.status_code}")
print(f"Telegram API response: {response.text}")
except requests.exceptions.RequestException as e:
print(f"Error sending message to Telegram: {e}")
print(f"HTTP Code: {response.status_code if 'response' in locals() else 'N/A'}")
if 'response' in locals():
print(f"Response text: {response.text}")
def get_weather_and_send_to_telegram():
"""
獲取天氣資料並發送到 Telegram。
"""
url = f"{CWA_API_URL}?Authorization={CWA_AUTHORIZATION_CODE}&format=JSON&locationName={LOCATION_NAME}"
print(f"Fetching weather data from: {url}")
try:
response = requests.get(url)
response.raise_for_status() # 檢查 HTTP 請求是否成功
print(f"HTTP Code: {response.status_code}")
payload = response.text
# print(payload) # 取消註解這行來查看原始 JSON 內容
doc = json.loads(payload) # 將 JSON 字串解析為 Python 字典
send_telegram_message("✨ 臺中市36小時天氣預報 ✨") # 先發送標題
# --- 關鍵修正點：JSON 路徑從 doc["records"]["locations"][0]["location"] 改為 doc["records"]["location"] ---
locations = doc["records"]["location"]
if not locations: # 檢查列表是否為空
print("Error: 'location' array not found in JSON or is empty.")
send_telegram_message("無法找到天氣地區資料。")
return
for location_data in locations:
if location_data.get("locationName") == LOCATION_NAME:
weather_elements = location_data.get("weatherElement")
if not weather_elements:
print(f"Error: 'weatherElement' array not found for {LOCATION_NAME}.")
send_telegram_message(f"無法找到{LOCATION_NAME}的天氣元素資料。")
return
# 先找到 Wx 元素，用它的時間來做主導遍歷
times_to_iterate = None
for element in weather_elements:
if element.get("elementName") == "Wx":
times_to_iterate = element.get("time")
break # 找到 Wx 元素後就跳出迴圈
if not times_to_iterate:
print("Error: Wx element or its time data not found.")
send_telegram_message("無法取得天氣現象時間資料。")
return
for time_period in times_to_iterate:
period_message = "--- 🗓️ 時段預報 ---\n"
period_message += f"時間: {time_period.get('startTime')} ~ {time_period.get('endTime')}\n"
# 遍歷所有天氣元素，找到對應時間段的資料
for element in weather_elements:
element_name = element.get("elementName")
inner_times = element.get("time")
if not inner_times:
continue # 如果該元素沒有時間數據，跳過
for inner_time in inner_times:
# 比較時間區間
if (inner_time.get("startTime") == time_period.get("startTime") and
inner_time.get("endTime") == time_period.get("endTime")):
parameter_name = inner_time.get("parameter", {}).get("parameterName")
parameter_unit = inner_time.get("parameter", {}).get("parameterUnit", "") # 獲取單位，如果沒有則為空字串
if element_name == "Wx":
period_message += f"天氣: {parameter_name}\n"
elif element_name == "MaxT":
period_message += f"最高溫: {parameter_name}°C\n"
elif element_name == "MinT":
period_message += f"最低溫: {parameter_name}°C\n"
elif element_name == "CI":
period_message += f"舒適度: {parameter_name}\n"
elif element_name == "PoP":
period_message += f"降雨機率: {parameter_name}%\n"
break # 找到對應時間的元素就跳出內層迴圈
send_telegram_message(period_message) # 發送每個時間段的訊息
time.sleep(0.2) # 稍作延遲，避免連續發送過快
except requests.exceptions.RequestException as e:
print(f"HTTP GET request failed: {e}")
send_telegram_message(f"HTTP GET request failed: {e}")
except json.JSONDecodeError as e:
print(f"Failed to parse JSON: {e}")
send_telegram_message(f"Failed to parse weather data (JSON error): {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
send_telegram_message(f"發生未知錯誤: {e}")
# --- 主程式執行點 ---
if __name__ == "__main__":
get_weather_and_send_to_telegram()
沒有留言:
張貼留言