package com.hdkj.lt.bf.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.hdkj.lt.bf.application.FeederGroupApplication;
import com.hdkj.lt.bf.common.SeMonitorThreshold;
import com.hdkj.lt.bf.common.SeSnapshotConstants;
import com.hdkj.lt.bf.entity.DwdShbDsCableSegmentBase;
import com.hdkj.lt.bf.entity.DwdShbDsSpanSegmentBase;
import com.hdkj.lt.bf.entity.DwdShbDsTransformerBase;
import com.hdkj.lt.bf.entity.FhzgSeCurrentEvent;
import com.hdkj.lt.bf.entity.FhzgSeLineLossDetail;
import com.hdkj.lt.bf.entity.FhzgSeMonitorCurrent;
import com.hdkj.lt.bf.entity.FhzgSeMonitorVoltage;
import com.hdkj.lt.bf.entity.FhzgSeSnapshotDetail;
import com.hdkj.lt.bf.entity.FhzgSeVoltageEvent;
import com.hdkj.lt.bf.event.MvtransOverVoltageEvent;
import com.hdkj.lt.bf.event.ReconTriggerEvent;
import com.hdkj.lt.bf.mapper.FeederTopoMapper;
import com.hdkj.lt.bf.mapper.FhzgSeCurrentEventMapper;
import com.hdkj.lt.bf.mapper.FhzgSeLineLossDetailMapper;
import com.hdkj.lt.bf.mapper.FhzgSeMonitorCurrentMapper;
import com.hdkj.lt.bf.mapper.FhzgSeMonitorVoltageMapper;
import com.hdkj.lt.bf.mapper.FhzgSeSnapshotDetailMapper;
import com.hdkj.lt.bf.mapper.FhzgSeVoltageEventMapper;
import com.hdkj.lt.bf.mapper.DwdShbDsCableSegmentBaseMapper;
import com.hdkj.lt.bf.mapper.DwdShbDsSpanSegmentBaseMapper;
import com.hdkj.lt.bf.mapper.DwdShbDsTransformerBaseMapper;
import com.hdkj.lt.bf.service.SeSnapshotService;
import com.hdkj.lt.core.bizms.modle.po.FeederTopo;
import com.hdkj.fhzg.optimization.response.FeederGroupTreeNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

@Slf4j
@Service
@RequiredArgsConstructor
public class SeSnapshotServiceImpl implements SeSnapshotService {

    private final FhzgSeSnapshotDetailMapper snapshotDetailMapper;
    private final FhzgSeMonitorCurrentMapper monitorCurrentMapper;
    private final FhzgSeMonitorVoltageMapper monitorVoltageMapper;
    private final FhzgSeCurrentEventMapper currentEventMapper;
    private final FhzgSeVoltageEventMapper voltageEventMapper;
    private final DwdShbDsTransformerBaseMapper transformerBaseMapper;
    private final FeederGroupApplication feederGroupApplication;
    private final ApplicationEventPublisher eventPublisher;
    private final FeederTopoMapper feederTopoMapper;
    private final FhzgSeLineLossDetailMapper lineLossDetailMapper;
    private final DwdShbDsSpanSegmentBaseMapper spanSegmentMapper;
    private final DwdShbDsCableSegmentBaseMapper cableSegmentMapper;

    private static final DateTimeFormatter DT_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @Override
    public void processSnapshots(String snapshotJsonArray) {
        JSONArray arr = JSON.parseArray(snapshotJsonArray);
        if (arr == null || arr.isEmpty()) return;
        for (int i = 0; i < arr.size(); i++) {
            processSnapshot(arr.getJSONObject(i).toJSONString());
        }
    }

