package com.rj.scheduler;

import com.rj.config.AppConfig;
import com.rj.config.YihangyiVllmAsrExecutorMode;
import com.rj.config.YihangyiVllmAsrProperties;
import com.rj.service.YihangyiVllmAsrService;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import java.nio.file.Path;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * yihangyi 音频 ASR 轮询。<strong>不使用 Spring {@code @Scheduled}</strong>，避免与全局定时线程池争抢；
 * 在独立单线程上按固定间隔执行（与 Python 常驻循环等价），批内仍可选 POOL 并行转写。
 */
@Slf4j
@Component
@RequiredArgsConstructor
@ConditionalOnProperty(name = "app.audio.upload.yihangyi.asr.enabled", havingValue = "true")
public class YihangyiVllmAsrScheduler {

    private final YihangyiVllmAsrProperties properties;
    private final YihangyiVllmAsrService yihangyiVllmAsrService;
    private final AppConfig appConfig;

    private ExecutorService workerPool;
    private ScheduledExecutorService pollExecutor;

    @PostConstruct
    void start() {
        if (properties.getExecutorMode() == YihangyiVllmAsrExecutorMode.POOL) {
            int n = Math.max(1, properties.getPoolSize());
            workerPool = Executors.newFixedThreadPool(n, r -> {
                Thread t = new Thread(r, "yihangyi-asr-worker");
                t.setDaemon(true);
                return t;
            });
            log.info("yihangyi ASR 批内使用线程池，线程数={}", n);
        } else {
            log.info("yihangyi ASR 批内单线程顺序处理");
        }
        long intervalMs = Math.max(1000L *60*3, properties.getPollIntervalMs());
        pollExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "yihangyi-asr-poll");
            t.setDaemon(true);
            return t;
        });
        pollExecutor.scheduleWithFixedDelay(this::pollSafe, 0, intervalMs, TimeUnit.MILLISECONDS);
        log.info("yihangyi ASR 独立轮询线程已启动，fixedDelay={}ms（不占 Spring @Scheduled 线程池）", intervalMs);
    }

    @PreDestroy
    void shutdown() {
        if (pollExecutor != null) {
            pollExecutor.shutdown();
            try {
                if (!pollExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
                    pollExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                pollExecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        if (workerPool != null) {
            workerPool.shutdown();
            try {
                if (!workerPool.awaitTermination(60, TimeUnit.SECONDS)) {
                    workerPool.shutdownNow();
                }
            } catch (InterruptedException e) {
                workerPool.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }

    private void pollSafe() {
        try {
            log.info("yihangyi ASR 音频-----》文本， 轮询开始");
            pollAndTranscribe();
        } catch (Throwable t) {
            log.error("yihangyi ASR 轮询未捕获异常", t);
        }
    }

    void pollAndTranscribe() {
        if (!appConfig.getScheduler().isStart()) {
            log.info("yihangyi ASR 调度跳过：app.scheduler.start=false");
            return;
        }
        if (properties.isSkipOnWindows()
                && System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win")) {
            log.info("yihangyi ASR 调度跳过：当前为 Windows 且 app.audio.upload.yihangyi.asr.skip-on-windows=true");
            return;
        }
        Path watchDir = Path.of(properties.getWatchDir());
        int maxBatch = Math.max(1, properties.getMaxBatch());
        long t0 = System.nanoTime();
        log.info(
                "yihangyi ASR 调度开始：watchDir={} executorMode={} maxBatch={} pollIntervalMs={}",
                watchDir.toAbsolutePath(),
                properties.getExecutorMode(),
                maxBatch,
                properties.getPollIntervalMs());

        if (!java.nio.file.Files.isDirectory(watchDir)) {
            log.warn("yihangyi ASR 调度结束（异常）：监视目录不存在或不是文件夹: {}", watchDir.toAbsolutePath());
            return;
        }
        int pendingTotal = yihangyiVllmAsrService.countPendingAudio(watchDir);
        log.info("yihangyi ASR 待转写音频文件总数: {}", pendingTotal);

        List<Path> pending = yihangyiVllmAsrService.listNewestPendingAudio(watchDir, maxBatch);
        if (pending.isEmpty()) {
            log.info(
                    "yihangyi ASR 调度结束：本批无待处理文件，待转写总数={}，耗时={}ms",
                    pendingTotal,
                    (System.nanoTime() - t0) / 1_000_000);
            return;
        }

        log.info("yihangyi ASR 本批将处理 {} 个文件: {}", pending.size(),
                pending.stream().map(p -> p.getFileName().toString()).toList());

        if (properties.getExecutorMode() == YihangyiVllmAsrExecutorMode.POOL && workerPool != null) {
            List<CompletableFuture<Void>> futures = pending.stream()
                    .map(p -> CompletableFuture.runAsync(() -> yihangyiVllmAsrService.processOneFile(p), workerPool))
                    .toList();
            CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])).join();
        } else {
            for (Path p : pending) {
                yihangyiVllmAsrService.processOneFile(p);
            }
        }
        log.info(
                "yihangyi ASR 调度结束：本批已处理 {} 个文件，待转写总数（处理前统计）={}，耗时={}ms",
                pending.size(),
                pendingTotal,
                (System.nanoTime() - t0) / 1_000_000);
    }
}
