Python自动化水文年鉴数据处理：从PDF/Excel到结构化数据的实战方法-平芜编程栈

1. 项目缘起：水文年鉴数据处理中的“脏活累活”

干了这么多年水文数据分析，最头疼的环节之一，就是从那些五花八门、格式各异的水文年鉴里把数据“抠”出来。这活儿，说难吧，原理不复杂；说简单吧，真干起来全是坑。你可能遇到过PDF扫描版，文字是图片，得靠OCR识别，结果“3.14”给你识别成“3.l4”；也可能遇到过Excel版，但表格合并单元格满天飞，表头跨了五六行，数据分散在几十个工作表里；还有更古老的，直接给你纸质版，让你自己看着办。

这些“脏活累累活”消耗了工程师大量的时间和精力，而且极易出错。一个数据抄错，可能整个模型的结果就南辕北辙。所以，我一直想整理一套相对通用的方法，用Python把这些流程自动化、标准化。所谓“通用”，不是指一个脚本通吃所有格式，那不现实；而是指一套可复用的方法论、工具链和应对策略，让你面对任何一种年鉴，都能快速找到切入点，构建起高效、可靠的数据提取流水线。

今天要聊的，就是这套方法的核心思路和实战工具箱。无论你是刚接触Python的水文新人，还是被数据整理折磨已久的老手，希望这些“踩坑”换来的经验，能帮你把时间更多地花在更有价值的分析建模上，而不是重复的复制粘贴。

2. 核心挑战与应对策略：拆解水文年鉴的“多样性”

在动手写代码之前，我们必须先理解对手。水文年鉴的数据组织方式虽然遵循一定的规范，但在数字化呈现上却千差万别。盲目开始编码，只会陷入无尽的调试和适配。我的策略是“先侦察，后作战”。

2.1 常见数据载体与侦察手段

首先，你需要判断你手头的年鉴是什么“体质”。

1. 结构化电子表格（如.xlsx, .xls）这是最理想的情况，但陷阱也最多。侦察要点：

工作表数量与命名：用pandas的ExcelFile对象快速查看。

import pandas as pd xls = pd.ExcelFile('水文年鉴.xlsx') print(xls.sheet_names) # 打印所有工作表名

表头结构：用pd.read_excel(‘水文年鉴.xlsx’, sheet_name=‘某站’, nrows=10)读取前10行，观察表头有几行、是否有合并单元格。合并单元格在pandas中通常表现为第一行有值，后续行为NaN。
数据起始行：真正的数据从第几行开始？表头下方是否有空行或说明行？
特殊格式：是否有用颜色、批注标记的异常值或备注信息？这些是后续清洗的重点。

2. 非结构化文档（如.pdf）这是最常见的硬骨头，分为两类：

文本型PDF：可以直接复制文字。用pdfplumber库打开，能直接提取文字和表格框线。

import pdfplumber with pdfplumber.open('水文年鉴.pdf') as pdf: page = pdf.pages[5] # 查看第6页 text = page.extract_text() print(text[:500]) # 打印前500字符看看 # 尝试提取表格 tables = page.extract_tables() for table in tables: print(table)

扫描图像型PDF：文字是图片，必须使用OCR（光学字符识别）。pdf2image库将PDF转为图片，再用pytesseract或paddleocr进行识别。这是精度和速度的权衡点。

3. 纯文本或数据库文件（如.txt, .csv, .mdb）这类相对友好，但需要注意编码问题（特别是中文）和数据分隔符。用文本编辑器（如VS Code）打开查看前几行，判断是固定宽度分隔还是逗号、制表符分隔。

侦察经验：永远不要相信文件后缀名！一个名为.xls的文件，用pandas打不开，用文本编辑器打开发现其实是HTML代码伪装的情况我遇到过。先用最简单的方法（如file命令，或直接文本编辑器查看文件头部）做个初步判断。

2.2 制定通用提取流程框架

基于侦察结果，我总结出一个四层流水线框架，它像一条生产线，数据从一端流入，经过层层处理，从另一端流出干净的结构化数据。

第一层：数据加载与初筛

目标：将原始数据读入Python，转化为可操作的对象（如DataFrame、文本列表、图像对象）。
工具选型：
- Excel/CSV:pandas.read_excel,pandas.read_csv
- PDF文本:pdfplumber
- PDF扫描件:pdf2image+paddleocr
- Access数据库(.mdb):pyodbc或mdbtools（Linux/macOS）
关键动作：这一步就要处理编码问题（指定encoding=‘gbk’或‘utf-8’）、跳过无关行（skiprows参数）、选择特定工作表。