    @Override
    public void processSnapshot(String snapshotJson) {
        JSONObject root = JSON.parseObject(snapshotJson);
        if (root == null) return;

        JSONArray pointTimes = root.getJSONArray("pointTime");
        if (pointTimes == null || pointTimes.isEmpty()) return;
        LocalDateTime snapTime = LocalDateTime.parse(pointTimes.getString(0), DT_FMT);

        Map<String, String[]> feederMap = buildFeederMapping();
        List<FhzgSeSnapshotDetail> batch = new ArrayList<>();

        // === 0. 批量预加载状态机（避免逐设备 selectOne 循环导致连接持有过久 / abandon） ===
        JSONArray feeders = root.getJSONArray("periodFeederSeResult");
        JSONArray mvtrans = root.getJSONArray("periodMVTransSeResult");
        Set<String> feederIds = new HashSet<>();
        Set<String> allDeviceIds = new HashSet<>();
        if (feeders != null) {
            for (int i = 0; i < feeders.size(); i++) {
                String id = feeders.getJSONObject(i).getString("psrId");
                if (id != null) { feederIds.add(id); allDeviceIds.add(id); }
            }
        }
        if (mvtrans != null) {
            for (int i = 0; i < mvtrans.size(); i++) {
                String id = mvtrans.getJSONObject(i).getString("psrId");
                if (id != null) allDeviceIds.add(id);
            }
        }
        Map<String, FhzgSeMonitorCurrent> currentStateMap = loadCurrentStateMap(feederIds);
        Map<String, FhzgSeMonitorVoltage> voltageStateMap = loadVoltageStateMap(allDeviceIds);

        // === 1. 馈线 ===
        if (feeders != null) {
            for (int i = 0; i < feeders.size(); i++) {
                JSONObject f = feeders.getJSONObject(i);
                FhzgSeSnapshotDetail d = parseFeeder(f, snapTime, feederMap);
                if (d != null) {
                    batch.add(d);
                    processFeederCurrent(f, d, snapTime, feederMap, currentStateMap);
                    processFeederVoltage(f, d, snapTime, feederMap, voltageStateMap);
                }
            }
        }

        // === 2. 配变 ===
        if (mvtrans != null && !mvtrans.isEmpty()) {
            // 先收集当前断面所有配变 ID，再一次性查名称（避免全表加载）
            Set<String> mvtransIds = new HashSet<>();
            for (int i = 0; i < mvtrans.size(); i++) {
                String id = mvtrans.getJSONObject(i).getString("psrId");
                if (id != null) mvtransIds.add(id);
            }
            Map<String, String> transformerNameMap = buildTransformerMapping(mvtransIds);
            for (int i = 0; i < mvtrans.size(); i++) {
                JSONObject m = mvtrans.getJSONObject(i);
                FhzgSeSnapshotDetail d = parseMvtrans(m, snapTime, feederMap);
                if (d != null) {
                    batch.add(d);
                    processMvtransMonitor(m, d, snapTime, feederMap, transformerNameMap, voltageStateMap);
                }
            }
        }

        // === 3. 线路线段 → 线损计算 ===
        JSONArray segments = root.getJSONArray("periodSegmentSeResult");
        if (segments != null && !segments.isEmpty()) {
            processSegmentsForLineLoss(segments, snapTime, feederMap);
        }

        if (!batch.isEmpty()) {
            // 防重：查该断面时刻已有的所有 device_id（精确到秒，无需再 IN 过滤）
            Set<String> existingIds = new HashSet<>();
            snapshotDetailMapper.selectList(
                    new LambdaQueryWrapper<FhzgSeSnapshotDetail>()
                            .select(FhzgSeSnapshotDetail::getDeviceId)
                            .eq(FhzgSeSnapshotDetail::getSnapshotTime, snapTime)
            ).forEach(e -> existingIds.add(e.getDeviceId()));

            List<FhzgSeSnapshotDetail> toInsert = new ArrayList<>();
            for (FhzgSeSnapshotDetail d : batch) {
                if (existingIds.contains(d.getDeviceId())) continue;
                toInsert.add(d);
            }

            int inserted = 0;
            if (!toInsert.isEmpty()) {
                inserted = insertBatchSharded(rows -> snapshotDetailMapper.insertBatch(rows), toInsert);
            }
            log.info("状估断面处理完成 snapTime={}, 写入{}条(去重跳过{}条)",
                    snapTime, inserted, batch.size() - inserted);
        }
    }

    // ============================================================
    // 馈线
    // ============================================================

    private FhzgSeSnapshotDetail parseFeeder(JSONObject f, LocalDateTime snapTime, Map<String, String[]> feederMap) {
        String feederId = f.getString("psrId");
        String[] loc = feederMap.getOrDefault(feederId, new String[]{"", "", ""});
        BigDecimal loadRate = firstValue(f.getJSONArray("loadRates"));
        BigDecimal voltage = firstValue(f.getJSONArray("seoUs"));
        BigDecimal voltageRatio = voltage != null
                ? voltage.divide(SeMonitorThreshold.FEEDER_BASE_KV, 4, RoundingMode.HALF_UP) : null;
        return FhzgSeSnapshotDetail.builder()
                .snapshotTime(snapTime).deviceId(feederId).deviceType("feeder")
                .feederId(feederId).feederName(loc.length > 2 ? loc[2] : null)
                .subsId(loc[0]).countyId(loc[1])
                .activePower(firstValue(f.getJSONArray("seoPs")))
                .reactivePower(firstValue(f.getJSONArray("seoQs")))
                .currentValue(firstValue(f.getJSONArray("seoIs")))
                .voltageValue(voltage).loadRate(loadRate).voltageRatio(voltageRatio)
                .currentStatus(calcCurrentStatus(loadRate))
                .voltageStatus(calcFeederVoltageStatus(voltageRatio))
                .build();
    }

