deploy_pickup/run_pickup_monitor_old.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
run_pickup_monitor.py
---------------------
拾音器异响检测主程序

功能说明：
该程序用于音频采集设备（拾音器）的实时异常检测。
与摄像头版本不同，本版本：
1. 没有视频/图片采集和上报
2. 上报时包含音频频谱图分析数据
3. 支持进水流量PLC数据读取
4. 每1分钟计算一次平均重建误差并上报

上报数据结构：
{
    "message": {
        "channelInfo": {"name": "设备信息"},
        "requestTime": 时间戳,
        "classification": {
            "level_one": 2,  // 音频检测大类
            "level_two": 6   // 异常类型小类: 6=未分类, 7=轴承, 8=气蚀, 9=松动/共振, 10=叶轮, 11=阀件
        },
        "skillInfo": {"name": "异响检测"},
        "sound_detection": {
            "abnormalwav": "",           // base64编码的音频
            "status": 0/1,               // 0=异常, 1=正常
            "condition": {
                "running_status": "运行中/停机中",  // 启停状态
                "inlet_flow": 0.0            // 进水流量
            },
            "score": {
                "abnormal_score": 0.0,   // 当前1分钟内8s平均重构误差
                "score_threshold": 0.0   // 该设备的异常阀值
            },
            "frequency": {
                "this_frequency": [],              // 当前1分钟的频谱图
                "normal_frequency_middle": [],    // 过去10分钟的频谱图平均
                "normal_frequency_upper": [],     // 上限（暂为空）
                "normal_frequency_lower": []      // 下限（暂为空）
            }
        }
    }
}

使用方法：
    python run_pickup_monitor.py

配置文件：
    config/rtsp_config.yaml
