From 85787fb57289317969ca741acbe2f4ac3f0b9d8d Mon Sep 17 00:00:00 2001 From: shunli Date: Wed, 21 Jan 2026 16:01:40 +0800 Subject: [PATCH] fix(web): solve p0 mapping, nested validation, and add health checks --- main.py | 29 +++++++++++++--- src/bridges.py | 5 ++- src/env_detect.py | 50 ++++++++++++++++++++++++--- src/scheduler.py | 2 +- src/security.py | 20 +++++++++++ src/validator.py | 63 +++++++++++++++++++++++++++------ tests/test_core.py | 86 +++++++++++++++++++++++++++++++++++++++++++++- 7 files changed, 232 insertions(+), 23 deletions(-) diff --git a/main.py b/main.py index 1e912ca..dc7c23a 100644 --- a/main.py +++ b/main.py @@ -178,10 +178,11 @@ def show_algo_library(): # 环境状态概览 with st.expander("🖥️ 当前环境能力", expanded=False): - col1, col2, col3 = st.columns(3) + col1, col2, col3, col4 = st.columns(4) col1.metric("Python", "✅ 可用") col2.metric("MATLAB/Octave", "✅ 可用" if "matlab" in available_types else "❌ 不可用") col3.metric("C++ 运行时", "✅ 可用" if "cpp" in available_types else "❌ 不可用") + col4.metric("Web 桥接", "✅ 可用" if "web" in available_types else "❌ 不可用") st.json(env) st.subheader("已注册算法") @@ -208,6 +209,16 @@ def show_algo_library(): st.write(f"**入口**: `{algo.get('lib_path', 'N/A')}::{algo.get('function', 'N/A')}`") elif algo_type == "matlab": st.write(f"**入口**: `{algo.get('m_path', 'N/A')}::{algo.get('function', 'N/A')}`") + elif algo_type == "web": + endpoint = algo.get('endpoint', 'N/A') + st.write(f"**端点**: `{endpoint}`") + if st.button(f"🌐 测试连接: {algo['name']}", key=f"ping_{algo['name']}"): + from src.env_detect import check_endpoint_active + active, msg = check_endpoint_active(endpoint) + if active: + st.success(f"✅ 联通正常: {msg}") + else: + st.error(f"❌ 无法访问: {msg}") else: st.write(f"**入口**: `{algo.get('function', 'N/A')}`") @@ -395,10 +406,18 @@ def show_scheduling(): param: (spec or {}).get("field") or param for param, spec in (algo.get("inputs") or {}).items() } - out_map = { - str((spec or {}).get("index", 0)): field - for field, spec in (algo.get("outputs") or {}).items() - } + if algo_type == "web": + # Web 类型使用 Key 映射 (如 transmission_loss -> vis_data.tl_result_path) + out_map = { + field_name: (spec or {}).get("field") or field_name + for field_name, spec in (algo.get("outputs") or {}).items() + } + else: + # 传统类型使用 Index 映射 (如 0 -> clean_data) + out_map = { + str((spec or {}).get("index", 0)): field + for field, spec in (algo.get("outputs") or {}).items() + } preprocessing = [] from src.bridges import PythonBridge, CppBridge diff --git a/src/bridges.py b/src/bridges.py index 58a3556..2af1fc6 100644 --- a/src/bridges.py +++ b/src/bridges.py @@ -468,7 +468,7 @@ class WebBridge(AlgorithmBridge): from src.resource_slots import slot_context from src.security import ( validate_endpoint_host, sanitize_output_params, - generate_run_id, version_tracker + generate_run_id, version_tracker, validate_critical_enums ) try: @@ -484,6 +484,9 @@ class WebBridge(AlgorithmBridge): if value is not None: request_params[http_field] = value + # 3. 业务逻辑检查：关键参数枚举校验 + validate_critical_enums(request_params) + # 3. 清洗输出参数（生成唯一 run_id） run_id = generate_run_id() if self.sanitize_output: diff --git a/src/env_detect.py b/src/env_detect.py index d4e8ebc..493819a 100644 --- a/src/env_detect.py +++ b/src/env_detect.py @@ -36,6 +36,15 @@ def _check_ld_paths() -> List[str]: return paths[:10] # 限制输出长度 +def _cpp_runtime_available() -> bool: + """检测 C++ 运行时能力（不依赖编译器）""" + try: + import ctypes # noqa: F401 + except Exception: + return False + return platform.system().lower() in {"linux", "darwin", "windows"} + + def collect_env_info() -> Dict[str, Any]: """收集完整的环境信息""" compilers = { @@ -45,6 +54,8 @@ def collect_env_info() -> Dict[str, Any]: "clang++": shutil.which("clang++"), "cl": shutil.which("cl"), # MSVC } + compile_available = any(compilers.values()) + runtime_available = _cpp_runtime_available() return { "python": { @@ -64,11 +75,15 @@ def collect_env_info() -> Dict[str, Any]: }, "cpp": { "compilers": {k: v for k, v in compilers.items() if v}, - "any_available": any(compilers.values()), + "compile_available": compile_available, + "runtime_available": runtime_available, + # Backward-compatible alias (historically used as "is cpp usable") + "any_available": runtime_available, }, "ld_paths": _check_ld_paths(), "scipy_available": _has_module("scipy"), "numpy_available": _has_module("numpy"), + "web_available": _has_module("requests"), } @@ -80,12 +95,33 @@ def get_available_algo_types() -> List[str]: if env["matlab"]["engine_available"] or env["matlab"]["oct2py_available"] or env["matlab"]["octave_cli"]: types.append("matlab") - if env["cpp"]["any_available"]: + # C++ 运行时可用即视为可执行（编译器仅影响“从源码编译”能力） + if env["cpp"]["runtime_available"]: types.append("cpp") + + if env["web_available"]: + types.append("web") return types +def check_endpoint_active(url: str, timeout: int = 5) -> Tuple[bool, str]: + """检查远程端点是否存活""" + try: + import requests + # 对根路径发起 GET 请求进行探活 + from urllib.parse import urljoin, urlparse + parsed = urlparse(url) + base_url = f"{parsed.scheme}://{parsed.netloc}" + + resp = requests.get(base_url, timeout=timeout) + if resp.status_code < 500: + return True, f"Active (HTTP {resp.status_code})" + return False, f"Server Error (HTTP {resp.status_code})" + except Exception as e: + return False, f"Disconnected: {str(e)}" + + def get_actionable_suggestions() -> List[Dict[str, str]]: """针对故障环境提供具体的修复指令建议""" env = collect_env_info() @@ -106,10 +142,16 @@ def get_actionable_suggestions() -> List[Dict[str, str]]: }) # C++ 建议 - if not env["cpp"]["any_available"]: + if not env["cpp"]["runtime_available"]: + suggestions.append({ + "target": "C++ Runtime", + "reason": "当前 Python 环境缺少 ctypes/动态链接支持", + "fix": "请使用包含 ctypes 的标准 Python 发行版，或联系交付方提供兼容运行环境。", + }) + elif not env["cpp"]["compile_available"]: suggestions.append({ "target": "C++ Compiler", - "reason": "未检测到 gcc/g++/clang 等编译器", + "reason": "未检测到 gcc/g++/clang 等编译器（仅影响“从源码编译”能力）", "fix": "建议安装基础构建工具: `sudo apt install build-essential`" }) diff --git a/src/scheduler.py b/src/scheduler.py index 8e55043..c4437b9 100644 --- a/src/scheduler.py +++ b/src/scheduler.py @@ -86,6 +86,6 @@ class CoreScheduler: # 按声明的输出类型填充哑元数据，让后续步骤能做类型/维度校验 spec = output_specs.get(attr) or {} expected_type = spec.get("type") or "ndarray" - setattr(temp_data, attr, _dummy_value(expected_type)) + FlowValidator._set_nested_field(temp_data, attr, _dummy_value(expected_type)) return issues diff --git a/src/security.py b/src/security.py index f2480f9..2e803af 100644 --- a/src/security.py +++ b/src/security.py @@ -87,6 +87,26 @@ def sanitize_filename(filename: str) -> str: return re.sub(r'[^\w\-.]', '_', filename) +def validate_critical_enums(params: Dict) -> None: + """ + 对关键参数进行枚举校验，防止远程服务静默回落 (Silent Fallback) + """ + # 建模时间段白名单 + ALLOWED_TIME_TYPES = { + "Annual", "January", "February", "March", "April", "May", "June", + "July", "August", "September", "October", "November", "December", + "Winter", "Spring", "Summer", "Autumn" + } + # 建模方法白名单 + ALLOWED_MODEL_TYPES = {"bellhop", "kraken", "ram"} + + if "timeType" in params and params["timeType"] not in ALLOWED_TIME_TYPES: + raise ValueError(f"不合法的时间段类型: {params['timeType']}") + + if "modelType" in params and params["modelType"] not in ALLOWED_MODEL_TYPES: + raise ValueError(f"不支持的声学模型类型: {params['modelType']}") + + class ServiceVersionTracker: """服务版本追踪器，用于审计与回滚""" diff --git a/src/validator.py b/src/validator.py index 35a294b..d76eef5 100644 --- a/src/validator.py +++ b/src/validator.py @@ -58,17 +58,57 @@ class FlowValidator: return errors + @staticmethod + def _get_nested_field(obj: Any, path: str) -> Any: + """支持 'sea_env.modeling.freq_hz' 这种嵌套取值""" + for key in path.split('.'): + if obj is None: + return None + if isinstance(obj, dict): + obj = obj.get(key) + else: + obj = getattr(obj, key, None) + return obj + + @staticmethod + def _set_nested_field(obj: Any, path: str, value: Any) -> None: + """支持 'vis_data.tl_result_path' 这种嵌套写入""" + keys = path.split('.') + for key in keys[:-1]: + if isinstance(obj, dict): + if key not in obj: + obj[key] = {} + obj = obj[key] + else: + if not hasattr(obj, key): + setattr(obj, key, {}) + next_obj = getattr(obj, key) + if not isinstance(next_obj, dict): + setattr(obj, key, {}) + next_obj = getattr(obj, key) + obj = next_obj + + final_key = keys[-1] + if isinstance(obj, dict): + obj[final_key] = value + else: + setattr(obj, final_key, value) + @staticmethod def check_field_existence(data: WaterSoundData, field_name: str) -> Tuple[bool, str, str]: - """1. 字段存在性校验""" - if not hasattr(data, field_name) or getattr(data, field_name) is None: - return False, f"缺少必要字段: {field_name}", f"建议：请在流程前置步骤中增加输出 '{field_name}' 的算法，或手动配置输入。" + """1. 字段存在性校验 (支持嵌套路径)""" + val = FlowValidator._get_nested_field(data, field_name) + if val is None: + return False, f"缺少必要字段/属性: {field_name}", f"建议：请在流程前置步骤中增加输出 '{field_name}' 的算法，或手动配置输入。" return True, "", "" @staticmethod def check_data_type(data: WaterSoundData, field_name: str, expected_type: str) -> Tuple[bool, str, str]: - """2. 类型校验""" - val = getattr(data, field_name) + """2. 类型校验 (支持嵌套路径)""" + val = FlowValidator._get_nested_field(data, field_name) + if val is None: + return True, "", "" # 存在性由上一个 check 负责 + actual_type = type(val).__name__ type_map = { @@ -85,9 +125,9 @@ class FlowValidator: @staticmethod def check_dimensions(data: WaterSoundData, field_name: str, expected_dim: List[Optional[int]]) -> Tuple[bool, str, str]: - """3. 维度校验""" - val = getattr(data, field_name) - if not isinstance(val, np.ndarray): + """3. 维度校验 (支持嵌套路径)""" + val = FlowValidator._get_nested_field(data, field_name) + if val is None or not isinstance(val, np.ndarray): return True, "", "" # 非数组不校验维度 actual_shape = val.shape @@ -115,12 +155,13 @@ class FlowValidator: return True, "", "" if algo_type == "cpp": - if env_info.get("cpp", {}).get("any_available"): + cpp_info = env_info.get("cpp", {}) + if cpp_info.get("runtime_available") or cpp_info.get("any_available"): return True, "", "" return ( False, - "当前环境未检测到可用的 C/C++ 工具链", - "建议：安装 MinGW/GCC/MSVC（或在交付包中提供已编译的动态库并更新检测策略）。", + "当前环境未检测到可用的 C++ 动态库运行时能力", + "建议：使用包含 ctypes 的标准 Python，并确保系统支持动态链接（Linux: glibc/dlopen）。若需从源码编译，再安装 build-essential/clang。", ) if algo_type == "matlab": diff --git a/tests/test_core.py b/tests/test_core.py index 7009ab3..b26eec6 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -84,6 +84,30 @@ def test_validation(): assert issues[0]['algo'] == "Clean" print("Scheduler validation flow passed.") +def test_nested_validation(): + from src.validator import FlowValidator + from src.models import WaterSoundData + + wsd = WaterSoundData() + wsd.sea_env = {"modeling": {"freq_hz": 100}} + + # 1. 存在性校验 + ok, msg, _ = FlowValidator.check_field_existence(wsd, "sea_env.modeling.freq_hz") + assert ok, f"Nested existence failed: {msg}" + + ok, _, _ = FlowValidator.check_field_existence(wsd, "sea_env.modeling.non_existent") + assert not ok, "Nested non-existence should fail" + + # 2. 类型校验 + ok, msg, _ = FlowValidator.check_data_type(wsd, "sea_env.modeling.freq_hz", "int") + assert ok, f"Nested type check failed: {msg}" + + # 3. 嵌套写入仿真 + FlowValidator._set_nested_field(wsd, "vis_data.tl_matrix.values", np.array([1, 2, 3])) + assert wsd.vis_data["tl_matrix"]["values"][1] == 2 + + print("Nested validation test passed.") + def test_sea_trial_pipeline(): print("Starting sea-trial pipeline test...") from src.sample_algorithms import dc_removal_and_bandpass, quality_check, snr_estimation @@ -113,6 +137,7 @@ def test_sea_trial_pipeline(): def test_cpp_bridge(): print("Starting C++ bridge test...") from src.bridges import CppBridge + from src.integrity import sha256_file lib_path = "src/sample_algos_cpp/simple_ops.so" if not os.path.exists(lib_path): @@ -127,7 +152,8 @@ def test_cpp_bridge(): lib_path, "multiply_by_factor_cpp", {"in_data": "clean_data"}, - {"0": "final_result"} + {"0": "final_result"}, + expected_sha256=sha256_file(lib_path), ) scheduler = CoreScheduler() @@ -143,6 +169,60 @@ def test_cpp_bridge(): assert np.allclose(result.final_result, raw * 1.5) print("C++ bridge test passed.") +def test_cpp_bridge_isolated(): + print("Starting C++ bridge isolated test...") + from src.bridges import CppBridge + from src.integrity import sha256_file + + lib_path = "src/sample_algos_cpp/simple_ops.so" + if not os.path.exists(lib_path): + print("C++ library not found, skipping isolated CPP test.") + return + + raw = np.array([1.0, 2.0, 3.0, 4.0, 5.0], dtype=np.float64) + wsd = WaterSoundData(clean_data=raw) + + cpp_bridge = CppBridge( + "GainCPP(Isolated)", + lib_path, + "multiply_by_factor_cpp", + {"in_data": "clean_data"}, + {"0": "final_result"}, + expected_sha256=sha256_file(lib_path), + isolate=True, + ) + + scheduler = CoreScheduler() + scheduler.set_flow([cpp_bridge]) + result = scheduler.execute(wsd) + + if result.status != "finished": + print(f"C++ Bridge Isolated Execution Failed: {result.error_info}") + for l in result.log: + print(f" Log: {l}") + + assert result.status == "finished" + assert np.allclose(result.final_result, raw * 1.5) + print("C++ bridge isolated test passed.") + +def test_python_loader_sha256(): + print("Starting Python loader SHA256 test...") + from src.algo_loader import load_algo_func + from src.integrity import sha256_file + + module_file = "src/sample_algorithms.py" + good_hash = sha256_file(module_file) + func = load_algo_func("src.sample_algorithms", "snr_estimation", expected_sha256=good_hash) + assert callable(func) + + bad_hash = "0" * 64 + try: + load_algo_func("src.sample_algorithms", "snr_estimation", expected_sha256=bad_hash) + assert False, "expected SHA256 mismatch to raise" + except Exception as e: + assert "SHA256" in str(e) + print("Python loader SHA256 test passed.") + def test_matlab_bridge_fallback(): print("Starting MATLAB bridge fallback test...") from src.bridges import MatlabBridge @@ -171,4 +251,8 @@ if __name__ == "__main__": test_validation() test_sea_trial_pipeline() test_cpp_bridge() + test_cpp_bridge_isolated() + test_python_loader_sha256() test_matlab_bridge_fallback() + test_nested_validation() + print("\nAll Tests Passed Successfully! 🚀")