    private void processFeederCurrent(JSONObject f, FhzgSeSnapshotDetail detail,
                                       LocalDateTime snapTime, Map<String, String[]> feederMap,
                                       Map<String, FhzgSeMonitorCurrent> currentStateMap) {
        if (detail.getCurrentStatus() == null) return;
        String feederId = detail.getFeederId();
        String[] loc = feederMap.get(feederId);
        String feederName = loc != null && loc.length > 2 ? loc[2] : null;
        String countyId = loc != null ? loc[1] : "";
        String subsId = loc != null ? loc[0] : "";
        BigDecimal loadRate = detail.getLoadRate();
        int status = detail.getCurrentStatus();
        boolean isOverLimit = status != 0;

        FhzgSeMonitorCurrent state = currentStateMap.get(feederId);

        if (state == null) {
            state = FhzgSeMonitorCurrent.builder()
                    .feederId(feederId).feederName(feederName)
                    .lastLoadRate(loadRate).lastStatus(status).lastSnapTime(snapTime)
                    .alarmTriggered(0).build();
            if (isOverLimit) state.setFirstOverTime(snapTime);
            monitorCurrentMapper.insert(state);
        } else {
            LocalDateTime lastSnap = state.getLastSnapTime();
            if (lastSnap != null && Duration.between(lastSnap, snapTime).toMinutes() > SeMonitorThreshold.SNAP_INTERVAL_MINUTES) {
                // 断档间隙：先关掉还在活跃的事件，再清空状态机引用
                if (state.getAlarmTriggered() == 1 && state.getAlarmEventId() != null) {
                    FhzgSeCurrentEvent close = new FhzgSeCurrentEvent();
                    close.setId(state.getAlarmEventId());
                    close.setStatus(2);
                    close.setRecoverTime(snapTime);
                    close.setLastLoadRate(loadRate);
                    currentEventMapper.updateById(close);
                }
                state.setFirstOverTime(null);
                state.setAlarmTriggered(0);
                state.setAlarmEventId(null);
            }
            BigDecimal maxLoadRate = state.getLastLoadRate();
            if (loadRate != null && (maxLoadRate == null || loadRate.compareTo(maxLoadRate) > 0))
                maxLoadRate = loadRate;

            if (isOverLimit) {
                if (state.getFirstOverTime() == null) state.setFirstOverTime(snapTime);
                if (state.getFirstOverTime() != null
                        && Duration.between(state.getFirstOverTime(), snapTime).toMinutes() >= SeMonitorThreshold.CURRENT_TRIGGER_DURATION_MINUTES
                        && state.getAlarmTriggered() == 0) {
                    // 触发时往回看持续窗口：有任意断面≥100%算过载，否则重载
                    boolean hasOverload = hasOverloadSnapshotInWindow(feederId, state.getFirstOverTime(), snapTime);
                    String alarmType = hasOverload ? "current_overload" : "current_heavy";
                    String countyName = lookupCountyName(countyId, feederMap);
                    FhzgSeCurrentEvent event = FhzgSeCurrentEvent.builder()
                            .feederId(feederId).feederName(feederName).alarmType(alarmType)
                            .countyId(countyId).countyName(countyName).subsId(subsId)
                            .firstOverTime(state.getFirstOverTime()).triggerTime(snapTime)
                            .triggerLoadRate(loadRate).maxLoadRate(maxLoadRate).lastLoadRate(loadRate).status(1)
                            .build();
                    currentEventMapper.insert(event);
                    state.setAlarmTriggered(1);
                    state.setAlarmEventId(event.getId());
                    // 取事件窗口内最大负载率断面时刻作为重构触发断面时刻
                    LocalDateTime maxLrTime = findMaxLoadRateTime(feederId, state.getFirstOverTime(), snapTime);
                    eventPublisher.publishEvent(new ReconTriggerEvent(
                            event.getId(), feederId, feederName, countyId, subsId,
                            alarmType, maxLrTime, state.getFirstOverTime()));
                }
                // 活跃事件：断面负载率比记录值高则刷新 maxLoadRate
                if (state.getAlarmTriggered() == 1 && state.getAlarmEventId() != null) {
                    FhzgSeCurrentEvent cur = currentEventMapper.selectById(state.getAlarmEventId());
                    if (cur != null) {
                        BigDecimal curMax = cur.getMaxLoadRate();
                        if (loadRate != null && (curMax == null || loadRate.compareTo(curMax) > 0)) {
                            FhzgSeCurrentEvent upd = new FhzgSeCurrentEvent();
                            upd.setId(state.getAlarmEventId());
                            upd.setMaxLoadRate(loadRate);
                            currentEventMapper.updateById(upd);
                        }
                    }
                }
            } else {
                if (state.getAlarmTriggered() == 1 && state.getAlarmEventId() != null) {
                    FhzgSeCurrentEvent cur = currentEventMapper.selectById(state.getAlarmEventId());
                    if (cur != null) {
                        BigDecimal trueMax = cur.getMaxLoadRate();
                        if (trueMax == null || (maxLoadRate != null && maxLoadRate.compareTo(trueMax) > 0)) {
                            trueMax = maxLoadRate;
                        }
                        FhzgSeCurrentEvent update = new FhzgSeCurrentEvent();
                        update.setId(state.getAlarmEventId());
                        update.setStatus(2); update.setRecoverTime(snapTime);
                        update.setLastLoadRate(loadRate); update.setMaxLoadRate(trueMax);
                        currentEventMapper.updateById(update);
                    }
                }
                state.setFirstOverTime(null); state.setAlarmTriggered(0); state.setAlarmEventId(null);
            }
            state.setLastLoadRate(loadRate); state.setLastStatus(status); state.setLastSnapTime(snapTime);
            monitorCurrentMapper.updateById(state);
        }
    }

    private void processFeederVoltage(JSONObject f, FhzgSeSnapshotDetail detail,
                                       LocalDateTime snapTime, Map<String, String[]> feederMap,
                                       Map<String, FhzgSeMonitorVoltage> voltageStateMap) {
        if (detail.getVoltageStatus() == null) return;
        String[] loc = feederMap.get(detail.getFeederId());
        String feederName = loc != null && loc.length > 2 ? loc[2] : null;
        checkVoltageMonitor(detail.getFeederId(), "feeder", feederName, detail.getVoltageRatio(), detail.getVoltageStatus(),
                snapTime, detail.getFeederId(), feederName,
                loc != null ? loc[0] : "", loc != null ? loc[1] : "", null, null, voltageStateMap);
    }

    // ============================================================
    // 配变
    // ============================================================

    private FhzgSeSnapshotDetail parseMvtrans(JSONObject m, LocalDateTime snapTime, Map<String, String[]> feederMap) {
        String mvtransId = m.getString("psrId");
        String feederId = m.getString("feederId");
        String[] loc = feederMap.getOrDefault(feederId, new String[]{"", "", ""});
        BigDecimal lvVoltageA = firstValue(m.getJSONArray("seiLvUas"));
        BigDecimal voltageRatio = lvVoltageA != null
                ? lvVoltageA.divide(SeMonitorThreshold.MVTRANS_BASE_KV, 4, RoundingMode.HALF_UP) : null;
        BigDecimal capacity = null;
        String capStr = m.getString("capacitys");
        if (capStr != null) { try { capacity = new BigDecimal(capStr); } catch (NumberFormatException ignored) {} }
        return FhzgSeSnapshotDetail.builder()
                .snapshotTime(snapTime).deviceId(mvtransId).deviceType("mvtrans")
                .feederId(feederId).subsId(loc[0]).countyId(loc[1])
                .lvVoltageA(lvVoltageA).capacity(capacity).voltageRatio(voltageRatio)
                .voltageStatus(calcMvtransVoltageStatus(voltageRatio))
                .build();
    }

