尧图网站建设 
简介基于Python网络爬虫的陕西省气象数据实时发布系统毕业设计资料包面向计算机及相关专业完成毕设、课程设计的学生提供完整项目源码与配套毕业论文。资源聚焦气象数据采集、处理与实时发布流程涵盖爬虫脚本、后端逻辑、前端界面及数据库配置等模块适合有Python基础并希望参考完整工程结构的读者。包体共684个文件约58.24MB以py源码、pyc编译文件、dll依赖库、xml配置、cs及pyd扩展等类型为主同时包含sln、csproj等工程文件整体目录清晰便于按功能查阅。目前已有199人学习下载。资料中除项目源码外还整理了毕业论文答辩模板知识点解析覆盖选题背景与目的、选题意义及研究现状、研究过程与方法、致谢等关键环节可帮助理清答辩思路、规范论文陈述。对需要快速上手真实毕设项目、缺少完整资料参考的同学而言这套资源兼具代码参考与写作指导双重价值。1. 气象实时发布系统不是只写爬虫那么简单拿到“基于Python网络爬虫的陕西省气象数据实时发布系统”这个毕业设计题目最容易踩的坑是把它当成一个纯爬虫作业。答辩评委盯的是整条链路数据从哪个源来、怎么定时落库、发布接口返回什么、前端图表是不是真的在滚动。一套完整的系统等于requests采集层加MySQL时序落库加APScheduler定时调度加Flask发布接口加ECharts可视化五个环节缺一个都要单独被问一轮。这套设计适合两类人Python方向选爬虫或数据分析题目的毕业生以及想补全“采集→存储→调度→展示”全流程的初级工程师。整个系统不依赖重型框架本地一台机器就能跑通完整演示环境只要Python 3.8加MySQL工程量正好卡在毕设的合理区间。如果后面想复用到隔壁省份只需要换站点字典表里的station_id列表其余代码一行不用改。2. 数据源选型与requests采集层会话、限速、字段解析2.1 三个数据源的取舍陕西省气象数据没有统一开放的免费实时API常见路子有三条。中国气象数据网提供官方下载入口但需要注册并申请接口权限实时数据的鉴权流程在答辩现场容易卡壳更适合用来补历史数据和写绪论背景中国天气网的城市级页面和JSON接口实时性好请求头模拟到位即可稳定抓取适合做主数据源和风天气这类第三方平台申请即用免费额度有限而且论文里写“数据来源”时不如官方站点有说服力。三者的对比关系很直接数据源获取方式实时性反爬难度在系统里的角色中国气象数据网注册申请接口/文件下载分钟级高需鉴权历史数据、论文引用中国天气网城市页面与JSON接口分钟级中常规请求头即可实时采集主数据源和风天气等第三方API Key分钟级低有免费额度限制备用通道、接口异常时兜底2.2 请求会话与三件套参数采集模块用requests实现重点不是把代码写得多花哨而是把headers、timeout、重试三件套做完整。很多毕设源码挂在“请求被拒绝”上根源就是缺了这些基础参数。import requests from fake_useragent import UserAgent def build_session(): session requests.Session() session.headers.update({ User-Agent: UserAgent().random, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,en-US;q0.5,en;q0.3, Connection: keep-alive, }) return session def fetch_url(session, url, retries3, timeout(5, 10)): for attempt in range(retries): try: resp session.get(url, timeouttimeout) resp.raise_for_status() resp.encoding resp.apparent_encoding return resp.text except (requests.ConnectionError, requests.Timeout) as e: if attempt retries - 1: raise return Nonetimeout写成元组(5, 10)表示连接等待5秒、读取等待10秒避免某个区县站点响应慢时把整个调度任务挂死。retries控制在3次配合raise_for_status()会在返回4xx/5xx状态码时直接抛异常。resp.encoding用apparent_encoding而不是硬编码utf-8因为部分区县页面meta标签声明的是gb2312硬编码会让中文站点名在入库后变成乱码。Session对象复用TCP连接连续抓取西安、榆林、宝鸡多个站点时比每次重新请求快不少。2.3 解析策略先找JSON接口再考虑BeautifulSoup中国天气网的城市数据如果直接抓HTML需要处理频繁变动的页面结构和标签层级。抓包之后会发现省市级数据藏在独立的JSON接口里请求直接返回结构化字段比解析HTML稳定一个量级。这个“先看接口、再写解析”的习惯在答辩时很加分说明你在做工程而不是在写作业。对于只能拿到HTML的旧版页面用BeautifulSoup也能兜底from bs4 import BeautifulSoup def parse_temperature(html): soup BeautifulSoup(html, html.parser) temp_node soup.select_one(p.tem span) if temp_node is None: return None raw temp_node.get_text().replace(℃, ).strip() return float(raw) if raw else Noneselect_one取第一个匹配节点返回None时直接走异常分支入库前统一过滤。get_text()比.text更安全它只提取文本节点并忽略内部标签。解析失败时把原始HTML前200个字符写入日志排错时一眼就能看出是站点改版还是编码错乱。2.4 全量采集与增量采集的触发条件气象站点数据周期性变化没必要每次抓全量。常见做法是在建库后跑一次全量补齐最近7天数据随后所有定时任务只走增量。实现上用一个字典表记录每个站点的last_sync_time每次采集前读取判断。模式触发时机请求量典型用途全量初始化建库、站点字典变更后大补历史趋势、重建数据增量定时任务触发默认10分钟一次小实时发布和页面刷新提示抓完一个站点后加time.sleep(0.5~1.5)随机限速既降低被限流概率又能让入库时间戳看起来更接近真实观测节奏。3. MySQL时序落库站点字典、唯一索引与入库校验3.1 两张表的职责划分站点字典表负责描述“在哪里观测”气象数据表负责记录“观测到什么”。拆成两张表而不是塞在一起是为了避免站点名调整或经纬度修正时逐行更新历史记录。数据表只通过station_id关联字典表职责边界清晰答辩画ER图也好讲。CREATE TABLE station ( station_id VARCHAR(20) NOT NULL COMMENT 站点编码如101110101, station_name VARCHAR(50) NOT NULL COMMENT 站点名称, city VARCHAR(30) NOT NULL COMMENT 所属地市, lat DECIMAL(8,5) NULL COMMENT 纬度, lon DECIMAL(8,5) NULL COMMENT 经度, is_active TINYINT NOT NULL DEFAULT 1 COMMENT 1启用 0停用, PRIMARY KEY (station_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT气象站点字典表;station_id字段直接用上游数据源的城市编码而不用自增id因为爬虫拿到的原始数据本身就带这套编码直接落库天然去重。经纬度用DECIMAL(8,5)float在传输和比较时容易产生精度偏差。is_active是给调度器用的开关某个站点接口临时异常时先置0不影响采集主流程。3.2 时序表与唯一索引设计CREATE TABLE weather_record ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, station_id VARCHAR(20) NOT NULL, record_time DATETIME NOT NULL COMMENT 观测时间, temp DECIMAL(4,1) NULL COMMENT 气温/摄氏度, humidity INT NULL COMMENT 相对湿度/%, wind_dir VARCHAR(10) NULL COMMENT 风向, wind_level VARCHAR(10) NULL COMMENT 风力等级, pressure INT NULL COMMENT 气压/hPa, precipitation DECIMAL(6,1) NULL COMMENT 降水量/mm, create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 写入时间, PRIMARY KEY (id), UNIQUE KEY uk_station_time (station_id, record_time), KEY idx_time (record_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT气象数据时序表;uk_station_time唯一索引是整个去重方案的基石同一站点同一观测时间只允许一条记录。后续入库不管是INSERT IGNORE还是ON DUPLICATE KEY UPDATE都必须依赖这个约束兜底。idx_time普通索引服务前端按时间范围查询否则“最近24小时”这类请求会走全表扫描。3.3 SQLAlchemy模型与冲突更新from sqlalchemy import create_engine, Column, String, DateTime, DECIMAL, Integer, BigInteger from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.dialects.mysql import insert Base declarative_base() class WeatherRecord(Base): __tablename__ weather_record id Column(BigInteger, primary_keyTrue, autoincrementTrue) station_id Column(String(20), nullableFalse) record_time Column(DateTime, nullableFalse) temp Column(DECIMAL(4, 1)) humidity Column(Integer) pressure Column(Integer) def upsert_records(engine, records): if not records: return 0 stmt insert(WeatherRecord).values(records) stmt stmt.on_duplicate_key_update( tempstmt.inserted.temp, humiditystmt.inserted.humidity, pressurestmt.inserted.pressure, ) with engine.begin() as conn: result conn.execute(stmt) return result.rowcounton_duplicate_key_update的右侧写成stmt.inserted.temp表示冲突时覆盖为新抓取的值适合修正上游数据源的可能修正如果想保留首次入库值把右侧改成WeatherRecord.temp。engine.begin()是事务上下文管理器自动提交或回滚不需要手动commit。连接串关闭echo参数否则每次入库都会把SQL打印到控制台日志文件很快被刷爆。3.4 入库前校验与字段合理性气象数据写入前必须做范围校验否则一个解析异常的温度值可能把整条曲线拉出画面答辩演示时非常难看。校验逻辑放在和数据库层分离的service模块里字段合理范围越界处理temp-40 ~ 50 °C丢弃该字段并写日志humidity0 ~ 100 %丢弃该字段并写日志pressure870 ~ 1085 hPa丢弃该字段并写日志precipitation0 ~ 999 mm丢弃该字段并写日志校验函数返回过滤后的记录解析出的None值字段直接跳过不入库。宁可空着显示“--”也不能用0占位0在气象语义里本身就是有效观测值。4. APScheduler定时调度触发器选型与任务日志追踪4.1 为什么不用系统crontab系统的crontab也能做定时任务但它是黑盒的代码评审和答辩时说不清楚调度状态。APScheduler写在应用生命周期里启动、停止、任务列表都可见还能在Flask进程内共享数据库连接池。三种触发器的适用场景差别很大触发器适用场景配置要点date一次性任务指定run_dateinterval固定间隔任务minutes/hours参数cron按自然钟点hour/minute/week等表达式4.2 调度器初始化与参数from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.interval import IntervalTrigger import logging logger logging.getLogger(scheduler) def schedule_jobs(app): scheduler BackgroundScheduler(timezoneAsia/Shanghai) scheduler.add_job( run_incremental_sync, triggerIntervalTrigger(minutes10), idincremental_sync, replace_existingTrue, max_instances1, coalesceTrue, misfire_grace_time30, ) scheduler.add_job( run_daily_cleanup, triggerCronTrigger(hour2, minute0), iddaily_cleanup, replace_existingTrue, ) scheduler.start() return schedulertimezone显式指定Asia/Shanghai如果服务器默认UTC会在夏令时或跨时区迁移时跑偏。max_instances1防止上次任务没跑完下次又启动一个并发线程coalesceTrue告诉调度器把错过多次的执行合并成一次补跑misfire_grace_time30表示任务延迟30秒内仍然补执行超过则丢弃。这三个参数是答辩时“工程意识”的主要得分点很多参考源码这里都不写。启动调度后需要在Flask的teardown钩子里调用scheduler.shutdown()否则开发模式重启会积累多个后台线程。常见做法是在app.run()之前调用schedule_jobs(app)并把返回的调度器实例挂在app.extensions里以便统一管理。4.3 任务函数与日志追踪def run_incremental_sync(): logger.info(incremental_sync started) try: stations get_active_stations() for station in stations: data fetch_station_data(station[station_id]) records parse_to_records(data) inserted upsert_records(engine, records) logger.info(station%s rows%d, station[station_id], inserted) except Exception: logger.exception(incremental_sync failed)logger.exception自带堆栈信息比logger.error只打一条消息有用得多排错时不用再回头翻代码。调度日志单独输出到logs/scheduler.log用RotatingFileHandler按大小滚动日志单文件无限膨胀在长期运行后会拖慢写入。任务入口和结束各留一条日志时间戳核对阶段全靠它。4.4 失败重试与数据补偿APScheduler本身不负责任务内部的重试。在任务函数里套while循环重试会阻塞调度线程不推荐。常见做法是把抓取失败的站点编码写进一张retry_queue表下一个调度周期优先处理队列里的站点。系统宕机恢复后调度器根据misfire_grace_time自动补跑数据最终一致不会出现大面积空洞。5. Flask发布与ECharts可视化查询接口与动态刷新5.1 发布端目录结构app.py models.py crawler/ __init__.py fetcher.py parser.py services/ scheduler_service.py station_service.py templates/ index.html static/ js/ dashboard.jsapp.py只负责创建Flask实例、注册蓝图、启动调度器crawler包拆成请求和解析两个模块后期换数据源只需要改parser.py的字段映射其它代码不用动。services层隔离调度与业务逻辑答辩讲架构分层时这层拆解是很好的切入点。5.2 查询接口与参数化SQLfrom flask import Flask, request, jsonify from sqlalchemy import text app Flask(__name__) app.route(/api/weather/realtime) def realtime_weather(): city request.args.get(city, default西安, typestr) limit request.args.get(limit, default12, typeint) limit min(limit, 48) sql text( SELECT w.temp, w.humidity, w.pressure, w.record_time FROM weather_record w JOIN station s ON w.station_id s.station_id WHERE s.city :city ORDER BY w.record_time DESC LIMIT :limit ) rows db.session.execute(sql, {city: city, limit: limit}).fetchall() data [ { temp: float(r.temp), humidity: r.humidity, pressure: r.pressure, record_time: r.record_time.strftime(%Y-%m-%d %H:%M:%S), } for r in rows ] return jsonify({code: 0, data: data})limit做了48的上限保护防止前端参数传一个超大值把数据库拖垮。查询用绑定参数而不是字符串拼接从根本上杜绝SQL注入这条在安全评审里是必查项。record_time在返回前用strftime统一成字符串格式避免前端各浏览器时区解析不一致带来的八小时偏移。5.3 ECharts时间轴与自动刷新div idchart stylewidth:100%;height:420px;/div script srchttps://cdn.jsdelivr.net/npm/echarts5/dist/echarts.min.js/script script async function load() { const resp await fetch(/api/weather/realtime?city西安limit24); const data (await resp.json()).data.reverse(); chart.setOption({ xAxis: { data: data.map(d d.record_time) }, series: [{ data: data.map(d d.temp), type: line, smooth: true }] }); } const chart echarts.init(document.getElementById(chart)); load(); setInterval(load, 60000); /script接口返回的是倒序数据最新在前前端先reverse()再交给ECharts时间轴才会从左往右正常递增。setInterval 60秒刷新一次是正式发布节奏答辩演示时把60000改成10000评委能看到坐标轴自己滚动视觉效果比静态截图好很多。smooth: true把折线变成平滑曲线数值趋势一眼扫过去更直观。5.4 debug模式下的双调度器问题Flask的debugTrue会启动两个进程APScheduler在子进程里也执行一次结果就是定时任务重复入库。解决思路是加环境变量开关import os if os.environ.get(WERKZEUG_RUN_MAIN) true: scheduler schedule_jobs(app)WERKZEUG_RUN_MAIN只在重载进程里为true主进程不会重复注册。如果用的是gunicorn多worker部署同理只在worker 0里启动调度器。6. 答辩前自查时间戳核对、接口验证与爬虫日志比对6.1 检查入库时间是否真的在滚动SELECT station_id, MAX(record_time) AS latest FROM weather_record GROUP BY station_id;每个站点的latest时间应该落在最近10~15分钟内调度间隔加抓取耗时的余量。如果某个站点明显滞后优先查它的接口返回是否正常其次是解析器是否抛了异常被过滤掉。页面上最新时间点和当前时间差超过一个调度周期评委一眼就能看出来是假实时。6.2 模拟前端请求验证接口curl http://127.0.0.1:5000/api/weather/realtime?city榆林limit6检查返回JSON里code是否为0、data条数是否与limit一致、record_time格式是否为YYYY-MM-DD HH:MM:SS。配合jq管道可以做数值校验比如判断temp字段是否为有限数字。这个命令在答辩现场跑一次很快比打开浏览器等页面加载更直观。6.3 时区一致性核对日志里看到任务在14:00运行数据库里的create_time却显示06:00大概率是MySQL会话时区没有对齐。连接串里加engine create_engine( mysqlpymysql://root:passlocalhost/weather?charsetutf8mb4, connect_args{init_command: SET time_zone 08:00}, )同时把APScheduler的timezone、Python侧datetime.now()的时区统一成Asia/Shanghai。三处不一致时前端展示和日志时间戳对不上排查会非常耗时。最后一个容易忽视的点定时任务间隔不要调到30秒以下。气象数据本身分钟级更新抓太频繁除了增加被限流风险还会让数据库写入压力变大。演示前手动触发一次全量采集垫底再开启调度器页面起始数据齐全时间轴滚动效果也更完整。校验时多留意日志里scheduler和fetcher两处时间戳的先后关系差距过大不是事务没提交就是网络IO卡住了。本文还有配套的精品资源点击获取 Python爬虫气象数据实时发布系统：从采集到可视化全流程
简介基于Python网络爬虫的陕西省气象数据实时发布系统毕业设计资料包面向计算机及相关专业完成毕设、课程设计的学生提供完整项目源码与配套毕业论文。资源聚焦气象数据采集、处理与实时发布流程涵盖爬虫脚本、后端逻辑、前端界面及数据库配置等模块适合有Python基础并希望参考完整工程结构的读者。包体共684个文件约58.24MB以py源码、pyc编译文件、dll依赖库、xml配置、cs及pyd扩展等类型为主同时包含sln、csproj等工程文件整体目录清晰便于按功能查阅。目前已有199人学习下载。资料中除项目源码外还整理了毕业论文答辩模板知识点解析覆盖选题背景与目的、选题意义及研究现状、研究过程与方法、致谢等关键环节可帮助理清答辩思路、规范论文陈述。对需要快速上手真实毕设项目、缺少完整资料参考的同学而言这套资源兼具代码参考与写作指导双重价值。1. 气象实时发布系统不是只写爬虫那么简单拿到“基于Python网络爬虫的陕西省气象数据实时发布系统”这个毕业设计题目最容易踩的坑是把它当成一个纯爬虫作业。答辩评委盯的是整条链路数据从哪个源来、怎么定时落库、发布接口返回什么、前端图表是不是真的在滚动。一套完整的系统等于requests采集层加MySQL时序落库加APScheduler定时调度加Flask发布接口加ECharts可视化五个环节缺一个都要单独被问一轮。这套设计适合两类人Python方向选爬虫或数据分析题目的毕业生以及想补全“采集→存储→调度→展示”全流程的初级工程师。整个系统不依赖重型框架本地一台机器就能跑通完整演示环境只要Python 3.8加MySQL工程量正好卡在毕设的合理区间。如果后面想复用到隔壁省份只需要换站点字典表里的station_id列表其余代码一行不用改。2. 数据源选型与requests采集层会话、限速、字段解析2.1 三个数据源的取舍陕西省气象数据没有统一开放的免费实时API常见路子有三条。中国气象数据网提供官方下载入口但需要注册并申请接口权限实时数据的鉴权流程在答辩现场容易卡壳更适合用来补历史数据和写绪论背景中国天气网的城市级页面和JSON接口实时性好请求头模拟到位即可稳定抓取适合做主数据源和风天气这类第三方平台申请即用免费额度有限而且论文里写“数据来源”时不如官方站点有说服力。三者的对比关系很直接数据源获取方式实时性反爬难度在系统里的角色中国气象数据网注册申请接口/文件下载分钟级高需鉴权历史数据、论文引用中国天气网城市页面与JSON接口分钟级中常规请求头即可实时采集主数据源和风天气等第三方API Key分钟级低有免费额度限制备用通道、接口异常时兜底2.2 请求会话与三件套参数采集模块用requests实现重点不是把代码写得多花哨而是把headers、timeout、重试三件套做完整。很多毕设源码挂在“请求被拒绝”上根源就是缺了这些基础参数。import requests from fake_useragent import UserAgent def build_session(): session requests.Session() session.headers.update({ User-Agent: UserAgent().random, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,en-US;q0.5,en;q0.3, Connection: keep-alive, }) return session def fetch_url(session, url, retries3, timeout(5, 10)): for attempt in range(retries): try: resp session.get(url, timeouttimeout) resp.raise_for_status() resp.encoding resp.apparent_encoding return resp.text except (requests.ConnectionError, requests.Timeout) as e: if attempt retries - 1: raise return Nonetimeout写成元组(5, 10)表示连接等待5秒、读取等待10秒避免某个区县站点响应慢时把整个调度任务挂死。retries控制在3次配合raise_for_status()会在返回4xx/5xx状态码时直接抛异常。resp.encoding用apparent_encoding而不是硬编码utf-8因为部分区县页面meta标签声明的是gb2312硬编码会让中文站点名在入库后变成乱码。Session对象复用TCP连接连续抓取西安、榆林、宝鸡多个站点时比每次重新请求快不少。2.3 解析策略先找JSON接口再考虑BeautifulSoup中国天气网的城市数据如果直接抓HTML需要处理频繁变动的页面结构和标签层级。抓包之后会发现省市级数据藏在独立的JSON接口里请求直接返回结构化字段比解析HTML稳定一个量级。这个“先看接口、再写解析”的习惯在答辩时很加分说明你在做工程而不是在写作业。对于只能拿到HTML的旧版页面用BeautifulSoup也能兜底from bs4 import BeautifulSoup def parse_temperature(html): soup BeautifulSoup(html, html.parser) temp_node soup.select_one(p.tem span) if temp_node is None: return None raw temp_node.get_text().replace(℃, ).strip() return float(raw) if raw else Noneselect_one取第一个匹配节点返回None时直接走异常分支入库前统一过滤。get_text()比.text更安全它只提取文本节点并忽略内部标签。解析失败时把原始HTML前200个字符写入日志排错时一眼就能看出是站点改版还是编码错乱。2.4 全量采集与增量采集的触发条件气象站点数据周期性变化没必要每次抓全量。常见做法是在建库后跑一次全量补齐最近7天数据随后所有定时任务只走增量。实现上用一个字典表记录每个站点的last_sync_time每次采集前读取判断。模式触发时机请求量典型用途全量初始化建库、站点字典变更后大补历史趋势、重建数据增量定时任务触发默认10分钟一次小实时发布和页面刷新提示抓完一个站点后加time.sleep(0.5~1.5)随机限速既降低被限流概率又能让入库时间戳看起来更接近真实观测节奏。3. MySQL时序落库站点字典、唯一索引与入库校验3.1 两张表的职责划分站点字典表负责描述“在哪里观测”气象数据表负责记录“观测到什么”。拆成两张表而不是塞在一起是为了避免站点名调整或经纬度修正时逐行更新历史记录。数据表只通过station_id关联字典表职责边界清晰答辩画ER图也好讲。CREATE TABLE station ( station_id VARCHAR(20) NOT NULL COMMENT 站点编码如101110101, station_name VARCHAR(50) NOT NULL COMMENT 站点名称, city VARCHAR(30) NOT NULL COMMENT 所属地市, lat DECIMAL(8,5) NULL COMMENT 纬度, lon DECIMAL(8,5) NULL COMMENT 经度, is_active TINYINT NOT NULL DEFAULT 1 COMMENT 1启用 0停用, PRIMARY KEY (station_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT气象站点字典表;station_id字段直接用上游数据源的城市编码而不用自增id因为爬虫拿到的原始数据本身就带这套编码直接落库天然去重。经纬度用DECIMAL(8,5)float在传输和比较时容易产生精度偏差。is_active是给调度器用的开关某个站点接口临时异常时先置0不影响采集主流程。3.2 时序表与唯一索引设计CREATE TABLE weather_record ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, station_id VARCHAR(20) NOT NULL, record_time DATETIME NOT NULL COMMENT 观测时间, temp DECIMAL(4,1) NULL COMMENT 气温/摄氏度, humidity INT NULL COMMENT 相对湿度/%, wind_dir VARCHAR(10) NULL COMMENT 风向, wind_level VARCHAR(10) NULL COMMENT 风力等级, pressure INT NULL COMMENT 气压/hPa, precipitation DECIMAL(6,1) NULL COMMENT 降水量/mm, create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 写入时间, PRIMARY KEY (id), UNIQUE KEY uk_station_time (station_id, record_time), KEY idx_time (record_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT气象数据时序表;uk_station_time唯一索引是整个去重方案的基石同一站点同一观测时间只允许一条记录。后续入库不管是INSERT IGNORE还是ON DUPLICATE KEY UPDATE都必须依赖这个约束兜底。idx_time普通索引服务前端按时间范围查询否则“最近24小时”这类请求会走全表扫描。3.3 SQLAlchemy模型与冲突更新from sqlalchemy import create_engine, Column, String, DateTime, DECIMAL, Integer, BigInteger from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.dialects.mysql import insert Base declarative_base() class WeatherRecord(Base): __tablename__ weather_record id Column(BigInteger, primary_keyTrue, autoincrementTrue) station_id Column(String(20), nullableFalse) record_time Column(DateTime, nullableFalse) temp Column(DECIMAL(4, 1)) humidity Column(Integer) pressure Column(Integer) def upsert_records(engine, records): if not records: return 0 stmt insert(WeatherRecord).values(records) stmt stmt.on_duplicate_key_update( tempstmt.inserted.temp, humiditystmt.inserted.humidity, pressurestmt.inserted.pressure, ) with engine.begin() as conn: result conn.execute(stmt) return result.rowcounton_duplicate_key_update的右侧写成stmt.inserted.temp表示冲突时覆盖为新抓取的值适合修正上游数据源的可能修正如果想保留首次入库值把右侧改成WeatherRecord.temp。engine.begin()是事务上下文管理器自动提交或回滚不需要手动commit。连接串关闭echo参数否则每次入库都会把SQL打印到控制台日志文件很快被刷爆。3.4 入库前校验与字段合理性气象数据写入前必须做范围校验否则一个解析异常的温度值可能把整条曲线拉出画面答辩演示时非常难看。校验逻辑放在和数据库层分离的service模块里字段合理范围越界处理temp-40 ~ 50 °C丢弃该字段并写日志humidity0 ~ 100 %丢弃该字段并写日志pressure870 ~ 1085 hPa丢弃该字段并写日志precipitation0 ~ 999 mm丢弃该字段并写日志校验函数返回过滤后的记录解析出的None值字段直接跳过不入库。宁可空着显示“--”也不能用0占位0在气象语义里本身就是有效观测值。4. APScheduler定时调度触发器选型与任务日志追踪4.1 为什么不用系统crontab系统的crontab也能做定时任务但它是黑盒的代码评审和答辩时说不清楚调度状态。APScheduler写在应用生命周期里启动、停止、任务列表都可见还能在Flask进程内共享数据库连接池。三种触发器的适用场景差别很大触发器适用场景配置要点date一次性任务指定run_dateinterval固定间隔任务minutes/hours参数cron按自然钟点hour/minute/week等表达式4.2 调度器初始化与参数from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.interval import IntervalTrigger import logging logger logging.getLogger(scheduler) def schedule_jobs(app): scheduler BackgroundScheduler(timezoneAsia/Shanghai) scheduler.add_job( run_incremental_sync, triggerIntervalTrigger(minutes10), idincremental_sync, replace_existingTrue, max_instances1, coalesceTrue, misfire_grace_time30, ) scheduler.add_job( run_daily_cleanup, triggerCronTrigger(hour2, minute0), iddaily_cleanup, replace_existingTrue, ) scheduler.start() return schedulertimezone显式指定Asia/Shanghai如果服务器默认UTC会在夏令时或跨时区迁移时跑偏。max_instances1防止上次任务没跑完下次又启动一个并发线程coalesceTrue告诉调度器把错过多次的执行合并成一次补跑misfire_grace_time30表示任务延迟30秒内仍然补执行超过则丢弃。这三个参数是答辩时“工程意识”的主要得分点很多参考源码这里都不写。启动调度后需要在Flask的teardown钩子里调用scheduler.shutdown()否则开发模式重启会积累多个后台线程。常见做法是在app.run()之前调用schedule_jobs(app)并把返回的调度器实例挂在app.extensions里以便统一管理。4.3 任务函数与日志追踪def run_incremental_sync(): logger.info(incremental_sync started) try: stations get_active_stations() for station in stations: data fetch_station_data(station[station_id]) records parse_to_records(data) inserted upsert_records(engine, records) logger.info(station%s rows%d, station[station_id], inserted) except Exception: logger.exception(incremental_sync failed)logger.exception自带堆栈信息比logger.error只打一条消息有用得多排错时不用再回头翻代码。调度日志单独输出到logs/scheduler.log用RotatingFileHandler按大小滚动日志单文件无限膨胀在长期运行后会拖慢写入。任务入口和结束各留一条日志时间戳核对阶段全靠它。4.4 失败重试与数据补偿APScheduler本身不负责任务内部的重试。在任务函数里套while循环重试会阻塞调度线程不推荐。常见做法是把抓取失败的站点编码写进一张retry_queue表下一个调度周期优先处理队列里的站点。系统宕机恢复后调度器根据misfire_grace_time自动补跑数据最终一致不会出现大面积空洞。5. Flask发布与ECharts可视化查询接口与动态刷新5.1 发布端目录结构app.py models.py crawler/ __init__.py fetcher.py parser.py services/ scheduler_service.py station_service.py templates/ index.html static/ js/ dashboard.jsapp.py只负责创建Flask实例、注册蓝图、启动调度器crawler包拆成请求和解析两个模块后期换数据源只需要改parser.py的字段映射其它代码不用动。services层隔离调度与业务逻辑答辩讲架构分层时这层拆解是很好的切入点。5.2 查询接口与参数化SQLfrom flask import Flask, request, jsonify from sqlalchemy import text app Flask(__name__) app.route(/api/weather/realtime) def realtime_weather(): city request.args.get(city, default西安, typestr) limit request.args.get(limit, default12, typeint) limit min(limit, 48) sql text( SELECT w.temp, w.humidity, w.pressure, w.record_time FROM weather_record w JOIN station s ON w.station_id s.station_id WHERE s.city :city ORDER BY w.record_time DESC LIMIT :limit ) rows db.session.execute(sql, {city: city, limit: limit}).fetchall() data [ { temp: float(r.temp), humidity: r.humidity, pressure: r.pressure, record_time: r.record_time.strftime(%Y-%m-%d %H:%M:%S), } for r in rows ] return jsonify({code: 0, data: data})limit做了48的上限保护防止前端参数传一个超大值把数据库拖垮。查询用绑定参数而不是字符串拼接从根本上杜绝SQL注入这条在安全评审里是必查项。record_time在返回前用strftime统一成字符串格式避免前端各浏览器时区解析不一致带来的八小时偏移。5.3 ECharts时间轴与自动刷新div idchart stylewidth:100%;height:420px;/div script srchttps://cdn.jsdelivr.net/npm/echarts5/dist/echarts.min.js/script script async function load() { const resp await fetch(/api/weather/realtime?city西安limit24); const data (await resp.json()).data.reverse(); chart.setOption({ xAxis: { data: data.map(d d.record_time) }, series: [{ data: data.map(d d.temp), type: line, smooth: true }] }); } const chart echarts.init(document.getElementById(chart)); load(); setInterval(load, 60000); /script接口返回的是倒序数据最新在前前端先reverse()再交给ECharts时间轴才会从左往右正常递增。setInterval 60秒刷新一次是正式发布节奏答辩演示时把60000改成10000评委能看到坐标轴自己滚动视觉效果比静态截图好很多。smooth: true把折线变成平滑曲线数值趋势一眼扫过去更直观。5.4 debug模式下的双调度器问题Flask的debugTrue会启动两个进程APScheduler在子进程里也执行一次结果就是定时任务重复入库。解决思路是加环境变量开关import os if os.environ.get(WERKZEUG_RUN_MAIN) true: scheduler schedule_jobs(app)WERKZEUG_RUN_MAIN只在重载进程里为true主进程不会重复注册。如果用的是gunicorn多worker部署同理只在worker 0里启动调度器。6. 答辩前自查时间戳核对、接口验证与爬虫日志比对6.1 检查入库时间是否真的在滚动SELECT station_id, MAX(record_time) AS latest FROM weather_record GROUP BY station_id;每个站点的latest时间应该落在最近10~15分钟内调度间隔加抓取耗时的余量。如果某个站点明显滞后优先查它的接口返回是否正常其次是解析器是否抛了异常被过滤掉。页面上最新时间点和当前时间差超过一个调度周期评委一眼就能看出来是假实时。6.2 模拟前端请求验证接口curl http://127.0.0.1:5000/api/weather/realtime?city榆林limit6检查返回JSON里code是否为0、data条数是否与limit一致、record_time格式是否为YYYY-MM-DD HH:MM:SS。配合jq管道可以做数值校验比如判断temp字段是否为有限数字。这个命令在答辩现场跑一次很快比打开浏览器等页面加载更直观。6.3 时区一致性核对日志里看到任务在14:00运行数据库里的create_time却显示06:00大概率是MySQL会话时区没有对齐。连接串里加engine create_engine( mysqlpymysql://root:passlocalhost/weather?charsetutf8mb4, connect_args{init_command: SET time_zone 08:00}, )同时把APScheduler的timezone、Python侧datetime.now()的时区统一成Asia/Shanghai。三处不一致时前端展示和日志时间戳对不上排查会非常耗时。最后一个容易忽视的点定时任务间隔不要调到30秒以下。气象数据本身分钟级更新抓太频繁除了增加被限流风险还会让数据库写入压力变大。演示前手动触发一次全量采集垫底再开启调度器页面起始数据齐全时间轴滚动效果也更完整。校验时多留意日志里scheduler和fetcher两处时间戳的先后关系差距过大不是事务没提交就是网络IO卡住了。本文还有配套的精品资源点击获取