feat: 实现配置热更新机制

数据库扩展:
- roi_configs 新增算法参数字段（working_hours, confirm_on_duty_sec等）
- 新增 config_update_log 表记录配置变更日志

Redis缓存:
- ROI/摄像头配置缓存到 Redis（TTL 1小时）
- sync_all_to_redis() 批量同步配置
- notify_config_change() 发布配置变更通知

热更新:
- AlgorithmManager 订阅 Redis config_update 频道
- load_from_redis() 从 Redis 加载算法参数
- reload_algorithm() 热更新单个算法
- reload_all_algorithms() 重新加载所有算法

配置模型:
- ROIInfo 添加算法参数字段

algorithms.py

@@ -12,56 +12,73 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 
 class LeavePostAlgorithm:
+    STATE_WAITING = "WAITING"
     STATE_ON_DUTY = "ON_DUTY"
-    STATE_OFF_DUTY_COUNTDOWN = "OFF_DUTY_COUNTDOWN"
+    STATE_LEAVING = "LEAVING"
+    STATE_OFF_DUTY = "OFF_DUTY"
     STATE_NON_WORK_TIME = "NON_WORK_TIME"
-    STATE_INIT = "INIT"
 
     def __init__(
         self,
-        threshold_sec: int = 300,
-        confirm_sec: int = 10,
-        return_sec: int = 30,
+        confirm_on_duty_sec: int = 10,
+        confirm_leave_sec: int = 10,
+        cooldown_sec: int = 300,
         working_hours: Optional[List[Dict]] = None,
+        target_class: Optional[str] = "person",
     ):
-        self.threshold_sec = threshold_sec
-        self.confirm_sec = confirm_sec
-        self.return_sec = return_sec
+        self.confirm_on_duty_sec = confirm_on_duty_sec
+        self.confirm_leave_sec = confirm_leave_sec
+        self.cooldown_sec = cooldown_sec
         self.working_hours = working_hours or []
+        self.target_class = target_class
 
         self.alert_cooldowns: Dict[str, datetime] = {}
-        self.cooldown_seconds = 300
 
-        self.state: str = self.STATE_INIT
+        self.state: str = self.STATE_WAITING
         self.state_start_time: Optional[datetime] = None
-        self.on_duty_window = deque()
-        self.alarm_sent: bool = False
-        self.last_person_seen_time: Optional[datetime] = None
-        self.last_detection_time: Optional[datetime] = None
-        self.init_start_time: Optional[datetime] = None
+        self.detection_history: deque = deque()
 
-    def is_in_working_hours(self, dt: Optional[datetime] = None) -> bool:
+        self.alarm_sent: bool = False
+        self.last_person_time: Optional[datetime] = None
+
+    def _is_in_working_hours(self, dt: Optional[datetime] = None) -> bool:
         if not self.working_hours:
             return True
-
         dt = dt or datetime.now()
         current_minutes = dt.hour * 60 + dt.minute
-
         for period in self.working_hours:
             start_minutes = period["start"][0] * 60 + period["start"][1]
             end_minutes = period["end"][0] * 60 + period["end"][1]
             if start_minutes <= current_minutes < end_minutes:
                 return True
-
         return False
 
-    def check_detection_in_roi(self, detection: Dict, roi_id: str) -> bool:
+    def _check_detection_in_roi(self, detection: Dict, roi_id: str) -> bool:
         matched_rois = detection.get("matched_rois", [])
         for roi in matched_rois:
             if roi.get("roi_id") == roi_id:
                 return True
         return False
 
+    def _check_target_class(self, detection: Dict, target_class: Optional[str]) -> bool:
+        if not target_class:
+            return True
+        return detection.get("class") == target_class
+
+    def _get_detection_window(self, current_time: datetime) -> List[bool]:
+        detections = []
+        while self.detection_history and (current_time - self.detection_history[0][0]).total_seconds() > max(self.confirm_on_duty_sec, self.confirm_leave_sec):
+            self.detection_history.popleft()
+        for _, has_person in self.detection_history:
+            detections.append(has_person)
+        return detections
+
+    def _get_latest_bbox(self, tracks: List[Dict], roi_id: str) -> List[float]:
+        for det in tracks:
+            if self._check_detection_in_roi(det, roi_id):
+                return det.get("bbox", [])
+        return []
+
     def process(
         self,
         roi_id: str,
@@ -71,142 +88,123 @@ class LeavePostAlgorithm:
     ) -> List[Dict]:
         current_time = current_time or datetime.now()
 
-        roi_has_person = False
-        for det in tracks:
-            if self.check_detection_in_roi(det, roi_id):
-                roi_has_person = True
-                break
-
-        in_work = self.is_in_working_hours(current_time)
+        in_work = self._is_in_working_hours(current_time)
         alerts = []
 
         if not in_work:
             self.state = self.STATE_NON_WORK_TIME