    private void processMvtransMonitor(JSONObject m, FhzgSeSnapshotDetail detail,
                                        LocalDateTime snapTime, Map<String, String[]> feederMap,
                                        Map<String, String> transformerNameMap,
                                        Map<String, FhzgSeMonitorVoltage> voltageStateMap) {
        if (detail.getVoltageStatus() == null) return;
        String mvtransId = detail.getDeviceId();
        String[] loc = feederMap.get(detail.getFeederId());
        int status = detail.getVoltageStatus();
        if (status == 0) recoverUserAlarms(mvtransId, snapTime);

        String transformerName = transformerNameMap != null
                ? transformerNameMap.getOrDefault(mvtransId, mvtransId) : mvtransId;
        checkVoltageMonitor(mvtransId, "mvtrans", transformerName, detail.getVoltageRatio(), status,
                snapTime, detail.getFeederId(), loc != null && loc.length > 2 ? loc[2] : null,
                loc != null ? loc[0] : "", loc != null ? loc[1] : "", mvtransId, null, voltageStateMap);

        if (status != 0) {
            eventPublisher.publishEvent(new MvtransOverVoltageEvent(
                    this, mvtransId, detail.getFeederId(),
                    loc != null ? loc[1] : "", loc != null ? loc[0] : "",
                    snapTime, detail.getVoltageRatio()));
        }
    }

    // ============================================================
    // 电压状态机
    // ============================================================

    private void checkVoltageMonitor(String deviceId, String deviceType, String deviceName,
                                      BigDecimal voltageRatio, int voltageStatus,
                                      LocalDateTime snapTime, String feederId, String feederName,
                                      String subsId, String countyId, String mvtransId, String consumerId,
                                      Map<String, FhzgSeMonitorVoltage> voltageStateMap) {
        FhzgSeMonitorVoltage state = voltageStateMap.get(deviceId + ":" + deviceType);
        boolean isOverLimit = voltageStatus != 0;

        if (state == null) {
            state = FhzgSeMonitorVoltage.builder()
                    .deviceId(deviceId).deviceType(deviceType)
                    .feederId(feederId).feederName(feederName)
                    .subsId(subsId).countyId(countyId)
                    .mvtransId(mvtransId).consumerId(consumerId)
                    .lastVoltageRatio(voltageRatio).lastVoltageStatus(voltageStatus)
                    .lastSnapTime(snapTime).alarmTriggered(0).build();
            if (isOverLimit) state.setFirstOverTime(snapTime);
            monitorVoltageMapper.insert(state);
        } else {
            LocalDateTime lastSnap = state.getLastSnapTime();
            if (lastSnap != null && Duration.between(lastSnap, snapTime).toMinutes() > SeMonitorThreshold.SNAP_INTERVAL_MINUTES) {
                // 断档间隙：先关掉还在活跃的事件，再清空状态机引用
                if (state.getAlarmTriggered() == 1 && state.getAlarmEventId() != null) {
                    FhzgSeVoltageEvent close = new FhzgSeVoltageEvent();
                    close.setId(state.getAlarmEventId());
                    close.setStatus(2); close.setRecoverTime(snapTime); close.setLastValue(voltageRatio);
                    voltageEventMapper.updateById(close);
                }
                state.setFirstOverTime(null); state.setAlarmTriggered(0); state.setAlarmEventId(null);
            }
            if (isOverLimit) {
                if (state.getFirstOverTime() == null) state.setFirstOverTime(snapTime);
                if (state.getFirstOverTime() != null
                        && Duration.between(state.getFirstOverTime(), snapTime).toMinutes() >= SeMonitorThreshold.TRIGGER_DURATION_MINUTES
                        && state.getAlarmTriggered() == 0) {
                            String countyName = lookupCountyName(countyId, null);

                            if ("feeder".equals(deviceType)) {
                                // === 馈线自身越限 → 创建 feeder 级独立事件 ===
                                String alarmType = resolveVoltageAlarmType(deviceType, voltageStatus);
                                BigDecimal threshold = resolveVoltageThreshold(deviceType, voltageStatus);
                                BigDecimal[] vr = queryVoltageRangeFromSnap(feederId, state.getFirstOverTime(), snapTime);
                                int[] mc = queryMvtransCounts(feederId, state.getFirstOverTime(), snapTime);
                                FhzgSeVoltageEvent event = FhzgSeVoltageEvent.builder()
                                        .deviceId(deviceId).deviceName(deviceName)
                                        .deviceType(deviceType).alarmLevel(deviceType).alarmType(alarmType)
                                        .feederId(feederId).feederName(feederName)
                                        .subsId(subsId).countyId(countyId).countyName(countyName)
                                        .threshold(threshold)
                                        .firstOverTime(state.getFirstOverTime()).triggerTime(snapTime)
                                        .triggerValue(voltageRatio)
                                        .voltageRangeMax(vr[0]).voltageRangeMin(vr[1])
                                        .overUpperMvtrans(mc[0]).overLowerMvtrans(mc[1])
                                        .lastValue(voltageRatio).status(1).build();
                                voltageEventMapper.insert(event);
                                state.setAlarmTriggered(1); state.setAlarmEventId(event.getId());
                                // 馈线越下限 → 异步触发重构算法
                                if ("feeder".equals(deviceType) && "voltage_under".equals(alarmType)) {
                                    eventPublisher.publishEvent(new ReconTriggerEvent(
                                            event.getId(), feederId, feederName, countyId, subsId,
                                            alarmType, snapTime, state.getFirstOverTime()));
                                }
                            } else if ("mvtrans".equals(deviceType)) {
                                // === 台区越限 → 创建 mvtrans 级独立事件 ===
                                String alarmType = resolveVoltageAlarmType(deviceType, voltageStatus);
                                BigDecimal threshold = resolveVoltageThreshold(deviceType, voltageStatus);
                                FhzgSeVoltageEvent event = FhzgSeVoltageEvent.builder()
                                        .deviceId(deviceId).deviceName(deviceName)
                                        .deviceType(deviceType).alarmLevel(deviceType).alarmType(alarmType)
                                        .feederId(feederId).feederName(feederName)
                                        .subsId(subsId).countyId(countyId).countyName(countyName)
                                        .mvtransId(mvtransId)
                                        .threshold(threshold)
                                        .firstOverTime(state.getFirstOverTime()).triggerTime(snapTime)
                                        .triggerValue(voltageRatio).lastValue(voltageRatio).status(1)
                                        .build();
                                voltageEventMapper.insert(event);
                                state.setAlarmTriggered(1);
                                state.setAlarmEventId(event.getId());
                            }
                            // consumer 级别暂走旧逻辑（如有可后续优化）
                        }
            } else {
                // 设备恢复正常（任何 deviceType 都走）：关自己的事件
                if (state.getAlarmTriggered() == 1 && state.getAlarmEventId() != null) {
                    FhzgSeVoltageEvent update = new FhzgSeVoltageEvent();
                    update.setId(state.getAlarmEventId());
                    update.setStatus(2); update.setRecoverTime(snapTime); update.setLastValue(voltageRatio);
                    voltageEventMapper.updateById(update);
                }
                state.setFirstOverTime(null); state.setAlarmTriggered(0); state.setAlarmEventId(null);
            }
            state.setLastVoltageRatio(voltageRatio); state.setLastVoltageStatus(voltageStatus);
            state.setLastSnapTime(snapTime);
            monitorVoltageMapper.updateById(state);
        }
    }

    // ============================================================
    // 配变→用户恢复
    // ============================================================

    private void recoverUserAlarms(String mvtransId, LocalDateTime snapTime) {
        FhzgSeVoltageEvent updateEvent = new FhzgSeVoltageEvent();
        updateEvent.setStatus(2); updateEvent.setRecoverTime(snapTime);
        voltageEventMapper.update(updateEvent,
                new LambdaQueryWrapper<FhzgSeVoltageEvent>()
                        .eq(FhzgSeVoltageEvent::getMvtransId, mvtransId)
                        .eq(FhzgSeVoltageEvent::getAlarmLevel, "consumer")
                        .eq(FhzgSeVoltageEvent::getStatus, 1));
        FhzgSeMonitorVoltage reset = new FhzgSeMonitorVoltage();
        reset.setFirstOverTime(null); reset.setAlarmTriggered(0);
        reset.setAlarmEventId(null); reset.setLastVoltageStatus(0);
        monitorVoltageMapper.update(reset,
                new LambdaQueryWrapper<FhzgSeMonitorVoltage>()
                        .eq(FhzgSeMonitorVoltage::getMvtransId, mvtransId)
                        .eq(FhzgSeMonitorVoltage::getDeviceType, "consumer"));
    }

    // ============================================================
    // 线路线损计算
    // ============================================================

