import threading # 放在你的其他 import 之前 if not hasattr(threading.Thread, "isAlive"): threading.Thread.isAlive = threading.Thread.is_alive # 给旧调试器一个别名 import warnings import os import re import numpy as np import pandas as pd from scipy import stats warnings.filterwarnings("ignore") def auto_detect_coord_columns(df: pd.DataFrame): """ 双重验证智能识别坐标列: 1. 严格正则匹配列名 2. 基于数值范围的地理学推断 """ lon_patterns = [ r'^lon', r'^lng', r'^longitude', r'经度', r'^x$', r'^utm_x$', r'^pixel_x$' ] lat_patterns = [ r'^lat', r'^latitude', r'纬度', r'^y$', r'^utm_y$', r'^pixel_y$' ] x_col, y_col = None, None for col in df.columns: col_str = str(col).lower().strip() if x_col is None and any(re.search(p, col_str) for p in lon_patterns): x_col = col if y_col is None and any(re.search(p, col_str) for p in lat_patterns): y_col = col if x_col and y_col: return x_col, y_col numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist() if len(numeric_cols) >= 2: col1, col2 = numeric_cols[0], numeric_cols[1] mean1 = df[col1].head(10).mean() mean2 = df[col2].head(10).mean() if abs(mean1) <= 90 and abs(mean2) > 90: y_col, x_col = col1, col2 elif abs(mean2) <= 90 and abs(mean1) > 90: x_col, y_col = col1, col2 else: if mean1 > mean2: x_col, y_col = col1, col2 else: x_col, y_col = col2, col1 print(f"💡 触发智能数值推断坐标列: X/经度->[{x_col}], Y/纬度->[{y_col}]") return x_col, y_col return df.columns[0], df.columns[1] def detect_outliers_iqr(data: pd.DataFrame, column: str) -> pd.Series: """使用 IQR 方法检测异常值,返回与 data 同索引的布尔序列""" s = pd.to_numeric(data[column], errors="coerce") q1 = s.quantile(0.25) q3 = s.quantile(0.75) iqr = q3 - q1 lower = q1 - 1.5 * iqr upper = q3 + 1.5 * iqr mask = (s < lower) | (s > upper) # 对 NaN 不判为异常 mask = mask.fillna(False) mask.index = data.index return mask def detect_outliers_zscore(data: pd.DataFrame, column: str, threshold: float = 3.0) -> pd.Series: """使用 Z-score 方法检测异常值,返回与 data 同索引的布尔序列""" s = pd.to_numeric(data[column], errors="coerce") z = pd.Series(stats.zscore(s.dropna()), index=s.dropna().index) mask = (z.abs() > threshold).reindex(data.index).fillna(False) return mask def read_csv_robust(path, **kwargs): """ 尝试多种编码读取 CSV;成功即返回 DataFrame。 kwargs 会透传给 pd.read_csv(比如 sep、dtype 等)。 """ # 按出现概率排序;优先无损 + 常见中文编码 encodings = [ "utf-8", "utf-8-sig", "gbk", "gb18030", "cp936", "utf-16", "utf-16le", "utf-16be", "cp1252", "big5", "latin1", # 最后兜底(能读但中文会变乱码) ] errors_modes = ["strict", "replace"] # 先严格,失败再替换非法字符 last_err = None for enc in encodings: for emode in errors_modes: try: return pd.read_csv(path, encoding=enc, **kwargs) except Exception as e: last_err = e continue # 如果全失败,抛出最后一个错误 raise last_err def _decimal_len(v) -> float: """计算数值或字符串小数点后的位数;若无法计算返回 NaN""" if pd.isna(v): return np.nan try: # 统一成字符串处理 s = str(v) if "." not in s: return 0 frac = s.split(".", 1)[1] # 去掉科学计数法中的尾随部分(如 '1.234e-05') frac = frac.split("e")[0].split("E")[0] return len(frac) except Exception: return np.nan def process_water_quality_data(input_file: str, output_file: str): """ 处理水质数据 CSV 文件 参数: input_file: 输入 CSV 文件路径 output_file: 输出 CSV 文件路径 """ # 0) 读取 print("正在读取 CSV 文件...") df = read_csv_robust(input_file) print(f"原始数据形状: {df.shape}") print(f"列名: {list(df.columns)}") # 0.5) 智能检测坐标列 x_col, y_col = auto_detect_coord_columns(df) print(f"坐标列检测结果: X/经度=[{x_col}], Y/纬度=[{y_col}]") # 1) 经纬度精度筛选(小数位 >= 7) print("\n正在筛选经纬度精度(小数位>=7)...") initial_count = len(df) for col in [y_col, x_col]: if col in df.columns: dec_len = df[col].apply(_decimal_len) keep_mask = dec_len >= 7 dropped = (~keep_mask).sum() df = df[keep_mask].copy() print(f"列 {col}: 去除了 {int(dropped)} 行(保留 {len(df)} 行)") after_coord_filter = len(df) print(f"经纬度精度筛选后剩余: {after_coord_filter} 行 (去除了 {initial_count - after_coord_filter} 行)") # 2) 异常值检测(IQR)- 只删除异常值,不删除整行 print("\n正在检测异常值(IQR)...") numeric_columns = df.select_dtypes(include=[np.number]).columns.tolist() exclude_columns = {"时间", "测量点", y_col, x_col} if "原始" in df.columns: exclude_columns.add("原始") columns_to_check = [c for c in numeric_columns if c not in exclude_columns] print(f"将检测以下列的异常值: {columns_to_check}") df_clean = df.copy() total_outliers_removed = 0 for column in columns_to_check: if column in df_clean.columns and df_clean[column].notna().sum() > 0: col_mask = detect_outliers_iqr(df_clean, column) outlier_count = int(col_mask.sum()) print(f'列 "{column}" 检测到 {outlier_count} 个异常值,将其设为 NaN') df_clean.loc[col_mask, column] = np.nan total_outliers_removed += outlier_count after_outlier_filter = len(df_clean) print(f"异常值处理完成: 保留 {after_outlier_filter} 行数据,共处理了 {total_outliers_removed} 个异常值") # 3) 去除 "原始" 列(若存在) if "原始" in df_clean.columns: df_clean = df_clean.drop(columns=["原始"]) print('已去除 "原始" 列') # 4) 字段类型处理:尽量把"时间"转为 datetime if "时间" in df_clean.columns: try: df_clean["时间"] = pd.to_datetime(df_clean["时间"], errors="coerce") except Exception: pass # 5) 按测量点统计平均值 print("\n正在按测量点统计平均值...") if "测量点" not in df_clean.columns: print('错误:未找到 "测量点" 列') return agg_dict = {} if "时间" in df_clean.columns and np.issubdtype(df_clean["时间"].dtype, np.datetime64): agg_dict["时间"] = "mean" elif "时间" in df_clean.columns: agg_dict["时间"] = lambda s: s.mode().iloc[0] if not s.mode().empty else s.dropna().iloc[0] if s.dropna().size else np.nan for col in [y_col, x_col]: if col in df_clean.columns: agg_dict[col] = "mean" for col in df_clean.select_dtypes(include=[np.number]).columns: if col not in {y_col, x_col}: agg_dict[col] = "mean" grouped = df_clean.groupby("测量点", as_index=False).agg(agg_dict) print(f"统计完成,共 {len(grouped)} 个测量点") print(f"输出数据形状: {grouped.shape}") # 6) 去除"时间"和"测量点"列 columns_to_drop = [] if "时间" in grouped.columns: columns_to_drop.append("时间") if "测量点" in grouped.columns: columns_to_drop.append("测量点") if columns_to_drop: grouped = grouped.drop(columns=columns_to_drop) print(f"已去除列: {columns_to_drop}") print(f"去除列后数据形状: {grouped.shape}") # 7) 保存 os.makedirs(os.path.dirname(output_file) or ".", exist_ok=True) grouped.to_csv(output_file, index=False, encoding="utf-8-sig") print(f"\n处理完成!结果已保存到: {output_file}") # 摘要 print("\n=== 处理结果摘要 ===") print(f"原始数据行数: {initial_count}") print(f"经纬度精度筛选后: {after_coord_filter}") print(f"异常值筛选后: {after_outlier_filter}") print(f"最终统计结果: {len(grouped)} 个测量点") return grouped def main(): """主函数""" input_file = r"D:\BaiduNetdiskDownload\yaobao\csv\input.csv" output_file =r"D:\BaiduNetdiskDownload\yaobao\csv\output_test.csv" if not output_file: output_file = "processed_water_quality.csv" try: _ = process_water_quality_data(input_file, output_file) except FileNotFoundError as e: print(f"文件未找到:{e}") except Exception as e: print(f"处理失败:{e}") if __name__ == "__main__": main()