package com.moka.gdtauto.service;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.moka.gdtauto.client.TencentAdsApiClientFactory;
import com.moka.gdtauto.common.Constant;
import com.moka.gdtauto.entity.AdPlan;
import com.moka.gdtauto.entity.AdPlanConf;
import com.moka.gdtauto.mapper.AdPlanConfMapper;
import com.moka.gdtauto.mapper.AdPlanMapper;
import com.tencent.ads.model.v3.DailyReportsGetResponseData;
import com.tencent.ads.model.v3.DailyReportsFilteringStruct;
import com.tencent.ads.model.v3.FilterOperator;
import com.tencent.ads.model.v3.ReportDateRange;
import com.tencent.ads.model.v3.ReportStruct;
import com.tencent.ads.model.v3.TimeLine;
import com.tencent.ads.v3.TencentAds;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 广告计划日报表同步服务
 * 按广告组级别同步日报数据到 ad_plan 表
 *
 * @author moka
 * @since 2026-03-10
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class AdPlanDailyReportService {

    private final TencentAdsApiClientFactory clientFactory;
    private final AdPlanConfMapper adPlanConfMapper;
    private final AdPlanMapper adPlanMapper;

    /** REQUEST_TIME 口径同步字段 */
    private static final List<String> REQ_FIELDS = Arrays.asList(
            "adgroup_id", "date", "cost", "income_val_1", "income_roi_1",
            "view_count", "valid_click_count", "ctr", "conversions_cost" ,"scan_follow_user_cost"
    );

    /** REPORTING_TIME 口径同步字段 */
    private static final List<String> REP_FIELDS = Arrays.asList(
            "adgroup_id", "date", "income_val_1", "income_roi_1"
    );

    private static final int PAGE_SIZE = 100;
    private static final int ADGROUP_BATCH_LIMIT = 100;

    @Async
    public void syncDailyReport(Long confId) {
        LocalDateTime start = LocalDateTime.now();
        LocalDateTime end = LocalDateTime.now();
        syncReqDailyReport(confId, start, end);
        syncRepDailyReport(confId, start, end);
    }

    /**
     * 同步 REQUEST_TIME 口径日报数据到 ad_plan 表
     *
     * @param startTime 开始时间
     * @param endTime   结束时间
     */
    public void syncReqDailyReport(Long confId, LocalDateTime startTime, LocalDateTime endTime) {
        log.info("[syncReqDailyReport] 开始同步 REQUEST_TIME 口径日报 [{}, {}]", startTime, endTime);
        doSyncDailyReport(confId, startTime, endTime, TimeLine.REQUEST_TIME.getValue(), true);
    }

    /**
     * 同步 REPORTING_TIME 口径日报数据到 ad_plan 表
     *
     * @param startTime 开始时间
     * @param endTime   结束时间
     */
    public void syncRepDailyReport(Long confId, LocalDateTime startTime, LocalDateTime endTime) {
        log.info("[syncRepDailyReport] 开始同步 REPORTING_TIME 口径日报 [{}, {}]", startTime, endTime);
        doSyncDailyReport(confId, startTime, endTime, TimeLine.REPORTING_TIME.getValue(), false);
    }

    /**
     * 执行同步逻辑
     *
     * @param startTime   开始时间
     * @param endTime     结束时间
     * @param timeLine    时间口径
     * @param isReqTime   是否为 REQUEST_TIME 口径（决定更新哪些字段）
     */
    private void doSyncDailyReport(Long confId, LocalDateTime startTime, LocalDateTime endTime, String timeLine, boolean isReqTime) {
        // 1. 分页查询 create_time 在范围内的 adPlanConf
        int confPageNum = 1;
        int totalUpdated = 0;
        Map<Long, Integer> confUpdatedCount = new ConcurrentHashMap<>();

        while (true) {
            LambdaQueryWrapper<AdPlanConf> confWrapper = new LambdaQueryWrapper<>();
            confWrapper
                    .eq(confId != null, AdPlanConf::getId, confId)
                    .eq(AdPlanConf::getDeleted, 0)
                    .orderByDesc(AdPlanConf::getId);
            Page<AdPlanConf> confPage = adPlanConfMapper.selectPage(new Page<>(confPageNum, PAGE_SIZE), confWrapper);
            List<AdPlanConf> confList = confPage.getRecords();

            if (confList.isEmpty()) {
                break;
            }

            // 2. 每个 conf 启动一个线程处理
            List<CompletableFuture<Void>> futures = confList.stream()
                    .map(conf -> CompletableFuture.runAsync(() -> {
                        try {
                            int updated = processConf(conf, startTime, endTime, timeLine, isReqTime);
                            confUpdatedCount.put(conf.getId(), updated);
                        } catch (Exception e) {
                            log.error("[doSyncDailyReport] 处理 confId={} 失败", conf.getId(), e);
                        }
                    }))
                    .collect(Collectors.toList());

            // 等待所有线程完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

            if (!confPage.hasNext()) {
                break;
            }
            confPageNum++;
        }

        totalUpdated = confUpdatedCount.values().stream().mapToInt(Integer::intValue).sum();
        log.info("[doSyncDailyReport] 同步完成 timeLine={} totalUpdated={}", timeLine, totalUpdated);
    }

    /**
     * 处理单个 AdPlanConf
     */
    private int processConf(AdPlanConf conf, LocalDateTime startTime, LocalDateTime endTime, String timeLine, boolean isReqTime) throws Exception {
        String jobNumber = conf.getJobNumber();
        TencentAds tencentAds = clientFactory.getTencentAds(jobNumber, Constant.MAIN_ORG_ACCOUNT_ID);

        String startStr = startTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
        String endStr = endTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
        ReportDateRange dateRange = new ReportDateRange();
        dateRange.setStartDate(startStr);
        dateRange.setEndDate(endStr);

        log.info("[广告日报] start={} end={} timeLine={} isReqTime={}", startStr, endStr, timeLine, isReqTime);

        List<String> groupBy = Arrays.asList("adgroup_id", "date");
        List<String> fields = isReqTime ? REQ_FIELDS : REP_FIELDS;

        int totalUpdated = 0;
        int planPageNum = 1;

        // 3. 分页查询该 conf 下的所有 adPlan
        while (true) {
            LambdaQueryWrapper<AdPlan> planWrapper = new LambdaQueryWrapper<>();
            planWrapper.eq(AdPlan::getAdPlanConfId, conf.getId())
                    .isNotNull(AdPlan::getAdgroupId)
                    .gt(AdPlan::getAdgroupId, 0)
                    .eq(AdPlan::getDeleted, 0)
                    .orderByAsc(AdPlan::getId);
            Page<AdPlan> planPage = adPlanMapper.selectPage(new Page<>(planPageNum, PAGE_SIZE), planWrapper);
            List<AdPlan> plans = planPage.getRecords();

            if (plans.isEmpty()) {
                break;
            }

            // 4. 按 accountId 分组，每 100 个 adgroup_id 一批调用 API
            Map<Long, List<AdPlan>> groupByAccount = plans.stream()
                    .filter(p -> p.getAccountId() != null && p.getAdgroupId() != null)
                    .collect(Collectors.groupingBy(AdPlan::getAccountId));

            for (Map.Entry<Long, List<AdPlan>> entry : groupByAccount.entrySet()) {
                Long accountId = entry.getKey();
                List<AdPlan> plansOfAccount = entry.getValue();

                // 按 100 条分批
                int batchCount = (int) Math.ceil((double) plansOfAccount.size() / ADGROUP_BATCH_LIMIT);
                for (int i = 0; i < batchCount; i++) {
                    int fromIdx = i * ADGROUP_BATCH_LIMIT;
                    int toIdx = Math.min(fromIdx + ADGROUP_BATCH_LIMIT, plansOfAccount.size());
                    List<AdPlan> batch = plansOfAccount.subList(fromIdx, toIdx);

                    try {
                        int updated = syncBatch(tencentAds, accountId, batch, dateRange, groupBy, fields, timeLine, isReqTime);
                        totalUpdated += updated;
                    } catch (Exception e) {
                        log.error("[processConf] 同步失败 accountId={} confId={} batch={}", accountId, conf.getId(), i, e);
                    }
                }
            }

            if (!planPage.hasNext()) {
                break;
            }
            planPageNum++;
        }

        return totalUpdated;
    }

    /**
     * 批量同步单批 adPlan
     */
    private int syncBatch(TencentAds tencentAds, Long accountId, List<AdPlan> batch,
                          ReportDateRange dateRange, List<String> groupBy, List<String> fields,
                          String timeLine, boolean isReqTime) throws Exception {

        // 构建 filtering: adgroup_id IN [...]
        List<String> adgroupIdValues = batch.stream()
                .map(p -> String.valueOf(p.getAdgroupId()))
                .collect(Collectors.toList());

        DailyReportsFilteringStruct filter = new DailyReportsFilteringStruct();
        filter.setField("adgroup_id");
        filter.setOperator(FilterOperator.IN);
        filter.setValues(adgroupIdValues);

        List<DailyReportsFilteringStruct> filtering = new ArrayList<>();
        filtering.add(filter);

        // 调用 API
        DailyReportsGetResponseData response = tencentAds.dailyReports().dailyReportsGet(
                "REPORT_LEVEL_ADGROUP",
                dateRange,
                groupBy,
                fields,
                accountId,
                filtering,
                null,  // orderBy
                timeLine,
                1L,
                (long) ADGROUP_BATCH_LIMIT,
                null   // modularDetailMode
        );

        if (response == null) {
            log.warn("[syncBatch] accountId={} 未返回日报数据", accountId);
            return 0;
        }

        // 构建更新数据
        List<AdPlan> toUpdate = new ArrayList<>();
        if (response.getList() != null) {
            for (ReportStruct rs : response.getList()) {
                if (rs.getAdgroupId() == null) {
                    continue;
                }
                AdPlan update = new AdPlan();
                update.setAdgroupId(rs.getAdgroupId());

                if (isReqTime) {
                    // REQUEST_TIME 口径：更新 cost, view_count, valid_click_count, ctr, conversions_cost, req_income_val_1, req_income_roi_1
                    update.setCost(rs.getCost());
                    update.setViewCount(rs.getViewCount());
                    update.setValidClickCount(rs.getValidClickCount());
                    update.setCtr(d2b(rs.getCtr()));
                    update.setConversionsCost(rs.getConversionsCost());
                    update.setReqIncomeVal1(rs.getIncomeVal1());
                    update.setReqIncomeRoi1(d2b(rs.getIncomeRoi1()));
                    update.setScanFollowUserCost(rs.getScanFollowUserCost());
                } else {
                    // REPORTING_TIME 口径：更新 rep_income_val_1, rep_income_roi_1
                    update.setRepIncomeVal1(rs.getIncomeVal1());
                    update.setRepIncomeRoi1(d2b(rs.getIncomeRoi1()));
                }
                update.setUpdateTime(LocalDateTime.now());
                toUpdate.add(update);
            }   
        }

        //无报表数据的，填充0值
        Map<Long, AdPlan> updateMap = toUpdate.stream().collect(Collectors.toMap(AdPlan::getAdgroupId, p -> p));
        batch.forEach(p -> {
            if (!updateMap.containsKey(p.getAdgroupId())) {
                AdPlan update = new AdPlan();
                update.setAdgroupId(p.getAdgroupId());

                if (isReqTime) { 
                    update.setCost(0L);
                    update.setViewCount(0L);
                    update.setValidClickCount(0L);
                    update.setCtr(BigDecimal.ZERO);
                    update.setConversionsCost(0L);
                    update.setReqIncomeVal1(0L);
                    update.setReqIncomeRoi1(BigDecimal.ZERO);
                    update.setScanFollowUserCost(0L);
                } else {
                    update.setRepIncomeVal1(0L);
                    update.setRepIncomeRoi1(BigDecimal.ZERO);
                }
                update.setUpdateTime(LocalDateTime.now());
                toUpdate.add(update);
            }
        });

        if (toUpdate.isEmpty()) {
            return 0;
        }

        // 批量更新
        if (isReqTime) {
            adPlanMapper.batchUpdateReqDailyReport(toUpdate);
        } else {
            adPlanMapper.batchUpdateRepDailyReport(toUpdate);
        }

        log.info("[syncBatch] accountId={} 批量更新 {} 条 timeLine={}", accountId, toUpdate.size(), timeLine);
        return toUpdate.size();
    }

    /** Double → BigDecimal 安全转换 */
    private BigDecimal d2b(Double d) {
        return d == null ? null : BigDecimal.valueOf(d);
    }
}