    /**
     * 处理线路线段断面数据，计算逐段线损功率并写入明细表
     * <p>
     * line_loss_kW = 3000 × I_kA² × r_Ω_per_km × (L_m / 1000)
     * 电阻率: dxd(导线段)=0.4132Ω/km, 0201(电缆段)=0.313Ω/km
     * 段长度从 dwd_shb_ds_span_segment_base（导线段）/ dwd_shb_ds_cable_segment_base（电缆段）的 length 取（单位：米）
     * SE 返回的 seoIs 为安培，先 /1000 换算成 kA
     */
    private void processSegmentsForLineLoss(JSONArray segments, LocalDateTime snapTime, Map<String, String[]> feederMap) {
        // 1. 按类型分组查段长度（档距段→span表，电缆段→cable表），过滤在运设备
        Map<String, BigDecimal> lengthMap = new HashMap<>();
        try {
            List<String> spanIds = new ArrayList<>();
            List<String> cableIds = new ArrayList<>();
            for (int i = 0; i < segments.size(); i++) {
                JSONObject seg = segments.getJSONObject(i);
                String pid = seg.getString("psrId");
                String pt = seg.getString("psrType");
                if (pid == null) continue;
                if ("dxd".equals(pt)) spanIds.add(pid);
                else if ("0201".equals(pt)) cableIds.add(pid);
            }
            if (!spanIds.isEmpty()) {
                spanSegmentMapper.selectList(
                        new LambdaQueryWrapper<DwdShbDsSpanSegmentBase>()
                                .select(DwdShbDsSpanSegmentBase::getPsrId, DwdShbDsSpanSegmentBase::getLength)
                                .in(DwdShbDsSpanSegmentBase::getPsrId, spanIds)
                                .eq(DwdShbDsSpanSegmentBase::getPsrState, "20"))
                        .forEach(r -> { if (r.getLength() != null) lengthMap.put(r.getPsrId(), r.getLength()); });
            }
            if (!cableIds.isEmpty()) {
                cableSegmentMapper.selectList(
                        new LambdaQueryWrapper<DwdShbDsCableSegmentBase>()
                                .select(DwdShbDsCableSegmentBase::getPsrId, DwdShbDsCableSegmentBase::getLength)
                                .in(DwdShbDsCableSegmentBase::getPsrId, cableIds)
                                .eq(DwdShbDsCableSegmentBase::getPsrState, "20"))
                        .forEach(r -> { if (r.getLength() != null) lengthMap.put(r.getPsrId(), r.getLength()); });
            }
        } catch (Exception e) {
            log.warn("查询段长度失败", e);
        }

        // 3. 逐段计算线损
        List<FhzgSeLineLossDetail> batch = new ArrayList<>();
        LocalDateTime now = LocalDateTime.now();
        for (int i = 0; i < segments.size(); i++) {
            JSONObject seg = segments.getJSONObject(i);
            String psrId = seg.getString("psrId");
            String psrType = seg.getString("psrType");
            String feederId = seg.getString("feederId");
            String[] loc = feederMap.getOrDefault(feederId, new String[]{"", "", ""});

            BigDecimal current = firstValue(seg.getJSONArray("seoIs"));
            if (current == null || current.compareTo(BigDecimal.ZERO) <= 0) continue;

            BigDecimal sbLength = lengthMap.get(psrId);
            if (sbLength == null || sbLength.compareTo(BigDecimal.ZERO) <= 0) continue;

            // 电阻率
            BigDecimal resistivity;
            if ("0201".equals(psrType)) {
                resistivity = new BigDecimal("0.313");      // 电缆段
            } else if ("dxd".equals(psrType)) {
                resistivity = new BigDecimal("0.4132");     // 导线段
            } else {
                log.debug("未知段类型跳过 psrType={}, psrId={}", psrType, psrId);
                continue;
            }

            // SE 返回电流为安培，换算成 kA：I_kA = I_A / 1000
            BigDecimal currentKa = current.divide(BigDecimal.valueOf(1000), 8, RoundingMode.HALF_UP);

            // 3000 × I_kA² × r × (L_m / 1000) = 3 × I_kA² × r × L_m
            BigDecimal lengthKm = sbLength.divide(BigDecimal.valueOf(1000), 8, RoundingMode.HALF_UP);
            BigDecimal loss = BigDecimal.valueOf(3000)
                    .multiply(currentKa.pow(2))
                    .multiply(resistivity)
                    .multiply(lengthKm)
                    .setScale(4, RoundingMode.HALF_UP);

            FhzgSeLineLossDetail detail = FhzgSeLineLossDetail.builder()
                    .snapTime(snapTime)
                    .feederId(feederId)
                    .psrId(psrId)
                    .psrType(psrType)
                    .countyId(loc.length > 1 ? loc[1] : "")
                    .subsId(loc.length > 0 ? loc[0] : "")
                    .currentKa(currentKa)
                    .sbLengthM(sbLength)
                    .resistivity(resistivity)
                    .lineLossKw(loss)
                    .createTime(now)
                    .build();
            batch.add(detail);
        }

        if (!batch.isEmpty()) {
            // 防重：查该断面时刻已有的段（避免重复跑同一断面时叠加）
            Set<String> existingPsrIds = new HashSet<>();
            lineLossDetailMapper.selectList(
                    new LambdaQueryWrapper<FhzgSeLineLossDetail>()
                            .select(FhzgSeLineLossDetail::getPsrId)
                            .eq(FhzgSeLineLossDetail::getSnapTime, snapTime)
            ).forEach(e -> existingPsrIds.add(e.getPsrId()));

            List<FhzgSeLineLossDetail> toInsert = new ArrayList<>();
            for (FhzgSeLineLossDetail d : batch) {
                if (existingPsrIds.contains(d.getPsrId())) continue;
                toInsert.add(d);
            }

            int inserted = 0;
            if (!toInsert.isEmpty()) {
                inserted = insertBatchSharded(rows -> lineLossDetailMapper.insertBatch(rows), toInsert);
            }
            int skippedNoData = segments.size() - batch.size();
            int skippedDup = batch.size() - toInsert.size();
            log.info("线路线损计算完成 snapTime={}, 写入{}条, 跳过{}条(无数据{}条/重复{}条)",
                    snapTime, inserted, skippedNoData + skippedDup, skippedNoData, skippedDup);
        }
    }

    // ============================================================
    // 查询辅助
    // ============================================================

    /**
     * 批量插入分片：避免单条巨型 INSERT（foreach 拼接全部行）
     * 超过 SQL 大小/执行时长限制会导致连接被 Druid removeAbandoned 回收（线损段量级上万行尤其明显）
     * 每片最多 BATCH_INSERT_SIZE 行，逐片插入
     */
    private static final int BATCH_INSERT_SIZE = 500;

    private <T> int insertBatchSharded(java.util.function.Function<List<T>, Integer> batchInserter, List<T> rows) {
        if (rows == null || rows.isEmpty()) return 0;
        int total = 0;
        for (int i = 0; i < rows.size(); i += BATCH_INSERT_SIZE) {
            int end = Math.min(i + BATCH_INSERT_SIZE, rows.size());
            total += batchInserter.apply(rows.subList(i, end));
        }
        return total;
    }

    /**
     * 批量加载电流状态机（按馈线 ID 集合一次性查询，key=feederId）
     * 避免 processFeederCurrent 逐设备 selectOne 造成海量 SQL 往返
     */
    private Map<String, FhzgSeMonitorCurrent> loadCurrentStateMap(Collection<String> feederIds) {
        if (feederIds == null || feederIds.isEmpty()) return Collections.emptyMap();
        List<FhzgSeMonitorCurrent> rows = monitorCurrentMapper.selectList(
                new LambdaQueryWrapper<FhzgSeMonitorCurrent>()
                        .in(FhzgSeMonitorCurrent::getFeederId, feederIds));
        Map<String, FhzgSeMonitorCurrent> map = new HashMap<>();
        for (FhzgSeMonitorCurrent s : rows) {
            map.put(s.getFeederId(), s);
        }
        return map;
    }

    /**
     * 批量加载电压状态机（按设备 ID 集合一次性查询，key=deviceId:deviceType）
     * 避免 checkVoltageMonitor 逐设备 selectOne 造成海量 SQL 往返
     */
    private Map<String, FhzgSeMonitorVoltage> loadVoltageStateMap(Collection<String> deviceIds) {
        if (deviceIds == null || deviceIds.isEmpty()) return Collections.emptyMap();
        List<FhzgSeMonitorVoltage> rows = monitorVoltageMapper.selectList(
                new LambdaQueryWrapper<FhzgSeMonitorVoltage>()
                        .in(FhzgSeMonitorVoltage::getDeviceId, deviceIds));
        Map<String, FhzgSeMonitorVoltage> map = new HashMap<>();
        for (FhzgSeMonitorVoltage s : rows) {
            map.put(s.getDeviceId() + ":" + s.getDeviceType(), s);
        }
        return map;
    }

