修复: 参数类型强制转换 + camelCase 防御性转换

algorithms.py

@@ -1551,6 +1551,30 @@ class NonMotorVehicleParkingAlgorithm(BaseAlgorithm):
 
 
 class AlgorithmManager:
+    # 参数类型定义，用于三级合并后的类型强制转换
+    _PARAM_TYPES = {
+        "leave_post": {
+            "confirm_on_duty_sec": int, "confirm_off_duty_sec": int,
+            "confirm_leave_sec": int, "leave_countdown_sec": int, "cooldown_sec": int,
+        },
+        "intrusion": {
+            "cooldown_seconds": int, "confirm_seconds": int,
+            "confirm_intrusion_seconds": int, "confirm_clear_seconds": int,
+        },
+        "illegal_parking": {
+            "confirm_vehicle_sec": int, "parking_countdown_sec": int,
+            "confirm_clear_sec": int, "cooldown_sec": int,
+        },
+        "vehicle_congestion": {
+            "count_threshold": int, "confirm_congestion_sec": int,
+            "confirm_clear_sec": int, "cooldown_sec": int,
+        },
+        "non_motor_vehicle_parking": {
+            "confirm_vehicle_sec": int, "parking_countdown_sec": int,
+            "confirm_clear_sec": int, "cooldown_sec": int,
+        },
+    }
+
     def __init__(self, working_hours: Optional[List[Dict]] = None):
         self.algorithms: Dict[str, Dict[str, Any]] = {}
         self.working_hours = working_hours or []
@@ -1609,6 +1633,19 @@ class AlgorithmManager:
             self._global_params = global_params_map or {}
             logger.info(f"全局参数已更新: {list(self._global_params.keys())}")
 
+    def _coerce_param_types(self, algorithm_type: str, params: dict) -> dict:
+        """强制转换参数类型，防止字符串型数字导致算法异常"""
+        type_map = self._PARAM_TYPES.get(algorithm_type, {})
+        for key, expected_type in type_map.items():
+            if key in params and params[key] is not None:
+                try:
+                    if not isinstance(params[key], expected_type):
+                        params[key] = expected_type(params[key])
+                except (ValueError, TypeError):
+                    logger.warning(f"参数类型转换失败: {algorithm_type}.{key}={params[key]!r}, 删除使用默认值")
+                    del params[key]
+        return params
+
     def get_min_alarm_duration(self, algorithm_type: str) -> Optional[int]:
         """从全局参数获取最小告警持续时间（秒）
 
@@ -1718,6 +1755,9 @@ class AlgorithmManager:
                 else:
                     params = {}
 
+                # 强制转换参数类型（防止字符串型数字）
+                params = self._coerce_param_types(algo_code, params)
+
                 if roi_id not in self.algorithms:
                     self.algorithms[roi_id] = {}
 
@@ -1918,6 +1958,9 @@ class AlgorithmManager:
             params = json.loads(params_str) if isinstance(params_str, str) else params_str
             algo_code = bind_config.get("algo_code")
 
+            # 强制转换参数类型（防止字符串型数字）
+            params = self._coerce_param_types(algo_code, params)
+
             # 获取现有算法实例
             existing_algo = self.algorithms[roi_id][key].get(algo_code)
 
@@ -2105,6 +2148,9 @@ class AlgorithmManager:
         if params:
             algo_params.update(params)
 
+        # 强制转换参数类型（防止字符串型数字）
+        algo_params = self._coerce_param_types(algorithm_type, algo_params)
+
         # 从 params 中提取告警等级（前端配置下发）
         configured_alarm_level = algo_params.get("alarm_level")
 

core/config_sync.py

@@ -15,6 +15,7 @@ import json
 import logging
 import os
 import platform
+import re
 import socket
 
 # 禁用系统代理（Clash 等代理工具会干扰 Redis TCP 长连接）
@@ -53,6 +54,12 @@ def _build_keepalive_options():
     return opts
 
 
+def _camel_to_snake(name: str) -> str:
+    """将 camelCase 转换为 snake_case"""
+    s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
+    return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()
+
+
 # ==================== Redis Key 常量 ====================
 
 # 云端 Redis Keys
@@ -648,6 +655,8 @@ class ConfigSyncManager:
             if global_params and isinstance(global_params, dict):
                 for algo_code, params_dict in global_params.items():
                     if isinstance(params_dict, dict):
+                        # 防御性转换：camelCase → snake_case
+                        params_dict = {_camel_to_snake(k): v for k, v in params_dict.items()}
                         self._db_manager.save_global_params(algo_code, params_dict)
                 logger.info(f"全局参数同步完成: {list(global_params.keys())}")