第二层：结构解析与定位

目标：在加载的数据中，精准定位到目标数据块。这是最需要“智慧”的一步。
核心策略：
1. 模式匹配：寻找标志性关键词。例如，寻找包含“站名”、“日期”、“水位”、“流量”等字段的行，这些行附近往往就是数据区的开始。
2. 表格探测：对于PDF，利用pdfplumber的表格检测功能；对于不规则文本，根据空格或制表符的规律手动划分。
3. 坐标定位（针对扫描件）：OCR后，每个识别出的文字都有其边界框坐标。通过寻找表头关键词的坐标，可以划定一个矩形区域，只识别该区域内的文字，提高精度和速度。
关键动作：编写灵活的查找函数，能够容忍一定的文字识别误差或格式微调。

第三层：数据清洗与规整

目标：将定位到的原始数据列表或数组，清洗成规整的DataFrame。
常见“脏数据”及处理：
- 缺失值与占位符：将“—”、“***”、“NaN”等统一替换为numpy.nan。
- 单位混杂：数据中混有“m³/s”、“(m)”等单位符号，需要用字符串方法（如.str.extract）分离数值和单位。
- 识别错误：OCR导致的“0”和“O”，“1”和“l”混淆，需要建立常见错误映射表进行替换。
- 格式不一致：日期可能是“2023-01-01”、“2023/1/1”、“20230101”，需要用pd.to_datetime统一，并设置errors=‘coerce’捕获异常格式。
- 多余空格与换行：使用.str.strip()清除。

第四层：验证与输出

目标：确保提取的数据逻辑正确，并输出为可用的格式。
验证方法：
- 范围校验：检查水位、流量值是否在历史合理范围内。
- 逻辑校验：例如，日降水量不应小于时段降水量；水位应有连续变化，不会剧烈跳变。
- 统计校验：与已知的统计特征（如月均值）进行粗略对比。
输出：清洗后的DataFrame输出为feather（速度快）、parquet（压缩比高）或标准的CSV文件，并附上一个简单的数据质量报告（如缺失值统计、异常值列表）。

这个框架是通用的思维模型，具体到每个项目，你需要为每一层选择合适的工具和参数。接下来，我们深入到最典型的两种场景中，看看具体怎么操作。

3. 实战场景一：从复杂Excel年鉴中提取多站数据

假设我们拿到一份省级水文年鉴的Excel文件，里面按河流或区域分了多个工作表，每个工作表里有多个测站的数据，表头可能有两三行。

3.1 逆向工程：手动解析文件结构

在写任何代码之前，我会先用Excel软件打开文件，花15-30分钟做以下事情：

记录所有工作表的名称和大致内容。
选择一个典型的工作表（如“长江干流”），观察其布局。表头有几行？每一列的含义是什么？数据区从第几行开始？
寻找规律。不同测站的数据是如何分隔的？是靠空行，还是靠一个合并了单元格的站名行？
记录下关键的模式。例如：“每个测站的数据块，总是以一个包含‘站名’、‘站码’的单元格开始，数据部分占据接下来的N行，直到下一个‘站名’单元格或文件结束。”

这个手动分析的过程至关重要，它直接决定了你自动化脚本的逻辑。

3.2 构建灵活的提取函数

基于上面的分析，我们来编写核心提取函数。这里的关键是利用pandas的块读取和条件判断能力，而不是一次性读入整个表。

