feat: validate uploaded pipeline data

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")

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]: