package com.yeechart.dotMatrix.oss;

import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSClient;
import com.aliyun.oss.model.ListObjectsRequest;
import com.aliyun.oss.model.OSSObjectSummary;
import com.aliyun.oss.model.ObjectListing;
import com.aliyun.oss.model.PutObjectResult;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;


@Component
@Slf4j
public class OSSUpload {
    private  final int THREAD_POOL_SIZE = 10; // 线程池大小
    private  final ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

    @Value("${oss.endpoint}")
    private String endpoint;
    @Value("${oss.accesskeyid}")
    private String accessKeyId;
    @Value("${oss.accesskeysecret}")
    private String accessKeySecret;
    @Value("${oss.bucket}")
    private String bucketName;

    public String getEndpoint() {
        return endpoint;
    }
    private final int MAX_RETRIES = 5;// 5个并发重试
    private final Semaphore semaphore = new Semaphore(MAX_RETRIES); // 并发重试
    private final BlockingQueue<OSSRetryTask> retryQueue = new ArrayBlockingQueue<>(400); // 假定的重试任务队列大小
    private final int MAX_RETRY_ATTEMPTS = 3; // 最大重试次数
    private final long SLEEP_INTERVAL_MS = 100; // 每次重试之间的延迟
    private final ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(10, new ThreadPoolExecutor.CallerRunsPolicy());
    protected Logger logger = LoggerFactory.getLogger(OSSUpload.class);


    private  OSSClient ossClient;//OSS
    @PostConstruct
    public  void init() {
        ossClient = new OSSClient(endpoint, accessKeyId, accessKeySecret);
    }


