package uploadusagedata;

import java.io.FileReader;
import java.io.IOException;
import java.io.Reader;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.time.Instant;
import java.util.Collections;
import java.util.List;
import java.util.TimeZone;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.opencsv.bean.CsvToBeanBuilder;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.transport.ElasticsearchTransport;

public class EsUploader {

  public static void main(String[] args) throws IOException {
    TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
    String csvFilePath = "/Users/prashantsavadi/Documents/dev1-workspace-spring-tool-suite-4-4.16.0.RELEASE/SandBoxJava/network_97b211b9-fd67-4142-ba22-3dc8956bdda6.csv";
    ObjectMapper mapper = new ObjectMapper();
    


  }

  public void processQueue(String inputDirPath, String processedDirPath) {
    Path inputDir = Paths.get(inputDirPath);
    Path processedDir = Paths.get(processedDirPath);

    // 1. Create the processed folder if it doesn't exist
    try {
      if (!Files.exists(processedDir)) Files.createDirectories(processedDir);
    } catch (IOException e) {
      System.err.println("Could not create processed directory: " + e.getMessage());
      return;
    }

    // 2. Scan all CSV files in the folder
    try (DirectoryStream<Path> stream = Files.newDirectoryStream(inputDir, "*.csv")) {
      for (Path filePath : stream) {
        String fileName = filePath.getFileName().toString();
        System.out.println(">>> STARTING: " + fileName);

        try (Reader reader = Files.newBufferedReader(filePath)) {
          // 3. Parse CSV into Beans
          List<NetworkStats> dataList = new CsvToBeanBuilder<NetworkStats>(reader)
              .withType(NetworkStats.class)
              .build()
              .parse();

          // 4. Send to Elasticsearch (Assuming this method returns true/false)
          boolean success = sendToElastic(dataList, fileName);

          if (success) {
            // 5. Move to processed folder on success
            Files.move(filePath, processedDir.resolve(fileName), StandardCopyOption.REPLACE_EXISTING);
            System.out.println(">>> COMPLETED & MOVED: " + fileName);
          } else {
            System.err.println(">>> FAILED: Data not accepted by Elastic for " + fileName);
          }

        } catch (Exception e) {
          System.err.println(">>> ERROR parsing " + fileName + ": " + e.getMessage());
          // File stays in input folder for manual inspection
        }
      }
    } catch (IOException e) {
      System.err.println("Error reading input directory: " + e.getMessage());
    }
  }

  private boolean sendToElastic(List<NetworkStats> dataList, String fileName) {
    
    ElasticsearchTransport transport = null;
    ElasticsearchClient client = ElasticsearchConfig.createClient();

    try {
      List<NetworkStats> dataList = new CsvToBeanBuilder<NetworkStats>(new FileReader(csvFilePath))
          .withType(NetworkStats.class)
          .build()
          .parse();

      BulkRequest.Builder br = new BulkRequest.Builder();
      System.out.println("POST /_bulk");

      for (NetworkStats networkStatistics : dataList) {
        Instant instant = Instant.parse(networkStatistics.getTimestamp());
        networkStatistics.setTimestamp(instant.toString());
        // stats.setFromTime(instant.toEpochMilli());
        // stats.setToTime(instant.toEpochMilli() + 3600000);
        networkStatistics.setType("metering");
        networkStatistics.setComponentGroup("ZT_EDGE_IDENTITIES");
        networkStatistics.setTags(Collections.singletonList("customer"));
        String yearMonth = networkStatistics.getTimestamp().substring(0, 7).replace("-", ".");
        String indexName = "nf_component_usage-" + yearMonth;
        // System.out.println("{ \"index\" : { \"_index\" : \"" + indexName + "\" } }");
        // System.out.println(mapper.writeValueAsString(stats));
        System.out.println("Timestamp for ES:" + instant.toString());
        br.operations(op -> op
            .index(idx -> idx
                .index(indexName)
                .document(networkStatistics)));
      }
      System.out.println("");
      transport = client._transport();
      BulkResponse bulk = client.bulk(br.build());
      bulk.items().forEach(data -> {
        System.out.println("Response " + data.get());

      });
      System.out.println("Successfully uploaded " + dataList.size() + " records to ES.");
      System.out.println("\n// Backfill text generated. Copy the lines above starting with POST /_bulk into Kibana Dev Tools.");


    } catch (Exception e) {
      e.printStackTrace();

    } finally {
      transport.close();


    }
    return false;
  }
}