    private BigDecimal[] queryVoltageRangeFromSnap(String feederId, LocalDateTime start, LocalDateTime end) {
        List<FhzgSeSnapshotDetail> list = snapshotDetailMapper.selectList(
                new LambdaQueryWrapper<FhzgSeSnapshotDetail>()
                        .eq(FhzgSeSnapshotDetail::getFeederId, feederId)
                        .eq(FhzgSeSnapshotDetail::getDeviceType, "feeder")
                        .isNotNull(FhzgSeSnapshotDetail::getVoltageRatio)
                        .ge(FhzgSeSnapshotDetail::getSnapshotTime, start)
                        .le(FhzgSeSnapshotDetail::getSnapshotTime, end));
        BigDecimal maxV = null, minV = null;
        for (FhzgSeSnapshotDetail s : list) {
            BigDecimal v = s.getVoltageRatio();
            if (v == null) continue;
            if (maxV == null || v.compareTo(maxV) > 0) maxV = v;
            if (minV == null || v.compareTo(minV) < 0) minV = v;
        }
        return new BigDecimal[]{maxV, minV};
    }

    /**
     * 查窗口内是否有任意过载断面(≥100%)
     */
    private boolean hasOverloadSnapshotInWindow(String feederId, LocalDateTime start, LocalDateTime end) {
        Long count = snapshotDetailMapper.selectCount(
                new LambdaQueryWrapper<FhzgSeSnapshotDetail>()
                        .eq(FhzgSeSnapshotDetail::getFeederId, feederId)
                        .eq(FhzgSeSnapshotDetail::getDeviceType, "feeder")
                        .ge(FhzgSeSnapshotDetail::getLoadRate, SeMonitorThreshold.LOAD_RATE_OVERLOAD)
                        .ge(FhzgSeSnapshotDetail::getSnapshotTime, start)
                        .le(FhzgSeSnapshotDetail::getSnapshotTime, end));
        return count != null && count > 0;
    }

    // ============================================================
    // 状态判断 / 阈值
    // ============================================================

    private int calcCurrentStatus(BigDecimal loadRate) {
        if (loadRate == null) return 0;
        if (loadRate.compareTo(SeMonitorThreshold.LOAD_RATE_OVERLOAD) >= 0) return 2;
        if (loadRate.compareTo(SeMonitorThreshold.LOAD_RATE_HEAVY) >= 0) return 1;
        return 0;
    }

    private int calcFeederVoltageStatus(BigDecimal ratio) {
        if (ratio == null || ratio.signum() <= 0) return 0;
        if (ratio.compareTo(SeMonitorThreshold.FEEDER_V_OVER) > 0) return 1;
        if (ratio.compareTo(SeMonitorThreshold.FEEDER_V_UNDER) < 0) return 2;
        return 0;
    }

    private int calcMvtransVoltageStatus(BigDecimal ratio) {
        if (ratio == null || ratio.signum() <= 0) return 0;
        if (ratio.compareTo(SeMonitorThreshold.MVTRANS_V_OVER) > 0) return 1;
        if (ratio.compareTo(SeMonitorThreshold.MVTRANS_V_UNDER) < 0) return 2;
        return 0;
    }

    private BigDecimal resolveVoltageThreshold(String deviceType, int status) {
        if ("feeder".equals(deviceType)) return status == 1 ? SeMonitorThreshold.FEEDER_V_OVER : SeMonitorThreshold.FEEDER_V_UNDER;
        if ("mvtrans".equals(deviceType)) return status == 1 ? SeMonitorThreshold.MVTRANS_V_OVER : SeMonitorThreshold.MVTRANS_V_UNDER;
        switch (status) {
            case 1: return SeMonitorThreshold.CONSUMER_GENERAL_V_OVER;
            case 2: return SeMonitorThreshold.CONSUMER_SEVERE_V_OVER;
            case 3: return SeMonitorThreshold.CONSUMER_GENERAL_V_UNDER;
            case 4: return SeMonitorThreshold.CONSUMER_SEVERE_V_UNDER;
            default: return null;
        }
    }

    private String resolveVoltageAlarmType(String deviceType, int status) {
        if ("consumer".equals(deviceType)) {
            switch (status) {
                case 1: return "voltage_over_general";
                case 2: return "voltage_over_severe";
                case 3: return "voltage_under_general";
                case 4: return "voltage_under_severe";
                default: return "voltage_over";
            }
        }
        return status == 1 ? "voltage_over" : "voltage_under";
    }

    /**
     * 查窗口内已有 mvtrans 越限事件数（用于 feeder 触发时预填 overUpper/overLowerMvtrans）
     */
    private int[] queryMvtransCounts(String feederId, LocalDateTime start, LocalDateTime end) {
        List<FhzgSeVoltageEvent> list = voltageEventMapper.selectList(
                new LambdaQueryWrapper<FhzgSeVoltageEvent>()
                        .eq(FhzgSeVoltageEvent::getFeederId, feederId)
                        .eq(FhzgSeVoltageEvent::getAlarmLevel, "mvtrans")
                        .eq(FhzgSeVoltageEvent::getStatus, 1)
                        .ge(FhzgSeVoltageEvent::getFirstOverTime, start)
                        .le(FhzgSeVoltageEvent::getFirstOverTime, end));
        int up = 0, down = 0;
        for (FhzgSeVoltageEvent e : list) {
            if (e.getAlarmType() != null && e.getAlarmType().contains("over")) up++;
            else if (e.getAlarmType() != null && e.getAlarmType().contains("under")) down++;
        }
        return new int[]{up, down};
    }