    @PreDestroy
    public void destroy() {
        if (ossClient != null) {
            ossClient.shutdown();
        }
        executor.shutdown();

        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }

        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(60, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
        }
    }

    /**
     * 上传文件
     *
     * @param data
     * @param originalFileName
     * @param onSuccessCallback
     * @param onFailureCallback
     * @return
     */
    public void enqueueUploadWithRetryAndCallback(byte[] data, String originalFileName,
                                                  Consumer<String> onSuccessCallback,
                                                  Consumer<Throwable> onFailureCallback) {

        OSSRetryTask task = new OSSRetryTask(data, originalFileName,
                onSuccessCallback, onFailureCallback);
        int retryAttempts = 0;
        while (retryAttempts <= MAX_RETRY_ATTEMPTS) {
            try {
                if (retryQueue.offer(task, 1, TimeUnit.SECONDS)) {
                    processRetryQueue();
                    return; // 成功加入队列，结束重试
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                onFailureCallback.accept(new RuntimeException("将任务添加到队列时中断", e));
                return;
            }

            // 如果队列满，等待一段时间后重试
            retryAttempts++;
            logger.warn("重试队列已满，将在 #{} 毫秒内重试。尝试 #{}", SLEEP_INTERVAL_MS, retryAttempts);
            try {
                LockSupport.parkNanos(100_000_000);
//                Thread.sleep(SLEEP_INTERVAL_MS);
            } catch (Exception e) {
                Thread.currentThread().interrupt();
                onFailureCallback.accept(new RuntimeException("等待重试时中断", e));
                return;
            }
        }

        // 达到最大重试次数后，执行失败回调
        onFailureCallback.accept(new Exception("Failed to enqueue task after maximum retry attempts"));


    }

    private void processRetryQueue() {
        if (executor.isShutdown()) {
            return;
        }
        executor.submit(this::processRetryTasks);
    }

    private void processRetryTasks() {
        while (true) {
            try {
                OSSRetryTask task = retryQueue.poll(1, TimeUnit.SECONDS);
                if (task == null) {
                    break;
                } // 队列为空则退出
                doUploadWithRetry(task);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    //private int count = 0;
    public void doUploadWithRetry(OSSRetryTask task) {
        if (task == null) {
            return;
        }

        String fileName = task.getFileName();

        InputStream fileStream = new ByteArrayInputStream(task.getData());
        Random random = new Random();

        try {
            for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
                try {
                    semaphore.acquire(); // 获取信号量，限制并发数

//                    PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, fileName, fileStream);
                    PutObjectResult putResult = ossClient.putObject(bucketName, fileName, fileStream);
                    if (putResult != null) {
                        // logger.info("OSS上传成功，次数为 {}，文件为 {} , 调用次数 {}", attempt, fileName,++count);
                        // 上传成功，释放信号量
                        semaphore.release();
                        String url = "http://" + endpoint + "/" + fileName;//"https://"  bucketName + "." +
                        task.getOnSuccessCallback().accept(url);
                        return;
                    } else {
                        logger.warn("OSS上传尝试 " + attempt + " 次 失败,无返回结果");
                    }
                } catch (Exception e) {
                    logger.warn("OSS上传尝试 {} 次失败，错误为 : {}", attempt, e.getMessage());
                } finally {
                    if (semaphore.availablePermits() < 1) {
                        semaphore.release(); // 确保信号量正确释放
                    }
                }

                if (attempt < MAX_RETRIES) {
                    long delayMs = random.nextInt(500) + 100; // 随机退避，范围100ms至600ms
                    scheduler.schedule(() -> doUploadWithRetry(task), delayMs, TimeUnit.MILLISECONDS);
                    return; // 退出当前方法，等待重试
                }
            }
        } finally {

            try {
                fileStream.close();

            } catch (IOException e) {

            }

        }

        logger.warn("OSS上传尝试 最大次数 失败");

        task.getOnFailureCallback().accept(new Exception("上传异常."));

    }


    /**
     * 删除指定路径的文件
     *
     * @param filePath 完整的文件路径（包含Bucket名称）
     */
    public void deleteFile(String filePath) {
        //取出文件路径
        URL url = null;
        try {
            url = new URL(filePath);
            String objectKey = url.getPath().substring(1); // 提取Object Key，去掉开头的'/'
            ossClient.deleteObject(bucketName, objectKey);
        } catch (Exception e) {
            logger.error("文件路径转换时，路径" + filePath + "错误" + e);
        }

    }

    /**
     * 获取指定前缀下所有文件的内容
     *
     * @param prefix 文件路径前缀
     * @return 匹配前缀的所有文件内容列表，每个元素为一个文件的字节内容
     */
    public CompletableFuture<Map<String, byte[]>> getFileContentsByPrefix(String prefix) throws IOException {

        return CompletableFuture.supplyAsync(() -> {
//            OSS ossClient = new OSSClientBuilder().build(endpoint, accessKeyId, accessKeySecret); // 确保endpoint等参数已定义

            ListObjectsRequest listObjectsRequest = new ListObjectsRequest(bucketName)
                    .withPrefix(prefix)
                    .withMaxKeys(1000); // 根据实际情况调整

            Map<String, byte[]> fileNameContentMap = new HashMap<>();

            try {
                ObjectListing objectListing;
                do {
                    objectListing = ossClient.listObjects(listObjectsRequest);

                    for (OSSObjectSummary objectSummary : objectListing.getObjectSummaries()) {
                        String fileName = extractFileNameWithFolder(objectSummary.getKey(), prefix);

                        byte[] content = downloadFileContent(ossClient, bucketName, objectSummary.getKey());

                        fileNameContentMap.put(fileName, content);
                    }

                    listObjectsRequest.setMarker(objectListing.getNextMarker());
                } while (objectListing.isTruncated());
            } catch (Exception e) {
                logger.error("OSS获取文件,出现异常" + e);
            }
//            finally {
//                ossClient.shutdown(); // 确保OSS客户端在最后被关闭
//            }

            return fileNameContentMap;

        });

    }

    /**
     * 提取文件名，保留上一层文件夹路径
     *
     * @param key
     * @param prefix
     * @return
     */
    public String extractFileNameWithFolder(String key, String prefix) {
        // 提取文件名并保留文件夹路径
        if (key.startsWith(prefix)) {
            key = key.substring(prefix.length());
        }
        return key;
    }

    /**
     * 只获取文件名称
     *
     * @param key
     * @return
     */
    public String extractFileNameFromKey(String key) {
        return key.substring(key.lastIndexOf("/") + 1);
    }

    public byte[] downloadFileContent(OSS ossClient, String bucketName, String objectKey) throws IOException {
        try (InputStream objectContent = ossClient.getObject(bucketName, objectKey).getObjectContent();
             ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
            byte[] buffer = new byte[1024];
            int bytesRead;
            while ((bytesRead = objectContent.read(buffer)) != -1) {
                baos.write(buffer, 0, bytesRead);
            }
            return baos.toByteArray();
        }
    }

}
