SmartBizTrip/utils/spider_weather.py

# 【数据准备 C：独立天气采集脚本】给数据库天气查询分支准备数据，不由网页每次查询自动触发。
# C1 读取配置 → C2 进入 __main__ 建表 → C3 执行一次 update_weather
# → C4 setup_scheduler 注册任务并持续检查是否到时。
# update_weather 内部：连库 → 遍历城市 → 查更新时间 → 判断是否过期
# → 按需请求天气 API → 逐日写库 → 关闭连接。

"""
需求：实现天气数据爬虫，定期从第三方API获取天气数据并存储到MySQL数据库
思路步骤：
1. 导入必要的模块和库
2. 配置API密钥、城市代码和数据库连接参数
3. 实现fetch_weather_data函数（从API获取天气数据）
4. 实现get_latest_update_time函数（获取数据库中最新更新时间）
5. 实现should_update_data函数（判断是否需要更新数据）
6. 实现store_weather_data函数（将天气数据存储到数据库）
7. 实现update_weather函数（更新指定城市的天气数据）
8. 实现setup_scheduler函数（设置定时任务）
9. 主函数（创建表、执行一次更新、启动定时任务）
"""

import gzip
import json
import os
import sys
import time
from datetime import datetime

import mysql.connector
import pytz
import requests
import schedule

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
from config import Config, env

# 【数据准备 C1】读取配置并解析时区；这属于模块顶层代码，导入时也会执行。
cfg = Config()
# 把配置中的时区名称转换为 pytz 时区对象；只有库识别的名称才能成功。
# 这是模块加载阶段执行的语句，名称无效会在进入下方 __main__ 之前抛异常。
TZ = pytz.timezone(cfg.weather_timezone)


def get_db_config():
    # 【连接内部步骤 1】组装 mysql.connector.connect 所需的关键字参数。
    # 由 connect_db 和主入口共用，统一数据库连接参数；charset 指定连接使用的字符集。
    # 返回字典不会连接数据库，连接动作要等 mysql.connector.connect(**字典)。
    return {
        "host": cfg.host,
        "user": cfg.user,
        "password": cfg.password,
        "database": cfg.database,
        "charset": "utf8mb4",
    }


def connect_db():
    # 【连接内部步骤 2】** 把字典展开成 host=... 等参数，返回数据库连接。
    # 返回 MySQL 连接对象，调用方再通过 conn.cursor 创建执行 SQL 的游标。
    # 本函数没有捕获连接异常，失败会交给调用方处理，不会返回一个伪造的空连接。
    return mysql.connector.connect(**get_db_config())


