diff --git a/app/prediction.py b/app/prediction.py index 86483c6..e55edaf 100644 --- a/app/prediction.py +++ b/app/prediction.py @@ -58,6 +58,39 @@ FEATURES = [ ID_COLUMN = "管道编号" PIPE_AGE_COLUMN = "管龄" +MATERIAL_COLUMN = "管材" +MATERIAL_CODE_OPTIONS = [ + (1, "镀锌"), + (2, "钢塑"), + (3, "铝塑"), + (4, "PPR"), + (5, "PE"), + (6, "UPVC"), + (7, "铸铁"), + (8, "预应力"), + (9, "自应力"), + (10, "玻璃钢夹砂"), + (11, "钢管"), + (12, "钢套混凝土管"), + (13, "球墨铸铁"), + (14, "其他"), +] +MATERIAL_ALIAS_TO_CODE = { + str(code): code + for code, _ in MATERIAL_CODE_OPTIONS +} +MATERIAL_ALIAS_TO_CODE.update( + { + name.casefold(): code + for code, name in MATERIAL_CODE_OPTIONS + } +) +MATERIAL_ALIAS_TO_CODE.update( + { + f"{code}-{name}".casefold(): code + for code, name in MATERIAL_CODE_OPTIONS + } +) SUPPORTED_EXTENSIONS = {".csv", ".xls", ".xlsx"} CHINESE_FONT_PROP = None @@ -150,13 +183,21 @@ def secure_upload_name(original_filename: str, run_id: str) -> tuple[str, str]: def read_input_file(path: Path, suffix: str) -> pd.DataFrame: try: if suffix == ".csv": - return pd.read_csv(path) - return pd.read_excel(path) + return pd.read_csv(path, dtype={ID_COLUMN: "string"}) + return pd.read_excel(path, dtype={ID_COLUMN: "string"}) except Exception as exc: logging.exception("文件解析失败: %s", exc) raise PredictionError("文件解析失败，请检查编码或表格格式。") +def normalize_id_value(value: Any, fallback: str) -> str: + if pd.isna(value): + return fallback + if isinstance(value, str): + return value.strip() + return str(value) + + def validate_input_frame(df: pd.DataFrame) -> None: if df.empty: raise PredictionError("上传文件没有可预测的数据。") @@ -166,6 +207,28 @@ def validate_input_frame(df: pd.DataFrame) -> None: raise PredictionError(f"缺少必要字段: {', '.join(missing)}") +def normalize_material_code(value: Any) -> int: + if pd.isna(value): + raise PredictionError("管材不能为空。") + if isinstance(value, str): + key = value.strip().casefold() + else: + numeric_value = float(value) + if not numeric_value.is_integer(): + raise PredictionError(f"管材编码无效: {value}") + key = str(int(numeric_value)) + code = MATERIAL_ALIAS_TO_CODE.get(key) + if code is None: + raise PredictionError(f"管材编码无效: {value}") + return code + + +def prepare_model_features(df: pd.DataFrame) -> pd.DataFrame: + x_test = df[FEATURES].copy() + x_test[MATERIAL_COLUMN] = x_test[MATERIAL_COLUMN].map(normalize_material_code) + return x_test + + def grade_info(probability: float) -> tuple[str, str, str]: if probability <= 0.2: return ("I级", "管道安全风险十分严重，需立刻进行抢修或更新改造", "bg-dangerSoft text-dangerText") @@ -318,7 +381,7 @@ def run_prediction(uploaded: FileStorage, user_id: int, model) -> PredictionArti try: df = read_input_file(original_path, suffix) validate_input_frame(df) - x_test = df[FEATURES].copy() + x_test = prepare_model_features(df) try: curves = model.predict_survival_function(x_test) except Exception as exc: @@ -365,7 +428,7 @@ def render_survival_chart(df: pd.DataFrame, curves, image_path: Path) -> tuple[l for i, curve in enumerate(curves): times = [float(x) for x in list(curve.x)] probs = [float(y) for y in list(curve.y)] - pipe_id = str(df.iloc[i][ID_COLUMN]) if pd.notna(df.iloc[i][ID_COLUMN]) else f"Pipe_{i+1:03d}" + pipe_id = normalize_id_value(df.iloc[i][ID_COLUMN], f"Pipe_{i+1:03d}") pipe_age = f"{df.iloc[i][PIPE_AGE_COLUMN]} 年" if PIPE_AGE_COLUMN in df.columns and pd.notna(df.iloc[i][PIPE_AGE_COLUMN]) else "-" health_probability = interpolate_probability(times, probs, 10) remaining_life = estimate_remaining_life(times, probs) @@ -386,15 +449,13 @@ def render_survival_chart(df: pd.DataFrame, curves, image_path: Path) -> tuple[l { "管道编号": pipe_id, "管龄": pipe_age, - "健康概率": health_probability, - "预计剩余寿命": remaining_life, "健康等级": grade_label, } ) plt.step(times, probs, where="post", linewidth=2, label=pipe_id) font_kwargs = chinese_font_kwargs() - plt.xlabel("预测时间轴（年）", **font_kwargs) + plt.xlabel("管龄（年）", **font_kwargs) plt.ylabel("生存概率", **font_kwargs) plt.title("预测分析图", **font_kwargs) plt.grid(alpha=0.18) @@ -416,22 +477,58 @@ def write_prediction_workbook( summary_rows: list[dict[str, Any]], summary_sheet_rows: list[dict[str, Any]], ) -> None: - sample_data_rows: list[dict[str, Any]] = [] + chart_times: set[float] = set() + chart_series: list[tuple[str, dict[float, float]]] = [] for i, curve in enumerate(curves): times = [float(x) for x in list(curve.x)] probs = [float(y) for y in list(curve.y)] pipe_id = summary_rows[i]["pipe_id"] - for time, probability in zip(times, probs): - sample_data_rows.append( - { - "管道编号": pipe_id, - "样本序号": i + 1, - "时间（年）": time, - "生存概率": probability, - "风险概率": 1 - probability, - } - ) + chart_times.update(times) + chart_series.append((pipe_id, dict(zip(times, probs)))) + + sorted_times = sorted(chart_times) + sample_columns = ["管龄（年）", *[pipe_id for pipe_id, _ in chart_series]] + sample_data_rows = [ + [time, *[series.get(time) for _, series in chart_series]] + for time in sorted_times + ] with pd.ExcelWriter(excel_path, engine="openpyxl") as writer: - pd.DataFrame(summary_sheet_rows).to_excel(writer, sheet_name="结果摘要", index=False) - pd.DataFrame(sample_data_rows).to_excel(writer, sheet_name="样本数据", index=False) + summary_df = pd.DataFrame(summary_sheet_rows, columns=[ID_COLUMN, PIPE_AGE_COLUMN, "健康等级"]) + sample_df = pd.DataFrame(sample_data_rows, columns=sample_columns) + if ID_COLUMN in summary_df.columns: + summary_df[ID_COLUMN] = summary_df[ID_COLUMN].astype("string") + + summary_df.to_excel(writer, sheet_name="结果摘要", index=False) + sample_df.to_excel(writer, sheet_name="样本数据", index=False) + + sample_worksheet = writer.book["样本数据"] + sample_worksheet.freeze_panes = "A2" + sample_worksheet.auto_filter.ref = sample_worksheet.dimensions + sample_worksheet.column_dimensions["A"].width = 12 + for column_cells in sample_worksheet.iter_cols(min_col=2, max_col=sample_worksheet.max_column): + header_cell = column_cells[0] + sample_worksheet.column_dimensions[header_cell.column_letter].width = max(12, len(str(header_cell.value)) + 2) + for cell in column_cells[1:]: + cell.number_format = "0.0%" + + for sheet_name in ("结果摘要", "样本数据"): + worksheet = writer.book[sheet_name] + header_cells = next(worksheet.iter_rows(min_row=1, max_row=1), []) + id_column_index = None + for cell in header_cells: + if cell.value == ID_COLUMN: + id_column_index = cell.column + break + if id_column_index is None: + continue + for cell in worksheet.iter_cols( + min_col=id_column_index, + max_col=id_column_index, + min_row=2, + max_row=worksheet.max_row, + ): + for item in cell: + if item.value is not None: + item.value = str(item.value) + item.number_format = "@" diff --git a/app/routes.py b/app/routes.py index add9fde..3b4be7d 100644 --- a/app/routes.py +++ b/app/routes.py @@ -30,6 +30,7 @@ from .time_utils import format_datetime_for_timezone bp = Blueprint("main", __name__) REFERENCE_PDF_NAME = "20260630标准文本——供水管道健康状态与剩余寿命评估技术导则.pdf" +TEMPLATE_EXCEL_NAME = "管道预测数据模板.xlsx" REGISTRATION_SETTING_KEY = "allow_registration" RECORDS_PER_PAGE = 10 @@ -375,7 +376,7 @@ def download_template(): template_path = BASE_DIR / "example.xlsx" if not template_path.exists(): abort(404) - return send_file(template_path, as_attachment=True, download_name="example.xlsx") + return send_file(template_path, as_attachment=True, download_name=TEMPLATE_EXCEL_NAME) @bp.route("/reference_pdf") diff --git a/example.xlsx b/example.xlsx index da9b083..60e345c 100644 Binary files a/example.xlsx and b/example.xlsx differ diff --git a/static/js/dashboard.js b/static/js/dashboard.js index 8feffc5..12db5f4 100644 --- a/static/js/dashboard.js +++ b/static/js/dashboard.js @@ -19,7 +19,6 @@ const resultImportanceWrap = document.getElementById('resultImportanceWrap'); const resultImportanceImage = document.getElementById('resultImportanceImage'); const excelBtn = document.getElementById('excelBtn'); const resultPageBtn = document.getElementById('resultPageBtn'); -const homeStateKey = 'pipelineLifetime.homeState'; let selectedFile = null; let alertTimer = null; let alertHideTimer = null; @@ -66,24 +65,6 @@ function updateSelectedFile(file) { selectedFile = file; selectedFileName.textContent = '已选择文件：' + file.name; selectedFileName.classList.remove('hidden'); - saveHomeState({ selectedFilename: file.name }); -} - -function readHomeState() { - try { - return JSON.parse(sessionStorage.getItem(homeStateKey)) || {}; - } catch (err) { - return {}; - } -} - -function saveHomeState(nextState) { - try { - const currentState = readHomeState(); - sessionStorage.setItem(homeStateKey, JSON.stringify({ ...currentState, ...nextState })); - } catch (err) { - // Ignore storage failures; prediction still works without client-side restore. - } } function renderResult(data) { @@ -102,17 +83,6 @@ function renderResult(data) { resultContent.classList.add('flex'); } -function restoreHomeState() { - const savedState = readHomeState(); - if (savedState.selectedFilename) { - selectedFileName.textContent = '上次选择文件：' + savedState.selectedFilename + '（需重新选择）'; - selectedFileName.classList.remove('hidden'); - } - if (savedState.result) { - renderResult(savedState.result); - } -} - fileInput.addEventListener('change', () => { const file = fileInput.files[0]; if (!file) return; @@ -176,15 +146,6 @@ form.addEventListener('submit', async (e) => { return; } renderResult(data); - saveHomeState({ - selectedFilename: data.original_filename || selectedFile.name, - result: { - image_url: data.image_url, - importance_url: data.importance_url || '', - excel_url: data.excel_url, - result_url: data.result_url - } - }); showAlert('预测完成，已生成图表与电子表格报告。', 'success'); inlineResult.scrollIntoView({ behavior: 'smooth', block: 'nearest' }); } catch (err) { @@ -197,5 +158,3 @@ form.addEventListener('submit', async (e) => { submitIcon = restoredIcon; } }); - -restoreHomeState(); diff --git a/templates/base.html b/templates/base.html index 3247b48..058804e 100644 --- a/templates/base.html +++ b/templates/base.html @@ -142,15 +142,6 @@ showAppNotification(message, category === 'error' ? 'error' : 'info'); } - document.querySelectorAll('[data-clear-home-state]').forEach((form) => { - form.addEventListener('submit', () => { - try { - sessionStorage.removeItem('pipelineLifetime.homeState'); - } catch (err) { - // Ignore storage failures during logout. - } - }); - }); })(); {% block scripts %}{% endblock %} diff --git a/templates/result.html b/templates/result.html index c9a1f71..08ca52a 100644 --- a/templates/result.html +++ b/templates/result.html @@ -51,12 +51,13 @@
-

样本预览

+

结果展示

- + + @@ -64,6 +65,7 @@ {% for item in result.summary_rows %} + {% endfor %} diff --git a/tests/test_auth_registration.py b/tests/test_auth_registration.py index 9bf55f5..9d516d5 100644 --- a/tests/test_auth_registration.py +++ b/tests/test_auth_registration.py @@ -130,6 +130,19 @@ class RegistrationRoutesTest(unittest.TestCase): self.assertIn('name="captcha"', html) self.assertIn(captcha, html) + def test_template_download_uses_chinese_filename(self) -> None: + with TemporaryDirectory() as temp_dir: + app = self.create_test_app(temp_dir, allow_registration=True) + + response = app.test_client().get("/download_template") + + self.assertEqual(response.status_code, 200) + self.assertIn("attachment", response.headers["Content-Disposition"]) + self.assertIn( + "filename*=UTF-8''%E7%AE%A1%E9%81%93%E9%A2%84%E6%B5%8B%E6%95%B0%E6%8D%AE%E6%A8%A1%E6%9D%BF.xlsx", + response.headers["Content-Disposition"], + ) + def test_register_post_does_not_create_user_when_registration_is_closed(self) -> None: with TemporaryDirectory() as temp_dir: app = self.create_test_app(temp_dir, allow_registration=False) diff --git a/tests/test_prediction.py b/tests/test_prediction.py index 8b51300..242cef5 100644 --- a/tests/test_prediction.py +++ b/tests/test_prediction.py @@ -3,16 +3,22 @@ from __future__ import annotations import unittest from pathlib import Path from tempfile import TemporaryDirectory +from unittest.mock import patch import pandas as pd +from openpyxl import load_workbook from app.prediction import ( FEATURES, ID_COLUMN, + PIPE_AGE_COLUMN, PredictionError, estimate_remaining_life, grade_info, interpolate_probability, + prepare_model_features, + read_input_file, + render_survival_chart, secure_upload_name, validate_input_frame, write_prediction_workbook, @@ -53,6 +59,37 @@ class PredictionHelpersTest(unittest.TestCase): self.assertIn("缺少必要字段", ctx.exception.message) + def test_read_input_file_preserves_text_pipe_ids(self) -> None: + with TemporaryDirectory() as temp_dir: + output_path = Path(temp_dir) / "input.xlsx" + pd.DataFrame({ID_COLUMN: ["00123"], FEATURES[0]: [1]}).to_excel(output_path, index=False) + + df = read_input_file(output_path, ".xlsx") + + self.assertEqual(df[ID_COLUMN].tolist(), ["00123"]) + + def test_prepare_model_features_maps_material_aliases_to_codes(self) -> None: + rows = [] + for value in ["镀锌", "2-钢塑", 13]: + row = {feature: 1 for feature in FEATURES} + row["管材"] = value + rows.append(row) + df = pd.DataFrame(rows) + + x_test = prepare_model_features(df) + + self.assertEqual(x_test["管材"].tolist(), [1, 2, 13]) + + def test_prepare_model_features_rejects_invalid_material_alias(self) -> None: + row = {feature: 1 for feature in FEATURES} + row["管材"] = "未知管材" + df = pd.DataFrame([row]) + + with self.assertRaises(PredictionError) as ctx: + prepare_model_features(df) + + self.assertIn("管材编码无效", ctx.exception.message) + def test_prediction_workbook_keeps_sample_data_in_one_sheet(self) -> None: curves = [ DummyCurve([1, 2], [0.9, 0.7]), @@ -60,8 +97,8 @@ class PredictionHelpersTest(unittest.TestCase): ] summary_rows = [{"pipe_id": "P001"}, {"pipe_id": "P002"}] summary_sheet_rows = [ - {"管道编号": "P001", "健康概率": 0.7}, - {"管道编号": "P002", "健康概率": 0.6}, + {"管道编号": "P001", "管龄": "10 年", "健康概率": 0.7, "预计剩余寿命": 2.0, "健康等级": "IV级"}, + {"管道编号": "P002", "管龄": "12 年", "健康概率": 0.6, "预计剩余寿命": 1.5, "健康等级": "III级"}, ] with TemporaryDirectory() as temp_dir: @@ -71,10 +108,51 @@ class PredictionHelpersTest(unittest.TestCase): workbook = pd.ExcelFile(output_path) self.assertEqual(workbook.sheet_names, ["结果摘要", "样本数据"]) + summary_data = pd.read_excel(output_path, sheet_name="结果摘要") + self.assertEqual(summary_data.columns.tolist(), [ID_COLUMN, PIPE_AGE_COLUMN, "健康等级"]) + sample_data = pd.read_excel(output_path, sheet_name="样本数据") - self.assertEqual(len(sample_data), 4) - self.assertEqual(sample_data["管道编号"].tolist(), ["P001", "P001", "P002", "P002"]) - self.assertIn("风险概率", sample_data.columns) + self.assertEqual(sample_data.columns.tolist(), ["管龄（年）", "P001", "P002"]) + self.assertEqual(len(sample_data), 2) + self.assertEqual(sample_data["管龄（年）"].tolist(), [1, 2]) + self.assertEqual(sample_data["P001"].tolist(), [0.9, 0.7]) + self.assertEqual(sample_data["P002"].tolist(), [0.8, 0.6]) + + openpyxl_workbook = load_workbook(output_path) + sample_worksheet = openpyxl_workbook["样本数据"] + self.assertEqual(sample_worksheet.freeze_panes, "A2") + self.assertEqual(sample_worksheet.auto_filter.ref, "A1:C3") + self.assertEqual(sample_worksheet["B2"].number_format, "0.0%") + + def test_prediction_workbook_writes_pipe_ids_as_excel_text(self) -> None: + curves = [DummyCurve([1], [0.9])] + summary_rows = [{"pipe_id": "00123"}] + summary_sheet_rows = [{ID_COLUMN: "00123", PIPE_AGE_COLUMN: "10 年", "健康等级": "V级"}] + + with TemporaryDirectory() as temp_dir: + output_path = Path(temp_dir) / "prediction.xlsx" + write_prediction_workbook(output_path, curves, summary_rows, summary_sheet_rows) + + workbook = load_workbook(output_path) + summary_worksheet = workbook["结果摘要"] + self.assertEqual(summary_worksheet["A2"].value, "00123") + self.assertEqual(summary_worksheet["A2"].data_type, "s") + self.assertEqual(summary_worksheet["A2"].number_format, "@") + + sample_worksheet = workbook["样本数据"] + self.assertEqual(sample_worksheet["B1"].value, "00123") + self.assertEqual(sample_worksheet["B1"].data_type, "s") + + def test_survival_chart_uses_pipe_age_x_axis_label(self) -> None: + df = pd.DataFrame({ID_COLUMN: ["P001"], PIPE_AGE_COLUMN: [12]}) + + with TemporaryDirectory() as temp_dir: + output_path = Path(temp_dir) / "chart.png" + with patch("app.prediction.plt.xlabel") as xlabel: + render_survival_chart(df, [DummyCurve([1, 2], [0.9, 0.7])], output_path) + + xlabel.assert_called_once() + self.assertEqual(xlabel.call_args.args[0], "管龄（年）") if __name__ == "__main__":
管道编号编号管龄 等级
{{ item.pipe_id }}{{ item.pipe_age }} {{ item.grade_label }}