-            self.last_person_seen_time = None
-            self.last_detection_time = None
-            self.on_duty_window.clear()
+            self.detection_history.clear()
             self.alarm_sent = False
-            self.init_start_time = None
-        else:
-            if self.state == self.STATE_NON_WORK_TIME:
-                self.state = self.STATE_INIT
-                self.init_start_time = current_time
-                self.on_duty_window.clear()
-                self.alarm_sent = False
+            return []
 
-            if self.state == self.STATE_INIT:
-                if roi_has_person:
-                    self.state = self.STATE_ON_DUTY
-                    self.state_start_time = current_time
-                    self.on_duty_window.clear()
-                    self.on_duty_window.append((current_time, True))
-                    self.last_person_seen_time = current_time
-                    self.last_detection_time = current_time
-                    self.init_start_time = None
-                else:
-                    if self.init_start_time is None:
-                        self.init_start_time = current_time
+        if self.state == self.STATE_NON_WORK_TIME:
+            self.state = self.STATE_WAITING
+            self.detection_history.clear()
+            self.alarm_sent = False
 
-                    elapsed_since_init = (current_time - self.init_start_time).total_seconds()
-                    if elapsed_since_init >= self.threshold_sec:
-                        self.state = self.STATE_OFF_DUTY_COUNTDOWN
-                        self.state_start_time = current_time
-                        self.alarm_sent = False
+        roi_has_person = False
+        for det in tracks:
+            if self._check_detection_in_roi(det, roi_id) and self._check_target_class(det, self.target_class):
+                roi_has_person = True
+                break
 
-            elif self.state == self.STATE_ON_DUTY:
-                if roi_has_person:
-                    self.last_person_seen_time = current_time
-                    self.last_detection_time = current_time
+        if self.state == self.STATE_WAITING:
+            if roi_has_person:
+                self.state = self.STATE_ON_DUTY
+                self.state_start_time = current_time
+                self.detection_history.clear()
+                self.detection_history.append((current_time, True))
+            else:
+                pass
 
-                    self.on_duty_window.append((current_time, True))
-                    while self.on_duty_window and (current_time - self.on_duty_window[0][0]).total_seconds() > self.confirm_sec:
-                        self.on_duty_window.popleft()
-                else:
-                    self.on_duty_window.append((current_time, False))
-                    while self.on_duty_window and (current_time - self.on_duty_window[0][0]).total_seconds() > self.confirm_sec:
-                        self.on_duty_window.popleft()
+        elif self.state == self.STATE_ON_DUTY:
+            self.detection_history.append((current_time, roi_has_person))
+            if not roi_has_person:
+                self.state = self.STATE_LEAVING
+                self.state_start_time = current_time
 
-                    hit_ratio = sum(1 for t, detected in self.on_duty_window if detected) / max(len(self.on_duty_window), 1)
+        elif self.state == self.STATE_LEAVING:
+            self.detection_history.append((current_time, roi_has_person))
+            elapsed = (current_time - self.state_start_time).total_seconds()
 
-                    if hit_ratio == 0:
-                        self.state = self.STATE_OFF_DUTY_COUNTDOWN
-                        self.state_start_time = current_time
-                        self.alarm_sent = False
+            if roi_has_person:
+                self.state = self.STATE_ON_DUTY
+                self.state_start_time = current_time
+            elif elapsed >= self.confirm_leave_sec:
+                self.state = self.STATE_OFF_DUTY
+                self.state_start_time = current_time
 
-            elif self.state == self.STATE_OFF_DUTY_COUNTDOWN:
-                elapsed = (current_time - self.state_start_time).total_seconds()
+        elif self.state == self.STATE_OFF_DUTY:
+            elapsed = (current_time - self.state_start_time).total_seconds()
 
-                if roi_has_person:
-                    self.state = self.STATE_ON_DUTY
-                    self.state_start_time = current_time
-                    self.on_duty_window.clear()
-                    self.on_duty_window.append((current_time, True))
-                    self.last_person_seen_time = current_time
-                    self.alarm_sent = False
-                elif elapsed >= self.threshold_sec:
-                    if not self.alarm_sent:
-                        cooldown_key = f"{roi_id}"
-                        if cooldown_key not in self.alert_cooldowns or (
-                            current_time - self.alert_cooldowns[cooldown_key]
-                        ).total_seconds() > self.cooldown_seconds:
-                            bbox = self.get_latest_bbox_in_roi(tracks, roi_id)
-                            elapsed_minutes = int(elapsed / 60)
-                            alerts.append({
-                                "track_id": roi_id,
-                                "bbox": bbox,
-                                "off_duty_duration": elapsed,
-                                "alert_type": "leave_post",
-                                "message": f"离岗超过 {elapsed_minutes} 分钟",
-                            })
-                            self.alarm_sent = True
-                            self.alert_cooldowns[cooldown_key] = current_time
+            if roi_has_person:
+                self.state = self.STATE_ON_DUTY
+                self.state_start_time = current_time
+            elif elapsed >= self.confirm_leave_sec:
+                cooldown_key = f"{camera_id}_{roi_id}"
+                now = datetime.now()
+                if cooldown_key not in self.alert_cooldowns or (now - self.alert_cooldowns[cooldown_key]).total_seconds() > self.cooldown_sec:
+                    bbox = self._get_latest_bbox(tracks, roi_id)
+                    elapsed_minutes = int(elapsed / 60)
+                    alerts.append({
+                        "track_id": roi_id,
+                        "camera_id": camera_id,
+                        "bbox": bbox,
+                        "off_duty_duration": elapsed,
+                        "alert_type": "leave_post",
+                        "message": f"离岗超过 {elapsed_minutes} 分钟",
+                    })
+                    self.alert_cooldowns[cooldown_key] = now
 
         return alerts
 