import pandas as pd import numpy as np def extract_station_data_from_excel(file_path, sheet_name, header_rows=2, station_keyword='站名'): """ 从具有复杂格式的Excel工作表中提取各测站数据。 参数: file_path: Excel文件路径。 sheet_name: 工作表名称。 header_rows: 表头所占的行数（通常为合并单元格）。 station_keyword: 标识一个新测站开始的单元格内容关键词（如‘站名’、‘站码’）。 返回: 一个字典，键为测站名称，值为该站数据的DataFrame。 """ # 首先，将整个工作表读成一个DataFrame，不设表头，以便我们观察原始结构 df_raw = pd.read_excel(file_path, sheet_name=sheet_name, header=None) stations_data = {} current_station = None data_start_row = None # 假设真正的数据列名在header_rows行（索引从0开始） # 我们先获取这一行作为潜在的列名 potential_columns = df_raw.iloc[header_rows - 1].tolist() # 例如header_rows=2，则取第1行（0-indexed） # 从表头行之后开始遍历 for idx, row in df_raw.iterrows(): if idx < header_rows: continue # 跳过表头行 # 检查当前行是否包含测站标识（例如，第一列包含‘站名’） # 这里需要根据实际情况调整判断逻辑，比如检查单元格是否包含关键词 cell_value = str(row.iloc[0]) if pd.notna(row.iloc[0]) else '' if station_keyword in cell_value or (cell_value and not cell_value.strip().isdigit()): # 发现一个新测站！首先保存上一个测站的数据 if current_station is not None and data_start_row is not None: # 提取上一个测站的数据块 station_df = df_raw.iloc[data_start_row:idx].copy() # 设置正确的列名 station_df.columns = potential_columns # 简单的清洗：重置索引，删除全为空的行 station_df.reset_index(drop=True, inplace=True) station_df.dropna(how='all', inplace=True) stations_data[current_station] = station_df # 开始记录新测站 current_station = cell_value.replace(station_keyword, '').strip() data_start_row = idx + 1 # 假设数据从标识行的下一行开始 # 循环结束后，处理最后一个测站 if current_station is not None and data_start_row is not None: station_df = df_raw.iloc[data_start_row:].copy() station_df.columns = potential_columns station_df.reset_index(drop=True, inplace=True) station_df.dropna(how='all', inplace=True) stations_data[current_station] = station_df return stations_data # 使用示例 file_path = '某省水文年鉴2022.xlsx' sheet_name = '长江干流' all_stations = extract_station_data_from_excel(file_path, sheet_name, header_rows=2, station_keyword='站名') for station_name, data_df in all_stations.items(): print(f"测站: {station_name}, 数据量: {len(data_df)}") # 这里可以将data_df保存为单独的文件或进行进一步处理 # data_df.to_csv(f'{station_name}.csv', index=False, encoding='utf-8-sig')

这个函数是一个基础框架，实际应用中需要根据年鉴的具体格式进行大量调整。例如，标识测站的可能不是“站名”，而是“站码”；数据块可能不是紧接着标识行，中间还有一行单位行等等。

3.3 处理合并单元格与多级表头

合并单元格是pandas读取时的噩梦，因为它只在左上角单元格有值。一个实用的技巧是：用openpyxl引擎读取，并设置header=None，然后自己用前向填充（ffill）的方法来补全合并的单元格。

def read_excel_with_merged_cells(file_path, sheet_name, fill_na=True): """ 读取包含合并单元格的Excel，并尝试补全表头。 """ # 使用openpyxl引擎，读取为原始网格 df = pd.read_excel(file_path, sheet_name=sheet_name, header=None, engine='openpyxl') if fill_na: # 对前两行（假设是表头区域）进行前向填充，补全合并单元格 df.iloc[:2] = df.iloc[:2].ffill(axis=1) # 现在可以将补全后的行作为表头 # 例如，将第0行和第1行组合成多级列索引（如果需要） # header = [df.iloc[0].tolist(), df.iloc[1].tolist()] # df = df.iloc[2:] # 数据从第2行开始 # df.columns = pd.MultiIndex.from_arrays(header) # 或者简单点，只取最后一行作为列名 df.columns = df.iloc[1] # 假设第1行（0-indexed）是真正的列名 df = df.iloc[2:].reset_index(drop=True) return df

踩坑心得：对于极其复杂的表格，有时“半自动化”更高效。即用pandas或openpyxl将数据读到DataFrame后，不追求一步到位的完美解析，而是先提取出一个“粗糙”但完整的数据块，然后利用DataFrame强大的查询和清洗功能（如df[df[‘站名’].notna()]）进行二次分割和整理。这比写一个复杂脆弱的解析逻辑要稳健得多。

4. 实战场景二：征服扫描版PDF年鉴

扫描版PDF是数据提取的“终极挑战”。我们的武器库是：pdf2image+PaddleOCR。我选择PaddleOCR而不是Tesseract，主要是因为其对中文排版和复杂场景的识别效果更好，且社区活跃。

4.1 环境搭建与OCR引擎初始化

首先，安装必要的库。这里强烈建议使用conda或venv创建独立环境。

# 安装 pdf2image，需要系统安装 poppler # Ubuntu/Debian: sudo apt-get install poppler-utils # macOS: brew install poppler # Windows: 下载 poppler for windows 并将 bin 目录加入 PATH pip install pdf2image paddlepaddle paddleocr opencv-python

初始化OCR引擎时，可以针对水文数据的特性进行优化：

from paddleocr import PaddleOCR import pdf2image import cv2 import numpy as np # 初始化PaddleOCR # 使用方向分类器（cls=True）有助于纠正倾斜文本，但速度稍慢 # 使用轻量级模型（use_angle_cls=True, lang='ch'）在精度和速度间取得平衡 ocr_engine = PaddleOCR(use_angle_cls=True, lang='ch', show_log=False) # 如果硬件允许（有GPU），可以启用GPU加速 # ocr_engine = PaddleOCR(use_angle_cls=True, lang='ch', use_gpu=True)

4.2 核心流程：从PDF到结构化数据

流程可以概括为：分页转图像 -> 定位数据区域 -> OCR识别 -> 文本后处理。

def extract_data_from_scanned_pdf(pdf_path, output_dir='output'): """ 从扫描版PDF水文年鉴中提取数据。 """ import os os.makedirs(output_dir, exist_ok=True) # 1. 将PDF转换为图像列表 # dpi越高，识别精度可能越高，但速度越慢，内存消耗越大。300-400 DPI是常用值。 print(f"正在转换PDF: {pdf_path}") images = pdf2image.convert_from_path(pdf_path, dpi=300) all_page_data = [] for page_num, image in enumerate(images): print(f" 处理第 {page_num + 1} / {len(images)} 页...") # 将PIL图像转换为OpenCV格式（numpy数组） img_cv = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR) # 2. (可选) 图像预处理，提升OCR效果 # 例如，转为灰度图、二值化、降噪等。但PaddleOCR内置了较强的预处理，通常不需要。 # gray = cv2.cvtColor(img_cv, cv2.COLOR_BGR2GRAY) # _, binary = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY | cv2.THRESH_OTSU) # 3. 执行OCR # result 是一个列表，每个元素对应一行识别结果，包含文本框坐标和识别文本 result = ocr_engine.ocr(img_cv, cls=True) # 4. 解析OCR结果，定位数据区域 # 假设我们知道数据表大概在页面中间，且表头包含“日期”和“水位” page_text_lines = [] data_region_bbox = None # 数据区域的边界框 [x_min, y_min, x_max, y_max] for line in result: if line is None: continue # line的结构: [[[x1,y1],[x2,y2],[x3,y3],[x4,y4]], (text, confidence)] box, (text, conf) = line # 将文本和其中心点坐标存储起来 center_x = np.mean([point[0] for point in box]) center_y = np.mean([point[1] for point in box]) page_text_lines.append((text, conf, center_x, center_y, box)) # 通过关键词寻找表头，从而确定数据区域 # 这是一个简化的逻辑，实际情况可能需要更复杂的模式匹配 if '日期' in text or '水位' in text: # 假设表头在这一行，数据区域在其下方 # 获取表头行的Y坐标（取文本框底部） header_bottom = max(point[1] for point in box) # 简单设定一个数据区域，例如从表头下方20像素到页面底部 height, width = img_cv.shape[:2] data_region_bbox = [50, header_bottom + 20, width - 50, height - 50] print(f" 在第{page_num+1}页定位到疑似表头: '{text}'， 数据区域Y轴起始于: {header_bottom + 20}") # 5. 提取数据区域内的文本行，并按Y坐标排序（模拟行的概念） if data_region_bbox: x_min, y_min, x_max, y_max = data_region_bbox data_lines = [] for text, conf, cx, cy, box in page_text_lines: # 检查文本中心点是否在数据区域内 if x_min <= cx <= x_max and y_min <= cy <= y_max: data_lines.append((cy, text, conf)) # 用Y坐标排序 # 按Y坐标分组，将相近Y坐标的文本视为同一行 data_lines.sort(key=lambda x: x[0]) grouped_lines = [] current_y = None current_line_texts = [] y_tolerance = 5 # Y坐标容差，认为在5个像素内是同一行 for y, text, conf in data_lines: if current_y is None or abs(y - current_y) > y_tolerance: if current_line_texts: # 将同一行的文本按X坐标排序后合并 grouped_lines.append(' '.join(current_line_texts)) current_line_texts = [text] current_y = y else: current_line_texts.append(text) if current_line_texts: grouped_lines.append(' '.join(current_line_texts)) # 6. 将识别出的文本行转换为表格（这里非常简化） # 实际中，你需要根据空格或固定宽度来分割列 table_data = [] for line in grouped_lines: # 假设数据列由多个空格分隔 columns = line.split() if len(columns) > 1: # 简单过滤掉可能不是数据的行 table_data.append(columns) if table_data: # 尝试将数据转为DataFrame try: df_page = pd.DataFrame(table_data[1:], columns=table_data[0]) # 第一行作为表头 all_page_data.append(df_page) print(f" 第{page_num+1}页提取出 {len(df_page)} 行数据。") except Exception as e: print(f" 第{page_num+1}页数据转换DataFrame失败: {e}") else: print(f" 第{page_num+1}页未定位到数据区域。") # 7. 合并所有页的数据 if all_page_data: final_df = pd.concat(all_page_data, ignore_index=True) output_path = os.path.join(output_dir, 'extracted_data.csv') final_df.to_csv(output_path, index=False, encoding='utf-8-sig') print(f"\n所有数据已提取并保存至: {output_path}") return final_df else: print("\n未提取到任何数据。") return None

