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‘某站’, nrows10)读取前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_csvPDF文本:pdfplumberPDF扫描件:pdf2imagepaddleocrAccess数据库(.mdb):pyodbc或mdbtoolsLinux/macOS关键动作这一步就要处理编码问题指定encoding‘gbk’或‘utf-8’、跳过无关行skiprows参数、选择特定工作表。第二层结构解析与定位目标在加载的数据中精准定位到目标数据块。这是最需要“智慧”的一步。核心策略模式匹配寻找标志性关键词。例如寻找包含“站名”、“日期”、“水位”、“流量”等字段的行这些行附近往往就是数据区的开始。表格探测对于PDF利用pdfplumber的表格检测功能对于不规则文本根据空格或制表符的规律手动划分。坐标定位针对扫描件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_rows2, station_keyword站名): 从具有复杂格式的Excel工作表中提取各测站数据。 参数: file_path: Excel文件路径。 sheet_name: 工作表名称。 header_rows: 表头所占的行数通常为合并单元格。 station_keyword: 标识一个新测站开始的单元格内容关键词如‘站名’、‘站码’。 返回: 一个字典键为测站名称值为该站数据的DataFrame。 # 首先将整个工作表读成一个DataFrame不设表头以便我们观察原始结构 df_raw pd.read_excel(file_path, sheet_namesheet_name, headerNone) stations_data {} current_station None data_start_row None # 假设真正的数据列名在header_rows行索引从0开始 # 我们先获取这一行作为潜在的列名 potential_columns df_raw.iloc[header_rows - 1].tolist() # 例如header_rows2则取第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(dropTrue, inplaceTrue) station_df.dropna(howall, inplaceTrue) 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(dropTrue, inplaceTrue) station_df.dropna(howall, inplaceTrue) 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_rows2, 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, indexFalse, encodingutf-8-sig)这个函数是一个基础框架实际应用中需要根据年鉴的具体格式进行大量调整。例如标识测站的可能不是“站名”而是“站码”数据块可能不是紧接着标识行中间还有一行单位行等等。3.3 处理合并单元格与多级表头合并单元格是pandas读取时的噩梦因为它只在左上角单元格有值。一个实用的技巧是用openpyxl引擎读取并设置headerNone然后自己用前向填充ffill的方法来补全合并的单元格。def read_excel_with_merged_cells(file_path, sheet_name, fill_naTrue): 读取包含合并单元格的Excel并尝试补全表头。 # 使用openpyxl引擎读取为原始网格 df pd.read_excel(file_path, sheet_namesheet_name, headerNone, engineopenpyxl) if fill_na: # 对前两行假设是表头区域进行前向填充补全合并单元格 df.iloc[:2] df.iloc[:2].ffill(axis1) # 现在可以将补全后的行作为表头 # 例如将第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(dropTrue) return df踩坑心得对于极其复杂的表格有时“半自动化”更高效。即用pandas或openpyxl将数据读到DataFrame后不追求一步到位的完美解析而是先提取出一个“粗糙”但完整的数据块然后利用DataFrame强大的查询和清洗功能如df[df[‘站名’].notna()]进行二次分割和整理。这比写一个复杂脆弱的解析逻辑要稳健得多。4. 实战场景二征服扫描版PDF年鉴扫描版PDF是数据提取的“终极挑战”。我们的武器库是pdf2imagePaddleOCR。我选择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 # 使用方向分类器clsTrue有助于纠正倾斜文本但速度稍慢 # 使用轻量级模型use_angle_clsTrue, langch在精度和速度间取得平衡 ocr_engine PaddleOCR(use_angle_clsTrue, langch, show_logFalse) # 如果硬件允许有GPU可以启用GPU加速 # ocr_engine PaddleOCR(use_angle_clsTrue, langch, use_gpuTrue)4.2 核心流程从PDF到结构化数据流程可以概括为分页转图像 - 定位数据区域 - OCR识别 - 文本后处理。def extract_data_from_scanned_pdf(pdf_path, output_diroutput): 从扫描版PDF水文年鉴中提取数据。 import os os.makedirs(output_dir, exist_okTrue) # 1. 将PDF转换为图像列表 # dpi越高识别精度可能越高但速度越慢内存消耗越大。300-400 DPI是常用值。 print(f正在转换PDF: {pdf_path}) images pdf2image.convert_from_path(pdf_path, dpi300) 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, clsTrue) # 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_num1}页定位到疑似表头: {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(keylambda 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:], columnstable_data[0]) # 第一行作为表头 all_page_data.append(df_page) print(f 第{page_num1}页提取出 {len(df_page)} 行数据。) except Exception as e: print(f 第{page_num1}页数据转换DataFrame失败: {e}) else: print(f 第{page_num1}页未定位到数据区域。) # 7. 合并所有页的数据 if all_page_data: final_df pd.concat(all_page_data, ignore_indexTrue) output_path os.path.join(output_dir, extracted_data.csv) final_df.to_csv(output_path, indexFalse, encodingutf-8-sig) print(f\n所有数据已提取并保存至: {output_path}) return final_df else: print(\n未提取到任何数据。) return None4.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, , regexTrue) return df def standardize_column_names(df): 标准化列名小写替换空格为下划线。 df.columns df.columns.str.lower().str.replace(r[\s\/\(\)], _, regexTrue).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], formatpattern, errorscoerce) # 如果成功转换的非空值超过一定比例就认为转换成功 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, , regexFalse) # 替换占位符为NaN s s.replace(placeholders, np.nan) # 尝试转换为数值 try: df[col] pd.to_numeric(s, errorscoerce) 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, subsetNone): 删除完全重复的行。 if subset is None: subset df.columns.difference([level_anomaly, discharge_anomaly], sortFalse) if {level_anomaly, discharge_anomaly}.issubset(df.columns) else df.columns initial_len len(df) df df.drop_duplicates(subsetsubset, keepfirst).reset_index(dropTrue) dropped initial_len - len(df) if dropped 0: print(f删除了 {dropped} 条重复记录。) return df5.2 生成数据质量报告清洗完成后一份自动生成的质量报告能让你对数据心中有数。def generate_data_quality_report(df, output_pathdata_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, encodingutf-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(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.FileHandler(extraction.log), logging.StreamHandler()]) logger logging.getLogger(__name__) def main(config_pathconfig.yaml): with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) output_dir Path(config[processing][output_dir]) output_dir.mkdir(parentsTrue, exist_okTrue) 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_rowssheet_config[header_rows], station_keywordsheet_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, indexFalse) elif config[processing][output_format] feather: df_clean.to_feather(output_file) else: df_clean.to_csv(output_file, indexFalse, encodingutf-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_dirstr(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剩下的就交给自动化流程。这不仅能解放你的双手更能保证数据处理过程的一致性和可追溯性让数据质量真正可控。