package com.influx;

import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.WriteApi;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.HealthCheck;
import com.influxdb.client.write.Point;
import com.config.AppConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.influxdb.client.QueryApi;
import java.time.Instant;
import java.time.format.DateTimeFormatter;


public class InfluxDBClient implements AutoCloseable {
    private static final Logger logger = LoggerFactory.getLogger(InfluxDBClient.class);

    private final com.influxdb.client.InfluxDBClient influxDB;
    private final WriteApi writeApi;
    private final WriteApiBlocking writeApiBlocking;
    private final String bucket;
    private final String org;
    private final QueryApi queryApi;

    public InfluxDBClient() {

        this.influxDB = InfluxDBClientFactory.create(
                AppConfig.getInfluxUrl(),
                AppConfig.getInfluxToken().toCharArray(),
                AppConfig.getInfluxOrg(),
                AppConfig.getInfluxBucket()
        );
        this.writeApi = influxDB.makeWriteApi(InfluxDBConfig.getWriteOptions());
        this.writeApiBlocking = influxDB.getWriteApiBlocking();
        this.bucket = AppConfig.getInfluxBucket();
        this.org = AppConfig.getInfluxOrg();
        this.queryApi = influxDB.getQueryApi();

        // 检查连接状态
        checkConnection();
    }

    private void checkConnection() {
        try {
            HealthCheck health = influxDB.health();
            if (health.getStatus() == HealthCheck.StatusEnum.PASS) {
                logger.info("InfluxDB connection successful: {}", health.getMessage());
            } else {
                logger.warn("InfluxDB health check: {}", health.getMessage());
            }
        } catch (Exception e) {
            logger.error("Failed to connect to InfluxDB: {}", e.getMessage());
            throw new RuntimeException("InfluxDB connection failed", e);
        }
    }

    /**
     * 写入数据点（异步）
     */
    public void writePoint(Point point) {
        try {
            writeApi.writePoint(point);
            // 优雅关闭 WriteApi，确保数据发送完成
        } catch (Exception e) {
            logger.error("Failed to write point to InfluxDB: {}", e.getMessage());
            throw e;
        }
    }

    /**
     * 写入数据点（同步，用于重要数据）
     */
    public void writePointBlocking(Point point) {
        try {
            writeApiBlocking.writePoint(point);
        } catch (Exception e) {
            logger.error("Failed to write point to InfluxDB (blocking): {}", e.getMessage());
            throw e;
        }

    }

    /**
     * 写入原始 Line Protocol（异步）
     */
    public void writeRecord(String lineProtocol) {
        try {
            writeApi.writeRecord(InfluxDBConfig.getWritePrecision(),lineProtocol);
        } catch (Exception e) {
            logger.error("Failed to write record to InfluxDB: {}", e.getMessage());
        }
    }
    /**
     * 基础查询 - 查询所有数据
     */
    public List<FluxTable> queryAllData(String measurement) {
        String fluxQuery = String.format(
                "from(bucket: \"%s\") " +
                        "|> range(start: -1d) " +
                        "|> filter(fn: (r) => r._field != \"uuid\") " + // 新增此行，过滤掉field列的uuid
                        "|> filter(fn: (r) => r._measurement == \"%s\")",
                bucket, measurement
        );

        QueryApi queryApi = influxDB.getQueryApi();
        return queryApi.query(fluxQuery, org);
    }


    /**
     * 执行Flux查询并返回结果列表
     */
    public List<Map<String, Object>> queryFlux(String fluxQuery) {
        List<Map<String, Object>> resultList = new ArrayList<>();
        try {
            List<FluxTable> tables = queryApi.query(fluxQuery, org);

            for (FluxTable table : tables) {
                for (FluxRecord record : table.getRecords()) {
                    Map<String, Object> dataMap = new HashMap<>();

                    // 添加时间戳
                    dataMap.put("time", record.getTime());

                    // 添加测量名称
                    dataMap.put("measurement", record.getMeasurement());

                    // 添加所有字段值
                    dataMap.putAll(record.getValues());

                    resultList.add(dataMap);
                }
            }
            logger.debug("Query executed successfully, returned {} records", resultList.size());
        } catch (Exception e) {
            logger.error("Failed to execute Flux query: {}", e.getMessage());
            throw new RuntimeException("Query execution failed", e);
        }
        return resultList;
    }

    /**
     * 按时间范围查询数据
     */
    public List<Map<String, Object>> queryByTimeRange(String measurement, Instant startTime, Instant stopTime) {
        String fluxQuery = String.format(
                "from(bucket: \"%s\") " +
                        "|> range(start: %s, stop: %s) " +
                        "|> filter(fn: (r) => r._measurement == \"%s\") " +
                        "|> filter(fn: (r) => r._field != \"uuid\") " + // 新增此行，过滤掉field列的uuid
                        "|> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")",
                bucket, startTime, stopTime, measurement
        );
        return queryFlux(fluxQuery);
    }

    /**
     * 按UUID查询特定记录
     */
    public List<Map<String, Object>> queryByUuid(String measurement, String uuid) {
        String fluxQuery = String.format(
                "from(bucket: \"%s\") " +
                        "|> range(start: 0) " +
                        "|> filter(fn: (r) => r._measurement == \"%s\") " +
                        "|> filter(fn: (r) => r.uuid == \"%s\") " +
                        "|> filter(fn: (r) => r._field != \"uuid\") " + // 新增此行，过滤掉field列的uuid
                        "|> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")",
                bucket, measurement, uuid
        );
        return queryFlux(fluxQuery);
    }

    /**
     * 按设备ID和时间范围查询
     */
    public List<Map<String, Object>> queryByDeviceIdAndTime(String measurement, String deviceId, Instant startTime, Instant stopTime) {
        String fluxQuery = String.format(
                "from(bucket: \"%s\") " +
                        "|> range(start: %s, stop: %s) " +
                        "|> filter(fn: (r) => r._measurement == \"%s\") " +
                        "|> filter(fn: (r) => r.device_id == \"%s\") " +
                        "|> filter(fn: (r) => r._field != \"uuid\") " + // 新增此行，过滤掉field列的uuid
                        "|> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")",
                bucket, startTime, stopTime, measurement, deviceId
        );
        return queryFlux(fluxQuery);
    }

    /**
     * 模糊查询message内容
     */
    public List<Map<String, Object>> queryByMessageContent(String measurement, Instant startTime, Instant stopTime, String keyword) {
        String fluxQuery = String.format(
                "from(bucket: \"%s\") " +
                        "|> range(start: %s, stop: %s) " +
                        "|> filter(fn: (r) => r._measurement == \"%s\") " +
                        "|> filter(fn: (r) => r._field == \"message\") " +
                        "|> filter(fn: (r) => r._value =~ /.*%s.*/) " +
                        "|> filter(fn: (r) => r._field != \"uuid\") " + // 新增此行，过滤掉field列的uuid
                        "|> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")",
                bucket, startTime, stopTime, measurement, keyword
        );
        return queryFlux(fluxQuery);
    }



    @Override
    public void close() {
        if (writeApi != null) {
            writeApi.close();
        }
        if (influxDB != null) {
            influxDB.close();
        }
        logger.info("InfluxDB client closed");
    }

    // Getter 方法
    public String getBucket() {
        return bucket;
    }

    public String getOrg() {
        return org;
    }
}