master

Branches (1)

Manage

Manage

master

PythonHomework
/
爬取天气自动提醒.py

import requests
from bs4 import BeautifulSoup
import logging
from apscheduler.schedulers.blocking import BlockingScheduler
import matplotlib
import matplotlib.pyplot as plt
import io
import base64
import re
import sys


PUSH_KEY = "SCT282155Tbsscmv2v2zHLna06jIuvwZgv"
PUSH_URL = f"https://sctapi.ftqq.com/{PUSH_KEY}.send"
CITY_CODE = "101010100"
BASE_URL = f"https://www.weather.com.cn/weather/{CITY_CODE}.shtml"
AIR_QUALITY_URL = f"https://www.weather.com.cn/air/{CITY_CODE}.html"
NOTIFICATION_HOUR = 21
NOTIFICATION_MINUTE =21
WIND_THRESHOLD = 6
AQI_THRESHOLD = 150
RAIN_KEYWORDS = ["雨", "雷", "雹", "雪", "雾", "霾"]
LOG_FILE = ".venv/weather_crawler.log"
MATPLOTLIB_FONT = "SimHei"


matplotlib.use("Agg")
plt.rcParams["font.family"] = MATPLOTLIB_FONT
plt.rcParams["axes.unicode_minus"] = False


def setup_logging():
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.INFO)
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    file_handler = logging.FileHandler(LOG_FILE, encoding="utf-8")
    stream_handler = logging.StreamHandler()
    logger.addHandler(file_handler)
    logger.addHandler(stream_handler)
    return logger


def fetch_weather_data(logger):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
        "Referer": "https://www.weather.com.cn/"
    }
    try:
        response = requests.get(BASE_URL, headers=headers, timeout=15)
        response.encoding = "utf-8"
        if response.status_code != 200:
            logger.error(f"天气数据请求失败，状态码：{response.status_code}")
            return None
        soup = BeautifulSoup(response.text, "lxml")
        container = soup.find("div", id="7d") or soup.find("div", class_="t")
        if not container:
            logger.error("未找到天气数据容器")
            return None
        items = container.find_all("li", limit=3)
        return parse_forecast_items(items, logger)
    except Exception as e:
        logger.error(f"获取天气数据失败：{str(e)}", exc_info=True)
        return None


def parse_forecast_items(items, logger):
    forecast = []
    for idx, item in enumerate(items):
        try:
            date = item.find("h1").get_text(strip=True)
            weather = item.find("p", class_="wea").get_text(strip=True)
            temp = item.find("p", class_="tem").get_text(strip=True)
            wind = item.find("p", class_="win").get_text(strip=True)

            max_temp, min_temp = parse_temperature(temp)
            wind_level = parse_wind_level(wind)

            forecast.append({
                "date": date,
                "weather": weather,
                "max_temp": max_temp,
                "min_temp": min_temp,
                "wind_level": wind_level,
                "day_index": idx
            })
        except Exception as e:
            logger.error(f"解析第{idx + 1}天数据失败：{str(e)}", exc_info=True)
            continue
    return forecast if forecast else None


def parse_temperature(temp_str):
    try:
        temps = temp_str.split("/")
        return int(temps[0].replace("℃", "")), int(temps[1].replace("℃", ""))
    except:
        return 0, 0


def parse_wind_level(wind_str):
    match = re.search(r"\d+", wind_str)
    return int(match.group()) if match else 0


def fetch_air_quality(logger):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
        "Referer": "https://www.weather.com.cn/air/",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Connection": "keep-alive"
    }

    try:
        logger.info(f"正在请求空气质量数据：{AIR_QUALITY_URL}")
        response = requests.get(AIR_QUALITY_URL, headers=headers, timeout=15)
        response.encoding = "utf-8"

        if response.status_code != 200:
            logger.warning(f"空气质量请求失败，状态码：{response.status_code}，使用默认值")
            return 100

        soup = BeautifulSoup(response.text, "lxml")
        # 最新页面AQI位于<div class="aqi_num">标签内
        aqi_elem = soup.find("div", class_="aqi_num")
        return int(aqi_elem.text) if aqi_elem else 100

    except Exception as e:
        logger.error(f"获取空气质量失败：{str(e)}", exc_info=True)
        return 100