4.3 精度提升与后处理技巧

OCR识别不可能100%准确，尤其是对扫描质量不佳、字体特殊、有污渍的文件。以下是一些提升精度的实战技巧：

区域聚焦：不要整页识别。先用程序或人工确定数据表格在页面上的大致坐标区域（data_region_bbox），只对这个区域进行OCR，能大幅减少干扰，提高速度和精度。
自定义字库：PaddleOCR支持自定义字典。将水文专业术语（如测站名、参数名）和常见单位整理成一个user_dict.txt文件，在初始化引擎时通过det_db_box_thresh和rec_char_dict_path等参数引入，能显著提升相关词汇的识别率。
后处理规则引擎：
- 数字校正：识别出的文本，用正则表达式（如r'[\d\.]+'）提取所有疑似数字的片段。对于像“125.”这样的结果，很可能是“125.0”识别错误。
- 常见错误映射：建立字典，替换常见OCR错误，如{‘l’: ‘1’, ‘O’: ‘0’, ‘,’: ‘.’, ‘ ’: ‘’}（替换空格）。
- 基于上下文的校验：例如，识别出的“水位”值，如果突然比前后值大一个数量级，很可能“17.5”被识别成了“175”，需要结合前后数据点进行插值或标记为可疑值。
人机结合校验：对于关键数据或OCR置信度低的行，可以输出到一个中间文件，用高亮标出，供人工二次确认。这比全部人工核对效率高得多。

血泪教训：不要试图用一个复杂的脚本解决所有问题。对于扫描版PDF，我现在的策略是：先跑通一个基础流程，拿到“能用但有点脏”的初稿数据。然后，集中精力编写针对性的清洗规则来处理这些“脏”数据。把OCR识别和逻辑清洗分开，会让你的代码更清晰，也更容易维护。有时候，花半天时间写一个精巧的清洗函数，比折腾OCR参数提升那1%的准确率要划算得多。

5. 数据清洗与质量控制的自动化策略

无论数据来源如何，清洗都是必经之路。自动化清洗的目标不是追求完美，而是建立一套可重复、可审计的规则流水线。

5.1 构建模块化的清洗管道

我习惯将清洗步骤封装成一个个小函数，然后用pandas的pipe方法将它们串联起来，形成一个清洗管道。

def clean_hydrological_data(df): """ 水文数据清洗管道。 """ df_clean = (df .pipe(remove_extra_spaces) .pipe(standardize_column_names) .pipe(convert_date_columns) .pipe(handle_numeric_columns) .pipe(flag_anomalies) .pipe(drop_duplicate_rows) ) return df_clean def remove_extra_spaces(df): """去除所有字符串列的首尾空格和中间多余空格。""" str_cols = df.select_dtypes(include=['object']).columns for col in str_cols: df[col] = df[col].astype(str).str.strip().str.replace(r'\s+', ' ', regex=True) return df def standardize_column_names(df): """标准化列名：小写，替换空格为下划线。""" df.columns = df.columns.str.lower().str.replace(r'[\s\/\(\)]+', '_', regex=True).str.strip('_') return df def convert_date_columns(df, date_patterns=['%Y-%m-%d', '%Y/%m/%d', '%Y%m%d']): """尝试将疑似日期的列转换为datetime类型。""" for col in df.columns: # 简单启发式：列名包含‘date’、‘time’或‘日期’、‘时间’ if any(keyword in col.lower() for keyword in ['date', 'time', '日期', '时间']): for pattern in date_patterns: try: df[col] = pd.to_datetime(df[col], format=pattern, errors='coerce') # 如果成功转换的非空值超过一定比例，就认为转换成功 if df[col].notna().sum() > len(df) * 0.5: print(f"列 '{col}' 已按格式 '{pattern}' 转换为日期。") break except: continue return df def handle_numeric_columns(df): """处理数值列：移除单位符号，转换类型，处理占位符。""" # 定义需要处理的占位符和单位 placeholders = ['—', '***', 'NaN', 'NA', 'NULL', ''] units_to_remove = ['m³/s', 'm', 'mm', '°C', '℃'] for col in df.columns: # 跳过日期列 if pd.api.types.is_datetime64_any_dtype(df[col]): continue # 如果是对象类型，尝试清理并转换 if df[col].dtype == 'object': s = df[col].astype(str) # 移除单位符号 for unit in units_to_remove: s = s.str.replace(unit, '', regex=False) # 替换占位符为NaN s = s.replace(placeholders, np.nan) # 尝试转换为数值 try: df[col] = pd.to_numeric(s, errors='coerce') print(f"列 '{col}' 已转换为数值类型。") except: # 转换失败，保持原样 pass return df def flag_anomalies(df): """基于业务逻辑标记异常值。""" # 示例：标记水位异常突变的行 if 'water_level' in df.columns: df['level_diff'] = df['water_level'].diff().abs() # 假设相邻时段水位变化超过5米为异常（根据实际情况调整） df['level_anomaly'] = df['level_diff'] > 5.0 # 示例：标记流量为负值的行 if 'discharge' in df.columns: df['discharge_anomaly'] = df['discharge'] < 0 return df def drop_duplicate_rows(df, subset=None): """删除完全重复的行。""" if subset is None: subset = df.columns.difference(['level_anomaly', 'discharge_anomaly'], sort=False) if {'level_anomaly', 'discharge_anomaly'}.issubset(df.columns) else df.columns initial_len = len(df) df = df.drop_duplicates(subset=subset, keep='first').reset_index(drop=True) dropped = initial_len - len(df) if dropped > 0: print(f"删除了 {dropped} 条重复记录。") return df

5.2 生成数据质量报告

清洗完成后，一份自动生成的质量报告能让你对数据心中有数。

def generate_data_quality_report(df, output_path='data_quality_report.txt'): """生成简单的数据质量报告。""" report_lines = [] report_lines.append("="*50) report_lines.append("水文数据质量报告") report_lines.append("="*50) report_lines.append(f"数据总行数: {len(df)}") report_lines.append(f"数据总列数: {len(df.columns)}") report_lines.append("\n--- 各列缺失值统计 ---") missing_stats = df.isnull().sum() missing_pct = (missing_stats / len(df) * 100).round(2) missing_df = pd.DataFrame({'缺失数量': missing_stats, '缺失百分比%': missing_pct}) report_lines.append(missing_df.to_string()) report_lines.append("\n--- 数值列统计摘要 ---") numeric_cols = df.select_dtypes(include=[np.number]).columns if len(numeric_cols) > 0: report_lines.append(df[numeric_cols].describe().round(2).to_string()) else: report_lines.append("(无非数值列)") report_lines.append("\n--- 异常值标记统计 ---") anomaly_cols = [col for col in df.columns if 'anomaly' in col] for col in anomaly_cols: if col in df.columns: count = df[col].sum() if df[col].dtype == bool else df[col].notna().sum() report_lines.append(f"{col}: {int(count)} 条") report_lines.append("\n--- 前5行数据预览 ---") report_lines.append(df.head().to_string()) report = '\n'.join(report_lines) with open(output_path, 'w', encoding='utf-8') as f: f.write(report) print(f"数据质量报告已生成: {output_path}") return report

这个报告虽然简单，但能快速告诉你数据的基本情况、缺失严重程度以及是否存在明显的异常值，是决定下一步分析或是否需要返工重新提取的重要依据。