"""

import subprocess
import time
import re
import signal
import sys
import threading
import logging
import base64
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from datetime import datetime, timedelta
from collections import defaultdict

# 用于计算FFT
import numpy as np


try:
    import librosa
except ImportError:
    print("错误：缺少librosa库，请安装: pip install librosa")
    sys.exit(1)

try:
    import requests
except ImportError:
    print("错误：缺少requests库，请安装: pip install requests")
    sys.exit(1)

# 导入预测器模块
from predictor import MultiModelPredictor, CFG

# 导入配置管理模块（SQLite）
from config.config_manager import ConfigManager
from config.config_api import app as config_app, init_config_api
from config.db_models import get_db_path

# 导入泵状态监控模块（用于检测启停过渡期）
try:
    from core.pump_state_monitor import PumpStateMonitor
    PUMP_STATE_MONITOR_AVAILABLE = True
except ImportError:
    PUMP_STATE_MONITOR_AVAILABLE = False

# 导入人体检测读取模块（用于抑制有人时的误报）
try:
    from core.human_detection_reader import HumanDetectionReader
    HUMAN_DETECTION_AVAILABLE = True
except ImportError:
    HUMAN_DETECTION_AVAILABLE = False

# 导入报警聚合器（跨设备聚合抑制 + 分类型冷却）
try:
    from core.alert_aggregator import AlertAggregator
    ALERT_AGGREGATOR_AVAILABLE = True
except ImportError:
    ALERT_AGGREGATOR_AVAILABLE = False


# ========================================
# 配置日志系统
# ========================================
def setup_logging():
    # 配置日志系统（RotatingFileHandler -> system.log）
    from logging.handlers import RotatingFileHandler

    # 如果根 logger 已经被上层调用者配置过，则直接复用
    root = logging.getLogger()
    if root.handlers:
        return logging.getLogger('PickupMonitor')

    # 日志配置
    log_dir = Path(__file__).parent / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)
    log_file = log_dir / "system.log"

    formatter = logging.Formatter(
        '%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )

    # 按文件大小轮转，最多保留 2 个备份（共 30MB）
    file_handler = RotatingFileHandler(
        log_file,
        maxBytes=10 * 1024 * 1024,
        backupCount=2,
        encoding='utf-8'
    )
    file_handler.setFormatter(formatter)

    # 控制台输出（前台运行时可见，后台运行时 stdout 已被丢弃不影响）
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(formatter)

    logging.basicConfig(
        level=logging.INFO,
        handlers=[file_handler, console_handler]
    )

    return logging.getLogger('PickupMonitor')


# 初始化日志系统
logger = setup_logging()

# 导入能量基线模块
try:
    from core.energy_baseline import EnergyBaseline
    ENERGY_BASELINE_AVAILABLE = True
except ImportError:
    ENERGY_BASELINE_AVAILABLE = False
    logger.warning("能量基线模块未找到，泵状态检测功能禁用")


class RTSPStreamConfig:
    """
    RTSP流配置类
    
    封装单个RTSP流的配置信息，包含拾音器特有字段
    """
    
    def __init__(self, plant_name, rtsp_url, channel, 
                 camera_name, device_code, pump_name,
                 flow_plc, project_id):
        """
        初始化RTSP流配置
        
        参数:
            plant_name: 区域名称（泵房-反渗透高压泵等）
            rtsp_url: RTSP流URL
            channel: 通道号
            camera_name: 设备名称
            device_code: 设备编号（如1#-1）
            pump_name: 泵名称（A/B/C/D），用于匹配流量PLC
            flow_plc: 流量PLC地址映射
        """
        self.plant_name = plant_name
        self.rtsp_url = rtsp_url
        self.channel = channel
        self.pump_id = f"ch{channel}"
        self.camera_name = camera_name or f"ch{channel}"
        self.device_code = device_code
        self.pump_name = pump_name
        self.flow_plc = flow_plc or {}
        self.project_id = project_id
    
    def get_flow_plc_address(self):
        """
        获取该设备对应的进水流量PLC地址
        
        返回:
            PLC地址字符串，不存在则返回空字符串
        """
        if self.pump_name and self.flow_plc:
            return self.flow_plc.get(self.pump_name, "")
        return ""
    
    def __repr__(self):
        return f"RTSPStreamConfig(plant='{self.plant_name}', camera='{self.camera_name}', pump='{self.pump_id}')"


class FFmpegProcess:
    """
    FFmpeg进程管理类
    
    负责启动和管理单个RTSP流的FFmpeg进程。
    进程将RTSP流转换为固定时长的WAV音频文件。
    
    文件名格式: {project_id}_{device_code}_{时间戳}.wav
    """
    
    def __init__(self, stream_config, output_dir, config=None):
        """
        初始化FFmpeg进程管理器
        
        参数:
            stream_config: RTSP流配置
            output_dir: 音频输出目录
            config: 全局配置字典
        """
        self.config_dict = config or {}
        self.stream_config = stream_config
        self.output_dir = output_dir
        self.process = None
        
        # 从配置中读取文件时长，默认8秒
        audio_cfg = self.config_dict.get('audio', {})
        self.file_duration = audio_cfg.get('file_duration', 8)
        
        # 获取project_id（从 stream_config 中读取）
        self.project_id = stream_config.project_id
    
    def start(self):
        """
        启动FFmpeg进程
        
        返回:
            bool: 启动成功返回True，失败返回False
        """
        # 获取设备编号
        device_code = self.stream_config.device_code or self.stream_config.pump_id
        
        # 创建输出目录（每个设备独立目录）
        # 结构: data/{device_code}/current/
        current_dir = self.output_dir / device_code / "current"
        current_dir.mkdir(parents=True, exist_ok=True)
        # 存放人工核查确认为正常的异常音频，增训时全量参与训练
        verified_dir = self.output_dir / device_code / "verified_normal"
        verified_dir.mkdir(parents=True, exist_ok=True)
        
        # 构建输出文件名模板
        # 格式: {project_id}_{device_code}_{时间戳}.wav
        # 例如: 92_1#-1_20251218142000.wav
        output_pattern = str(current_dir / f"{self.project_id}_{device_code}_%Y%m%d%H%M%S.wav")
        
        # 构建FFmpeg命令
        # 添加内存限制参数，防止 RTSP 缓冲区无限增长导致 OOM
        cmd = [
            "ffmpeg",
            # RTSP 输入参数（内存限制）
            "-rtsp_transport", "tcp",           # 使用TCP传输
            "-probesize", "1000000",            # 限制探测大小为1MB（默认5MB）
            "-analyzeduration", "1000000",      # 限制分析时长为1秒（默认5秒）
            "-max_delay", "500000",             # 最大延迟500ms
            "-fflags", "nobuffer",              # 禁用输入缓冲
            "-flags", "low_delay",              # 低延迟模式
            "-i", self.stream_config.rtsp_url,  # 输入RTSP流
            # 音频输出参数
            "-vn",                              # 不处理视频
            "-ac", "1",                         # 单声道
            "-ar", str(CFG.SR),                 # 采样率（16000Hz）
            "-f", "segment",                    # 分段模式
            "-segment_time", str(int(self.file_duration)),  # 每段时长
            "-strftime", "1",                   # 启用时间格式化
            "-loglevel", "error",               # 只输出错误日志
            "-y",                               # 覆盖已存在的文件
            output_pattern,
        ]
        
        try:
            # 启动FFmpeg进程
            self.process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            
            logger.info(f"FFmpeg已启动: {device_code} | {self.stream_config.camera_name} | "
                  f"文件时长: {self.file_duration}秒 | PID={self.process.pid}")
            return True
            
        except FileNotFoundError:
            logger.error("FFmpeg错误: 未找到ffmpeg命令，请确保已安装FFmpeg")
            return False
        except Exception as e:
            logger.error(f"FFmpeg启动失败: {device_code} | 错误: {e}")
            return False
    
    def is_running(self):
        """
        检查FFmpeg进程是否在运行
        
        返回:
            bool: 进程运行中返回True，否则返回False
        """
        if self.process is None:
            return False
        return self.process.poll() is None
    
    def stop(self):
        """
        停止FFmpeg进程
        """
        if self.process is not None and self.is_running():
            logger.info(f"FFmpeg停止: {self.stream_config.plant_name} | {self.stream_config.camera_name} | PID={self.process.pid}")
            self.process.terminate()
            try:
                self.process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                logger.warning(f"FFmpeg强制终止: PID={self.process.pid}")
                self.process.kill()


class PickupMonitor:
    """
    拾音器监控线程类
    
    监控音频目录，调用预测器检测异常，推送告警。
    
    主要功能：
    1. 不截取视频帧（纯音频）
    2. 计算并上报频谱图数据（this_frequency + normal_frequency_middle）
    3. 每分钟汇总上报一次
    4. 使用SCADA API获取进水流量
    """
    
    def __init__(self, audio_dir, multi_predictor, 
                 stream_configs,
                 check_interval=1.0, config=None,
                 config_manager=None):
        """
        初始化监控器
        
        Args:
            audio_dir: 音频根目录
            multi_predictor: 多模型预测器实例
            stream_configs: 所有RTSP流配置
            check_interval: 检查间隔（秒）
            config: 配置字典
            config_manager: ConfigManager 实例（用于热更新，为 None 时禁用热更新）
        """
        # 音频根目录（各设备目录在其下）
        self.audio_dir = audio_dir
        self.multi_predictor = multi_predictor
        self.predictor = None  # 兼容性保留，已废弃
        self.stream_configs = stream_configs
        self.check_interval = check_interval
        self.config = config or {}
        
        # 热更新：持有 ConfigManager 引用，定期从 DB 刷新配置
        self._config_manager = config_manager
        self._last_config_reload = 0       # 上次配置刷新时间戳
        self._config_reload_interval = 30  # 配置刷新间隔（秒）
        
        # project_id
        self.project_id = self.config.get('platform', {}).get('project_id', 92)
        
        # 构建device_code到stream_config的映射
        self.device_map = {}
        for cfg in stream_configs:
            if cfg.device_code:
                self.device_map[cfg.device_code] = cfg
        
        # 已处理文件集合
        self.seen_files = set()
        
        # 每个设备的检测结果缓存（用于1分钟汇总）
        # key: device_code（如 "1#-1"）
        self.device_cache = defaultdict(lambda: {
            "errors": [],          # 每8秒的重建误差列表
            "last_upload": None,   # 上次上报时间
            "audio_data": [],      # 用于计算频谱图的音频数据
            "status": None         # 最近的运行状态
        })
        
        # 频谱图历史缓存（用于计算normal_frequency_middle）
        # key: device_code, value: list of (timestamp, freq_db)
        freq_cfg = self.config.get('prediction', {}).get('frequency_history', {})
        self.freq_history_enabled = freq_cfg.get('enabled', True)
        self.freq_history_minutes = freq_cfg.get('history_minutes', 10)
        self.freq_history = defaultdict(list)
        
        # 上一次上报状态（True=异常，False=正常，None=初始）
        # 用于状态变更时去重，防止持续报警
        self.last_report_status = {}
        
        # 上报周期（秒）
        audio_cfg = self.config.get('audio', {})
        self.segment_duration = audio_cfg.get('segment_duration', 60)
        
        # 异常音频保存配置
        save_cfg = self.config.get('prediction', {}).get('save_anomaly_audio', {})
        self.save_anomaly_enabled = save_cfg.get('enabled', True)
        self.save_anomaly_dir = Path(__file__).parent / save_cfg.get('save_dir', 'data/anomaly_detected')
        if self.save_anomaly_enabled:
            self.save_anomaly_dir.mkdir(parents=True, exist_ok=True)
        
        # 异常推送配置
        push_cfg = self.config.get('push_notification', {})
        self.push_enabled = push_cfg.get('enabled', False)
        self.alert_enabled = push_cfg.get('alert_enabled', True)  # 是否启用异常告警（false=暂时禁用异常上报）
        self.push_timeout = push_cfg.get('timeout', 30)
        self.push_retry_count = push_cfg.get('retry_count', 2)
        # 推送基地址列表（统一用 base_url/{project_id} 拼接，可无限扩展）
        raw_urls = push_cfg.get('push_base_urls', [])
        self.push_base_urls = [
            {"label": item.get("label", f"target-{i}"), "url": item.get("url", "").rstrip("/")}
            for i, item in enumerate(raw_urls)
            if item.get("url")  # 跳过空URL
        ]
        # 推送失败记录文件路径
        failed_log_path = push_cfg.get('failed_push_log', 'data/push_failures.jsonl')
        self.failed_push_log = Path(__file__).parent / failed_log_path
        self.failed_push_log.parent.mkdir(parents=True, exist_ok=True)
        
        # 如果 alert_enabled 为 False，记录日志提醒
        if not self.alert_enabled:
            logger.warning("异常告警已禁用（alert_enabled=false），仅上报心跳状态")
        
        # ========================================
        # 报警聚合器（替代原有固定 cooldown_minutes）
        # ----------------------------------------
        # 功能1：跨设备聚合抑制 - 同一水厂多设备同时报警 -> 环境噪声，全部抑制
        # 功能2：分类型冷却 - 同类型24h，不同类型1h
        # ========================================
        agg_cfg = push_cfg.get('alert_aggregate', {})
        self.alert_aggregator = None
        if ALERT_AGGREGATOR_AVAILABLE:
            self.alert_aggregator = AlertAggregator(
                push_callback=self._push_detection_result,
                aggregate_enabled=agg_cfg.get('enabled', True),
                window_seconds=agg_cfg.get('window_seconds', 300),
                min_devices=agg_cfg.get('min_devices', 2),
                cooldown_same_type_hours=push_cfg.get('cooldown_same_type_hours', 24),
                cooldown_diff_type_hours=push_cfg.get('cooldown_diff_type_hours', 1)
            )
        else:
            logger.warning("报警聚合器模块未找到，使用默认报警逻辑")
        
        # 上次异常音频保存时间（用于保存冷却时间计算）
        self.last_anomaly_save_time = {}
        # 异常保存冷却时间（分钟），同一设备连续异常时，每N分钟只保存一次
        self.anomaly_save_cooldown_minutes = save_cfg.get('cooldown_minutes', 10)
        
        # 当前异常分类结果锁定（持续异常期间保持分类结果不变）
        # key: device_code, value: (anomaly_type_code, type_name)
        self.locked_anomaly_type = {}
        
        # 滑动窗口投票配置（5次中有3次异常才判定为异常）
        voting_cfg = self.config.get('prediction', {}).get('voting', {})
        self.voting_enabled = voting_cfg.get('enabled', True)
        self.voting_window_size = voting_cfg.get('window_size', 5)   # 窗口大小
        self.voting_threshold = voting_cfg.get('threshold', 3)       # 异常阈值（>=3次则判定异常）
        self.detection_history = {}          # 每个设备的检测历史（True=异常）
        
        # 阈值容差区间配置（避免边界值反复跳变）
        # 误差在 threshold*(1-tolerance) ~ threshold*(1+tolerance) 范围内为灰区，维持上一状态
        self.tolerance_ratio = voting_cfg.get('tolerance_ratio', 0.05)  # 默认5%容差
        self.last_single_anomaly = {}  # 每个设备上一次的单周期判定结果
        
        # 阈值现在由 multi_predictor 管理，每个设备从其对应模型目录加载
        
        # 能量检测配置
        energy_cfg = self.config.get('prediction', {}).get('energy_detection', {})
        self.energy_detection_enabled = ENERGY_BASELINE_AVAILABLE and energy_cfg.get('enabled', True)
        self.skip_detection_when_stopped = energy_cfg.get('skip_when_stopped', True)
        
        # 能量基线检测器（每个设备一个）
        if self.energy_detection_enabled:
            self.energy_baselines = {}
        else:
            self.energy_baselines = None
        
        # SCADA API配置（用于获取泵状态和进水流量）
        scada_cfg = self.config.get('scada_api', {})
        self.scada_enabled = scada_cfg.get('enabled', False)
        self.scada_url = scada_cfg.get('base_url', '')  # 历史数据接口（备用）
        self.scada_realtime_url = scada_cfg.get('realtime_url', '')  # 实时数据接口（主用）
        self.scada_jwt = scada_cfg.get('jwt_token', '')
        self.scada_timeout = scada_cfg.get('timeout', 10)
        
        # 泵状态监控器（用于检测启停过渡期）
        self.pump_state_monitor = None
        self.pump_status_plc_configs = {}  # {pump_name: [{point, name}, ...]}
        if PUMP_STATE_MONITOR_AVAILABLE and self.scada_enabled:
            # 初始化泵状态监控器
            # 获取第一个启用水厂的project_id
            project_id = 0
            for plant in self.config.get('plants', []):
                if plant.get('enabled', False):
                    project_id = plant.get('project_id', 0)
                    # 加载泵状态点位配置
                    pump_status_plc = plant.get('pump_status_plc', {})
                    self.pump_status_plc_configs = pump_status_plc
                    break
            
            if project_id > 0:
                # 使用实时接口 URL 进行泵状态查询
                self.pump_state_monitor = PumpStateMonitor(
                    scada_url=self.scada_realtime_url,  # 使用实时数据接口
                    scada_jwt=self.scada_jwt,
                    project_id=project_id,
                    timeout=self.scada_timeout,
                    transition_window_minutes=15  # 启停后15分钟内视为过渡期
                )
                logger.info(f"泵状态监控器已启用 (project_id={project_id}, 过渡期窗口=15分钟, 使用实时接口)")
        
        # 线程控制
        self.running = False
        self.thread = None
        # 推送线程池（避免推送超时阻塞主监控循环）
        self._push_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="push")
        
        # 启动时间（用于跳过启动期间的状态变化日志）
        self.startup_time = None
        self.startup_warmup_seconds = 120  # 启动后120秒内不记录状态变化
        
        # ========================================
        # 异常上下文捕获配置
        # ========================================
        context_cfg = save_cfg.get('context_capture', {})
        self.context_pre_minutes = context_cfg.get('pre_minutes', 2)
        self.context_post_minutes = context_cfg.get('post_minutes', 2)

        # 音频文件历史缓存（用于捕获异常前的音频）
        # key: device_code, value: deque of (timestamp, file_path)
        from collections import deque
        # 计算需要保留的历史文件数量（按分钟计算，每分钟约1个文件）
        history_size = self.context_pre_minutes + 2  # 多保留2个作为缓冲
        self.audio_file_history = defaultdict(lambda: deque(maxlen=history_size))

        # 异常上下文捕获状态
        # key: device_code, value: dict
        self.anomaly_capture_state = {}

        logger.info(f"异常上下文捕获: 前{self.context_pre_minutes}分钟 + 后{self.context_post_minutes}分钟")
        
        # ========================================
        # 人体检测报警抑制配置
        # ========================================
        # 只要任意摄像头在冷却时间内检测到人，所有设备的异常报警都会被抑制
        human_cfg = self.config.get('human_detection', {})
        self.human_detection_enabled = human_cfg.get('enabled', False) and HUMAN_DETECTION_AVAILABLE
        self.human_reader = None
        
        if self.human_detection_enabled:
            db_path = human_cfg.get('db_path', '')
            cooldown_minutes = human_cfg.get('cooldown_minutes', 5)
            self.human_reader = HumanDetectionReader(
                db_path=db_path,
                cooldown_minutes=cooldown_minutes
            )
            logger.info(f"人体检测报警抑制已启用 (冷却时间={cooldown_minutes}分钟, 数据库={db_path})")
        
        # ========================================
        # 远程异响检测调度配置
        # ========================================
        # 通过定时 GET 请求远程接口，根据返回的 mode_type、model_list、时间窗口
        # 动态控制 alert_enabled 开关
        schedule_cfg = self.config.get('detection_schedule', {})
        self._detection_schedule_url = schedule_cfg.get('url', '')
        self._detection_schedule_interval = schedule_cfg.get('poll_interval', 60)
        self._detection_schedule_timeout = schedule_cfg.get('request_timeout', 10)
        # 独立计时器，与配置热刷新的 30 秒间隔解耦
        self._last_detection_schedule_check = 0
        # 远程调度控制的暂停标志（优先级最高，True 时跳过整个检测流程，包括模型推理）
        self.detection_paused = False
        
        if self._detection_schedule_url:
            logger.info(f"远程异响调度已启用 | URL={self._detection_schedule_url} | 间隔={self._detection_schedule_interval}秒")
    
    def start(self):
        """
        启动监控线程
        """
        if self.running:
            return
        
        # 启动前清理current目录中的遗留文件
        self._cleanup_current_on_startup()
        
        self.running = True
        self.startup_time = datetime.now()  # 记录启动时间
        self.thread = threading.Thread(target=self._monitor_loop, daemon=True)
        self.thread.start()
        # 打印监控的设备列表
        device_codes = list(self.device_map.keys())
        logger.info(f"监控线程已启动 | 目录: {self.audio_dir} | 监控设备: {device_codes}")
    
    def _cleanup_current_on_startup(self):
        """
        启动时清理current目录中的遗留文件
        
        删除上次运行遗留的文件，避免混入新采集数据影响检测准确性
        """
        cleaned_count = 0
        
        for device_code in self.device_map.keys():
            current_dir = self.audio_dir / device_code / "current"
            if not current_dir.exists():
                continue
            
            for wav_file in current_dir.glob("*.wav"):
                try:
                    wav_file.unlink()
                    cleaned_count += 1
                except Exception as e:
                    logger.warning(f"清理遗留文件失败: {wav_file.name} | {e}")
        
        if cleaned_count > 0:
            logger.info(f"启动清理: 已删除current目录中 {cleaned_count} 个遗留文件")
    
    def stop(self):
        """
        停止监控线程
        """
        if not self.running:
            return
        
        self.running = False
        if self.thread is not None:
            self.thread.join(timeout=5)
        # 关闭推送线程池，等待已提交的推送任务完成
        self._push_executor.shutdown(wait=True, cancel_futures=False)
        logger.info(f"监控线程已停止")
    
    def _reload_hot_config(self):
        # 从 DB 热加载可变配置项，无需重启服务
        # 只刷新运行时可安全变更的参数，不影响 FFmpeg 进程和流映射
        if self._config_manager is None:
            return
        
        now = time.time()
        # 未达到刷新间隔则跳过
        if now - self._last_config_reload < self._config_reload_interval:
            return
        self._last_config_reload = now
        
        try:
            # 从 DB 读取最新配置
            fresh = self._config_manager.get_full_config()
            self.config = fresh
            
            # 刷新推送配置
            push_cfg = fresh.get('push_notification', {})
            self.push_enabled = push_cfg.get('enabled', False)
            self.alert_enabled = push_cfg.get('alert_enabled', True)
            self.push_timeout = push_cfg.get('timeout', 30)
            self.push_retry_count = push_cfg.get('retry_count', 2)
            raw_urls = push_cfg.get('push_base_urls', [])
            self.push_base_urls = [
                {"label": item.get("label", f"target-{i}"), "url": item.get("url", "").rstrip("/")}
                for i, item in enumerate(raw_urls)
                if item.get("url")
            ]
            
            # 刷新投票配置
            voting_cfg = fresh.get('prediction', {}).get('voting', {})
            self.voting_enabled = voting_cfg.get('enabled', True)
            self.voting_window_size = voting_cfg.get('window_size', 5)
            self.voting_threshold = voting_cfg.get('threshold', 3)
            self.tolerance_ratio = voting_cfg.get('tolerance_ratio', 0.05)
            
            # 刷新能量检测配置
            energy_cfg = fresh.get('prediction', {}).get('energy_detection', {})
            self.skip_detection_when_stopped = energy_cfg.get('skip_when_stopped', True)
            
            # 刷新人体检测配置
            human_cfg = fresh.get('human_detection', {})
            new_human_enabled = human_cfg.get('enabled', False)
            if new_human_enabled != self.human_detection_enabled:
                logger.info(f"人体检测抑制配置变更: {self.human_detection_enabled} -> {new_human_enabled}")
                self.human_detection_enabled = new_human_enabled
            
            # 刷新远程异响调度配置
            schedule_cfg = fresh.get('detection_schedule', {})
            self._detection_schedule_url = schedule_cfg.get('url', '')
            self._detection_schedule_interval = schedule_cfg.get('poll_interval', 60)
            self._detection_schedule_timeout = schedule_cfg.get('request_timeout', 10)
            
            logger.debug("配置热刷新完成")
        except Exception as e:
            logger.error(f"配置热刷新失败: {e}")
    
    def _check_detection_schedule(self):
        # 远程异响检测调度检查
        # 独立于 DB 配置热刷新，有自己的 60 秒计时器
        # 条件全部满足才开启 alert_enabled：
        #   1. model_list 中包含 "异响"
        #   2. mode_type == 1
        #   3. start_time <= 当前时间 <= end_time
        if not self._detection_schedule_url:
            return
        
        now_ts = time.time()
        if now_ts - self._last_detection_schedule_check < self._detection_schedule_interval:
            return
        self._last_detection_schedule_check = now_ts
        
        try:
            resp = requests.get(
                self._detection_schedule_url,
                params={'project_id': self.project_id},
                timeout=self._detection_schedule_timeout
            )
            resp.raise_for_status()
            data = resp.json()
            
            model_list = data.get('model_list', [])
            mode_type = int(data.get('mode_type', 0))
            start_str = data.get('start_time', '')
            end_str = data.get('end_time', '')
            
            # model_list 中没有"异响"，说明本项目不涉及异响检测，保持现状不做干预
            if '异响' not in model_list:
                return
            
            if not start_str or not end_str:
                logger.warning("远程调度时间窗口不完整，保持当前状态")
                return
            
            # 解析时间，支持多种常见格式
            now = datetime.now()
            start_dt = self._parse_schedule_time(start_str)
            end_dt = self._parse_schedule_time(end_str)
            
            if start_dt is None or end_dt is None:
                logger.warning(f"时间解析失败: start='{start_str}', end='{end_str}'")
                return
            
            # 核心判定：mode_type==1 且当前时间在窗口内 -> 检测启用
            # 反之 -> 暂停整个检测流程（不做模型推理，不上报）
            should_detect = (mode_type == 1) and (start_dt <= now <= end_dt)
            new_paused = not should_detect
            
            if new_paused != self.detection_paused:
                logger.info(
                    f"异响检测调度变更: {'暂停' if new_paused else '恢复'} | "
                    f"mode_type={mode_type} | "
                    f"窗口=[{start_str}, {end_str}] | 当前={now.strftime('%Y-%m-%d %H:%M:%S')}"
                )
                self.detection_paused = new_paused
        except Exception as e:
            # 请求失败保持当前状态不变
            logger.warning(f"远程异响调度请求失败，保持当前状态: {e}")
    
    @staticmethod
    def _parse_schedule_time(time_str):
        # 尝试多种格式解析时间字符串，全部失败返回 None
        for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M',
                    '%Y-%m-%dT%H:%M:%S', '%Y-%m-%dT%H:%M'):
            try:
                return datetime.strptime(time_str, fmt)
            except ValueError:
                continue
        return None
    
    def _monitor_loop(self):
        """
        监控循环（线程主函数）
        
        持续检查各设备的音频目录，处理新生成的WAV文件。
        每分钟汇总一次结果进行上报。
        """
        # 确保根目录存在
        self.audio_dir.mkdir(parents=True, exist_ok=True)
        
        while self.running:
            try:
                # 热更新：定期从 DB 刷新可变配置
                self._reload_hot_config()
                # 远程异响调度：每60秒检查一次远程接口，控制 detection_paused
                self._check_detection_schedule()

                # 远程调度暂停时，从源头跳过整个扫描+推理+上报流程
                # 只保留热更新和调度检查，连目录遍历都不做
                if self.detection_paused:
                    time.sleep(self.check_interval)
                    continue

                # 扫描所有设备目录下的current文件夹
                for device_code in self.device_map.keys():
                    device_current_dir = self.audio_dir / device_code / "current"
                    
                    # 目录不存在则跳过
                    if not device_current_dir.exists():
                        continue
                    
                    for wav_file in device_current_dir.glob("*.wav"):
                        # 跳过已处理的文件
                        if wav_file in self.seen_files:
                            continue

                        # 检查文件是否完整
                        try:
                            stat_info = wav_file.stat()
                            file_age = time.time() - stat_info.st_mtime
                            file_size = stat_info.st_size

                            # 文件还在写入中，下次再检查（不加入 seen_files）
                            if file_age < 12.0:
                                continue

                            # 文件大小异常处理（60秒 × 16000Hz × 2字节 ≈ 1.9MB）
                            if file_size < 500_000:
                                if file_age > 20.0:
                                    # 确认是损坏文件，删除并标记
                                    try:
                                        wav_file.unlink()
                                        logger.debug(f"删除异常小文件: {wav_file.name} ({file_size / 1000:.1f}KB)")
                                    except Exception as e:
                                        logger.error(f"删除文件失败: {wav_file.name} | {e}")
                                    self.seen_files.add(wav_file)
                                continue

                            if file_size > 3_000_000:
                                if file_age > 20.0:
                                    # 文件过大，直接归档到日期目录（不丢弃，不加 seen）
                                    logger.warning(f"文件过大，直接归档: {wav_file.name} ({file_size / 1000:.1f}KB)")
                                    self._move_audio_to_date_dir(wav_file)
                                    self.seen_files.add(wav_file)
                                continue

                        except Exception as e:
                            logger.error(f"文件状态检查失败: {wav_file.name} | {e}")
                            continue

                        # 处理新文件
                        self._process_new_file(wav_file)

                        # 标记为已处理
                        self.seen_files.add(wav_file)
                
                # 检查是否需要进行周期性上报
                self._check_periodic_upload()
                
                # 检查聚合窗口是否到期，到期则执行聚合判定并推送
                if self.alert_aggregator:
                    self.alert_aggregator.check_and_flush()
                
                # 清理过大的已处理文件集合
                if len(self.seen_files) > 10000:
                    recent_files = sorted(self.seen_files, 
                                        key=lambda f: f.stat().st_mtime if f.exists() else 0)[-5000:]
                    self.seen_files = set(recent_files)
                
            except Exception as e:
                logger.error(f"监控循环错误: {e}")
            
            # 等待下一次检查
            time.sleep(self.check_interval)
    
    def _process_new_file(self, wav_file):
        """
        处理新的音频文件
        
        文件名格式: {project_id}_{device_code}_{时间戳}.wav
        例如: 92_1#-1_20251218142000.wav
        
        流程：
        1. 加载音频
        2. 计算能量判断设备状态
        3. 进行AE异常检测，记录重建误差
        4. 保存音频数据用于计算频谱图
        """
        try:
            # 从文件名解析device_code
            # 格式: {project_id}_{device_code}_{时间戳}.wav
            try:
                parts = wav_file.stem.split('_')
                if len(parts) >= 3:
                    device_code = parts[1]  # 第二部分是device_code
                else:
                    device_code = "unknown"
            except:
                device_code = "unknown"
            
            # ========================================
            # 泵启停状态检测（基于 PLC 查询）- 优先于模型检查
            # ----------------------------------------
            # 逻辑：通过 SCADA API 查询 PLC 点位判断泵是否运行
            # 作用：
            #   1. 冷启动模式下也能过滤停机时的音频（保证训练数据质量）
            #   2. 泵停机时跳过异常检测（避免无意义的检测）
            #   3. 记录设备状态用于后续上报
            # 依赖：pump_state_monitor + rtsp_config 中的 pump_status_plc 配置
            # ========================================
            device_status = "未知"
            pump_is_running = True  # 默认认为运行中（PLC 查询失败时的保守策略）
            
            # 获取该设备对应的流配置
            stream_config = self.device_map.get(device_code)
            
            if self.pump_state_monitor and stream_config:
                # 根据设备的 pump_name 找到关联的 PLC 点位配置
                pump_name = stream_config.pump_name
                pump_configs = self.pump_status_plc_configs.get(pump_name, [])
                
                if pump_configs:
                    # 遍历所有关联泵，只要有一个运行就认为设备在工作
                    any_pump_running = False
                    for pump_cfg in pump_configs:
                        point = pump_cfg.get("point", "")
                        name = pump_cfg.get("name", point)
                        pump_id = point
                        
                        # 查询泵状态（带 60 秒缓存，不会频繁请求 SCADA）
                        is_running, _ = self.pump_state_monitor.update_pump_state(pump_id, point, name)
                        if is_running:
                            any_pump_running = True
                    
                    pump_is_running = any_pump_running
                    device_status = "开机" if pump_is_running else "停机"
                    
                    # 泵全部停机时跳过（可通过配置禁用此行为）
                    # 冷启动和正常模式都适用，确保训练数据质量
                    if self.skip_detection_when_stopped and not pump_is_running:
                        logger.info(f"泵停机(PLC): {device_code} | 归档到过渡期目录（不用于训练）")
                        self._move_audio_to_transition_dir(wav_file, "stopped")
                        return
                    
                    # ========================================
                    # 过渡期检测（泵启停后15分钟内）
                    # ----------------------------------------
                    # 目的：过滤启停过程中的不稳定音频
                    # 确保训练数据只包含稳定运行期的音频
                    # ========================================
                    if self.skip_detection_when_stopped:
                        pump_in_transition, transition_pump_names = \
                            self.pump_state_monitor.check_pumps_transition(pump_configs)
                        
                        if pump_in_transition:
                            logger.info(f"泵过渡期: {device_code} | 归档到过渡期目录（不用于训练） | "
                                       f"过渡期泵: {', '.join(transition_pump_names)}")
                            self._move_audio_to_transition_dir(wav_file, "transition")
                            return
            
            # 获取该设备的预测器（懒加载模型）
            device_predictor = self.multi_predictor.get_predictor(device_code)
            if device_predictor is None:
                logger.info(f"冷启动模式（设备 {device_code} 无模型）: 归档 {wav_file.name}")
                self._move_audio_to_date_dir(wav_file)
                return
            # 加载音频
            try:
                y, sr = librosa.load(str(wav_file), sr=CFG.SR, mono=True)
            except Exception as e:
                logger.error(f"音频加载失败: {wav_file.name} | {e}")
                return
            
            # 记录状态到缓存（用于周期上报）
            # 注意：泵状态检测已在前面完成，这里只记录状态
            self.device_cache[device_code]["status"] = device_status
            
            # ========================================
            # 计算重建误差
            # ========================================
            error = self._compute_reconstruction_error(wav_file, device_predictor)
            
            if error is not None:
                self.device_cache[device_code]["errors"].append(error)
            
            # ========================================
            # 保存音频数据用于计算频谱图
            # ========================================
            self.device_cache[device_code]["audio_data"].append(y)
            
            # ========================================
            # 暂存文件路径，等待周期聚合判定后再归档
            # ========================================
            if "pending_files" not in self.device_cache[device_code]:
                self.device_cache[device_code]["pending_files"] = []
            self.device_cache[device_code]["pending_files"].append(wav_file)
            
            # ========================================
            # 记录到音频历史缓存（用于异常上下文捕获）
            # ========================================
            self.audio_file_history[device_code].append((datetime.now(), wav_file))
            
            # 初始化上次上报时间
            if self.device_cache[device_code]["last_upload"] is None:
                self.device_cache[device_code]["last_upload"] = datetime.now()
            
            # 获取阈值并判断结果
            threshold = self._get_threshold(device_code)
            
            # ========================================
            # 快速通道：连续多个文件误差极高时快速预警（暂时关闭）
            # 不走投票窗口，用于捕获突发性严重故障
            # ========================================
            # if error is not None and threshold is not None and threshold > 0:
            #     # 维护快速通道缓冲区
            #     if "fast_alert_buffer" not in self.device_cache[device_code]:
            #         self.device_cache[device_code]["fast_alert_buffer"] = []
            #     
            #     fast_buf = self.device_cache[device_code]["fast_alert_buffer"]
            #     # 连续性检查：误差超过 2x 阈值时记录，否则清空缓冲区
            #     if error > threshold * 2.0:
            #         fast_buf.append(error)
            #     else:
            #         fast_buf.clear()
            #     
            #     # 连续 3 个文件（~24秒）都超过 2x 阈值 → 触发快速预警
            #     FAST_ALERT_CONSECUTIVE = 3
            #     if len(fast_buf) >= FAST_ALERT_CONSECUTIVE:
            #         # 检查是否已触发过（避免重复告警）
            #         last_fast = self.device_cache[device_code].get("last_fast_alert_time")
            #         now = datetime.now()
            #         can_fast_alert = (last_fast is None or
            #                          (now - last_fast).total_seconds() > 300)  # 5分钟冷却
            #         
            #         if can_fast_alert:
            #             # 快速通道同样受抑制逻辑约束
            #             suppress = False
            #             # 泵过渡期抑制
            #             if self.pump_state_monitor and stream_config:
            #                 pump_name = stream_config.pump_name
            #                 pump_configs = self.pump_status_plc_configs.get(pump_name, [])
            #                 if pump_configs:
            #                     in_transition, _ = self.pump_state_monitor.check_pumps_transition(pump_configs)
            #                     if in_transition:
            #                         suppress = True
            #             # 人体检测抑制
            #             if self.human_detection_enabled and self.human_reader:
            #                 if self.human_reader.is_in_cooldown():
            #                     suppress = True
            #             # alert_enabled 开关
            #             if not self.alert_enabled:
            #                 suppress = True
            #             
            #             if not suppress:
            #                 avg_fast = float(np.mean(fast_buf))
            #                 logger.warning(
            #                     f"[!!] 快速通道触发: {device_code} | "
            #                     f"连续{len(fast_buf)}个文件异常 | "
            #                     f"平均误差={avg_fast:.6f} 阈值={threshold:.6f}")
            #                 self.device_cache[device_code]["last_fast_alert_time"] = now
            #                 fast_buf.clear()
            #                 # 标记快速预警，在下次周期上报时一并处理
            #                 self.device_cache[device_code]["fast_alert_pending"] = True
            
            if error is not None and threshold is not None:
                is_anomaly = error > threshold
                result_tag = "!!" if is_anomaly else "OK"
                logger.info(f"[{result_tag}] {device_code} | {wav_file.name} | "
                           f"误差={error:.6f} 阈值={threshold:.6f}")
            elif error is not None:
                logger.debug(f"文件预测: {wav_file.name} | 误差={error:.6f} | 阀值=未设置")
            else:
                logger.warning(f"预测跳过: {wav_file.name} | 误差计算失败")
            
        except Exception as e:
            logger.error(f"处理文件失败: {wav_file.name} | 错误: {e}")
    
    def _compute_reconstruction_error(self, wav_file, device_predictor):
        """
        计算单个音频文件的重建误差（Min-Max 标准化）
        
        使用8秒窗口、4秒步长切割音频，提取多个patches分别计算误差后取平均值。
        
        参数:
            wav_file: 音频文件路径
            device_predictor: 设备预测器实例
        
        返回:
            重建误差值（所有patches的平均MSE），失败返回None
        """
        try:
            import torch
            from predictor.utils import align_to_target
            
            # Min-Max 标准化参数
            global_min = device_predictor.global_min
            global_max = device_predictor.global_max
            
            # 加载音频
            y, _ = librosa.load(str(wav_file), sr=CFG.SR, mono=True)
            
            win_samples = int(CFG.WIN_SEC * CFG.SR)
            hop_samples = int(CFG.HOP_SEC * CFG.SR)
            
            if len(y) < win_samples:
                logger.warning(f"音频太短，无法提取patches: {wav_file.name}")
                return None
            
            patches = []
            for start in range(0, len(y) - win_samples + 1, hop_samples):
                window = y[start:start + win_samples]
                
                S = librosa.feature.melspectrogram(
                    y=window, sr=CFG.SR, n_fft=CFG.N_FFT,
                    hop_length=CFG.HOP_LENGTH, n_mels=CFG.N_MELS, power=2.0
                )
                S_db = librosa.power_to_db(S, ref=np.max)
                
                if S_db.shape[1] < CFG.TARGET_FRAMES:
                    S_db = np.pad(S_db, ((0, 0), (0, CFG.TARGET_FRAMES - S_db.shape[1])))
                else:
                    S_db = S_db[:, :CFG.TARGET_FRAMES]
                
                # Min-Max 标准化
                S_norm = (S_db - global_min) / (global_max - global_min + 1e-6)
                patches.append(S_norm.astype(np.float32))
            
            if not patches:
                logger.warning(f"未能提取任何patches: {wav_file.name}")
                return None
            
            arr = np.stack(patches, 0)
            arr = np.expand_dims(arr, 1)
            tensor = torch.from_numpy(arr)
            
            torch_device = device_predictor.torch_device
            tensor = tensor.to(torch_device)
            
            with torch.no_grad():
                recon = device_predictor.model(tensor)
                recon = align_to_target(recon, tensor)
                mse_per_patch = torch.mean((recon - tensor) ** 2, dim=[1, 2, 3])
                mean_mse = torch.mean(mse_per_patch).item()
            
            logger.debug(f"重建误差: {wav_file.name} | patches={len(patches)} | MSE={mean_mse:.6f}")
            return mean_mse
            
        except Exception as e:
            logger.error(f"计算重建误差失败: {wav_file.name} | {e}")
            return None
    
    def _check_periodic_upload(self):
        """
        检查是否需要进行周期性上报
        
        每分钟汇总一次各设备的检测结果并上报
        """
        # 远程调度暂停时跳过上报，与 _monitor_loop 中的推理拦截保持一致
        if self.detection_paused:
            return
        
        now = datetime.now()
        
        for device_code, cache in self.device_cache.items():
            # 检查上报时间间隔
            last_upload = cache.get("last_upload")
            if last_upload is None:
                continue
            
            elapsed = (now - last_upload).total_seconds()
            
            # 达到上报周期
            if elapsed >= self.segment_duration:
                # 获取该设备的流配置
                stream_config = self.device_map.get(device_code)
                
                # 计算平均重建误差
                errors = cache.get("errors", [])
                avg_error = float(np.mean(errors)) if errors else 0.0
                
                # 获取阈值
                threshold = self._get_threshold(device_code)
                
                # 判断当前周期是否异常（带容差区间）
                # 容差区间：避免边界值反复跳变，但灰区内仍与阈值比较
                if threshold:
                    upper_bound = threshold * (1 + self.tolerance_ratio)  # 确定异常边界
                    lower_bound = threshold * (1 - self.tolerance_ratio)  # 确定正常边界
                    
                    if avg_error > upper_bound:
                        # 超过上边界 -> 确定异常
                        is_current_anomaly = True
                    elif avg_error < lower_bound:
                        # 低于下边界 -> 确定正常
                        is_current_anomaly = False
                    else:
                        # 灰区 -> 与阈值比较（避免异常状态延长）
                        is_current_anomaly = avg_error > threshold
                    
                    # 记录本次判定结果
                    self.last_single_anomaly[device_code] = is_current_anomaly
                else:
                    is_current_anomaly = False
                
                # ========================================
                # 滑动窗口投票：5次中有3次异常才判定为异常
                # ========================================
                if device_code not in self.detection_history:
                    self.detection_history[device_code] = []
                
                # 记录本次检测结果
                self.detection_history[device_code].append(is_current_anomaly)
                
                # 保持窗口大小
                if len(self.detection_history[device_code]) > self.voting_window_size:
                    self.detection_history[device_code].pop(0)
                
                # 投票判定最终异常状态
                if self.voting_enabled and len(self.detection_history[device_code]) >= self.voting_window_size:
                    anomaly_count = sum(self.detection_history[device_code])
                    is_anomaly = anomaly_count >= self.voting_threshold
                    window_info = f"窗口[{anomaly_count}/{self.voting_window_size}]"
                else:
                    is_anomaly = is_current_anomaly
                    window_info = "窗口未满"
                
                # ========================================
                # 状态变更检测
                # trigger_alert: 仅在 正常(或初始) -> 异常 时为True
                # 冷却逻辑已移至 AlertAggregator 内部处理
                # ========================================
                last_is_anomaly = self.last_report_status.get(device_code)
                trigger_alert = False
                
                if is_anomaly:
                    # 只有状态变化(正常->异常) 才触发报警流程
                    # 泵过渡期检查已在 _process_new_file 中完成，过渡期内的文件不会进入 pending
                    if last_is_anomaly is None or not last_is_anomaly:
                        if self.alert_enabled:
                            # 人体检测抑制：任意摄像头检测到人则不报警
                            if self.human_detection_enabled and self.human_reader:
                                if self.human_reader.is_in_cooldown():
                                    trigger_alert = False
                                    status_info = self.human_reader.get_status_info()
                                    logger.info(f"人体检测抑制: {device_code} | {status_info}，跳过报警")
                                else:
                                    trigger_alert = True
                            else:
                                trigger_alert = True
                        else:
                            trigger_alert = False
                            logger.debug(f"异常告警已禁用，跳过告警: {device_code}")
                
                # 获取运行状态
                running_status = cache.get("status", "未知")
                
                # 获取进水流量
                inlet_flow = self._get_inlet_flow(stream_config) if stream_config else 0.0
                
                # 计算本次频谱图
                audio_data = cache.get("audio_data", [])
                freq_db = self._compute_frequency_spectrum(audio_data)
                
                # 保存到频谱图历史（只保存dB值）
                if self.freq_history_enabled and freq_db:
                    self.freq_history[device_code].append((now, freq_db))
                    # 清理过期历史
                    cutoff = now - timedelta(minutes=self.freq_history_minutes)
                    self.freq_history[device_code] = [
                        (t, d) for t, d in self.freq_history[device_code]
                        if t > cutoff
                    ]
                
                # 计算历史频谱图平均值（normal_frequency_middle）
                freq_middle_db = self._compute_frequency_middle(device_code)
                
                # ========================================
                # 异常分类：只在状态从正常变为异常时进行分类
                # 持续异常期间沿用上次分类结果，保持一致性
                # ========================================
                anomaly_type_code = 6  # 默认：未分类异常
                type_name = "未分类异常"
                
                if is_anomaly and audio_data:
                    # 检查是否是新的异常（从正常变为异常）
                    is_new_anomaly = (last_is_anomaly is None or not last_is_anomaly)
                    
                    if is_new_anomaly:
                        # 新异常：进行分类并锁定结果
                        try:
                            from core.anomaly_classifier import classify_anomaly
                            if len(audio_data) > 0:
                                y = audio_data[-1] if isinstance(audio_data[-1], np.ndarray) else np.array(audio_data[-1])
                                anomaly_type_code, type_name, confidence = classify_anomaly(y, sr=16000)
                                # 锁定分类结果
                                self.locked_anomaly_type[device_code] = (anomaly_type_code, type_name)
                                logger.info(f"异常分类（新异常）: {type_name} (code={anomaly_type_code}, 置信度={confidence:.2f})")
                        except Exception as e:
                            logger.warning(f"异常分类失败: {e}")
                    else:
                        # 持续异常：沿用锁定的分类结果
                        if device_code in self.locked_anomaly_type:
                            anomaly_type_code, type_name = self.locked_anomaly_type[device_code]
                            logger.debug(f"异常分类（沿用）: {type_name} (code={anomaly_type_code})")
                else:
                    # 状态正常时清除锁定的分类结果
                    if device_code in self.locked_anomaly_type:
                        del self.locked_anomaly_type[device_code]
                
                # 上报逻辑
                if self.push_enabled and stream_config:
                    # 预读异常音频base64：在文件被归档/清空之前立即读取
                    # 解决竞态问题：异步推送或聚合器延迟推送时文件可能已被移走
                    pre_read_wav_b64 = ""
                    if trigger_alert:
                        try:
                            current_pending = cache.get("pending_files", [])
                            if current_pending and current_pending[0].exists():
                                with open(current_pending[0], "rb") as f:
                                    pre_read_wav_b64 = base64.b64encode(f.read()).decode('utf-8')
                                logger.debug(f"预读异常音频成功: {current_pending[0].name} | size={len(pre_read_wav_b64)}")
                        except Exception as e:
                            logger.warning(f"预读异常音频失败: {e}")

                    if trigger_alert and self.alert_aggregator:
                        # 报警走聚合器：跨设备聚合判定 + 分类型冷却
                        # 聚合器会在窗口到期后决定是否真正推送
                        self.alert_aggregator.submit_alert(
                            plant_name=stream_config.plant_name,
                            device_code=device_code,
                            anomaly_type_code=anomaly_type_code,
                            push_kwargs=dict(
                                stream_config=stream_config,
                                device_code=device_code,
                                is_anomaly=is_anomaly,
                                trigger_alert=True,
                                abnormal_score=avg_error,
                                score_threshold=threshold,
                                running_status=running_status,
                                inlet_flow=inlet_flow,
                                freq_db=freq_db,
                                freq_middle_db=freq_middle_db,
                                anomaly_type_code=anomaly_type_code,
                                abnormal_wav_b64=pre_read_wav_b64
                            )
                        )
                    else:
                        # 非报警（心跳）或聚合器不可用 -> 提交到线程池异步推送
                        self._push_executor.submit(
                            self._push_detection_result,
                            stream_config=stream_config,
                            device_code=device_code,
                            is_anomaly=is_anomaly,
                            trigger_alert=trigger_alert,
                            abnormal_score=avg_error,
                            score_threshold=threshold,
                            running_status=running_status,
                            inlet_flow=inlet_flow,
                            freq_db=freq_db,
                            freq_middle_db=freq_middle_db,
                            anomaly_type_code=anomaly_type_code,
                            abnormal_wav_b64=pre_read_wav_b64
                        )
                    
                    # 更新上一次状态
                    self.last_report_status[device_code] = is_anomaly
                
                # 日志记录
                thr_str = f"{threshold:.6f}" if threshold else "未设置"
                # 报警去向说明
                if trigger_alert and self.alert_aggregator:
                    alert_dest = "-> 聚合器"
                elif trigger_alert:
                    alert_dest = "-> 直接推送"
                else:
                    alert_dest = ""
                
                # 使用设备名作为标识，增加视觉分隔
                cam_label = ""
                if stream_config:
                    cam_label = f"({stream_config.camera_name})"
                
                result_emoji = "!!" if is_anomaly else "OK"
                alert_str = f"报警=是 {alert_dest}" if trigger_alert else "报警=否"
                
                logger.info(
                    f"[{result_emoji}] {device_code}{cam_label} | "
                    f"误差={avg_error:.6f} 阈值={thr_str} | "
                    f"{window_info} | {running_status} | "
                    f"{'异常' if is_anomaly else '正常'} | {alert_str}"
                )
                
                # ========================================
                # 根据投票结果归档文件
                # 用投票后的 is_anomaly 决定归档，避免单次波动误归
                # ========================================
                pending_files = cache.get("pending_files", [])

                # 检查是否是新的异常（从正常变为异常）
                is_new_anomaly = is_anomaly and (last_is_anomaly is None or not last_is_anomaly)

                # 更新异常上下文捕获状态机
                self._update_anomaly_capture_state(device_code, is_anomaly, is_new_anomaly,
                                                   avg_error, threshold, now, pending_files)

                if pending_files:
                    if is_anomaly:
                        # 异常文件由上下文捕获状态机统一管理
                        # 状态机会在收集完前+后文件后一次性保存到 anomaly_detected
                        pass
                    else:
                        # 正常 -> 移到日期目录归档
                        for f in pending_files:
                            self._move_audio_to_date_dir(f)
                        # 状态恢复正常时，清除保存冷却时间
                        if device_code in self.last_anomaly_save_time:
                            del self.last_anomaly_save_time[device_code]
                
                # 重置缓存
                cache["errors"] = []
                cache["audio_data"] = []
                cache["pending_files"] = []
                cache["last_upload"] = now
                logger.info("─" * 60)
    
    def _update_anomaly_capture_state(self, device_code, is_anomaly,
                                       is_new_anomaly, avg_error,
                                       threshold, now, pending_files):
        """
        异常上下文捕获状态机

        状态流转：
        1. 无状态 + 新异常 -> 触发捕获，回溯 audio_file_history 获取前置文件
        2. 已触发 + 异常持续 -> 持续收集异常文件
        3. 已触发 + 异常结束 -> 开始收集后续文件（post阶段）
        4. post阶段 + 时间到 -> 保存所有文件到 anomaly_detected

        所有异常文件（从触发到结束）统一保存到 anomaly_detected 目录，
        包含 metadata.json 记录误差/阈值等信息。
        """
        state = self.anomaly_capture_state.get(device_code)

        if is_new_anomaly and state is None:
            # 新异常触发：回溯获取前置文件
            anomaly_cutoff = now - timedelta(minutes=1)
            pre_cutoff = now - timedelta(minutes=self.context_pre_minutes + 1)

            pre_files = []
            anomaly_files = []

            for ts, fpath in self.audio_file_history[device_code]:
                if not fpath.exists():
                    continue
                if ts >= anomaly_cutoff:
                    anomaly_files.append(fpath)
                elif ts >= pre_cutoff:
                    pre_files.append(fpath)

            # 兜底：如果回溯没找到，把当前 pending 加入
            if not anomaly_files and pending_files:
                anomaly_files = [f for f in pending_files if f.exists()]

            self.anomaly_capture_state[device_code] = {
                "trigger_time": now,
                "avg_error": avg_error,
                "threshold": threshold,
                "pre_files": pre_files,
                "anomaly_files": anomaly_files,
                "post_files": [],
                "anomaly_ended": False,
                "post_start_time": None
            }

            logger.info(f"异常上下文捕获已触发: {device_code} | "
                       f"前置={len(pre_files)}个 | 异常={len(anomaly_files)}个")

        elif state is not None:
            if is_anomaly and not state["anomaly_ended"]:
                # 异常持续中：继续收集异常文件
                for f in pending_files:
                    if f.exists():
                        state["anomaly_files"].append(f)
            else:
                # 异常结束（或之前已结束，在收集后续文件）
                if not state["anomaly_ended"]:
                    state["anomaly_ended"] = True
                    state["post_start_time"] = now
                    logger.info(f"异常结束，开始收集后续文件: {device_code} | "
                               f"异常文件共{len(state['anomaly_files'])}个 | "
                               f"等待{self.context_post_minutes}分钟")

                # 收集后续文件
                for f in pending_files:
                    if f.exists():
                        state["post_files"].append(f)

                # 检查 post 阶段是否到时间
                elapsed_post = (now - state["post_start_time"]).total_seconds() / 60
                if elapsed_post >= self.context_post_minutes:
                    self._save_anomaly_context(device_code, state)
                    del self.anomaly_capture_state[device_code]
                    self.last_anomaly_save_time[device_code] = now
    
    def _save_anomaly_context(self, device_code: str, state: dict):
        """
        保存异常上下文文件到独立目录
        
        目录结构:
        data/anomaly_detected/{device_code}/{异常文件名(不含扩展名)}/
        ├── 92_1#-1_20260130140313.wav  (保持原文件名)
        ├── ...
        └── metadata.json
        
        metadata.json 字段说明:
        - before_trigger: 触发前 1~(N+1) 分钟的文件（正常时期，用于对比）
        - at_trigger: 触发时刻前 1 分钟内的文件（异常开始出现的时期）
        - after_trigger: 触发后 N 分钟的文件（不管异常是否恢复）
        
        参数:
            device_code: 设备编号
            state: 捕获状态字典
        """
        import shutil
        import json
        
        try:
            # 获取第一个异常文件名作为文件夹名
            anomaly_files = state.get("anomaly_files", [])
            if not anomaly_files:
                logger.warning(f"无异常文件，跳过保存: {device_code}")
                return
            
            # 用第一个异常文件名（不含扩展名）作为文件夹名
            first_anomaly = anomaly_files[0]
            folder_name = first_anomaly.stem  # 如 92_1#-1_20260130140313
            save_dir = self.save_anomaly_dir / device_code / folder_name
            save_dir.mkdir(parents=True, exist_ok=True)
            
            # 收集所有文件名（使用新命名）
            all_files = {"before_trigger": [], "at_trigger": [], "after_trigger": []}
            
            # 移动前置文件（来自日期目录，移动后原位置不再保留）
            for fpath in state.get("pre_files", []):
                if fpath.exists():
                    dest = save_dir / fpath.name
                    shutil.move(str(fpath), str(dest))  # 移动而非复制，避免重复
                    all_files["before_trigger"].append(fpath.name)
            
            # 移动触发时刻文件（保持原名）
            for fpath in anomaly_files:
                if fpath.exists():
                    dest = save_dir / fpath.name
                    shutil.move(str(fpath), str(dest))
                    all_files["at_trigger"].append(fpath.name)
            
            # 移动后续文件（保持原名）
            for fpath in state.get("post_files", []):
                if fpath.exists():
                    dest = save_dir / fpath.name
                    shutil.move(str(fpath), str(dest))
                    all_files["after_trigger"].append(fpath.name)
            
            # 生成精简的元数据文件
            trigger_time = state.get("trigger_time")
            metadata = {
                "device_code": device_code,
                "trigger_time": trigger_time.strftime("%Y-%m-%d %H:%M:%S") if trigger_time else None,
                "avg_error": round(state.get("avg_error", 0.0), 6),
                "threshold": round(state.get("threshold", 0.0), 6),
                "files": all_files
            }
            
            metadata_path = save_dir / "metadata.json"
            with open(metadata_path, 'w', encoding='utf-8') as f:
                json.dump(metadata, f, ensure_ascii=False, indent=2)
            
            total = sum(len(v) for v in all_files.values())
            logger.warning(f"异常上下文已保存: {device_code}/{folder_name} | "
                          f"共{total}个文件 (前{len(all_files['before_trigger'])}+异常{len(all_files['at_trigger'])}+后{len(all_files['after_trigger'])})")
            
        except Exception as e:
            logger.error(f"保存异常上下文失败: {device_code} | {e}")
    
    def _compute_frequency_middle(self, device_code):
        """
        计算历史频谱图平均值（normal_frequency_middle）
        
        参数:
            device_code: 设备编号
        
        返回:
            dB值列表的平均值
        """
        history = self.freq_history.get(device_code, [])
        if not history or len(history) < 2:
            return []
        
        try:
            # 收集所有历史dB值（只保留长度一致的）
            all_db = []
            ref_len = len(history[0][1])  # 使用第一条记录的长度作为参考
            
            for _, db in history:
                if len(db) == ref_len:
                    all_db.append(db)
            
            if not all_db:
                return []
            
            # 计算各频率点的平均dB
            avg_db = np.mean(all_db, axis=0).tolist()
            return avg_db
            
        except Exception as e:
            logger.error(f"计算频谱图历史平均失败: {e}")
            return []
    
    def _get_threshold(self, device_code):
        """
        获取指定设备的阈值
        
        优先级：
        1. 从 multi_predictor 获取设备对应模型的阈值
        2. 使用配置文件中的默认阈值
        3. 返回0.0（不进行异常判定）
        
        参数:
            device_code: 设备编号（如 "LT-1"）
        
        返回:
            阈值，未找到返回默认值或0.0
        """
        # 方式1：从 multi_predictor 获取设备阈值
        thr = self.multi_predictor.get_threshold(device_code)
        if thr is not None:
            logger.debug(f"阈值来源: {device_code} -> 设备模型阈值 = {thr:.6f}")
            return thr
        
        # 方式2：使用配置中的默认阈值
        default_threshold = self.config.get('prediction', {}).get('default_threshold', 0.0)
        
        if default_threshold > 0:
            logger.debug(f"阈值来源: {device_code} -> 配置默认值 = {default_threshold:.6f}")
            return default_threshold
        
        # 首次找不到阈值时记录警告
        if device_code not in getattr(self, '_threshold_warned', set()):
            if not hasattr(self, '_threshold_warned'):
                self._threshold_warned = set()
            self._threshold_warned.add(device_code)
            logger.warning(f"未找到阈值: {device_code}，将跳过异常判定")
        
        return 0.0
    
    def _get_inlet_flow(self, stream_config: RTSPStreamConfig) -> float:
        """
        获取进水流量（使用实时数据接口）
        
        使用 current-data 接口直接获取最新一条数据
        
        参数:
            stream_config: 流配置
        
        返回:
            进水流量值，失败返回0.0
        """
        if not self.scada_enabled or not stream_config:
            logger.debug(f"流量跳过: scada_enabled={self.scada_enabled}, stream_config={stream_config}")
            return 0.0
        
        # 获取PLC数据点位
        plc_address = stream_config.get_flow_plc_address()
        if not plc_address:
            logger.debug(f"流量跳过(无PLC地址): {stream_config.device_code} | pump_name='{stream_config.pump_name}'")
            return 0.0
        
        # 使用水厂配置的 project_id
        project_id = stream_config.project_id
        
        try:
            # 当前时间戳（毫秒）
            now_ms = int(datetime.now().timestamp() * 1000)
            
            # 请求参数
            params = {"time": now_ms}
            
            # 请求体：使用实时数据接口格式
            request_body = [
                {
                    "deviceId": "1",
                    "deviceItems": plc_address,
                    "deviceName": f"流量_{stream_config.pump_name}",
                    "project_id": project_id
                }
            ]
            
            # 请求头
            headers = {
                "Content-Type": "application/json",
                "JWT-TOKEN": self.scada_jwt
            }
            
            logger.debug(f"流量请求: {stream_config.device_code} | project_id={project_id} | plc={plc_address}")
            
            # 发送 POST 请求到实时接口
            response = requests.post(
                self.scada_realtime_url,
                params=params,
                json=request_body,
                headers=headers,
                timeout=self.scada_timeout
            )
            
            if response.status_code == 200:
                data = response.json()
                if data.get("code") == 200:
                    if data.get("data"):
                        # 获取第一条数据（实时接口只返回最新一条）
                        latest = data["data"][0]
                        if "val" in latest:
                            flow = float(latest["val"])
                            logger.debug(f"流量获取成功: {stream_config.device_code} | 流量={flow}")
                            return flow
                        else:
                            logger.debug(f"流量数据无val字段: {stream_config.device_code}")
                    else:
                        # API正常返回但无数据
                        logger.debug(f"流量查询无数据: {stream_config.device_code}")
                else:
                    logger.warning(f"流量API返回异常: {stream_config.device_code} | code={data.get('code')} | msg={data.get('msg')}")
            else:
                logger.warning(f"流量HTTP错误: {stream_config.device_code} | status={response.status_code}")
            
        except Exception as e:
            logger.warning(f"流量获取异常: {stream_config.device_code} | {e}")
        
        return 0.0
    
    def _compute_frequency_spectrum(self, audio_data):
        """
        计算频谱图数据
        
        将1分钟内的多个8秒音频片段合并，计算整体FFT
        
        参数:
            audio_data: 音频数据列表
        
        返回:
            dB值列表（0-8000Hz均匀分布，共400个点）
        """
        if not audio_data:
            return []
        
        try:
            # 合并所有音频片段
            combined = np.concatenate(audio_data)
            
            # 计算FFT
            n = len(combined)
            fft_result = np.fft.rfft(combined)
            freqs = np.fft.rfftfreq(n, 1.0 / CFG.SR)
            
            # 计算幅度（转换为dB）
            magnitude = np.abs(fft_result)
            # 避免log(0)
            magnitude = np.maximum(magnitude, 1e-10)
            db = 20 * np.log10(magnitude)
            
            # 降采样到400个点（0-8000Hz均匀分布）
            max_freq = 8000
            num_points = 400
            
            db_list = []
            
            for i in range(num_points):
                # 计算目标频率（0到8000Hz均匀分布）
                target_freq = (i / (num_points - 1)) * max_freq
                # 找到最接近的频率索引
                idx = np.argmin(np.abs(freqs - target_freq))
                if idx < len(freqs):
                    db_list.append(float(db[idx]))
            
            return db_list
            
        except Exception as e:
            logger.error(f"计算频谱图失败: {e}")
            return []
    
    def _push_detection_result(self, stream_config, device_code,
                               is_anomaly, trigger_alert, abnormal_score, score_threshold,
                               running_status, inlet_flow,
                               freq_db, freq_middle_db=None,
                               anomaly_type_code=6,
                               abnormal_wav_b64=""):
        """
        推送检测结果到远程服务器
        
        上报格式符合用户要求：
        - 包含频谱图数据（this_frequency + normal_frequency_middle）
        - 包含启停状态和进水流量
        - sound_detection.status 和 abnormalwav 仅在 trigger_alert=True 时上报异常(0)和base64，否则为正常(1)和空
        
        参数:
            stream_config: 流配置
            device_code: 设备编号
            is_anomaly: 实际检测是否异常
            trigger_alert: 是否触发报警（仅在新异常产生时为True）
            abnormal_score: 平均重建误差
            score_threshold: 阈值
            running_status: 启停状态
            inlet_flow: 进水流量
            freq_db: 本次频谱图dB值列表
            freq_middle_db: 历史平均频谱图dB值列表
            abnormal_wav_b64: 预读的异常音频base64编码（在文件归档前读取，避免竞态）
        """
        try:
            import time as time_module
            
            # 获取设备名称
            camera_name = stream_config.camera_name
            
            # 从 push_base_urls 列表构建推送目标（每个 base_url + /{project_id}）
            if not self.push_base_urls:
                logger.warning(f"未配置push_base_urls: {device_code}")
                return
            
            # 获取该设备对应的 project_id，用于拼接最终推送URL
            project_id = stream_config.project_id
            
            # 构建推送消息
            request_time = int(time_module.time() * 1000)
            
            # 异常音频已通过参数 abnormal_wav_b64 传入（调用方在文件归档前预读）
            # 如果调用方未提供，记录警告以便排查
            if trigger_alert and not abnormal_wav_b64:
                logger.warning(f"报警推送但无异常音频数据: {device_code}")

            # 决定 sound_detection.status
            # 只有在 trigger_alert=True 时才报 0 (异常)
            # 其他情况（正常、持续异常）都报 1 (正常/空) -- 根据用户要求: "正常情况下 ... 是空 就行"
            report_status = 0 if trigger_alert else 1
            
            payload = {
                "message": {
                    # 通道信息
                    "channelInfo": {"name": camera_name},
                    # 请求时间戳
                    "requestTime": request_time,
                    # 分类信息
                    "classification": {
                        "level_one": 2,                   # 音频检测大类
                        "level_two": anomaly_type_code    # 异常类型小类（6=未分类, 7=轴承, 8=气蚀, 9=松动, 10=叶轮, 11=阀件）
                    },
                    # 技能信息
                    "skillInfo": {"name": "异响检测"},
                    # 声音检测数据
                    "sound_detection": {
                        # 异常音频
                        "abnormalwav": abnormal_wav_b64,
                        # 状态：0=异常（仅新异常）, 1=正常（或持续异常）
                        "status": report_status,
                        # 设备状态信息
                        "condition": {
                            "running_status": "运行中" if running_status == "开机" else "停机中",
                            "inlet_flow": inlet_flow
                        },
                        # 得分信息
                        "score": {
                            "abnormal_score": abnormal_score,    # 当前1分钟平均重构误差
                            "score_threshold": score_threshold   # 该设备异常阀值
                        },
                        # 频谱图数据
                        "frequency": {
                            "this_frequency": freq_db,                    # 当前1分钟的频谱图
                            "normal_frequency_middle": freq_middle_db or [],  # 过去10分钟的频谱图平均
                        }
                    }
                }
            }
            
            # 遍历所有推送基地址，逐个拼接 project_id 发送（各目标互不影响）
            push_targets = [
                (f"{item['url']}/{project_id}", item['label'])
                for item in self.push_base_urls
            ]
            
            # 逐个目标推送（各目标互不影响）
            for target_url, target_label in push_targets:
                self._send_to_target(
                    target_url, target_label, payload,
                    device_code, camera_name, trigger_alert, abnormal_score
                )
            
        except Exception as e:
            logger.error(f"推送通知异常: {e}")
    
    def _send_to_target(self, target_url, target_label, payload,
                        device_code, camera_name, trigger_alert, abnormal_score):
        # 向单个推送目标发送数据（含重试逻辑）
        # 各目标独立调用，某个目标失败不影响其他目标
        import time as time_module
        
        push_success = False
        for attempt in range(self.push_retry_count + 1):
            try:
                response = requests.post(
                    target_url,
                    json=payload,
                    timeout=self.push_timeout,
                    headers={"Content-Type": "application/json"}
                )
                
                if response.status_code == 200:
                    alert_tag = "报警" if trigger_alert else "心跳"
                    logger.info(
                        f"    [{alert_tag}][{target_label}] {device_code}({camera_name}) | "
                        f"误差={abnormal_score:.6f}"
                    )
                    push_success = True
                    break
                else:
                    logger.warning(
                        f"推送失败[{target_label}]: {device_code} | URL={target_url} | "
                        f"状态码={response.status_code} | 内容={response.text[:100]}"
                    )
                    
            except requests.exceptions.Timeout:
                logger.warning(f"推送超时[{target_label}]: {device_code} | URL={target_url} | 尝试 {attempt + 1}/{self.push_retry_count + 1}")
            except requests.exceptions.RequestException as e:
                logger.warning(f"推送异常[{target_label}]: {device_code} | URL={target_url} | {e}")
            
            # 重试间隔
            if attempt < self.push_retry_count:
                time_module.sleep(1)
        
        if not push_success:
            logger.error(f"推送失败[{target_label}]: {device_code} | URL={target_url} | 已达最大重试次数")
    
    def _move_audio_to_date_dir(self, wav_file):
        """
        将音频移动到日期目录归档
        
        目录结构：
        deploy_pickup/data/{device_code}/{日期}/{文件名}
        
        参数:
            wav_file: 音频文件路径
        """
        try:
            import shutil
            
            # 从文件名提取device_code和日期
            # 格式: 92_1#-1_20251218142000.wav
            match = re.match(r'\d+_(.+)_(\d{8})\d{6}\.wav', wav_file.name)
            if not match:
                logger.warning(f"无法从文件名提取信息: {wav_file.name}")
                return
            
            device_code = match.group(1)  # 如 1#-1
            date_str = match.group(2)     # YYYYMMDD
            
            # 构建目标目录: data/{device_code}/{日期}/
            date_dir = self.audio_dir / device_code / date_str
            date_dir.mkdir(parents=True, exist_ok=True)
            
            # 移动文件
            dest_file = date_dir / wav_file.name
            shutil.move(str(wav_file), str(dest_file))
            logger.debug(f"音频已归档: {device_code}/{date_str}/{wav_file.name}")
            
        except Exception as e:
            logger.error(f"移动音频失败: {wav_file.name} | 错误: {e}")
    
    def _move_audio_to_transition_dir(self, wav_file, reason):
        """
        将泵停机/过渡期的音频移动到过渡期目录
        
        目录结构：
        deploy_pickup/data/{device_code}/pump_transition/{文件名}
        
        这些音频不会被用于模型训练，但保留用于分析调试
        
        参数:
            wav_file: 音频文件路径
            reason: 原因标识（stopped=停机, transition=过渡期）
        """
        try:
            import shutil
            
            # 从文件名提取device_code
            # 格式: 92_1#-1_20251218142000.wav
            match = re.match(r'\d+_(.+)_\d{14}\.wav', wav_file.name)
            if not match:
                logger.warning(f"无法从文件名提取信息: {wav_file.name}")
                return
            
            device_code = match.group(1)  # 如 1#-1
            
            # 构建目标目录: data/{device_code}/pump_transition/
            transition_dir = self.audio_dir / device_code / "pump_transition"
            transition_dir.mkdir(parents=True, exist_ok=True)
            
            # 移动文件
            dest_file = transition_dir / wav_file.name
            shutil.move(str(wav_file), str(dest_file))
            logger.debug(f"过渡期音频已归档: {device_code}/pump_transition/{wav_file.name} ({reason})")
            
        except Exception as e:
            logger.error(f"移动过渡期音频失败: {wav_file.name} | 错误: {e}")
    
    def _cleanup_old_files(self, days: int = 7):
        """
        清理超过指定天数的正常音频文件
        
        清理规则：
        - 只清理日期归档目录（data/{device_code}/{日期}/）
        - 保留current目录和anomaly_detected目录
        - 超过days天的文件删除
        
        参数:
            days: 保留天数，默认7天
        """
        try:
            import shutil
            from datetime import datetime, timedelta
            
            # 计算截止日期
            cutoff_date = datetime.now() - timedelta(days=days)
            cutoff_str = cutoff_date.strftime("%Y%m%d")
            
            deleted_count = 0
            
            # 遍历每个设备目录
            for device_dir in self.audio_dir.iterdir():
                if not device_dir.is_dir():
                    continue
                
                # 检查是否是日期目录（跳过current和其他特殊目录）
                for subdir in device_dir.iterdir():
                    if not subdir.is_dir():
                        continue
                    
                    # 跳过current目录
                    if subdir.name == "current":
                        continue
                    
                    # 检查是否为日期目录（YYYYMMDD格式）
                    if not re.match(r'^\d{8}$', subdir.name):
                        continue
                    
                    # 如果日期早于截止日期，删除整个目录
                    if subdir.name < cutoff_str:
                        try:
                            shutil.rmtree(subdir)
                            deleted_count += 1
                            logger.debug(f"清理过期目录: {device_dir.name}/{subdir.name}")
                        except Exception as e:
                            logger.error(f"删除目录失败: {subdir} | {e}")
            
            if deleted_count > 0:
                logger.info(f"清理完成: 共删除{deleted_count}个过期目录")
                
        except Exception as e:
            logger.error(f"清理过期文件失败: {e}")


class PickupMonitoringSystem:
    """
    拾音器监控系统
    
    管理FFmpeg进程和监控线程
    """
    
    def __init__(self, db_path=None, yaml_config=None):
        """
        初始化监控系统

        参数:
            db_path: SQLite 数据库路径（为 None 时使用默认路径 config/pickup_config.db）
            yaml_config: 若不为 None，则直接使用该 dict 作为配置（跳过 DB）
        """
        self.db_path = db_path
        self.config_manager = None

        if yaml_config is not None:
            # 直接使用传入的 YAML 配置字典
            self.config = yaml_config
            print(f"配置源: YAML (外部传入)")
        else:
            # 从 SQLite DB 加载
            actual_db = Path(db_path) if db_path else get_db_path()
            if not actual_db.exists():
                raise FileNotFoundError(
                    f"\u914d\u7f6e\u6570\u636e\u5e93\u4e0d\u5b58\u5728: {actual_db}\n"
                    f"\u8bf7\u5148\u8fd0\u884c\u8fc1\u79fb\u811a\u672c: python tool/migrate_yaml_to_db.py"
                )
            self.config_manager = ConfigManager(str(actual_db))
            print(f"配置源: SQLite ({actual_db})")
            self.config = self._load_config()
        
        # 冷启动模式标记
        self.cold_start_mode = False
        
        # 初始化多模型预测器（支持每个设备独立模型）
        self.multi_predictor = MultiModelPredictor()
        self.predictor = None  # 兼容性保留，已废弃
        
        # 从配置中注册所有设备的模型目录映射
        print("\n正在初始化多模型预测器...")
        for plant in self.config.get('plants', []):
            if not plant.get('enabled', False):
                continue
            for stream in plant.get('rtsp_streams', []):
                device_code = stream.get('device_code', '')
                model_subdir = stream.get('model_subdir', device_code)
                if device_code and model_subdir:
                    self.multi_predictor.register_device(device_code, model_subdir)
                    print(f"  注册设备: {device_code} -> models/{model_subdir}/")
        
        print(f"已注册 {len(self.multi_predictor.registered_devices)} 个设备模型映射")
        
        # 进程和监控器列表
        self.ffmpeg_processes = []
        self.monitors = []
        
        # 信号处理
        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)
    
    def _load_config(self):
        """
        从 SQLite DB 加载配置
        
        返回:
            Dict: 配置字典
        """
        config = self.config_manager.get_full_config()
        logger.info("配置已从 SQLite 加载")
        return config
    
    def _parse_rtsp_streams(self):
        """
        解析配置文件中的RTSP流信息
        
        返回:
            List[RTSPStreamConfig]: RTSP流配置列表
        """
        streams = []
        
        plants = self.config.get('plants', [])
        if not plants:
            raise ValueError("配置文件中未找到水厂配置")
        
        for plant in plants:
            plant_name = plant.get('name')
            if not plant_name:
                print("警告: 跳过未命名的区域配置")
                continue
            
            # 检查是否启用该水厂（默认启用以兼容旧配置）
            if not plant.get('enabled', True):
                logger.debug(f"跳过禁用的水厂: {plant_name}")
                continue
            
            # 获取流量PLC配置
            flow_plc = plant.get('flow_plc', {})
            
            # 获取该水厂的project_id（每个plant有自己的project_id）
            project_id = plant.get('project_id', 92)
            logger.info(f"加载区域配置: {plant_name} | project_id={project_id}")
            
            rtsp_streams = plant.get('rtsp_streams', [])
            for stream in rtsp_streams:
                url = stream.get('url')
                channel = stream.get('channel')
                camera_name = stream.get('name', '')
                device_code = stream.get('device_code', '')
                pump_name = stream.get('pump_name', '')
                
                if not url or channel is None:
                    print(f"警告: 跳过不完整的RTSP流配置 (区域: {plant_name})")
                    continue
                
                streams.append(RTSPStreamConfig(
                    plant_name=plant_name,
                    rtsp_url=url,
                    channel=channel,
                    camera_name=camera_name,
                    device_code=device_code,
                    pump_name=pump_name,
                    flow_plc=flow_plc,
                    project_id=project_id
                ))
        
        return streams
    
    def start(self):
        """
        启动监控系统
        """
        print("=" * 70)
        print("拾音器异响检测系统")
        print("=" * 70)
        
        # 解析流配置
        streams = self._parse_rtsp_streams()
        print(f"\n共配置 {len(streams)} 个拾音设备:")
        for stream in streams:
            print(f"  - {stream.device_code} | {stream.camera_name}")
        
        # 启动FFmpeg进程
        print("\n启动FFmpeg进程...")
        for stream in streams:
            ffmpeg = FFmpegProcess(stream, CFG.AUDIO_DIR, self.config)
            if ffmpeg.start():
                self.ffmpeg_processes.append(ffmpeg)
            else:
                print(f"警告: FFmpeg启动失败，跳过该流: {stream}")
        
        if not self.ffmpeg_processes:
            print("\n错误: 所有FFmpeg进程均启动失败")
            sys.exit(1)
        
        print(f"\n成功启动 {len(self.ffmpeg_processes)}/{len(streams)} 个FFmpeg进程")
        
        # 收集所有流配置（用于PickupMonitor）
        all_stream_configs = [p.stream_config for p in self.ffmpeg_processes]
        
        # 启动监控线程（统一一个监控器）
        print("\n启动监控线程...")
        check_interval = self.config.get('prediction', {}).get('check_interval', 1.0)
        
        monitor = PickupMonitor(
            audio_dir=CFG.AUDIO_DIR,
            multi_predictor=self.multi_predictor,
            stream_configs=all_stream_configs,
            check_interval=check_interval,
            config=self.config,
            config_manager=self.config_manager
        )
        monitor.start()
        self.monitors.append(monitor)
        
        print(f"\n成功启动监控线程")
        print("\n" + "=" * 70)
        print("系统已启动，开始监控...")
        print("按 Ctrl+C 停止系统")
        print("=" * 70 + "\n")
        
        # FFmpeg重启配置
        max_restart_attempts = 5        # 单个进程最大重启次数
        restart_interval_base = 30    # 基础重启间隔（秒）-> 改为30s
        restart_counts = {id(p): 0 for p in self.ffmpeg_processes}  # 重启计数
        
        # 主循环（带自动重启）
        try:
            while True:
                # 检查每个FFmpeg进程状态
                for ffmpeg in self.ffmpeg_processes:
                    if not ffmpeg.is_running():
                        pid = id(ffmpeg)
                        device_code = ffmpeg.stream_config.device_code
                        
                        # 检查重启次数
                        if restart_counts.get(pid, 0) < max_restart_attempts:
                            # 计算等待时间（指数退避）
                            wait_time = restart_interval_base * (2 ** restart_counts.get(pid, 0))
                            logger.warning(f"FFmpeg进程停止: {device_code} | 将在{wait_time}秒后重启 | "
                                         f"重试次数: {restart_counts.get(pid, 0) + 1}/{max_restart_attempts}")
                            
                            time.sleep(wait_time)
                            
                            # 尝试重启
                            if ffmpeg.start():
                                logger.debug(f"FFmpeg重启成功: {device_code}")
                                restart_counts[pid] = 0  # 重置计数
                            else:
                                restart_counts[pid] = restart_counts.get(pid, 0) + 1
                                logger.error(f"FFmpeg重启失败: {device_code}")
                        else:
                            logger.error(f"FFmpeg达到最大重启次数: {device_code} | 已放弃重启")
                
                # 检查是否所有进程都已放弃
                running_count = sum(1 for p in self.ffmpeg_processes if p.is_running())
                all_abandoned = all(restart_counts.get(id(p), 0) >= max_restart_attempts 
                                   for p in self.ffmpeg_processes if not p.is_running())
                
                if running_count == 0 and all_abandoned:
                    print("\n错误: 所有FFmpeg进程均已停止且无法重启")
                    break
                
                # 每天0点执行7天清理
                current_hour = datetime.now().hour
                current_date = datetime.now().strftime("%Y%m%d")
                
                if not hasattr(self, '_last_cleanup_date'):
                    self._last_cleanup_date = ""
                
                # 在0点且今天还没清理过时执行
                if current_hour == 0 and self._last_cleanup_date != current_date:
                    logger.info("执行7天过期文件清理...")
                    for monitor in self.monitors:
                        monitor._cleanup_old_files(days=7)
                    self._last_cleanup_date = current_date
                
                # 定期打印RTSP状态（每分钟一次）
                if not hasattr(self, '_last_status_log'):
                    self._last_status_log = datetime.now()
                
                if (datetime.now() - self._last_status_log).total_seconds() >= 60:
                    running = sum(1 for p in self.ffmpeg_processes if p.is_running())
                    total = len(self.ffmpeg_processes)
                    logger.info(f"RTSP状态: {running}/{total} 个FFmpeg进程运行中")
                    logger.info("─" * 60)
                    self._last_status_log = datetime.now()
                
                time.sleep(10)
        except KeyboardInterrupt:
            print("\n\n收到停止信号，正在关闭系统...")
        finally:
            self.stop()
    
    def stop(self):
        """
        停止监控系统
        """
        print("\n正在停止监控系统...")
        
        print("停止监控线程...")
        for monitor in self.monitors:
            monitor.stop()
        
        print("停止FFmpeg进程...")
        for ffmpeg in self.ffmpeg_processes:
            ffmpeg.stop()
        
        print("系统已完全停止")
    
    def _signal_handler(self, signum, frame):
        """
        信号处理函数
        """
        print(f"\n\n收到信号 {signum}，正在关闭系统...")
        self.stop()
        sys.exit(0)


def _start_config_api_server(config_manager, multi_predictor=None, port=18080):
    # 在后台线程中启动 FastAPI 配置管理 API
    try:
        import uvicorn
        init_config_api(config_manager, multi_predictor)
        logger.info(f"启动配置管理 API: http://0.0.0.0:{port}")
        uvicorn.run(config_app, host="0.0.0.0", port=port, log_level="warning")
    except ImportError:
        logger.warning("uvicorn 未安装，配置管理 API 无法启动。请安装: pip install uvicorn")
    except Exception as e:
        logger.error(f"配置管理 API 启动失败: {e}")


def main():
    """
    主函数
    """
    # 检测 DB 是否存在
    db_path = get_db_path(Path(__file__).parent / "config")
    if not db_path.exists():
        print(f"错误: 配置数据库不存在: {db_path}")
        print(f"\n请先运行迁移脚本: python tool/migrate_yaml_to_db.py")
        sys.exit(1)
    
    try:
        system = PickupMonitoringSystem()
        
        # 后台线程启动配置管理 API
        api_port = 18080
        api_thread = threading.Thread(
            target=_start_config_api_server,
            args=(system.config_manager, system.multi_predictor, api_port),
            daemon=True,
            name="config-api"
        )
        api_thread.start()
        system.start()
    except FileNotFoundError as e:
        print(f"\n错误: {e}")
        print("\n请确保:")
        print("1. 已完成训练并计算阈值")
        print("2. 已复制必要的模型文件到 models/ 目录")
        sys.exit(1)
    except Exception as e:
        print(f"\n严重错误: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()