def check_running_status(forecast, aqi):
    tomorrow = next((d for d in forecast if d["day_index"] == 1), None)
    if not tomorrow:
        return "⚠️ 未获取到明日天气数据，跑操状态未知"

    has_rain = any(keyword in tomorrow["weather"] for keyword in RAIN_KEYWORDS)
    has_strong_wind = tomorrow["wind_level"] >= WIND_THRESHOLD
    has_pollution = aqi >= AQI_THRESHOLD

    if has_rain or has_strong_wind or has_pollution:
        reasons = []
        if has_rain: reasons.append(f"天气{tomorrow['weather']}")
        if has_strong_wind: reasons.append(f"风力{tomorrow['wind_level']}级")
        if has_pollution: reasons.append(f"AQI{aqi}污染")
        return f"🚫 明日不跑操！原因：{', '.join(reasons)}"
    else:
        return "✅ 明日正常跑操"


def send_notification(forecast, aqi, logger):
    if not forecast or len(forecast) < 2:
        logger.warning("有效数据不足，跳过通知")
        return False

    city_name = "北京"
    title = f"{city_name} 明日天气及跑操提醒"
    content = f"""
🌞 今日天气（{forecast[0]['date']}）
天气：{forecast[0]['weather']}
温度：{forecast[0]['min_temp']}~{forecast[0]['max_temp']}℃
风力：{forecast[0]['wind_level']}级

🌡️ 明日天气（{forecast[1]['date']}）
天气：{forecast[1]['weather']}
温度：{forecast[1]['min_temp']}~{forecast[1]['max_temp']}℃
风力：{forecast[1]['wind_level']}级
空气质量：{"良好" if aqi < AQI_THRESHOLD else f"污染(AQI{aqi})"}

{check_running_status(forecast, aqi)}
    """.strip()

    payload = {"title": title, "desp": content}
    try:
        logger.info("正在发送微信通知...")
        response = requests.post(PUSH_URL, data=payload, timeout=15)
        result = response.json()
        if result.get("code") == 0:
            logger.info("通知发送成功")
            return True
        else:
            logger.error(f"通知发送失败：{result.get('message')}")
            return False
    except Exception as e:
        logger.error(f"发送通知时出错：{str(e)}", exc_info=True)
        return False


def daily_notification_job(logger):
    logger.info("===== 天气提醒任务开始执行 =====")

    weather_data = fetch_weather_data(logger)
    aqi = fetch_air_quality(logger)

    if weather_data:
        send_notification(weather_data, aqi, logger)
    else:
        logger.warning("未获取到有效天气数据，任务终止")

    logger.info("===== 任务执行完毕 =====\n")


if __name__ == "__main__":
    logger = setup_logging()

    if len(sys.argv) > 1 and sys.argv[1] == "--test":
        mock_forecast = [
            {"date": "6月8日", "weather": "晴", "max_temp": 32, "min_temp": 22, "wind_level": 3, "day_index": 0},
            {"date": "6月9日", "weather": "雷阵雨", "max_temp": 28, "min_temp": 20, "wind_level": 7, "day_index": 1}
        ]
        send_notification(mock_forecast, aqi=180, logger=logger)
    else:
        scheduler = BlockingScheduler()
        scheduler.add_job(
            daily_notification_job,
            "cron",
            hour=NOTIFICATION_HOUR,
            minute=NOTIFICATION_MINUTE,
            args=[logger],
            name="Daily Weather Notification"
        )

        logger.info(f"定时任务已设置：每天{NOTIFICATION_HOUR}:{NOTIFICATION_MINUTE}自动发送通知")
        logger.info("程序运行中... (按Ctrl+C退出)")
        scheduler.start()