# 【数据准备 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) # # 检查 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"]}}