Automating Weather Alerts to WeChat Contacts with Python
This implementation utilizes Python to fetch localized meteorological data and disseminate it to specific contacts within a messaging platform. The solution relies on the wxpy library for bot integration, tkinter for a graphical user interface, and requests for external API communication.
Core Dependencies
The system requires the following dependencies:
- Python 3.x
- wxpy: Facilitates WeChat bot interaction.
- tkinter: Built-in GUI framework.
- requests: Handles HTTP requests to weather services.
Data Retrieval Logic
Weather information is aggregated via the Baidu Map Telematics API. The function below handles the HTTP request, parses the JSON payload, and constructs a formatted summary. Error handling ensures that if the provided city does not match, a default location (e.g., Haidian) is queried instead of crashing.
The PM2.5 value is evaluated against thresholds to categorize air quality levels.
import requests
BAIDU_API_KEY = 'YOUR_BAIDU_API_KEY_HERE'
BASE_URL = "http://api.map.baidu.com/telematics/v3/weather?location={}&output=json&ak={}"
def format_pollution_level(pm_value):
try:
value = int(pm_value) if pm_value else 0
except ValueError:
value = 0
if value < 35:
return 'Good'
elif 35 <= value < 75:
return 'Moderate'
elif 75 <= value < 115:
return 'Lightly Polluted'
elif 115 <= value < 150:
return 'Moderately Polluted'
elif 150 <= value < 250:
return 'Heavily Polluted'
else:
return 'Severely Polluted'
def get_weather_report(city_name: str) -> str:
url = BASE_URL.format(city_name, BAIDU_API_KEY)
response = requests.get(url, timeout=5)
response.raise_for_status()
data = response.json()
# Fallback logic for invalid cities
if data.get('error') != 0:
url = BASE_URL.format('Haidian', BAIDU_API_KEY)
response = requests.get(url, timeout=5)
data = response.json()
result = data.get('results', [{}])[0]
current_data = result.get('weather_data', [{}])[0]
index_info = result.get('index', [])
report_lines = [
f"Weather Forecast Report",
f"Location: {result.get('currentCity')}",
f"PM2.5 Level: {result.get('pm25')} ({format_pollution_level(result.get('pm25'))})",
f"Current Temperature: {current_data.get('temperature')}°C",
f"Condition: {current_data.get('weather')}",
f"Wind Direction: {current_data.get('wind')}",
]
# Append lifestyle suggestions
if len(index_info) > 4:
report_lines.append(f"Attire Advice: {index_info[0].get('des')}")
report_lines.append(f"Health Reminder: {index_info[2].get('des')}")
report_lines.append(f"Outdoor Exercise: {index_info[3].get('des')}")
report_lines.append(f"UV Index: {index_info[4].get('des')}")
# Future forecast
future_data = result.get('weather_data', [])
forecast_text = "Three-Day Outlook:\n"
for day in future_data[:3]:
forecast_text += f"- Date: {day.get('date')}, Temp: {day.get('temperature')}°, Weather: {day.get('weather')}, Wind: {day.get('wind')}\n"
report_lines.append(forecast_text)
return '\n'.join(report_lines)
User Interface Implementation
A Tkinter-based desktop window presants a list of bot connections or friends. The interface displays the contact name and their registered city status. Upon selecting a friend and executing the action, the system triggers the reporting function.
import tkinter as tk
from tkinter import messagebox
# Assuming bot and friends are initialized from wxpy module before this script runs
# from wxpy import Bot
# bot = Bot(cache_path=True)
# my_friends = bot.friends()
def create_gui():
root = tk.Tk()
root.title("WeChat Weather Notify System")
root.geometry("800x600")
frame_control = tk.Frame(root)
frame_display = tk.Frame(root)
# Listbox for contacts
lb_contacts = tk.Listbox(frame_control, selectmode=tk.SINGLE, height=15, width=40)
lb_contacts.pack(side=tk.LEFT, fill=tk.BOTH)
# Output display box
txt_output = tk.Text(frame_display, height=15, width=60)
txt_output.pack(fill=tk.BOTH)
# Populate contact list
def refresh_friends_list():
lb_contacts.delete(0, tk.END)
try:
for friend in my_friends:
city = friend.city if friend.city else "Unknown City"
name = friend.name or friend.remark_name
lb_contacts.insert(tk.END, f"{name} | {city}")
except Exception as e:
txt_output.insert(tk.END, f"Error loading friends: {e}\n")
refresh_friends_list()
# Send Functionality
def dispatch_weather():
selection = lb_contacts.curselection()
if not selection:
return
content = lb_contacts.get(selection[0])
parts = content.split('|')
target_city = parts[-1].strip()
target_name = parts[0].strip()
if target_city == "Unknown City":
target_city = "Default_City_Here"
print(f"No city set for {target_name}, using default.")
try:
weather_msg = get_weather_report(target_city)
txt_output.insert(tk.END, f"--- Report for {target_name} ---\n")
txt_output.insert(tk.END, weather_msg + '\n')
# Find friend object by search string
found_friend = list(my_friends.search(target_name))[0] if any(name in m for m in my_friends.values() if hasattr(m, 'name')) else None
if found_friend:
found_friend.send(weather_msg)
print(f"Sent to {found_friend}")
else:
print(f"Friend not found dynamically: {target_name}")
except Exception as e:
txt_output.insert(tk.END, f"Failed to generate/send report: {e}\n")
btn_send = tk.Button(frame_control, text="Send Report", command=dispatch_weather)
btn_send.pack(pady=10)
frame_control.pack(side=tk.LEFT, fill=tk.BOTH)
frame_display.pack(side=tk.RIGHT, fill=tk.BOTH)
root.mainloop()
Integration Workflow
- Initialization: The bot logs into WeChat and caches the session state for efficiency.
- Synchronization: The application queries the friend list (
bot.friends()). - Data Filtering: For each entry, the program inspects the
cityattribute attached to the profile. - Trigger: A UI event handler retrieves the selected item, extracts the city parameter, calls
get_weather_report, and sends the resulting string via the API method.
The process repeats automatically upon execution of the main script loop.