6. 工程化与持续集成：让流程可持续

对于需要定期处理新版年鉴（如每年一次）的任务，将上述流程工程化至关重要。目标是：新数据来了，点一下按钮或跑一个命令，就能自动完成从提取到清洗到报告的全过程。

6.1 使用配置文件管理变量

不要将文件路径、表头行数、关键词等硬编码在脚本里。使用一个配置文件（如config.yaml或config.json）来管理它们。

# config.yaml data_sources: 2022: excel_path: "./raw_data/水文年鉴2022.xlsx" sheets: - name: "长江干流" header_rows: 2 station_keyword: "站名" data_start_keywords: ["日期", "水位"] - name: "黄河中游" header_rows: 3 station_keyword: "站码" data_start_keywords: ["时间", "流量"] 2021: pdf_path: "./raw_data/水文年鉴2021_scanned.pdf" dpi: 350 data_region: page1: [100, 200, 800, 1000] # x1, y1, x2, y2 page2: [100, 150, 800, 1100] processing: output_dir: "./processed_data" quality_report: true output_format: "parquet" # 可选 csv, feather, parquet

然后在主脚本中读取这个配置，你的代码就变成了一个可适配不同年份、不同格式的通用框架。

6.2 编写主控脚本与日志记录

主控脚本（如main.py）负责协调整个流程。

# main.py import yaml import logging from pathlib import Path from extraction import extract_station_data_from_excel, extract_data_from_scanned_pdf from cleaning import clean_hydrological_data, generate_data_quality_report # 设置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler('extraction.log'), logging.StreamHandler()]) logger = logging.getLogger(__name__) def main(config_path='config.yaml'): with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) output_dir = Path(config['processing']['output_dir']) output_dir.mkdir(parents=True, exist_ok=True) for year, source_config in config['data_sources'].items(): logger.info(f"开始处理 {year} 年数据...") if 'excel_path' in source_config: # 处理Excel for sheet_config in source_config['sheets']: logger.info(f" 处理工作表: {sheet_config['name']}") data_dict = extract_station_data_from_excel( source_config['excel_path'], sheet_config['name'], header_rows=sheet_config['header_rows'], station_keyword=sheet_config['station_keyword'] ) for station, df in data_dict.items(): if df is not None and not df.empty: df_clean = clean_hydrological_data(df) output_file = output_dir / f"{year}_{sheet_config['name']}_{station}.{config['processing']['output_format']}" if config['processing']['output_format'] == 'parquet': df_clean.to_parquet(output_file, index=False) elif config['processing']['output_format'] == 'feather': df_clean.to_feather(output_file) else: df_clean.to_csv(output_file, index=False, encoding='utf-8-sig') logger.info(f" 已保存测站 '{station}' 数据至 {output_file}") if config['processing']['quality_report']: report_path = output_dir / f"{year}_{sheet_config['name']}_{station}_quality_report.txt" generate_data_quality_report(df_clean, report_path) elif 'pdf_path' in source_config: # 处理PDF logger.info(f" 处理PDF文件: {source_config['pdf_path']}") df_all = extract_data_from_scanned_pdf( source_config['pdf_path'], output_dir=str(output_dir) ) if df_all is not None: df_clean = clean_hydrological_data(df_all) output_file = output_dir / f"{year}_extracted.{config['processing']['output_format']}" # ... 保存数据 logger.info(f" PDF数据已保存至 {output_file}") logger.info("所有数据处理完成！") if __name__ == '__main__': main()

6.3 利用任务调度实现自动化

对于年度任务，可以结合操作系统的任务计划（Windows任务计划程序、Linux的cron）或Python的schedule库，实现定期自动运行。更工程化的做法是使用Apache Airflow或Prefect这样的工作流管理平台，可以可视化地监控任务执行状态、处理依赖关系和失败重试。

最终，一个理想的水文年鉴数据处理项目，应该是一个结构清晰的代码仓库，包含config/（配置文件）、src/（核心模块）、tests/（单元测试）、data/raw/（原始数据）、data/processed/（处理结果）和logs/（运行日志）。每次新的年鉴数据到来，你只需要更新配置文件，然后运行python main.py，剩下的就交给自动化流程。这不仅能解放你的双手，更能保证数据处理过程的一致性和可追溯性，让数据质量真正可控。