-    def get_latest_bbox_in_roi(self, tracks: List[Dict], roi_id: str) -> List[float]:
-        for det in tracks:
-            if self.check_detection_in_roi(det, roi_id):
-                return det.get("bbox", [])
-        return []
-
     def reset(self):
-        self.state = self.STATE_INIT
+        self.state = self.STATE_WAITING
         self.state_start_time = None
-        self.on_duty_window.clear()
+        self.detection_history.clear()
         self.alarm_sent = False
-        self.last_person_seen_time = None
-        self.last_detection_time = None
-        self.init_start_time = None
+        self.last_person_time = None
         self.alert_cooldowns.clear()
 
-    def get_state(self, track_id: str) -> Optional[Dict[str, Any]]:
+    def get_state(self, roi_id: str) -> Dict[str, Any]:
         return {
             "state": self.state,
             "alarm_sent": self.alarm_sent,
-            "last_person_seen_time": self.last_person_seen_time,
+            "last_person_time": self.last_person_time,
         }
 
 
 class IntrusionAlgorithm:
-    def __init__(self, cooldown_seconds: int = 300):
+    def __init__(
+        self,
+        cooldown_seconds: int = 120,
+        confirm_seconds: int = 5,
+        target_class: Optional[str] = None,
+    ):
         self.cooldown_seconds = cooldown_seconds
-        self.last_alert_time: Dict[str, float] = {}
-        self.alert_triggered: Dict[str, bool] = {}
+        self.confirm_seconds = confirm_seconds
+        self.target_class = target_class
 
-    def is_roi_has_person(self, tracks: List[Dict], roi_id: str) -> bool:
-        for det in tracks:
-            matched_rois = det.get("matched_rois", [])
-            for roi in matched_rois:
-                if roi.get("roi_id") == roi_id:
-                    return True
+        self.last_alert_time: Dict[str, datetime] = {}
+        self.alert_triggered: Dict[str, bool] = {}
+        self.detection_start: Dict[str, Optional[datetime]] = {}
+
+    def _check_detection_in_roi(self, detection: Dict, roi_id: str) -> bool:
+        matched_rois = detection.get("matched_rois", [])
+        for roi in matched_rois:
+            if roi.get("roi_id") == roi_id:
+                return True
         return False
 
-    def get_latest_bbox_in_roi(self, tracks: List[Dict], roi_id: str) -> List[float]:
+    def _check_target_class(self, detection: Dict, target_class: Optional[str]) -> bool:
+        if not target_class:
+            return True
+        return detection.get("class") == target_class
+
+    def _get_latest_bbox(self, tracks: List[Dict], roi_id: str) -> List[float]:
         for det in tracks:
-            matched_rois = det.get("matched_rois", [])
-            for roi in matched_rois:
-                if roi.get("roi_id") == roi_id:
-                    return det.get("bbox", [])
+            if self._check_detection_in_roi(det, roi_id):
+                return det.get("bbox", [])
         return []
 
     def process(
@@ -216,30 +214,41 @@ class IntrusionAlgorithm:
         tracks: List[Dict],
         current_time: Optional[datetime] = None,
     ) -> List[Dict]:
-        roi_has_person = self.is_roi_has_person(tracks, roi_id)
-
-        if not roi_has_person:
-            return []
-
-        now = time.monotonic()
+        current_time = current_time or datetime.now()
         key = f"{camera_id}_{roi_id}"
 
-        if key not in self.last_alert_time:
-            self.last_alert_time[key] = 0
-            self.alert_triggered[key] = False
+        roi_has_person = False
+        for det in tracks:
+            if self._check_detection_in_roi(det, roi_id) and self._check_target_class(det, self.target_class):
+                roi_has_person = True
+                break
 
-        if now - self.last_alert_time[key] >= self.cooldown_seconds:
-            self.last_alert_time[key] = now
+        if not roi_has_person:
+            self.detection_start.pop(key, None)
             self.alert_triggered[key] = False
-
-        if self.alert_triggered[key]:
             return []
 
