From 82a8b4187a13e51dbdca03efa8d303294f552940 Mon Sep 17 00:00:00 2001 From: Huarch Date: Mon, 3 Aug 2026 15:50:26 +0800 Subject: [PATCH] feat: validate uploaded pipeline data --- app/prediction.py | 91 +++++++++++++++++++++++++++++++++++++++- tests/test_prediction.py | 52 +++++++++++++++++++++++ 2 files changed, 141 insertions(+), 2 deletions(-) diff --git a/app/prediction.py b/app/prediction.py index 83b8352..73b9ce0 100644 --- a/app/prediction.py +++ b/app/prediction.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import math import sys import uuid from dataclasses import dataclass @@ -290,6 +291,12 @@ def fill_defects_from_detail_sheet(path: Path, df: pd.DataFrame) -> pd.DataFrame for index in df.index[needs_fill]: excel_row = int(df.at[index, "_excel_row"]) + template_id = normalize_id_value(df.at[index, ID_COLUMN], "") if ID_COLUMN in df.columns else "" + detail_id = normalize_id_value(defect_sheet.cell(excel_row, 1).value, "") + if template_id and template_id != detail_id: + raise PredictionError( + f"Template 与缺陷计算工作表的管道编号不一致，Excel行：{excel_row}。" + ) if pd.isna(df.at[index, "结构缺陷"]): value = numeric_or_none(defect_sheet.cell(excel_row, 5).value) if value is None: @@ -310,13 +317,91 @@ def fill_defects_from_detail_sheet(path: Path, df: pd.DataFrame) -> pd.DataFrame return df -def validate_input_frame(df: pd.DataFrame) -> None: +def blank_values(series: pd.Series) -> pd.Series: + return series.isna() | series.astype("string").str.strip().eq("") + + +def source_rows(df: pd.DataFrame, mask: pd.Series) -> list[int]: + rows: list[int] = [] + for index in df.index[mask]: + if "_excel_row" in df.columns and pd.notna(df.at[index, "_excel_row"]): + rows.append(int(df.at[index, "_excel_row"])) + else: + rows.append(int(df.index.get_loc(index)) + 2) + return rows + + +def raise_for_rows(message: str, df: pd.DataFrame, mask: pd.Series) -> None: + if mask.any(): + raise PredictionError(f"{message}，Excel行：{source_rows(df, mask)}") + + +def validate_input_frame(df: pd.DataFrame, max_pipe_age: float | None = None) -> None: if df.empty: raise PredictionError("上传文件没有可预测的数据。") missing = [col for col in REQUIRED_INPUT_COLUMNS if col not in df.columns] if missing: raise PredictionError(f"缺少必要字段: {', '.join(missing)}") + blank_id = blank_values(df[ID_COLUMN]) + raise_for_rows("管道编号不能为空", df, blank_id) + normalized_ids = df[ID_COLUMN].astype("string").str.strip() + duplicate_ids = normalized_ids.duplicated(keep=False) + raise_for_rows("管道编号不能重复", df, duplicate_ids) + + for column in (PIPE_AGE_COLUMN, MATERIAL_COLUMN, "管径"): + raise_for_rows(f"{column}不能为空", df, blank_values(df[column])) + + numeric_columns = [ + PIPE_AGE_COLUMN, + "管径", + "流速", + "压力", + "温度", + "降雨量", + "结构缺陷", + "功能缺陷", + ] + numeric_values: dict[str, pd.Series] = {} + for column in numeric_columns: + if column not in df.columns: + continue + blank = blank_values(df[column]) + numeric = pd.to_numeric(df[column], errors="coerce") + finite = numeric.map(lambda value: math.isfinite(value) if pd.notna(value) else False) + invalid = ~blank & (numeric.isna() | ~finite) + raise_for_rows(f"{column}必须为有限数值", df, invalid) + numeric_values[column] = numeric + + pipe_age = numeric_values[PIPE_AGE_COLUMN] + raise_for_rows("管龄（年）不能小于0", df, pipe_age < 0) + if max_pipe_age is not None: + raise_for_rows( + f"管龄（年）不能超过模型支持上限{max_pipe_age:g}年", + df, + pipe_age > max_pipe_age, + ) + diameter = numeric_values["管径"] + raise_for_rows("管径必须大于0", df, diameter <= 0) + + if "结构缺陷" in numeric_values: + structural = numeric_values["结构缺陷"] + raise_for_rows( + "结构缺陷必须在0～5范围内", + df, + structural.notna() & ~structural.between(0, 5), + ) + else: + structural = pd.Series(pd.NA, index=df.index, dtype="Float64") + if "功能缺陷" in numeric_values: + functional = numeric_values["功能缺陷"] + raise_for_rows( + "功能缺陷必须在0～5范围内", + df, + functional.notna() & ~functional.between(0, 5), + ) + raise_for_rows("填写功能缺陷时必须同时填写结构缺陷", df, functional.notna() & structural.isna()) + def normalize_category_value( value: Any, @@ -429,7 +514,9 @@ def run_prediction(uploaded: FileStorage, user_id: int, model) -> PredictionArti try: df = read_input_file(original_path, suffix) - validate_input_frame(df) + display_age_range = getattr(model, "config", {}).get("display_age_range", []) + max_pipe_age = float(display_age_range[1]) if len(display_age_range) > 1 else None + validate_input_frame(df, max_pipe_age=max_pipe_age) x_test = prepare_model_features(df) try: predictions = model.predict(x_test, variant="defect_sensitive") diff --git a/tests/test_prediction.py b/tests/test_prediction.py index da9f641..9063f59 100644 --- a/tests/test_prediction.py +++ b/tests/test_prediction.py @@ -93,6 +93,42 @@ class PredictionHelpersTest(unittest.TestCase): validate_input_frame(df) + def test_validate_input_frame_rejects_blank_pipe_id_with_source_row(self) -> None: + df = pd.DataFrame([{column: 1 for column in INPUT_COLUMNS}]) + df[ID_COLUMN] = [" "] + df["_excel_row"] = [7] + + with self.assertRaisesRegex(PredictionError, "管道编号不能为空，Excel行：\\[7\\]"): + validate_input_frame(df) + + def test_validate_input_frame_rejects_duplicate_pipe_ids_after_trimming(self) -> None: + df = pd.DataFrame([{column: 1 for column in INPUT_COLUMNS}] * 2) + df[ID_COLUMN] = ["P001", " P001 "] + + with self.assertRaisesRegex(PredictionError, "管道编号不能重复，Excel行：\\[2, 3\\]"): + validate_input_frame(df) + + def test_validate_input_frame_rejects_non_numeric_optional_value(self) -> None: + df = pd.DataFrame([{column: 1 for column in INPUT_COLUMNS}]) + df["压力"] = ["未知"] + + with self.assertRaisesRegex(PredictionError, "压力必须为有限数值，Excel行：\\[2\\]"): + validate_input_frame(df) + + def test_validate_input_frame_rejects_functional_defect_above_five(self) -> None: + df = pd.DataFrame([{column: 1 for column in INPUT_COLUMNS}]) + df["功能缺陷"] = [6] + + with self.assertRaisesRegex(PredictionError, "功能缺陷必须在0～5范围内，Excel行：\\[2\\]"): + validate_input_frame(df) + + def test_validate_input_frame_rejects_pipe_age_above_model_limit(self) -> None: + df = pd.DataFrame([{column: 1 for column in INPUT_COLUMNS}]) + df[PIPE_AGE_COLUMN] = [74] + + with self.assertRaisesRegex(PredictionError, "管龄（年）不能超过模型支持上限73年，Excel行：\\[2\\]"): + validate_input_frame(df, max_pipe_age=73) + def test_example_workbook_contains_backend_contract_columns(self) -> None: workbook = load_workbook(Path(__file__).resolve().parents[1] / "example.xlsx", read_only=True) worksheet = workbook["Template"] @@ -139,6 +175,22 @@ class PredictionHelpersTest(unittest.TestCase): self.assertAlmostEqual(float(df.loc[0, "结构缺陷"]), 3.8) self.assertAlmostEqual(float(df.loc[0, "功能缺陷"]), 1.6) + def test_read_input_file_rejects_misaligned_detail_sheet_id(self) -> None: + with TemporaryDirectory() as temp_dir: + output_path = Path(temp_dir) / "input.xlsx" + workbook = Workbook() + template = workbook.active + template.title = "Template" + template.append(INPUT_COLUMNS) + template.append(["P001", 5, 1, 100, 1, 1.2, 0.4, 20, 800, None, None]) + detail = workbook.create_sheet("缺陷计算") + detail.append([ID_COLUMN] + ["缺陷"] * 9) + detail.append(["P002"] + [None] * 9) + workbook.save(output_path) + + with self.assertRaisesRegex(PredictionError, "工作表的管道编号不一致，Excel行：2"): + read_input_file(output_path, ".xlsx") + def test_prepare_model_features_maps_material_aliases_to_rc1_names(self) -> None: rows = [] for value in ["镀锌", "2-钢塑", 11]: