package com.Modules.NormalData;

import com.influx.InfluxDBClient;
import com.common.util.SyslogParser;
import com.config.AppProperties;
import com.config.KafkaConsumerProperties;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.client.WriteApi;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.time.LocalDate;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;

import com.common.util.Sm4Util;
@Slf4j
@Component
public class SysLogProcessor {

    @Autowired
    @Qualifier("logNormalProcessorExecutor")
    private Executor taskExecutor;

    @Value("${app.processor.batch-size:100}")
    private int batchSize;

    @Value("${app.processor.process-timeout-ms:30000}")
    private long processTimeoutMs;

    @Autowired
    private AppProperties appProperties;

    @Autowired
    private KafkaConsumerProperties kafkaConsumerProperties;

    @Autowired
    private LogNormalProcessor logNormalProcessor;

    @Autowired
    private com.influx.InfluxDBClient influxClient;

    private final AtomicInteger totalProcessed = new AtomicInteger(0);
    private final AtomicInteger currentBatchCount = new AtomicInteger(0);



    /**
     * 方案一：直接多线程并行处理（推荐）
     * 单线程消费，每条消息独立提交给线程池处理
     */
    public void listen(List<String> records, Acknowledgment ack) {
        try {
            // 处理业务逻辑
            for (String record : records) {
                System.out.println(record);
            }
            // 手动提交偏移量
            ack.acknowledge();  // 告诉Kafka这条消息已成功处理
        } catch (Exception e) {
            // 不提交偏移量，消息会被重新消费
        }
    }
    @KafkaListener(
            topics = "${spring.kafka.consumer.topic:agent-syslog-topic}",
            concurrency = "${spring.kafka.listener.concurrency}"  // 多线程消费
    )
    public void consumeAndParallelProcess(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        if (records.isEmpty()) {
            ack.acknowledge();
            return;
        }
        log.info("开始处理批次消息，数量: {}", records.size());
        currentBatchCount.set(records.size());

        try {
            List<CompletableFuture<Void>> processingFutures = new ArrayList<>();

            // 为每条消息创建独立的处理任务
            for (ConsumerRecord<String, String> record : records) {
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    try {
                        // 异步处理单条消息
                        log.info("收到syslogmessage："+ Sm4Util.decryptCbc(record.value(), appProperties.getSm4Key()));
                        processSingleMessageAsync(record);
                    } catch (Exception e) {
                        log.error("处理消息失败, topic: {}, partition: {}, offset: {}",
                                record.topic(), record.partition(), record.offset(), e);
                    } finally {
                        // 计数器递减
                        if (currentBatchCount.decrementAndGet() == 0) {
                            log.debug("批次所有消息处理完成");
                        }
                    }
                }, taskExecutor);

                processingFutures.add(future);
            }