-        bbox = self.get_latest_bbox_in_roi(tracks, roi_id)
+        if self.alert_triggered.get(key, False):
+            elapsed_since_alert = (current_time - self.last_alert_time.get(key, datetime.min)).total_seconds()
+            if elapsed_since_alert < self.cooldown_seconds:
+                return []
+            self.alert_triggered[key] = False
+
+        if self.detection_start.get(key) is None:
+            self.detection_start[key] = current_time
+
+        elapsed = (current_time - self.detection_start[key]).total_seconds()
+        if elapsed < self.confirm_seconds:
+            return []
+
+        bbox = self._get_latest_bbox(tracks, roi_id)
+        self.last_alert_time[key] = current_time
         self.alert_triggered[key] = True
+        self.detection_start[key] = None
 
         return [{
             "roi_id": roi_id,
+            "camera_id": camera_id,
             "bbox": bbox,
             "alert_type": "intrusion",
             "message": "检测到周界入侵",
@@ -248,52 +257,194 @@ class IntrusionAlgorithm:
     def reset(self):
         self.last_alert_time.clear()
         self.alert_triggered.clear()
+        self.detection_start.clear()
 
 
 class AlgorithmManager:
     def __init__(self, working_hours: Optional[List[Dict]] = None):
         self.algorithms: Dict[str, Dict[str, Any]] = {}
         self.working_hours = working_hours or []
+        self._update_lock = threading.Lock()
 
         self.default_params = {
             "leave_post": {
-                "threshold_sec": 300,
-                "confirm_sec": 10,
-                "return_sec": 30,
+                "confirm_on_duty_sec": 10,
+                "confirm_leave_sec": 10,
+                "cooldown_sec": 300,
+                "target_class": "person",
             },
             "intrusion": {
-                "cooldown_seconds": 300,
+                "cooldown_seconds": 120,
+                "confirm_seconds": 5,
+                "target_class": None,
             },
         }
-
+        
+        self._pubsub = None
+        self._pubsub_thread = None
+        self._running = False
+    
+    def start_config_subscription(self):
+        """启动配置变更订阅"""
+        try:
+            from config.settings import get_settings
+            settings = get_settings()
+            redis_client = redis.Redis(
+                host=settings.redis.host,
+                port=settings.redis.port,
+                db=settings.redis.db,
+                password=settings.redis.password,
+                decode_responses=True,
+            )
+            
+            self._pubsub = redis_client.pubsub()
+            self._pubsub.subscribe("config_update")
+            
+            self._running = True
+            self._pubsub_thread = threading.Thread(
+                target=self._config_update_worker,
+                name="ConfigUpdateSub",
+                daemon=True
+            )
+            self._pubsub_thread.start()
+            logger.info("已启动配置变更订阅")
+        except Exception as e:
+            logger.error(f"启动配置订阅失败: {e}")
+    
+    def _config_update_worker(self):
+        """配置更新订阅工作线程"""
+        try:
+            for message in self._pubsub.listen():
+                if not self._running:
+                    break
+                if message["type"] == "message":
+                    try:
+                        import json
+                        data = json.loads(message["data"])
+                        if data.get("type") == "roi":
+                            roi_ids = data.get("ids", [])
+                            if roi_ids:
+                                for roi_id in roi_ids:
+                                    self.reload_algorithm(roi_id)
+                            else:
+                                self.reload_all_algorithms()
+                    except Exception as e:
+                        logger.error(f"处理配置更新消息失败: {e}")
+        except Exception as e:
+            logger.error(f"配置订阅线程异常: {e}")
+    
+    def stop_config_subscription(self):
+        """停止配置变更订阅"""
+        self._running = False
+        if self._pubsub:
+            self._pubsub.close()
+        if self._pubsub_thread and self._pubsub_thread.is_alive():
+            self._pubsub_thread.join(timeout=5)
+        logger.info("配置订阅已停止")
+    
+    def load_from_redis(self, roi_id: str) -> bool:
+        """从Redis加载单个ROI的算法配置"""
+        try:
+            from core.config_sync import get_config_sync_manager
+            config_manager = get_config_sync_manager()
+            roi_config = config_manager.get_roi_from_redis(roi_id)
+            
+            if not roi_config:
+                return False
+            
+            with self._update_lock:
+                algorithm_type = roi_config.get("algorithm_type", "leave_post")
+                
+                if algorithm_type == "leave_post":
+                    params = {
+                        "working_hours": roi_config.get("working_hours"),
+                        "confirm_on_duty_sec": roi_config.get("confirm_on_duty_sec", 10),
+                        "confirm_leave_sec": roi_config.get("confirm_leave_sec", 10),
+                        "cooldown_sec": roi_config.get("cooldown_sec", 300),
+                        "target_class": roi_config.get("target_class", "person"),
+                    }
+                    if roi_id in self.algorithms and "leave_post" in self.algorithms[roi_id]:
+                        algo = self.algorithms[roi_id]["leave_post"]
+                        algo.confirm_on_duty_sec = params["confirm_on_duty_sec"]
+                        algo.confirm_leave_sec = params["confirm_leave_sec"]
+                        algo.cooldown_sec = params["cooldown_sec"]
+                        algo.target_class = params["target_class"]
+                        if params["working_hours"]:
+                            algo.working_hours = params["working_hours"]
+                        logger.info(f"已热更新算法参数: {roi_id}")
+                    else:
+                        self.register_algorithm(roi_id, "leave_post", params)
+                        logger.info(f"已从Redis加载算法: {roi_id}")
+                else:
+                    params = {}
+                    if roi_id in self.algorithms and algorithm_type in self.algorithms[roi_id]:
+                        pass
+                    else:
+                        self.register_algorithm(roi_id, algorithm_type, params)
+                        logger.info(f"已从Redis加载算法: {roi_id}")
+            
+            return True
+        except Exception as e:
+            logger.error(f"从Redis加载算法配置失败: {e}")
+            return False
+    
+    def reload_algorithm(self, roi_id: str) -> bool:
+        """重新加载单个ROI的算法配置"""
+        if roi_id not in self.algorithms:
+            return self.load_from_redis(roi_id)
+        
+        self.reset_algorithm(roi_id)
+        return self.load_from_redis(roi_id)
+    
+    def reload_all_algorithms(self) -> int:
+        """重新加载所有算法配置"""
+        count = 0
+        try:
+            from core.config_sync import get_config_sync_manager
+            config_manager = get_config_sync_manager()
+            roi_configs = config_manager.get_all_roi_configs()
+            
+            for roi_config in roi_configs:
+                roi_id = roi_config.get("roi_id")
+                if self.reload_algorithm(roi_id):
+                    count += 1
+            
+            logger.info(f"已重新加载 {count} 个算法配置")
+            return count
+        except Exception as e:
+            logger.error(f"重新加载所有算法配置失败: {e}")
+            return count
+    
     def register_algorithm(
         self,
         roi_id: str,
         algorithm_type: str,
         params: Optional[Dict[str, Any]] = None,
     ):
-        if roi_id in self.algorithms:
-            if algorithm_type in self.algorithms[roi_id]:
-                return
+        if roi_id in self.algorithms and algorithm_type in self.algorithms[roi_id]:
+            return
 
         if roi_id not in self.algorithms:
             self.algorithms[roi_id] = {}
 
-        algo_params = self.default_params.get(algorithm_type, {})
+        algo_params = self.default_params.get(algorithm_type, {}).copy()
         if params:
             algo_params.update(params)
 
         if algorithm_type == "leave_post":
             roi_working_hours = algo_params.get("working_hours") or self.working_hours
             self.algorithms[roi_id]["leave_post"] = LeavePostAlgorithm(
-                threshold_sec=algo_params.get("threshold_sec", 300),
-                confirm_sec=algo_params.get("confirm_sec", 10),
-                return_sec=algo_params.get("return_sec", 30),
+                confirm_on_duty_sec=algo_params.get("confirm_on_duty_sec", 10),
+                confirm_leave_sec=algo_params.get("confirm_leave_sec", 10),
+                cooldown_sec=algo_params.get("cooldown_sec", 300),
                 working_hours=roi_working_hours,
+                target_class=algo_params.get("target_class", "person"),
             )
         elif algorithm_type == "intrusion":
             self.algorithms[roi_id]["intrusion"] = IntrusionAlgorithm(
-                cooldown_seconds=algo_params.get("cooldown_seconds", 300),
+                cooldown_seconds=algo_params.get("cooldown_seconds", 120),
+                confirm_seconds=algo_params.get("confirm_seconds", 5),
+                target_class=algo_params.get("target_class"),
             )
 
     def process(
@@ -348,11 +499,11 @@ class AlgorithmManager:
             for algo_type, algo in self.algorithms[roi_id].items():
                 if algo_type == "leave_post":
                     status[algo_type] = {
-                        "state": getattr(algo, "state", "INIT_STATE"),
+                        "state": getattr(algo, "state", "WAITING"),
                         "alarm_sent": getattr(algo, "alarm_sent", False),
                     }
                 else:
                     status[algo_type] = {
-                        "track_count": len(getattr(algo, "track_states", {})),
+                        "detection_count": len(getattr(algo, "detection_start", {})),
                     }
         return status

config/config_models.py

@@ -104,6 +104,11 @@ class ROIInfo:
     alert_cooldown: int = 300
     enabled: bool = True
     extra_params: Optional[Dict[str, Any]] = None
+    working_hours: Optional[List[Dict]] = None  # 工作时间段
+    confirm_on_duty_sec: int = 10  # 在岗确认时间
+    confirm_leave_sec: int = 10  # 离岗确认时间
+    cooldown_sec: int = 300  # 告警冷却时间
+    target_class: str = "person"  # 目标类别
     
     def to_dict(self) -> Dict[str, Any]:
         """转换为字典"""
@@ -117,6 +122,11 @@ class ROIInfo:
             "alert_cooldown": self.alert_cooldown,
             "enabled": self.enabled,
             "extra_params": self.extra_params,
+            "working_hours": self.working_hours,
+            "confirm_on_duty_sec": self.confirm_on_duty_sec,
+            "confirm_leave_sec": self.confirm_leave_sec,
+            "cooldown_sec": self.cooldown_sec,
+            "target_class": self.target_class,
         }
     
     @classmethod
@@ -128,6 +138,14 @@ class ROIInfo:
         algo_type_str = data.get("algorithm_type", "leave_post")
         algo_type = AlgorithmType(algo_type_str) if algo_type_str in [e.value for e in AlgorithmType] else AlgorithmType.LEAVE_POST
         
+        working_hours = data.get("working_hours")
+        if isinstance(working_hours, str):
+            import json
+            try:
+                working_hours = json.loads(working_hours)
+            except:
+                working_hours = None
+        
         return cls(
             roi_id=data.get("roi_id", ""),
             camera_id=data.get("camera_id", ""),
@@ -138,6 +156,11 @@ class ROIInfo:
             alert_cooldown=data.get("alert_cooldown", 300),
             enabled=data.get("enabled", True),
             extra_params=data.get("extra_params"),
+            working_hours=working_hours,
+            confirm_on_duty_sec=data.get("confirm_on_duty_sec", 10),
+            confirm_leave_sec=data.get("confirm_leave_sec", 10),
+            cooldown_sec=data.get("cooldown_sec", 300),
+            target_class=data.get("target_class", "person"),
         )
     
     def is_point_inside(self, point: List[float]) -> bool:

config/database.py

@@ -162,6 +162,23 @@ class SQLiteManager:
                 alert_cooldown INTEGER DEFAULT 300,
                 enabled BOOLEAN DEFAULT 1,
                 extra_params TEXT,
+                working_hours TEXT,
+                confirm_on_duty_sec INTEGER DEFAULT 10,
+                confirm_leave_sec INTEGER DEFAULT 10,
+                cooldown_sec INTEGER DEFAULT 300,
+                target_class TEXT DEFAULT 'person',
+                updated_at TEXT
+            )
+        """)
+        
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS config_update_log (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                config_type TEXT NOT NULL,
+                config_id TEXT,
+                old_value TEXT,
+                new_value TEXT,
+                updated_by TEXT,
                 updated_at TEXT
             )
         """)
@@ -469,14 +486,21 @@ class SQLiteManager:
             cursor.execute("""
                 INSERT OR REPLACE INTO roi_configs (
                     roi_id, camera_id, roi_type, coordinates, algorithm_type,
-                    alert_threshold, alert_cooldown, enabled, extra_params, updated_at
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    alert_threshold, alert_cooldown, enabled, extra_params,
+                    working_hours, confirm_on_duty_sec, confirm_leave_sec,
+                    cooldown_sec, target_class, updated_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """, (
                 roi_id, camera_id, roi_type, str(coordinates), algorithm_type,
                 kwargs.get('alert_threshold', 3),
                 kwargs.get('alert_cooldown', 300),
                 kwargs.get('enabled', True),
                 str(kwargs.get('extra_params')) if kwargs.get('extra_params') else None,
+                str(kwargs.get('working_hours')) if kwargs.get('working_hours') else None,
+                kwargs.get('confirm_on_duty_sec', 10),
+                kwargs.get('confirm_leave_sec', 10),
+                kwargs.get('cooldown_sec', 300),
+                kwargs.get('target_class', 'person'),
                 now
             ))
             self._conn.commit()
@@ -494,12 +518,19 @@ class SQLiteManager:
             if row:
                 columns = ['roi_id', 'camera_id', 'roi_type', 'coordinates',
                           'algorithm_type', 'alert_threshold', 'alert_cooldown',
-                          'enabled', 'extra_params', 'updated_at']
+                          'enabled', 'extra_params', 'working_hours',
+                          'confirm_on_duty_sec', 'confirm_leave_sec', 'cooldown_sec',
+                          'target_class', 'updated_at']
                 result = dict(zip(columns, row))
                 try:
                     result['coordinates'] = eval(result['coordinates'])
                 except:
                     pass
+                try:
+                    if result.get('working_hours'):
+                        result['working_hours'] = eval(result['working_hours'])
+                except:
+                    pass
                 return result
             return None
         except Exception as e:
@@ -513,7 +544,9 @@ class SQLiteManager:
             cursor.execute("SELECT * FROM roi_configs WHERE camera_id = ?", (camera_id,))
             columns = ['roi_id', 'camera_id', 'roi_type', 'coordinates',
                       'algorithm_type', 'alert_threshold', 'alert_cooldown',
-                      'enabled', 'extra_params', 'updated_at']
+                      'enabled', 'extra_params', 'working_hours',
+                      'confirm_on_duty_sec', 'confirm_leave_sec', 'cooldown_sec',
+                      'target_class', 'updated_at']
             results = []
             for row in cursor.fetchall():
                 r = dict(zip(columns, row))
@@ -521,6 +554,11 @@ class SQLiteManager:
                     r['coordinates'] = eval(r['coordinates'])
                 except:
                     pass
+                try:
+                    if r.get('working_hours'):
+                        r['working_hours'] = eval(r['working_hours'])
+                except:
+                    pass
                 results.append(r)
             return results
         except Exception as e:
@@ -534,7 +572,9 @@ class SQLiteManager:
             cursor.execute("SELECT * FROM roi_configs ORDER BY camera_id, roi_id")
             columns = ['roi_id', 'camera_id', 'roi_type', 'coordinates',
                       'algorithm_type', 'alert_threshold', 'alert_cooldown',
-                      'enabled', 'extra_params', 'updated_at']
+                      'enabled', 'extra_params', 'working_hours',
+                      'confirm_on_duty_sec', 'confirm_leave_sec', 'cooldown_sec',
+                      'target_class', 'updated_at']
             results = []
             for row in cursor.fetchall():
                 r = dict(zip(columns, row))
@@ -542,6 +582,11 @@ class SQLiteManager:
                     r['coordinates'] = eval(r['coordinates'])
                 except:
                     pass
+                try:
+                    if r.get('working_hours'):
+                        r['working_hours'] = eval(r['working_hours'])
+                except:
+                    pass
                 results.append(r)
             return results
         except Exception as e:
@@ -558,6 +603,58 @@ class SQLiteManager:
         except Exception as e:
             logger.error(f"删除ROI配置失败: {e}")
             return False
+    
+    def log_config_update(
+        self,
+        config_type: str,
+        config_id: Optional[str],
+        old_value: Any,
+        new_value: Any,
+        updated_by: str = "system"
+    ):
+        """记录配置更新日志"""
+        try:
+            cursor = self._conn.cursor()
+            now = datetime.now().isoformat()
+            cursor.execute("""
+                INSERT INTO config_update_log (
+                    config_type, config_id, old_value, new_value, updated_by, updated_at
+                ) VALUES (?, ?, ?, ?, ?, ?)
+            """, (
+                config_type,
+                config_id,
+                str(old_value) if old_value else None,
+                str(new_value) if new_value else None,
+                updated_by,
+                now
+            ))
+            self._conn.commit()
+            logger.info(f"配置更新日志已记录: {config_type}/{config_id}")
+        except Exception as e:
+            logger.error(f"记录配置更新日志失败: {e}")
+    
+    def get_config_update_log(
+        self,
+        config_type: Optional[str] = None,
+        limit: int = 100
+    ) -> List[Dict[str, Any]]:
+        """获取配置更新日志"""
+        try:
+            cursor = self._conn.cursor()
+            query = "SELECT * FROM config_update_log WHERE 1=1"
+            params = []
+            if config_type:
+                query += " AND config_type = ?"
+                params.append(config_type)
+            query += " ORDER BY id DESC LIMIT ?"
+            params.append(limit)
+            cursor.execute(query, params)
+            columns = ['id', 'config_type', 'config_id', 'old_value', 'new_value',
+                      'updated_by', 'updated_at']
+            return [dict(zip(columns, row)) for row in cursor.fetchall()]
+        except Exception as e:
+            logger.error(f"获取配置更新日志失败: {e}")
+            return []
 
 
 def get_sqlite_manager() -> SQLiteManager:

core/config_sync.py

@@ -383,6 +383,168 @@ class ConfigSyncManager:
             ),
         }
     
+    def _cache_roi_to_redis(self, roi_config: Dict[str, Any]) -> bool:
+        """将ROI配置缓存到Redis"""
+        if not self._redis_client:
+            return False
+        
+        try:
+            roi_id = roi_config.get("roi_id")
+            key = f"config:roi:{roi_id}"
+            
+            self._redis_client.hset(key, mapping={
+                "roi_id": roi_id,
+                "camera_id": roi_config.get("camera_id", ""),
+                "roi_type": roi_config.get("roi_type", ""),
+                "coordinates": str(roi_config.get("coordinates", [])),
+                "algorithm_type": roi_config.get("algorithm_type", ""),
+                "working_hours": str(roi_config.get("working_hours", [])),
+                "confirm_on_duty_sec": str(roi_config.get("confirm_on_duty_sec", 10)),
+                "confirm_leave_sec": str(roi_config.get("confirm_leave_sec", 10)),
+                "cooldown_sec": str(roi_config.get("cooldown_sec", 300)),
+                "target_class": roi_config.get("target_class", "person"),
+                "enabled": str(roi_config.get("enabled", True)),
+            })
+            
+            self._redis_client.expire(key, 3600)
+            logger.debug(f"ROI配置已缓存到Redis: {key}")
+            return True
+        except Exception as e:
+            logger.error(f"缓存ROI配置到Redis失败: {e}")
+            return False
+    
+    def _cache_camera_to_redis(self, camera_config: Dict[str, Any]) -> bool:
+        """将摄像头配置缓存到Redis"""
+        if not self._redis_client:
+            return False
+        
+        try:
+            camera_id = camera_config.get("camera_id")
+            key = f"config:camera:{camera_id}"
+            
+            self._redis_client.hset(key, mapping={
+                "camera_id": camera_id,
+                "rtsp_url": camera_config.get("rtsp_url", ""),
+                "camera_name": camera_config.get("camera_name", ""),
+                "enabled": str(camera_config.get("enabled", True)),
+                "location": camera_config.get("location", ""),
+            })
+            
+            self._redis_client.expire(key, 3600)
+            logger.debug(f"摄像头配置已缓存到Redis: {key}")
+            return True
+        except Exception as e:
+            logger.error(f"缓存摄像头配置到Redis失败: {e}")
+            return False
+    
+    def sync_all_to_redis(self) -> int:
+        """同步所有配置到Redis缓存"""
+        if not self._redis_client:
+            return 0
+        
+        count = 0
+        try:
+            cameras = self._db_manager.get_all_camera_configs()
+            for camera in cameras:
+                if self._cache_camera_to_redis(camera):
+                    count += 1
+            
+            rois = self._db_manager.get_all_roi_configs()
+            for roi in rois:
+                if self._cache_roi_to_redis(roi):
+                    count += 1
+            
+            logger.info(f"已同步 {count} 条配置到Redis缓存")
+            return count
+        except Exception as e:
+            logger.error(f"同步配置到Redis失败: {e}")
+            return count
+    
+    def get_roi_from_redis(self, roi_id: str) -> Optional[Dict[str, Any]]:
+        """从Redis获取ROI配置"""
+        if not self._redis_client:
+            return None
+        
+        try:
+            key = f"config:roi:{roi_id}"
+            data = self._redis_client.hgetall(key)
+            if data:
+                if data.get('coordinates'):
+                    data['coordinates'] = eval(data['coordinates'])
+                if data.get('working_hours'):
+                    data['working_hours'] = eval(data['working_hours'])
+                data['confirm_on_duty_sec'] = int(data.get('confirm_on_duty_sec', 10))
+                data['confirm_leave_sec'] = int(data.get('confirm_leave_sec', 10))
+                data['cooldown_sec'] = int(data.get('cooldown_sec', 300))
+                data['enabled'] = data.get('enabled', 'True') == 'True'
+                return data
+            return None
+        except Exception as e:
+            logger.error(f"从Redis获取ROI配置失败: {e}")
+            return None
+    
+    def get_camera_from_redis(self, camera_id: str) -> Optional[Dict[str, Any]]:
+        """从Redis获取摄像头配置"""
+        if not self._redis_client:
+            return None
+        
+        try:
+            key = f"config:camera:{camera_id}"
+            data = self._redis_client.hgetall(key)
+            if data:
+                data['enabled'] = data.get('enabled', 'True') == 'True'
+                return data
+            return None
+        except Exception as e:
+            logger.error(f"从Redis获取摄像头配置失败: {e}")
+            return None
+    
+    def notify_config_change(self, config_type: str, config_ids: List[str]):
+        """通知配置变更（发布到Redis频道）"""
+        if not self._redis_client:
+            return
+        
+        try:
+            message = {
+                "type": config_type,
+                "ids": config_ids,
+                "timestamp": datetime.now().isoformat(),
+            }
+            self._redis_client.publish("config_update", json.dumps(message))
+            logger.info(f"已发布配置变更通知: {config_type} - {config_ids}")
+        except Exception as e:
+            logger.error(f"发布配置变更通知失败: {e}")
+    
+    def clear_redis_cache(self, config_type: Optional[str] = None):
+        """清理Redis缓存"""
+        if not self._redis_client:
+            return
+        
+        try:
+            if config_type == "roi":
+                keys = self._redis_client.keys("config:roi:*")
+                if keys:
+                    self._redis_client.delete(*keys)
+            elif config_type == "camera":
+                keys = self._redis_client.keys("config:camera:*")
+                if keys:
+                    self._redis_client.delete(*keys)
+            else:
+                keys = self._redis_client.keys("config:*")
+                if keys:
+                    self._redis_client.delete(*keys)
+            logger.info(f"已清理Redis缓存: {config_type or 'all'}")
+        except Exception as e:
+            logger.error(f"清理Redis缓存失败: {e}")
+    
+    def reload_algorithms(self):
+        """重新加载所有算法配置"""
+        self.invalidate_all_cache()
+        self.clear_redis_cache()
+        count = self.sync_all_to_redis()
+        self.notify_config_change("roi", [])
+        logger.info(f"算法配置已重新加载，更新了 {count} 条缓存")
+    
     def close(self):
         """关闭管理器"""
         self.stop_config_subscription()