    // ============================================================
    // 辅助
    // ============================================================

    private BigDecimal firstValue(JSONArray arr) {
        if (arr == null || arr.isEmpty()) return null;
        String v = arr.getString(0);
        if (v == null || v.isEmpty()) return null;
        try { return new BigDecimal(v); } catch (NumberFormatException e) { return null; }
    }

    /**
     * 查事件窗口 [start, end] 内该馈线最大负载率的那条断面的 snapshotTime
     */
    private LocalDateTime findMaxLoadRateTime(String feederId, LocalDateTime start, LocalDateTime end) {
        if (feederId == null || start == null) return end;
        List<FhzgSeSnapshotDetail> list = snapshotDetailMapper.selectList(
                new LambdaQueryWrapper<FhzgSeSnapshotDetail>()
                        .eq(FhzgSeSnapshotDetail::getFeederId, feederId)
                        .eq(FhzgSeSnapshotDetail::getDeviceType, "feeder")
                        .isNotNull(FhzgSeSnapshotDetail::getLoadRate)
                        .ge(FhzgSeSnapshotDetail::getSnapshotTime, start)
                        .le(FhzgSeSnapshotDetail::getSnapshotTime, end != null ? end : LocalDateTime.now())
                        .orderByDesc(FhzgSeSnapshotDetail::getLoadRate)
                        .last("LIMIT 1"));
        if (list != null && !list.isEmpty()) {
            return list.get(0).getSnapshotTime();
        }
        return end;
    }

    /**
     * 按 countyId 从 SeSnapshotConstants.COUNTIES 查 countyName。
     * COUNTIES 的 key=countyId(=maintOrg UUID)，value=[countySid, countyName]
     */
    private String lookupCountyName(String countyId, Map<String, String[]> feederMap) {
        if (countyId == null) return null;
        for (Map.Entry<String, String[]> entry : SeSnapshotConstants.COUNTIES) {
            if (countyId.equals(entry.getKey())) {
                return entry.getValue()[1];
            }
        }
        return null;
    }

    private Map<String, String[]> buildFeederMapping() {
        Map<String, String[]> map = new HashMap<>();
        try {
            List<FeederGroupTreeNode> tree = feederGroupApplication.getFeederGroupTree();
            if (tree != null) buildFeederMappingRecursive(tree, map, "", "");
        } catch (Exception e) {
            log.warn("构建馈线映射失败", e);
        }
        return map;
    }

    /**
     * 按配变 ID 集合批量查名称（只查当前断面出现的，不加载全表）
     * 查不到的设备以 ID 代替
     */
    private Map<String, String> buildTransformerMapping(Collection<String> ids) {
        if (ids == null || ids.isEmpty()) return Collections.emptyMap();
        Map<String, String> map = new HashMap<>();
        try {
            List<DwdShbDsTransformerBase> rows = transformerBaseMapper.selectList(
                                new LambdaQueryWrapper<DwdShbDsTransformerBase>()
                                        .in(DwdShbDsTransformerBase::getPsrId, ids)
                                        .eq(DwdShbDsTransformerBase::getPsrState, "20")
                                        .isNotNull(DwdShbDsTransformerBase::getName));
            for (DwdShbDsTransformerBase row : rows) {
                map.put(row.getPsrId(), row.getName());
            }
            log.info("配变名称映射加载完成: 配变{}台, 命中{}条, 未命中{}条",
                    ids.size(), map.size(), ids.size() - map.size());
        } catch (Exception e) {
            log.warn("加载配变名称映射失败", e);
        }
        return map;
    }

    private void buildFeederMappingRecursive(List<FeederGroupTreeNode> nodes, Map<String, String[]> map,
                                              String parentSubsId, String parentCountyId) {
        if (nodes == null) return;
        for (FeederGroupTreeNode node : nodes) {
            String subsId = parentSubsId, countyId = parentCountyId;
            if (node.getType() == 2) countyId = node.getId();
            else if (node.getType() == 3) subsId = node.getId();
            else if (node.getType() == 4) map.put(node.getId(), new String[]{subsId, countyId, node.getName()});
            if (node.getChildren() != null && !node.getChildren().isEmpty())
                buildFeederMappingRecursive(node.getChildren(), map, subsId, countyId);
        }
    }

    // ============================================================
    // 覆盖更新
    // ============================================================

    @Override
    public void processSnapshotOverwrite(String snapshotJson) {
        JSONObject root = JSON.parseObject(snapshotJson);
        if (root == null) return;

        JSONArray pointTimes = root.getJSONArray("pointTime");
        if (pointTimes == null || pointTimes.isEmpty()) return;
        LocalDateTime snapTime = LocalDateTime.parse(pointTimes.getString(0), DT_FMT);

        // 先删该断面时刻的存量明细（覆盖更新；无大事务，删除独立提交）
        int deleted = snapshotDetailMapper.delete(
                new LambdaQueryWrapper<FhzgSeSnapshotDetail>()
                        .eq(FhzgSeSnapshotDetail::getSnapshotTime, snapTime));
        // 线损明细同步删除（否则防重会跳过插入，保留旧值导致与断面不一致）
        int deletedLoss = lineLossDetailMapper.delete(
                new LambdaQueryWrapper<FhzgSeLineLossDetail>()
                        .eq(FhzgSeLineLossDetail::getSnapTime, snapTime));
        log.info("状估覆盖更新 snapTime={}, 删除断面{}条, 删除线损{}条", snapTime, deleted, deletedLoss);

        // 重新走正常流程写入（此时防重逻辑不会跳过任何记录）
        // 注意：不包裹大事务——手动补断面耗时可能超 Druid removeAbandonedTimeout，
        // 若删+写整体回滚会导致断面永远无法写入；宁可中途失败下次再补
        processSnapshot(snapshotJson);
    }
}
