Fading Coder

One Final Commit for the Last Sprint

Home > Tech > Content

Automating Weather Alerts to WeChat Contacts with Python

Tech 2

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

  1. Initialization: The bot logs into WeChat and caches the session state for efficiency.
  2. Synchronization: The application queries the friend list (bot.friends()).
  3. Data Filtering: For each entry, the program inspects the city attribute attached to the profile.
  4. 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.

Related Articles

Understanding Strong and Weak References in Java

Strong References Strong reference are the most prevalent type of object referencing in Java. When an object has a strong reference pointing to it, the garbage collector will not reclaim its memory. F...

Comprehensive Guide to SSTI Explained with Payload Bypass Techniques

Introduction Server-Side Template Injection (SSTI) is a vulnerability in web applications where user input is improper handled within the template engine and executed on the server. This exploit can r...

Implement Image Upload Functionality for Django Integrated TinyMCE Editor

Django’s Admin panel is highly user-friendly, and pairing it with TinyMCE, an effective rich text editor, simplifies content management significantly. Combining the two is particular useful for bloggi...

Leave a Comment

Anonymous

◎Feel free to join the discussion and share your thoughts.