def fetch_weather_data(city, location):
    """
    从第三方API拿取天气数据。
    :param city:
    :param location:
    :return: 天气数据
    """
    # 【更新内部步骤 4.1】用配置的城市代码请求天气；timeout=10 限制本次网络等待。
    # 输入 city 用于日志说明，location 是天气服务的城市代码；二者不能互相替代。
    # 成功时返回天气 JSON 对应的字典；已捕获的请求/解析异常通常返回 None，供调用方跳过写入。
    headers = {"X-QW-Api-Key": cfg.weather_api_key, "Accept-Encoding": "gzip"}
    url = f"{cfg.weather_base_url}?location={location}"
    try:
        response = requests.get(url, headers=headers, timeout=10)
        print(response)
        # <Response [200]>
        # 检查 HTTP 是否返回失败状态，失败则抛请求异常，交给下方 requests.RequestException 分支。
        # HTTP 成功不等于天气接口的业务 code 成功，业务 code 在 store_weather_data 里另外判断。
        response.raise_for_status()
        print(222222222, response.raise_for_status())
        # 222222222 None
        print(11111111, response.headers.get("Content-Encoding"))
        # 11111111 gzip
        # 【更新内部步骤 4.2】解析返回体。requests 可能已经解压响应，重复解压失败时走 BadGzipFile 回退。
        if response.headers.get("Content-Encoding") == "gzip":
            print("gzip")
            # gzip
            # 使用.decompress方法时候报错了
            data = gzip.decompress(response.content).decode("utf-8")
            print(1111, data)
        else:
            data = response.text

        # exit()
        # 把解码后的 JSON 文本转成字典；字典中的 daily 是后续逐日入库的原始数据。
        return json.loads(data)
    except requests.RequestException as e:
        print(f"请求 {city} 天气数据失败: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"{city} JSON 解析错误: {e}, 响应内容: {response.text[:500]}...")
        return None
    except gzip.BadGzipFile:
        print(f"{city} 数据未正确解压，尝试直接解析: {response.text[:500]}...")
        return json.loads(response.text) if response.text else None


def get_latest_update_time(cursor, city):
    """
    拿取 这个 城市city 在数据表weather_data中最新更新时间。
    :param cursor:
    :param city:
    :return: datetime对象 或 None

    fetchone() 返回单个元组，元素顺序与 SELECT 列顺序一致：
        # cursor.execute("SELECT city, MAX(update_time) FROM weather_data WHERE city = %s", (city,))
        # result = cursor.fetchone()
        # result => ("北京", datetime.datetime(2026, 4, 28, 10, 30, 0))
        # result[0] => "北京"
        # result[1] => datetime.datetime(2026, 4, 28, 10, 30, 0)

        # 如果查询无结果（表中无该城市数据）：
        # result => (None,)

    fetchall() 返回元组列表，适用于多行结果：
        # cursor.execute("SELECT city, update_time FROM weather_data WHERE city = %s", (city,))
        # results = cursor.fetchall()
        # results => [
        #     ("北京", datetime.datetime(2026, 4, 27, 10, 30, 0)),
        #     ("北京", datetime.datetime(2026, 4, 28, 10, 30, 0))
        # ]
    """
    # 【更新内部步骤 2】只读该城市最新 update_time，供下一步判断是否需要更新。
    # 输入 cursor 是已有游标，city 用于过滤该城市；输出最近更新时间或 None。
    # MAX 是 SQL 聚合，所以正常查询只得到一行一个值，用 fetchone 而不是 fetchall。
    cursor.execute("SELECT MAX(update_time) FROM weather_data WHERE city = %s", (city,))
    # 结果形如 (时间值,)；即使该城市没有记录，MAX 通常仍得到一行 (None,)。
    # 下方读取 result[0]，把没有可用时间的情况统一返回 None。
    result = cursor.fetchone()
    # 本查询只返回一个聚合值，result 示例 => (datetime.datetime(2026, 4, 28, 10, 30, 0),)
    # 若无数据 => (None,)
    return result[0] if result[0] else None


def should_update_data(latest_time, force_update=False):
    """
    如果没有更新时间 或者 更新时间大于24小时 或者 force_update=True，就返回True，就去更新数据库中的这个城市的天气数据。
    其他情况返回False，表示不更新数据库中的数据。
    :param latest_time: 某个城市最新的更新时间
    :param force_update: 默认否
    :return: 是 或这 否 true or false
    """
    # 【更新内部步骤 3】强制更新或没有记录时直接返回 True，否则检查是否已满 24 小时。
    # 调用者 update_weather 为每个城市调用一次。force_update=True 强制返回 True；否则看最近更新时间。
    # 返回布尔值只是更新决定，本函数不发 HTTP 请求，也不写入天气记录。
    if force_update:
        return True
    if not latest_time:
        return True
    current_time = datetime.now(TZ)
    # 数据库的无时区时间按本地时区解释；已有时区的时间转换到同一时区再比较。
    latest_time = TZ.localize(latest_time) if latest_time.tzinfo is None else latest_time.astimezone(TZ)
    # 时间差先换算成秒，再除以 3600 得到小时；满 24 小时才返回 True，不是每次启动都强制更新。
    return (current_time - latest_time).total_seconds() / 3600 >= 24


def store_weather_data(conn, cursor, city, data):
    """
    把从第三方拿到的天气数据（json）解析出来，组装成可以插入到mysql天气表中的insert语句。然后执行插入动作。
    :param conn:
    :param cursor:
    :param city:
    :param data:
    :return:
    """

    # {"code":"200",
    # "updateTime":"2026-02-04T10:42+08:00",
    # "fxLink":"https://www.qweather.com/weather/beijing-101010100.html",
    # "daily":[
    #     {"fxDate":"2026-02-04","sunrise":"07:21","sunset":"17:37","moonrise":"20:21",
    #     "moonset":"08:41","moonPhase":"亏凸月","moonPhaseIcon":"805","tempMax":"12",
    #     "tempMin":"-2","iconDay":"100","textDay":"晴","iconNight":"150","textNight":"晴",
    #     "wind360Day":"225","windDirDay":"西南风","windScaleDay":"1-3","windSpeedDay":"3",
    #     "wind360Night":"0","windDirNight":"北风","windScaleNight":"1-3","windSpeedNight":"16",
    #     "humidity":"29","precip":"0.0","pressure":"1016","vis":"25","cloud":"0","uvIndex":"3"}
    #     。。。。
    # "refer":{"sources":["QWeather"],"license":["QWeather Developers License"]}}

    # 【更新内部步骤 5.1】先检查接口业务码；无效数据提前 return，不执行写库。
    # 输入 conn 用于提交/回滚，cursor 用于执行 SQL，city 标识写入城市，data 是接口返回字典。
    # 没有直接返回成功列表；通过逐条 SQL 写入和日志体现结果，由 update_weather 继续下一城市。
    # or 有短路行为：data 为空时不再调用 data.get，直接进入跳过分支。
    # 接口 code 在这里按字符串 "200" 比较，与 HTTP status_code 的整数 200 是不同字段。
    if not data or data.get("code") != "200":
        print(f"{city} 数据无效，跳过存储。")
        return

    # 取逐日预报列表；没有 daily 就得到空列表，下面 for 不执行，不会写入任何天气行。
    daily_data = data.get("daily", [])
    update_time = datetime.fromisoformat(
        data.get("updateTime").replace("+08:00", "+08:00")
    ).astimezone(TZ).replace(tzinfo=None)

    # 【更新内部步骤 5.2】按 daily 中的每一天组装字段，通过城市+日期的唯一键插入或更新。
    for day in daily_data:
        # {"fxDate":"2026-02-04","sunrise":"07:21","sunset":"17:37","moonrise":"20:21",
        #     #     "moonset":"08:41","moonPhase":"亏凸月","moonPhaseIcon":"805","tempMax":"12",
        #     #     "tempMin":"-2","iconDay":"100","textDay":"晴","iconNight":"150","textNight":"晴",
        #     #     "wind360Day":"225","windDirDay":"西南风","windScaleDay":"1-3","windSpeedDay":"3",
        #     #     "wind360Night":"0","windDirNight":"北风","windScaleNight":"1-3","windSpeedNight":"16",
        #     #     "humidity":"29","precip":"0.0","pressure":"1016","vis":"25","cloud":"0","uvIndex":"3"}
        fx_date = datetime.strptime(day["fxDate"], "%Y-%m-%d").date()
        # 元组中的值按后面 INSERT 列的顺序排列；参数顺序与占位符数量都要对应。
        # SQL 字符串本身包含 %s，占位符的实际值由 cursor.execute 的第二个参数提供。
        values = (
            city,
            fx_date,
            day.get("sunrise"),
            day.get("sunset"),
            day.get("moonrise"),
            day.get("moonset"),
            day.get("moonPhase"),
            day.get("moonPhaseIcon"),
            day.get("tempMax"),
            day.get("tempMin"),
            day.get("iconDay"),
            day.get("textDay"),
            day.get("iconNight"),
            day.get("textNight"),
            day.get("wind360Day"),
            day.get("windDirDay"),
            day.get("windScaleDay"),
            day.get("windSpeedDay"),
            day.get("wind360Night"),
            day.get("windDirNight"),
            day.get("windScaleNight"),
            day.get("windSpeedNight"),
            day.get("precip"),
            day.get("uvIndex"),
            day.get("humidity"),
            day.get("pressure"),
            day.get("vis"),
            day.get("cloud"),
            update_time,
        )
        # upsert (Update + Insert): 插入或更新操作，一条SQL同时处理两种情况
        # 原理：依赖表的唯一键约束（本表中为 UNIQUE KEY unique_city_date(city, fx_date)）
        #   - INSERT 路径：表中不存在该 (city, fx_date) 组合 → 直接插入新行
        #   - UPDATE 路径：表中已存在该 (city, fx_date) 组合 → 用新值覆盖旧值（ON DUPLICATE KEY UPDATE）
        #
        # 示例：
        #   首次拉取 ("北京", 2026-04-28) → INSERT 新行
        #   再次拉取 ("北京", 2026-04-28) → UPDATE 该行所有天气字段，同时更新 update_time
        #
        # VALUES(col) 表示 "要插入的那份新值"，而非表中原有的旧值
        insert_query = """
                       INSERT INTO weather_data (city, fx_date, sunrise, sunset, moonrise, moonset, moon_phase,
                                                 moon_phase_icon,
                                                 temp_max, temp_min, icon_day, text_day, icon_night, text_night,
                                                 wind360_day, wind_dir_day, wind_scale_day, wind_speed_day,
                                                 wind360_night, wind_dir_night, wind_scale_night, wind_speed_night,
                                                 precip, uv_index, humidity, pressure, vis, cloud, update_time)
                       VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                               %s, %s, %s, %s, %s, %s, %s)
                       ON DUPLICATE KEY UPDATE sunrise          = VALUES(sunrise),
                                               sunset           = VALUES(sunset),
                                               moonrise         = VALUES(moonrise),
                                               moonset          = VALUES(moonset),
                                               moon_phase       = VALUES(moon_phase),
                                               moon_phase_icon  = VALUES(moon_phase_icon),
                                               temp_max         = VALUES(temp_max),
                                               temp_min         = VALUES(temp_min),
                                               icon_day         = VALUES(icon_day),
                                               text_day         = VALUES(text_day),
                                               icon_night       = VALUES(icon_night),
                                               text_night       = VALUES(text_night),
                                               wind360_day      = VALUES(wind360_day),
                                               wind_dir_day     = VALUES(wind_dir_day),
                                               wind_scale_day   = VALUES(wind_scale_day),
                                               wind_speed_day   = VALUES(wind_speed_day),
                                               wind360_night    = VALUES(wind360_night),
                                               wind_dir_night   = VALUES(wind_dir_night),
                                               wind_scale_night = VALUES(wind_scale_night),
                                               wind_speed_night = VALUES(wind_speed_night),
                                               precip           = VALUES(precip),
                                               uv_index         = VALUES(uv_index),
                                               humidity         = VALUES(humidity),
                                               pressure         = VALUES(pressure),
                                               vis              = VALUES(vis),
                                               cloud            = VALUES(cloud),
                                               update_time      = VALUES(update_time) \
                       """
        try:
            # 【更新内部步骤 5.3】逐天写入并提交；本天失败则回滚本天事务，不撤销此前已提交日期。
            cursor.execute(insert_query, values)
            print(
                f"{city} {fx_date} 数据写入/更新成功: {day.get('textDay')}, 影响行数: {cursor.rowcount}"
            )
            # 本天 INSERT/UPDATE 成功后提交事务，下一天的写入属于后续操作。
            conn.commit()
            print(f"{city} 事务提交完成。")
        except mysql.connector.Error as e:
            print(f"{city} {fx_date} 数据库错误: {e}")
            # 本天执行异常时撤销尚未提交的修改；except 结束后外层 for 仍可继续处理下一天。
            conn.rollback()
            print(f"{city} 事务回滚。")


def update_weather(force_update=False):
    """
    遍历配置中的所有城市，判断是否需要拉取最新天气数据并入库。

    流程图：
        开始
         │
         ├── connect_db() ──► 建立 MySQL 连接
         │
         ├── 遍历 cfg.weather_city_codes 中的每个 (city, location)
         │    │
         │    ├── get_latest_update_time(cursor, city)
         │    │    └── SELECT MAX(update_time) FROM weather_data
         │    │         └──► latest_time = datetime / None
         │    │
         │    ├── should_update_data(latest_time, force_update) 判断：
         │    │    ├── force_update == True       ──► True（强制更新）
         │    │    ├── latest_time == None         ──► True（首次拉取）
         │    │    └── 距今 > 24 小时              ──► True（数据过期）
         │    │    └── 其他情况                    ──► False（跳过）
         │    │
         │    ├── True 分支（需要更新）:
         │    │    ├── fetch_weather_data(city, location)
         │    │    │    └──► HTTP GET → 和风天气 API → gzip 解压 → 返回 JSON dict
         │    │    │
         │    │    ├── data 有效？
         │    │    │    ├── 是 ──► store_weather_data(conn, cursor, city, data)
         │    │    │    │          └──► 解析 daily 数组 → 逐条 UPSERT → commit
         │    │    │    └── 否 ──► 跳过该城市（打印警告）
         │    │
         │    └── False 分支（数据已最新）:
         │         └──► 打印 "数据已为最新，无需更新"
         │
         └── cursor.close() / conn.close() ──► 结束
    """
    # 【数据准备 C3，内部步骤 1】建立连接并遍历配置城市；每个城市分别判断更新条件。
    # 由主入口先调用一次，之后可能由 schedule 在到期时调用。force_update 默认 False。
    # 城市之间串行处理：获取更新时间 → 决定是否拉取 → 检查返回 → 存储，再处理下一城市。
    conn = connect_db()
    cursor = conn.cursor()

    for city, location in cfg.weather_city_codes.items():
        # 拿取 这个 城市city 在数据表weather_data中最新更新时间。
        latest_time = get_latest_update_time(cursor, city)
        # 如果没有更新时间 或者 更新时间大于24小时 或者 force_update=True，就返回True，就去更新数据库中的这个城市的天气数据。
        # 其他情况返回False，表示不更新数据库中的数据。
        # True 才进入 HTTP 拉取分支；False 只打印“已最新”，然后循环自然进入下一个城市。
        if should_update_data(latest_time, force_update):
            print(f"开始更新 {city} 天气数据...")
            # 从第三方API拿取天气数据
            data = fetch_weather_data(city, location)
            # 先检查是否取得非空返回值；真正的天气业务 code 仍由 store_weather_data 继续验证。
            if data:
                # 把从第三方拿到的天气数据（json）解析出来，组装成可以插入到mysql天气表中的insert语句。然后执行插入动作。
                store_weather_data(conn, cursor, city, data)
        else:
            print(f"{city} 数据已为最新，无需更新。最新更新时间: {latest_time}")

    # 【更新内部步骤 6】正常完成循环后释放游标和连接。
    cursor.close()
    conn.close()


def setup_scheduler():
    # 仅当 weather_source 等于 "api" 时跳过定时调度；这不会取消此前已执行的更新。
    # 【数据准备 C4.1】api 模式仅跳过这里的定时循环；__main__ 前面的建表和首次更新仍会执行。
    # 由主入口在首次更新以后调用。api 模式立即 return，其他模式注册每日任务并进入常驻循环。
    # 函数只有一直运行，schedule.run_pending 才有机会检查到期任务；注册动作不会创建系统定时任务。
    if cfg.weather_source == "api":
        return

    # 【数据准备 C4.2】注册每天的运行时间；当前 .at 未显式传入时区，按运行进程的本地时间调度。
    # do 接收的是函数对象 update_weather，没有括号；只有任务到期时才会调用它。
    # 如果写成 do(update_weather()) 就会先立即执行函数，再把返回值交给调度器，这与当前逻辑不同。
    schedule.every().day.at(cfg.weather_schedule_time).do(update_weather)
    # 【数据准备 C4.3】每 60 秒检查到期任务，不是每 60 秒都请求一次天气。进程退出后调度停止。
    while True:
        schedule.run_pending()
        # 暂停当前线程 60 秒，降低检查频率；这会阻塞此采集脚本线程，不是 async 的协程暂停。
        time.sleep(60)


if __name__ == "__main__":
    # 初始检查和更新
    # 【数据准备 C2】直接运行时先确保 weather_data 表存在，再调用首次更新和调度器。
    with mysql.connector.connect(**get_db_config()) as conn:
        cursor = conn.cursor()
        cursor.execute("""
                       CREATE TABLE IF NOT EXISTS weather_data
                       (
                           id               INT AUTO_INCREMENT PRIMARY KEY,
                           city             VARCHAR(50) NOT NULL COMMENT '城市名称',
                           fx_date          DATE        NOT NULL COMMENT '预报日期',
                           sunrise          TIME COMMENT '日出时间',
                           sunset           TIME COMMENT '日落时间',
                           moonrise         TIME COMMENT '月升时间',
                           moonset          TIME COMMENT '月落时间',
                           moon_phase       VARCHAR(20) COMMENT '月相名称',
                           moon_phase_icon  VARCHAR(10) COMMENT '月相图标代码',
                           temp_max         INT COMMENT '最高温度',
                           temp_min         INT COMMENT '最低温度',
                           icon_day         VARCHAR(10) COMMENT '白天天气图标代码',
                           text_day         VARCHAR(20) COMMENT '白天天气描述',
                           icon_night       VARCHAR(10) COMMENT '夜间天气图标代码',
                           text_night       VARCHAR(20) COMMENT '夜间天气描述',
                           wind360_day      INT COMMENT '白天风向360角度',
                           wind_dir_day     VARCHAR(20) COMMENT '白天风向',
                           wind_scale_day   VARCHAR(10) COMMENT '白天风力等级',
                           wind_speed_day   INT COMMENT '白天风速 (km/h)',
                           wind360_night    INT COMMENT '夜间风向360角度',
                           wind_dir_night   VARCHAR(20) COMMENT '夜间风向',
                           wind_scale_night VARCHAR(10) COMMENT '夜间风力等级',
                           wind_speed_night INT COMMENT '夜间风速 (km/h)',
                           precip           DECIMAL(5, 1) COMMENT '降水量 (mm)',
                           uv_index         INT COMMENT '紫外线指数',
                           humidity         INT COMMENT '相对湿度 (%)',
                           pressure         INT COMMENT '大气压强 (hPa)',
                           vis              INT COMMENT '能见度 (km)',
                           cloud            INT COMMENT '云量 (%)',
                           update_time      DATETIME COMMENT '数据更新时间',
                           UNIQUE KEY unique_city_date (city, fx_date)
                       ) ENGINE = InnoDB
                         DEFAULT CHARSET = utf8mb4
                         COLLATE = utf8mb4_unicode_ci COMMENT ='天气数据表'
                       """)
        conn.commit()

    # 立即执行一次更新
    update_weather()
    # 我们全国有2000个城市

    # 启动定时任务
    setup_scheduler()

# 以下是另一种按查询触发的缓存设计思路，当前脚本未实现：
# 1 用户来查询某个城市 比如 北京的天气数据。
# 2 去数据库中查询北京天气数据，select weather_content,update_time from weather_data where city = "北京" and date = "2026-03-04"
# 3 检验这条数据的update_time，如果是1小时之前的，就从第三方api拉取最新数据，返回给用户。同时更新mysql数据。
# 4 如果这条数据是1小时之内更新的，就直接返回给用户。
# 5 做一个缓存，缓存的key是 city+date，设置过期时间ttl，当前时间-更新时间（1小时之内）

# 以上是第一种方式
# 其他方式：
# 从redis中查，有就用，没有就查第三方api，缓存redis，设置ttl=1小时。


# {"code":"200",
# "updateTime":"2026-02-04T10:42+08:00",
# "fxLink":"https://www.qweather.com/weather/beijing-101010100.html",
# "daily":[
#     {"fxDate":"2026-02-04","sunrise":"07:21","sunset":"17:37","moonrise":"20:21",
#     "moonset":"08:41","moonPhase":"亏凸月","moonPhaseIcon":"805","tempMax":"12",
#     "tempMin":"-2","iconDay":"100","textDay":"晴","iconNight":"150","textNight":"晴",
#     "wind360Day":"225","windDirDay":"西南风","windScaleDay":"1-3","windSpeedDay":"3",
#     "wind360Night":"0","windDirNight":"北风","windScaleNight":"1-3","windSpeedNight":"16",
#     "humidity":"29","precip":"0.0","pressure":"1016","vis":"25","cloud":"0","uvIndex":"3"}
#     。。。。
# "refer":{"sources":["QWeather"],"license":["QWeather Developers License"]}}