            // 等待所有处理任务完成（带超时）
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                    processingFutures.toArray(new CompletableFuture[0])
            );

            try {
                allFutures.get(processTimeoutMs, TimeUnit.MILLISECONDS);
                log.info("批次处理完成，总数: {}", records.size());
            } catch (java.util.concurrent.TimeoutException e) {
                log.warn("批次处理超时，已处理: {}/{}",
                        records.size() - currentBatchCount.get(), records.size());
            }

            // 提交偏移量
            ack.acknowledge();
            totalProcessed.addAndGet(records.size());

        } catch (Exception e) {
            log.error("消费处理异常", e);
            ack.acknowledge(); // 异常情况下也提交偏移量，避免阻塞
        }
    }

    /**
     * 方案二：分批多线程处理
     * 将大批次拆分为小批次进行并行处理
     */
    /**
    @KafkaListener(
            topics = "${kafka.topic.log-normal-batch:log-normal-topic-batch}",
            concurrency = "1"
    )
     */
    public void consumeAndBatchProcess(List<ConsumerRecord<String, String>> records,
                                       Acknowledgment ack) {
        if (records.isEmpty()) {
            ack.acknowledge();
            return;
        }

        log.info("开始分批处理消息，总数: {}", records.size());

        try {
            List<CompletableFuture<Void>> batchFutures = new ArrayList<>();

            // 将消息列表拆分为多个批次进行并行处理
            for (int i = 0; i < records.size(); i += batchSize) {
                int end = Math.min(i + batchSize, records.size());
                final List<ConsumerRecord<String, String>> batch = records.subList(i, end);
                final int batchIndex = i / batchSize + 1;

                CompletableFuture<Void> batchFuture = CompletableFuture.runAsync(() -> {
                    try {
                        processMessageBatch(batch, batchIndex);
                    } catch (Exception e) {
                        log.error("批次处理异常, 批次: {}", batchIndex, e);
                    }
                }, taskExecutor);

                batchFutures.add(batchFuture);
            }

            // 等待所有批次处理完成
            CompletableFuture<Void> allBatches = CompletableFuture.allOf(
                    batchFutures.toArray(new CompletableFuture[0])
            );

            allBatches.get(processTimeoutMs, TimeUnit.MILLISECONDS);

            // 提交偏移量
            ack.acknowledge();
            log.info("全部分批处理完成，总数: {}", records.size());
            totalProcessed.addAndGet(records.size());

        } catch (Exception e) {
            log.error("分批消费处理异常", e);
            ack.acknowledge();
        }
    }

    /**
     * 方案三：带优先级的处理
     * 根据消息内容决定处理优先级
     */
    /**
    @KafkaListener(
            topics = "${kafka.topic.log-normal-priority:log-normal-topic-priority}",
            concurrency = "1"
    )
     */
    public void consumeWithPriority(List<ConsumerRecord<String, String>> records,
                                    Acknowledgment ack) {
        if (records.isEmpty()) {
            ack.acknowledge();
            return;
        }

        log.info("开始优先级处理消息，数量: {}", records.size());

        try {
            List<CompletableFuture<Void>> highPriorityFutures = new ArrayList<>();
            List<CompletableFuture<Void>> normalPriorityFutures = new ArrayList<>();

            // 根据消息内容分类处理
            for (ConsumerRecord<String, String> record : records) {
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    processSingleMessageAsync(record);
                }, taskExecutor);

                // 根据消息内容判断优先级（这里简单根据key判断）
                if (isHighPriorityMessage(record)) {
                    highPriorityFutures.add(future);
                } else {
                    normalPriorityFutures.add(future);
                }
            }

            // 优先处理高优先级任务
            if (!highPriorityFutures.isEmpty()) {
                CompletableFuture.allOf(
                        highPriorityFutures.toArray(new CompletableFuture[0])
                ).get(processTimeoutMs / 2, TimeUnit.MILLISECONDS);
                log.info("高优先级消息处理完成: {}", highPriorityFutures.size());
            }

            // 然后处理普通优先级任务
            if (!normalPriorityFutures.isEmpty()) {
                CompletableFuture.allOf(
                        normalPriorityFutures.toArray(new CompletableFuture[0])
                ).get(processTimeoutMs / 2, TimeUnit.MILLISECONDS);
                log.info("普通优先级消息处理完成: {}", normalPriorityFutures.size());
            }

            ack.acknowledge();
            totalProcessed.addAndGet(records.size());

        } catch (Exception e) {
            log.error("优先级消费处理异常", e);
            ack.acknowledge();
        }
    }

    /**
     * 异步处理单条消息
     */
    @Async("logNormalProcessorExecutor")
    public void processSingleMessageAsync(ConsumerRecord<String, String> record) {
        long startTime = System.currentTimeMillis();

        try {
            //Message进行SM4解密
            String Sm4message=Sm4Util.decryptCbc(record.value(), appProperties.getSm4Key());

            String sysLogUUID =getSysLogUUID();
            String strDeviceInfo= SyslogParser.substringBeforeFirstChar(Sm4message,']');
            Map<String,String> mapdev  =SyslogParser.parseKeyValuePairs(strDeviceInfo);

            // 初始化 InfluxDB 客户端
            Point point = Point.measurement("syslog_security_"+sysLogUUID.substring(0,8) )
                    .addTag("device_id", mapdev.get("device_id")) // 添加设备ID标签
                    .addTag("device_collect_id", mapdev.get("device_collect_id")) // 添加探针ID标签
                    .addTag("uuid", sysLogUUID)   //syslog uuid
                    .addTag("topic", kafkaConsumerProperties.getTopic())  //kafka topic
                    .addField("message", Sm4message) // 添加字段
                    .addField("receive_time", mapdev.get("receive_time")) // 添加字段
                    .addField("uuid", sysLogUUID)
                    .time(System.currentTimeMillis(), WritePrecision.MS) ;// 毫秒级时间戳
            influxClient.writePoint(point);

            System.out.println("influxdb wirte syslog ,value:"+  record.key());
            // 日志信息插入pg XdrHoneypot 表
            //insertSingleRecord( record.value());

            String syslogMessage= Sm4message;
            //使用注入的 Spring Bean 进行标准化处理
            logNormalProcessor.process(syslogMessage, sysLogUUID, null);
            System.out.println("insert  postgres syslog ,value:"+  record.key());
            long costTime = System.currentTimeMillis() - startTime;
            log.debug("消息处理完成, offset: {}, 耗时: {}ms", record.offset(), costTime);

        } catch (Exception e) {
            log.error("异步处理消息失败, offset: {}", record.offset(), e);
            // 这里可以添加重试逻辑或死信队列处理
        }
    }





    /**
     * 处理消息批次
     */
    private void processMessageBatch(List<ConsumerRecord<String, String>> batch, int batchIndex) {
        long startTime = System.currentTimeMillis();
        log.debug("开始处理批次: {}, 数量: {}", batchIndex, batch.size());

        try {
            List<CompletableFuture<Void>> futures = new ArrayList<>();

            // 批次内并行处理
            for (ConsumerRecord<String, String> record : batch) {
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    processSingleMessageAsync(record);
                }, taskExecutor);
                futures.add(future);
            }

            // 等待批次内所有任务完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                    .get(processTimeoutMs, TimeUnit.MILLISECONDS);

            long costTime = System.currentTimeMillis() - startTime;
            log.info("批次处理完成: {}, 数量: {}, 总耗时: {}ms",
                    batchIndex, batch.size(), costTime);

        } catch (Exception e) {
            log.error("批次处理异常: {}", batchIndex, e);
        }
    }

    /**
     * 判断是否为高优先级消息
     */
    private boolean isHighPriorityMessage(ConsumerRecord<String, String> record) {
        // 简单的优先级判断逻辑
        // 可以根据key、header、或消息内容来判断
        return record.key() != null &&
                (record.key().contains("ERROR") || record.key().contains("URGENT"));
    }

    /**
     * 消息解析（模拟）
     */
    private String parseMessage(ConsumerRecord<String, String> record) {
        // 模拟解析耗时
        try {
            Thread.sleep(1);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return record.value();
    }

    /**
     * 业务逻辑处理（模拟）
     */
    private void processBusinessLogic(String message) {
        // 模拟业务处理耗时
        try {
            Thread.sleep(5); // 模拟5ms处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        // 这里可以添加具体的业务逻辑
        // 例如：数据转换、验证、丰富等
    }

    /**
     * 获取处理统计信息
     */
    public ProcessingStats getProcessingStats() {
        return new ProcessingStats(
                totalProcessed.get(),
                currentBatchCount.get()
        );
    }

    /**
     * 处理统计信息类
     */
    public static class ProcessingStats {
        private final int totalProcessed;
        private final int currentBatchCount;

        public ProcessingStats(int totalProcessed, int currentBatchCount) {
            this.totalProcessed = totalProcessed;
            this.currentBatchCount = currentBatchCount;
        }

        public int getTotalProcessed() { return totalProcessed; }
        public int getCurrentBatchCount() { return currentBatchCount; }
    }

    /**
     * 获取日志信息UUID，格式: yyyyMMddxxxxxxxx
     * @return
     */
    private static String  getSysLogUUID ()
    {
        // 获取当前日期
        LocalDate currentDate = LocalDate.now();
        // 定义格式 (yyyyMMdd)
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
        // 格式化日期
        String formattedDate = currentDate.format(formatter);
        return formattedDate +"-"+ UUID.randomUUID() ;
    }

}
