简介：本资源是一套面向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人学习下载，资源结构清晰、模块解耦明确：从requests+BeautifulSoup爬取解析，到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. 从页面结构逆向定位真实数据源：先看懂水文站页面的三层嵌套逻辑

水文站点实时数据页面普遍采用“三级嵌套架构”：最外层是固定导航栏（含站名筛选器），中间是动态加载的表格容器（id="data-table"），最内层是每行数据的 DOM 节点（class="station-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 一致，否则 403
• X-Requested-With: 多数平台要求为XMLHttpRequest
• Cookie: 含 JSESSIONID 或 token，必须复用浏览器登录态

提示：有些平台接口地址带时间戳参数（如?t=1715892345678），这是防缓存机制，不能硬编码，需用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(options=chrome_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?stationId=HANKOU&startTime=2024-05-15T00:00:00Z&endTime=2024-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/Shanghai"

3.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( id=station_cfg["id"], start=start_utc.strftime("%Y-%m-%dT%H:%M:%SZ"), end=end_utc.strftime("%Y-%m-%dT%H:%M:%SZ") ) resp = session.get(url, timeout=15) 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", encoding="utf-8") as f: stations = yaml.safe_load(f) session = get_session_with_cookies() # 复用登录态 now = datetime.now() start_time = now - timedelta(hours=1) # 拉取最近 1 小时 results = [] with ThreadPoolExecutor(max_workers=3) 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_workers=3：经实测，该平台对单 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，强制对齐：

import 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, utc=True) 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(drop=True) 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"]] = df

4.2 时间序列对齐：用 resample 填补高频缺失点

水文站上报频率不一：汉口站每 10 分钟一报，南京站每小时一报。若直接画图，X 轴刻度混乱。标准做法是以最高频站为基准（如 10 分钟），对所有站做 reindex + ffill：

# 假设汉口站是最高频（10 分钟） ref_idx = pd.date_range( start=all_data[0]["data"][0].get("time", ""), end=datetime.now(pytz.UTC), freq="10T" ) for sid, df in normalized_dfs.items(): if df.empty: continue # 设置时间索引并重采样 df = df.set_index("datetime_utc").reindex(ref_idx, method="ffill") df = df.reset_index().rename(columns={"index": "datetime_utc"}) normalized_dfs[sid] = df

注意：“ffill” 是水利场景首选——水位不会突变，用前值填充比插值更符合物理规律；而降雨量必须用bfill或0填充，因“无记录”不等于“正在下雨”。

5. 避坑指南：那些让我凌晨三点改代码的血泪经验（附现象→原因→解决）

这类项目最耗时的从来不是写代码，而是排查平台侧的“静默变更”。以下是我在 11 个月运维中踩过的 5 个高频坑，每一条都配真实日志片段。

5.1 现象：某天所有站返回空数据，但 status_code=200

原因：平台悄悄升级了 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(options=chrome_options, service=Service(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", encoding="utf-8-sig", index=False)，utf-8-sig的 BOM 头能被 Excel 正确识别。

5.5 现象：matplotlib 折线图 Y 轴刻度全是科学计数法（1e6），看不出水位变化

原因：flow_m3s数值过大（28400），而water_level_m很小（15.23），共用 Y 轴导致缩放失衡。
解决：不用twinx()，改用subplots(3,1,sharex=True)分三行画，每行独立 Y 轴，加plt.gca().yaxis.set_major_formatter(ScalarFormatter(useOffset=False))关闭科学计数。

提示：所有这些坑，都在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"], color="blue", linewidth=2, label="水位 (m)") ax1.set_ylabel("水位 (m)", color="blue") ax1.tick_params(axis='y', labelcolor="blue") # 右轴：流量（红色虚线） ax2 = ax1.twinx() ax2.plot(df["datetime_utc"], df["flow_m3s"], color="red", linewidth=2, linestyle="--", label="流量 (m³/s)") ax2.set_ylabel("流量 (m³/s)", color="red") ax2.tick_params(axis='y', labelcolor="red") # 降雨柱状图（底部） ax3 = ax1.twinx() ax3.spines["bottom"].set_position(("axes", -0.2)) ax3.bar(df["datetime_utc"], df["rain_mm"], width=0.02, alpha=0.6, color="green", label="降雨量 (mm)") ax3.set_ylabel("降雨量 (mm)", color="green") ax3.tick_params(axis='y', labelcolor="green") # 添加警戒线（示例：汉口站警戒水位27.30m） if "HANKOU" in station_name: ax1.axhline(y=27.30, color="orange", linestyle="-.", linewidth=1.5, label="警戒水位 (27.30m)") # 格式化 X 轴时间 ax1.xaxis.set_major_formatter(DateFormatter("%m-%d %H:%M")) plt.setp(ax1.get_xticklabels(), rotation=30) # 图例合并 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, loc="upper left") plt.title(f"{station_name} 近24小时水文趋势", fontsize=14, pad=20) plt.tight_layout() plt.savefig(output_path, dpi=150, bbox_inches="tight") plt.close() # 调用示例 for sid, df in normalized_dfs.items(): if not df.empty: plot_station_trend(df, normalized_dfs[sid].iloc[0]["station_name"], f"output/{sid}_trend.png")

6.2 关键细节说明（为什么这样设）

最后说句实在话：这套方案上线后，我把它打包成 systemd service，设为每天 00:05 自动运行，至今没人工干预过。但它绝不是“一次写完永不动”的银弹——每月至少有一次平台前端微调，导致 XPath 失效；每季度可能有一次接口字段变更。真正的工程能力，不在于写出多炫的代码，而在于把“应对变更”的成本压到最低：YAML 配置管字段，日志分级管异常，脚本化部署管更新。我现在每次改stations.yaml，心里都踏实——因为知道，只要字段名对得上，后面所有环节自动跟上。希望帮到你。

本文还有配套的精品资源，点击获取