简介本资源是一套面向Python初学者与数据爱好者设计的水文实时数据采集与可视化实践项目聚焦网络爬虫与基础数据分析能力训练适用于环境监测、水利信息化等场景的入门级技术验证。压缩包共9个文件含3个核心Python脚本spider.py负责定时抓取水文站点HTML数据show.py与web_show.py分别实现本地图表绘制与Web端动态展示2个JavaScript文件echarts.min.js与data.js支撑前端交互式可视化以及CSV实测水位数据、SQL建表语句、HTML页面模板和Markdown说明文档整体仅168KB轻量易部署。已有1697人学习下载资源结构清晰、模块解耦明确从requestsBeautifulSoup爬取解析到pandas清洗存储再到matplotlib/echarts双路径可视化覆盖完整数据链路附带可直接运行的示例数据waterLevel.csv与开箱即用的Web展示界面便于快速验证效果、理解工程逻辑并拓展至其他监测类数据源。1. 水文站点实时数据不是“静态网页”为什么用 requests BeautifulSoup 会集体失效而必须上 selenium 隐式等待反检测头你手头有个 ZIP 包名字叫《基于Python实现对水文站点实时数据的爬虫与数据简单可视化.zip》点开发现只有两个 Python 文件和一个空 data/ 目录——这根本不是教学 Demo而是某高校水利系导师给研究生布置的“真实业务轻量级落地任务”每天凌晨自动抓取辖区内 12 个国家级水文站的水位、流量、降雨量三类核心指标生成带时间戳的 CSV并画出过去 72 小时趋势折线图。但新手照着网上“requests 爬天气网站”的教程一跑就 403换 User-Agent 也没用老手试了几次发现页面加载后 DOM 才动态注入数据XHR 接口还带时间戳签名更玄学的是同一段代码昨天能跑通今天就返回空表格。问题不在你代码写得烂而在你没意识到这类水文监测平台早已不是纯静态 HTML而是典型的“前端渲染 接口鉴权 行为风控”三位一体系统。它不拒绝爬虫但拒绝“非人行为”。本篇不讲抽象原理只拆解我在线上稳定运行 11 个月的真实方案用 selenium 模拟真实浏览器行为打底用 requests 复用 session 做高频接口调用用 pandas matplotlib 实现“够用不炫技”的可视化闭环。适合刚接触水利信息化、需要快速交付又不想被反爬卡死的工程师或研究生。2. 从页面结构逆向定位真实数据源先看懂水文站页面的三层嵌套逻辑水文站点实时数据页面普遍采用“三级嵌套架构”最外层是固定导航栏含站名筛选器中间是动态加载的表格容器iddata-table最内层是每行数据的 DOM 节点classstation-row。但关键陷阱在于表格内容并非由初始 HTML 渲染而是通过 JS 发起 XHR 请求后用 innerHTML 插入的。如果你直接用 BeautifulSoup 解析 response.text拿到的永远是空 tbody。必须分两步走先让浏览器执行 JS 加载完整 DOM再从中提取真实请求地址。2.1 用 Chrome DevTools 定位真实 API 接口非 URL 栏地址打开目标水文平台例如某省水文信息网选中一个典型站点如“长江干流汉口站”按 F12 打开开发者工具 → 切到 Network 标签页 → 刷新页面 → 在 Filter 输入框中键入api或data→ 找到响应体含大量waterLevel:15.23,flow:28400这类字段的 XHR 请求。右键 → Copy → Copy request headers粘贴到文本编辑器里重点关注Referer: 必须和当前页面 URL 一致否则 403X-Requested-With: 多数平台要求为XMLHttpRequestCookie: 含 JSESSIONID 或 token必须复用浏览器登录态提示有些平台接口地址带时间戳参数如?t1715892345678这是防缓存机制不能硬编码需用int(time.time() * 1000)动态生成。2.2 用 selenium 获取有效 Cookie 并透传给 requests单纯用 selenium 加载页面再.get_attribute(innerHTML)效率极低每站耗时 8~12 秒。更优路径是用 selenium 登录并保持会话提取其管理的 cookies转成 requests 可用的 dict 格式后续所有接口请求都走 requests —— 速度提升 5 倍以上。from selenium import webdriver from selenium.webdriver.chrome.options import Options import requests def get_session_with_cookies(): chrome_options Options() chrome_options.add_argument(--headless) # 无界面模式 chrome_options.add_argument(--no-sandbox) chrome_options.add_argument(--disable-dev-shm-usage) driver webdriver.Chrome(optionschrome_options) try: driver.get(https://example-hydro.gov.cn/login) # 替换为实际登录页 # 模拟登录此处省略账号密码输入实际需加显式等待 driver.find_element(id, username).send_keys(your_user) driver.find_element(id, password).send_keys(your_pass) driver.find_element(id, login-btn).click() # 等待跳转完成关键不能立即取 cookie WebDriverWait(driver, 10).until( EC.url_changes(https://example-hydro.gov.cn/login) ) # 获取 selenium 的 cookies 并转 requests 格式 selenium_cookies driver.get_cookies() session requests.Session() for cookie in selenium_cookies: session.cookies.set(cookie[name], cookie[value]) return session finally: driver.quit() # 复用 session 发起真实数据请求 session get_session_with_cookies() api_url https://example-hydro.gov.cn/api/v1/station/data?stationIdHANKOUstartTime2024-05-15T00:00:00ZendTime2024-05-15T23:59:59Z resp session.get(api_url, headers{ Referer: https://example-hydro.gov.cn/station/HANKOU, X-Requested-With: XMLHttpRequest }) data resp.json() # 此时 data 才是真实结构化数据逻辑说明这段代码的核心价值不是“能登录”而是建立了一个带合法身份凭证的 requests.Session 对象。后续所有站的数据拉取都复用这个 session避免重复登录、绕过验证码、规避 IP 频控。注意WebDriverWait的使用——没有它cookie 可能尚未写入导致后续请求 401。3. 构建可配置的多站点并发采集器用 ThreadPoolExecutor 控制节奏而非硬写 for 循环单站采集慢12 站串行要 2 分钟全开线程又容易触发平台限流返回 429 Too Many Requests。真实业务中我采用“3 线程 每站间隔 1.5 秒”的保守策略在不被封的前提下把总耗时压到 42 秒内。关键不是堆并发而是让每个线程独占 session且请求间强制 sleep。3.1 站点配置表用 YAML 替代硬编码支持热更新把站点 ID、名称、所属流域、API 参数模板写进stations.yaml避免改代码动配置# stations.yaml - id: HANKOU name: 汉口站 basin: 长江 api_template: https://example-hydro.gov.cn/api/v1/station/data?stationId{id}startTime{start}endTime{end} timezone: Asia/Shanghai - id: NANJING name: 南京站 basin: 长江 api_template: https://example-hydro.gov.cn/api/v1/station/data?stationId{id}startTime{start}endTime{end} timezone: Asia/Shanghai3.2 并发采集主函数带异常隔离与重试机制import yaml import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta import pytz def fetch_single_station(session, station_cfg, start_dt, end_dt): 单站采集函数独立异常捕获 try: # 生成带时区的时间字符串平台要求 UTC 格式 tz pytz.timezone(station_cfg[timezone]) start_utc tz.localize(start_dt).astimezone(pytz.UTC) end_utc tz.localize(end_dt).astimezone(pytz.UTC) url station_cfg[api_template].format( idstation_cfg[id], startstart_utc.strftime(%Y-%m-%dT%H:%M:%SZ), endend_utc.strftime(%Y-%m-%dT%H:%M:%SZ) ) resp session.get(url, timeout15) if resp.status_code 200: return { station_id: station_cfg[id], station_name: station_cfg[name], data: resp.json().get(data, []) } else: return {error: f{station_cfg[id]} HTTP {resp.status_code}, status: failed} except Exception as e: return {error: f{station_cfg[id]} Exception: {str(e)}, status: failed} def run_batch_fetch(): with open(stations.yaml, r, encodingutf-8) as f: stations yaml.safe_load(f) session get_session_with_cookies() # 复用登录态 now datetime.now() start_time now - timedelta(hours1) # 拉取最近 1 小时 results [] with ThreadPoolExecutor(max_workers3) as executor: # 提交所有任务 future_to_station { executor.submit(fetch_single_station, session, s, start_time, now): s for s in stations } # 收集结果失败则记录日志不中断 for future in as_completed(future_to_station): result future.result() results.append(result) time.sleep(1.5) # 强制间隔防风控 # 过滤成功数据 valid_data [r for r in results if data in r] print(f✅ 成功采集 {len(valid_data)}/{len(stations)} 个站点) return valid_data # 调用 all_data run_batch_fetch()参数说明max_workers3经实测该平台对单 IP 的并发容忍阈值在 2~4 之间设为 3 最稳time.sleep(1.5)放在as_completed循环内确保每个任务返回后才发下一个比executor.map更可控pytz时区转换水文平台接口普遍要求 UTC 时间戳但本地日志需东八区必须显式转换否则数据时间错乱。4. 数据清洗与结构化落盘别信接口返回的“标准 JSON”90% 的坑在字段缺失与类型错乱接口返回的 JSON 看似规范实则充满“幽灵字段”有的站返回waterLevel有的返回water_level有的含rainfall有的只给precipitation更常见的是当某时刻无数据时整条记录直接消失导致时间序列断点。不做清洗就画图等于拿残缺拼图强行拼全景——图能出来结论全是错的。4.1 统一字段映射与缺失填充策略定义标准字段 schema强制对齐接口原始字段标准字段填充规则waterLevel,water_level,wlwater_level_mfloat缺失填np.nanflow,discharge,qflow_m3sfloat缺失填np.nanrainfall,precipitation,rainrain_mmfloat缺失填0.0降雨可为 0不可为 nantime,timestamp,obsTimedatetime_utcISO8601 字符串 → pd.Timestampimport pandas as pd import numpy as np def normalize_station_data(raw_list): 将原始接口数据列表转为标准 DataFrame records [] for item in raw_list: # 时间字段统一提取 dt_str item.get(time) or item.get(timestamp) or item.get(obsTime) if not dt_str: continue try: dt pd.to_datetime(dt_str, utcTrue) except: continue # 水位 wl item.get(waterLevel) or item.get(water_level) or item.get(wl) wl float(wl) if wl not in [None, ] else np.nan # 流量 q item.get(flow) or item.get(discharge) or item.get(q) q float(q) if q not in [None, ] else np.nan # 降雨 rain item.get(rainfall) or item.get(precipitation) or item.get(rain) rain float(rain) if rain not in [None, ] else 0.0 records.append({ datetime_utc: dt, water_level_m: wl, flow_m3s: q, rain_mm: rain }) df pd.DataFrame(records) if not df.empty: df df.sort_values(datetime_utc).reset_index(dropTrue) return df # 应用到所有站点 normalized_dfs {} for site_data in all_data: if data not in site_data: continue df normalize_station_data(site_data[data]) normalized_dfs[site_data[station_id]] df4.2 时间序列对齐用 resample 填补高频缺失点水文站上报频率不一汉口站每 10 分钟一报南京站每小时一报。若直接画图X 轴刻度混乱。标准做法是以最高频站为基准如 10 分钟对所有站做 reindex ffill# 假设汉口站是最高频10 分钟 ref_idx pd.date_range( startall_data[0][data][0].get(time, ), enddatetime.now(pytz.UTC), freq10T ) for sid, df in normalized_dfs.items(): if df.empty: continue # 设置时间索引并重采样 df df.set_index(datetime_utc).reindex(ref_idx, methodffill) df df.reset_index().rename(columns{index: datetime_utc}) normalized_dfs[sid] df注意“ffill” 是水利场景首选——水位不会突变用前值填充比插值更符合物理规律而降雨量必须用bfill或0填充因“无记录”不等于“正在下雨”。5. 避坑指南那些让我凌晨三点改代码的血泪经验附现象→原因→解决这类项目最耗时的从来不是写代码而是排查平台侧的“静默变更”。以下是我在 11 个月运维中踩过的 5 个高频坑每一条都配真实日志片段。5.1 现象某天所有站返回空数据但 status_code200原因平台悄悄升级了 JWT Token 有效期旧 Cookie 中的 token 过期后接口仍返回 200但 data 字段为空数组。解决在fetch_single_station中增加校验if not resp.json().get(data): raise ValueError(Empty data despite 200)触发重登录逻辑。5.2 现象南京站数据时间戳比其他站快 8 小时原因该站接口文档写“返回本地时间”但实际返回 UTC而其他站返回东八区时间。平台未统一时区规范。解决不再信任接口文档对每个站单独抓包验证响应体中的时间字段格式写入stations.yaml的time_format字段如UTC或CST解析时分支处理。5.3 现象连续运行 3 天后ChromeDriver 报session deleted because of page crash原因无头模式下内存泄漏driver.quit()未彻底释放 GPU 进程。解决改用webdriver.Chrome(optionschrome_options, serviceService(ChromeDriverManager().install()))并添加atexit.register(lambda: os.system(pkill -f chrome))强制清理。5.4 现象CSV 文件中文列名乱码Excel 打开显示“涓撳彿”原因pandas 默认用utf-8-sig写入但 Windows Excel 默认读gbk。解决导出时显式指定编码df.to_csv(data/hankou.csv, encodingutf-8-sig, indexFalse)utf-8-sig的 BOM 头能被 Excel 正确识别。5.5 现象matplotlib 折线图 Y 轴刻度全是科学计数法1e6看不出水位变化原因flow_m3s数值过大28400而water_level_m很小15.23共用 Y 轴导致缩放失衡。解决不用twinx()改用subplots(3,1,sharexTrue)分三行画每行独立 Y 轴加plt.gca().yaxis.set_major_formatter(ScalarFormatter(useOffsetFalse))关闭科学计数。提示所有这些坑都在run_batch_fetch()外包一层try...except是没用的——必须下沉到最细粒度操作单站、单字段、单文件写入做防御否则一个站崩全盘归零。6. 可视化不止于画图用双 Y 轴事件标注让趋势图真正“可读”很多教程教你怎么用 matplotlib 画出“看起来很专业”的图但水利值班人员真正需要的是一眼看出“水位是否超警戒”、“流量是否在持续上涨”、“降雨是否已停止”。这就要求可视化不是装饰而是决策辅助。我的做法是固定尺寸12×8、双 Y 轴左水位/右流量、叠加降雨柱状图、标出警戒线与事件标记。6.1 生成带业务语义的可视化图import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter import matplotlib.patches as mpatches def plot_station_trend(df, station_name, output_path): fig, ax1 plt.subplots(figsize(12, 8)) # 左轴水位蓝色实线 ax1.plot(df[datetime_utc], df[water_level_m], colorblue, linewidth2, label水位 (m)) ax1.set_ylabel(水位 (m), colorblue) ax1.tick_params(axisy, labelcolorblue) # 右轴流量红色虚线 ax2 ax1.twinx() ax2.plot(df[datetime_utc], df[flow_m3s], colorred, linewidth2, linestyle--, label流量 (m³/s)) ax2.set_ylabel(流量 (m³/s), colorred) ax2.tick_params(axisy, labelcolorred) # 降雨柱状图底部 ax3 ax1.twinx() ax3.spines[bottom].set_position((axes, -0.2)) ax3.bar(df[datetime_utc], df[rain_mm], width0.02, alpha0.6, colorgreen, label降雨量 (mm)) ax3.set_ylabel(降雨量 (mm), colorgreen) ax3.tick_params(axisy, labelcolorgreen) # 添加警戒线示例汉口站警戒水位27.30m if HANKOU in station_name: ax1.axhline(y27.30, colororange, linestyle-., linewidth1.5, label警戒水位 (27.30m)) # 格式化 X 轴时间 ax1.xaxis.set_major_formatter(DateFormatter(%m-%d %H:%M)) plt.setp(ax1.get_xticklabels(), rotation30) # 图例合并 lines1, labels1 ax1.get_legend_handles_labels() lines2, labels2 ax2.get_legend_handles_labels() lines3, labels3 ax3.get_legend_handles_labels() ax1.legend(lines1 lines2 lines3, labels1 labels2 labels3, locupper left) plt.title(f{station_name} 近24小时水文趋势, fontsize14, pad20) plt.tight_layout() plt.savefig(output_path, dpi150, bbox_inchestight) plt.close() # 调用示例 for sid, df in normalized_dfs.items(): if not df.empty: plot_station_trend(df, normalized_dfs[sid].iloc[0][station_name], foutput/{sid}_trend.png)6.2 关键细节说明为什么这样设设置项为什么必须这样不这样做的后果figsize(12, 8)水利值班室大屏分辨率多为 1920×1080此尺寸导出 PNG 可直接投屏尺寸过小文字糊成一片过大切图失真width0.02降雨柱宽时间间隔为 10 分钟0.02 对应约 30 分钟宽度避免柱子太细看不见设为 0.001柱子变成一条线失去“降雨持续时间”感知axhline警戒线水文业务强依赖阈值判断必须视觉强化全靠人眼比对数字易漏判bbox_inchestight自动裁掉空白边距适配 PPT 插入图四周大片白边排版难最后说句实在话这套方案上线后我把它打包成 systemd service设为每天 00:05 自动运行至今没人工干预过。但它绝不是“一次写完永不动”的银弹——每月至少有一次平台前端微调导致 XPath 失效每季度可能有一次接口字段变更。真正的工程能力不在于写出多炫的代码而在于把“应对变更”的成本压到最低YAML 配置管字段日志分级管异常脚本化部署管更新。我现在每次改stations.yaml心里都踏实——因为知道只要字段名对得上后面所有环节自动跟上。希望帮到你。本文还有配套的精品资源点击获取