package com.tzld.piaoquan.api.job.contentplatform;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.aliyun.odps.data.Record;
import com.tzld.piaoquan.api.dao.mapper.contentplatform.ext.ContentPlatformDemandVideoMapperExt;
import com.tzld.piaoquan.api.model.po.contentplatform.ContentPlatformDemandVideo;
import com.tzld.piaoquan.growth.common.component.HttpPoolClient;
import com.tzld.piaoquan.growth.common.component.ODPSManager;
import com.tzld.piaoquan.growth.common.model.bo.VideoDetail;
import com.tzld.piaoquan.growth.common.service.MessageAttachmentService;
import com.tzld.piaoquan.growth.common.utils.DateUtil;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Component
public class ContentPlatformDemandVideoJob {

    private static final String DEMAND_MATCH_API_URL = "http://api-internal.piaoquantv.com/videoVector/videoSearch/queryDemandMatchResult";

    @Autowired
    private HttpPoolClient httpPoolClient;

    @Autowired
    private ContentPlatformDemandVideoMapperExt demandVideoMapperExt;

    @Autowired
    private MessageAttachmentService messageAttachmentService;

    @Autowired
    private ODPSManager odpsManager;

    private static final int ODPS_CATEGORY_CHUNK = 5000;

    @ApolloJsonValue("${sync.channel.names:[\"公众号合作-即转-稳定\",\"群/企微合作-稳定\",\"公众号合作-Daily-自选\"]}")
    private List<String> syncChannelNames;

    /** 默认记录所属 channel_name 标记,用于与正常渠道数据隔离,便于幂等清理 */
    private static final String DEFAULT_RECORD_CHANNEL = "default";

    @XxlJob("syncContentPlatformDemandVideoJob")
    public ReturnT<String> syncContentPlatformDemandVideoJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        log.info("syncContentPlatformDemandVideoJob start, dt={}", dt);

        // 单次任务执行内复用 videoId -> VideoDetail 的查询结果，避免跨渠道重复请求
        Map<Long, VideoDetail> videoDetailCache = new HashMap<>();
        Set<Long> queriedVideoIds = new HashSet<>();

        for (String syncChannelName : syncChannelNames) {
            try {
                syncByChannel(dt, syncChannelName, videoDetailCache, queriedVideoIds);
            } catch (Exception e) {
                log.error("syncContentPlatformDemandVideoJob error, dt={}, channelName={}", dt, syncChannelName, e);
                return ReturnT.FAIL;
            }
        }

        try {
            insertDefaultRecords(dt);
        } catch (Exception e) {
            log.error("syncContentPlatformDemandVideoJob insertDefaultRecords error, dt={}", dt, e);
            return ReturnT.FAIL;
        }

        log.info("syncContentPlatformDemandVideoJob finish, dt={}", dt);
        return ReturnT.SUCCESS;
    }

    /**
     * 写入三条默认记录:仅含匹配实验id及三个匹配相关字段,其它字段为空。
     * 用 channel_name=DEFAULT_RECORD_CHANNEL 与正常渠道数据隔离,先按 (dt, channel_name) 删除后批量插入,保证幂等。
     * 仅当该 dt 已存在数据时才执行,避免在无任何渠道数据的日期下凭空插入默认记录。
     */
    private void insertDefaultRecords(String dt) {
        int existCount = demandVideoMapperExt.countByCondition(dt, null, null, null, null);
        if (existCount <= 0) {
            log.info("insertDefaultRecords skip, dt={} has no data", dt);
            return;
        }

        Long now = System.currentTimeMillis();
        List<ContentPlatformDemandVideo> defaults = new ArrayList<>();
        defaults.add(buildDefaultRecord(dt, now,
                "vector_search", "视频库_解构特征_向量相似匹配 ", "相似度>=0.8", "向量相似度+全局推荐rov", "平台搜索"));
        defaults.add(buildDefaultRecord(dt, now,
                "manager_search", "视频库_标题_包含匹配", "关键词匹配", "", "平台搜索"));
        defaults.add(buildDefaultRecord(dt, now,
                "hot", "", "", "全局推荐rov", "平台热门"));

        demandVideoMapperExt.deleteByDtAndChannelName(dt, DEFAULT_RECORD_CHANNEL);
        demandVideoMapperExt.batchInsert(defaults);
        log.info("insertDefaultRecords success, dt={}, count={}", dt, defaults.size());
    }

    private ContentPlatformDemandVideo buildDefaultRecord(String dt, Long now,
                                                          String experimentId,
                                                          String matchMethod,
                                                          String matchVideoFilter,
                                                          String matchSort,
                                                          String demandStrategy) {
        ContentPlatformDemandVideo record = new ContentPlatformDemandVideo();
        record.setDt(dt);
        record.setChannelName(DEFAULT_RECORD_CHANNEL);
        record.setExperimentId(experimentId);
        record.setMatchMethod(matchMethod);
        record.setMatchVideoFilter(matchVideoFilter);
        record.setMatchSort(matchSort);
        record.setDemandStrategy(demandStrategy);
        record.setStatus(1);
        record.setCreateTimestamp(now);
        return record;
    }

    private static final int PAGE_SIZE = 10000;

    private void syncByChannel(String dt, String syncChannelName,
                               Map<Long, VideoDetail> videoDetailCache,
                               Set<Long> queriedVideoIds) throws Exception {
        log.info("syncByChannel start, dt={}, channelName={}", dt, syncChannelName);

        Long now = System.currentTimeMillis();
        List<ContentPlatformDemandVideo> saveList = new ArrayList<>();

        int pageNum = 1;
        int totalPages = 1;

        while (pageNum <= totalPages) {
            // 调用分页接口获取需求匹配结果
            JSONObject requestParam = new JSONObject();
            requestParam.put("dt", dt);
            requestParam.put("channelName", syncChannelName);
            requestParam.put("pageNum", pageNum);
            requestParam.put("pageSize", PAGE_SIZE);

            String response = httpPoolClient.post(DEMAND_MATCH_API_URL, requestParam.toJSONString());
            if (!StringUtils.hasText(response)) {
                log.error("syncByChannel response is empty, dt={}, channelName={}, pageNum={}", dt, syncChannelName, pageNum);
                throw new RuntimeException("syncByChannel response is empty");
            }

            JSONObject result = JSONObject.parseObject(response);
            if (result.getInteger("code") != 0) {
                log.error("syncByChannel api error, dt={}, channelName={}, pageNum={}, msg={}", dt, syncChannelName, pageNum, result.getString("msg"));
                throw new RuntimeException("syncByChannel api error: " + result.getString("msg"));
            }

            JSONObject dataObj = result.getJSONObject("data");
            if (dataObj == null) {
                log.info("syncByChannel no data, dt={}, channelName={}, pageNum={}", dt, syncChannelName, pageNum);
                break;
            }

            totalPages = dataObj.getIntValue("totalPages");
            JSONArray records = dataObj.getJSONArray("records");
            if (records == null || records.isEmpty()) {
                log.info("syncByChannel no records, dt={}, channelName={}, pageNum={}", dt, syncChannelName, pageNum);
                break;
            }

            for (int i = 0; i < records.size(); i++) {
                JSONObject demandItem = records.getJSONObject(i);
                String channelName = demandItem.getString("channelName");
                String crowdSegment = demandItem.getString("crowdSegment");
                String dimension = demandItem.getString("dimension");
                String pointType = demandItem.getString("pointType");
                String standardElement = demandItem.getString("standardElement");
                String elementDimension = demandItem.getString("elementDimension");
                String categoryName = demandItem.getString("categoryName");
                String demandId = demandItem.getString("demandId");
                String crowdPackage = demandItem.getString("crowdPackage");
                String conversionTarget = demandItem.getString("conversionTarget");
                String partner = demandItem.getString("partner");
                String account = demandItem.getString("account");
                String sceneValue = demandItem.getString("sceneValue");
                String demandStrategy = demandItem.getString("demandStrategy");
                String driveDimensionTime = demandItem.getString("driveDimensionTime");
                String demandFilterSortStrategy = demandItem.getString("demandFilterSortStrategy");
                String demandType = demandItem.getString("demandType");
                String demandContentId = demandItem.getString("demandContentId");
                String demandContentTitle = demandItem.getString("demandContentTitle");
                String demandContentTopic = demandItem.getString("demandContentTopic");
                Integer crowdCount = demandItem.getInteger("crowdCount");
                Integer videoCount = demandItem.getInteger("videoCount");
                Long visitUv = demandItem.getLong("visitUv");
                Double uvRatio = demandItem.getDouble("uvRatio");
                Double totalRov = demandItem.getDouble("totalRov");
                String onlineAction = demandItem.getString("onlineAction");
                String channelLevel3 = demandItem.getString("channelLevel3");
                String matchMethod = demandItem.getString("matchMethod");
                String matchVideoFilter = demandItem.getString("matchVideoFilter");
                String matchSort = demandItem.getString("matchSort");
                Double sceneSumRov = demandItem.getDouble("sceneSumRov");

                JSONArray matchedVideos = demandItem.getJSONArray("matchedVideos");
                if (matchedVideos == null || matchedVideos.isEmpty()) {
                    continue;
                }

                for (int j = 0; j < matchedVideos.size(); j++) {
                    JSONObject videoItem = matchedVideos.getJSONObject(j);
                    ContentPlatformDemandVideo demandVideo = new ContentPlatformDemandVideo();
                    demandVideo.setDt(dt);
                    demandVideo.setChannelName(channelName != null ? channelName : "");
                    demandVideo.setCrowdSegment(crowdSegment != null ? crowdSegment : "");
                    demandVideo.setDimension(dimension != null ? dimension : "");
                    demandVideo.setPointType(pointType != null ? pointType : "");
                    demandVideo.setStandardElement(standardElement != null ? standardElement : "");
                    demandVideo.setElementDimension(elementDimension != null ? elementDimension : "");
                    demandVideo.setCategoryName(categoryName != null ? categoryName : "");
                    demandVideo.setDemandId(demandId != null ? demandId : "");
                    demandVideo.setCrowdPackage(crowdPackage != null ? crowdPackage : "");
                    demandVideo.setConversionTarget(conversionTarget != null ? conversionTarget : "");
                    demandVideo.setPartner(partner != null ? partner : "");
                    demandVideo.setAccount(account != null ? account : "");
                    demandVideo.setSceneValue(sceneValue != null ? sceneValue : "");
                    demandVideo.setDemandStrategy(demandStrategy != null ? demandStrategy : "");
                    demandVideo.setDriveDimensionTime(driveDimensionTime != null ? driveDimensionTime : "");
                    demandVideo.setDemandFilterSortStrategy(demandFilterSortStrategy != null ? demandFilterSortStrategy : "");
                    demandVideo.setDemandType(demandType != null ? demandType : "");
                    demandVideo.setDemandContentId(demandContentId != null ? demandContentId : "");
                    demandVideo.setDemandContentTitle(demandContentTitle != null ? demandContentTitle : "");
                    demandVideo.setDemandContentTopic(demandContentTopic != null ? demandContentTopic : "");
                    demandVideo.setCrowdCount(crowdCount);
                    demandVideo.setVideoCount(videoCount);
                    demandVideo.setVisitUv(visitUv);
                    demandVideo.setUvRatio(uvRatio);
                    demandVideo.setTotalRov(totalRov);
                    demandVideo.setOnlineAction(onlineAction != null ? onlineAction : "");
                    demandVideo.setChannelLevel3(channelLevel3 != null ? channelLevel3 : "");
                    demandVideo.setMatchMethod(matchMethod != null ? matchMethod : "");
                    demandVideo.setMatchVideoFilter(matchVideoFilter != null ? matchVideoFilter : "");
                    demandVideo.setMatchSort(matchSort != null ? matchSort : "");
                    demandVideo.setSceneSumRov(sceneSumRov);
                    demandVideo.setVideoId(videoItem.getLong("videoId"));
                    demandVideo.setConfigCode(videoItem.getString("configCode") != null ? videoItem.getString("configCode") : "");
                    demandVideo.setScore(videoItem.getDouble("score"));
                    demandVideo.setSim(videoItem.getDouble("sim"));
                    demandVideo.setRov(videoItem.getDouble("rov"));
                    demandVideo.setMatchText(videoItem.getString("text") != null ? videoItem.getString("text") : "");
                    demandVideo.setExperimentId(videoItem.getString("experimentId"));
                    demandVideo.setMatchExposurePv(videoItem.getLong("matchExposurePv"));
                    demandVideo.setMatchHeadSingleReturnRate(videoItem.getDouble("matchHeadSingleReturnRate"));
                    demandVideo.setMatchHeadDistributionSingleReturnRate(videoItem.getDouble("matchHeadDistributionSingleReturnRate"));
                    demandVideo.setMatchGeneralizedElement(videoItem.getString("matchGeneralizedElement"));
                    demandVideo.setMatchGeneralizedPointType(videoItem.getString("matchGeneralizedPointType"));
                    demandVideo.setStatus(1);
                    demandVideo.setCreateTimestamp(now);
                    if (Objects.isNull(demandVideo.getRov()) || demandVideo.getRov() == 0.0) {
                        continue;
                    }
                    saveList.add(demandVideo);
                }
            }

            pageNum++;
        }

        if (CollectionUtils.isEmpty(saveList)) {
            log.info("syncByChannel no matched videos, dt={}, channelName={}", dt, syncChannelName);
            return;
        }

        // 获取视频详情（标题、封面、视频URL）：仅请求当前缓存中尚未查询过的 videoId
        List<Long> videoIds = saveList.stream().map(ContentPlatformDemandVideo::getVideoId)
                .filter(Objects::nonNull)
                .distinct().collect(Collectors.toList());
        List<Long> uncachedVideoIds = videoIds.stream()
                .filter(id -> !queriedVideoIds.contains(id))
                .collect(Collectors.toList());
        for (List<Long> partition : Lists.partition(uncachedVideoIds, 20)) {
            Set<Long> ids = new HashSet<>(partition);
            videoDetailCache.putAll(messageAttachmentService.getVideoDetail(ids));
            // 无论是否命中，都标记为已查询，避免后续渠道再次发起请求
            queriedVideoIds.addAll(ids);
        }

        // 填充视频详情
        for (ContentPlatformDemandVideo demandVideo : saveList) {
            VideoDetail detail = videoDetailCache.get(demandVideo.getVideoId());
            if (Objects.nonNull(detail)) {
                demandVideo.setTitle(detail.getTitle());
                String cover = detail.getCover();
                if (StringUtils.hasText(cover) && cover.contains("/watermark")) {
                    cover = cover.substring(0, cover.indexOf("/watermark"));
                }
                demandVideo.setCover(cover);
                demandVideo.setVideo(detail.getVideoPath());
            }
        }

        // 过滤未获取到视频详情的记录
        saveList = saveList.stream()
                .filter(v -> StringUtils.hasText(v.getTitle()))
                .collect(Collectors.toList());

        // 从 ODPS video_merge_tag 拉取 merge_leve2(二级品类),填到 category 列,供下发侧黑名单过滤使用
        Map<Long, String> videoIdToCategory = fetchVideoCategoryMap(
                saveList.stream().map(ContentPlatformDemandVideo::getVideoId).distinct().collect(Collectors.toList()));
        for (ContentPlatformDemandVideo demandVideo : saveList) {
            String cat = videoIdToCategory.get(demandVideo.getVideoId());
            if (StringUtils.hasText(cat)) {
                demandVideo.setCategory(cat);
            }
        }

//        // 按crowd_segment粒度去重，相同videoId保留分数最高的一条
//        saveList = saveList.stream()
//                .collect(Collectors.groupingBy(v -> v.getCrowdSegment() + "_"
//                        + v.getChannelLevel3() + "_" + v.getDimension() + "_" + v.getVideoId()))
//                .values().stream()
//                .map(group -> group.stream()
//                        .max(Comparator.comparingDouble(v -> v.getScore() != null ? v.getScore() : 0.0))
//                        .orElse(null))
//                .filter(Objects::nonNull)
//                .collect(Collectors.toList());

        // 先删除当天该渠道数据，再批量插入
        demandVideoMapperExt.deleteByDtAndChannelName(dt, syncChannelName);
        for (List<ContentPlatformDemandVideo> partition : Lists.partition(saveList, 1000)) {
            demandVideoMapperExt.batchInsert(partition);
        }

        log.info("syncByChannel success, dt={}, channelName={}, count={}", dt, syncChannelName, saveList.size());
    }

    /**
     * 批查 loghubods.video_merge_tag 拿 videoid → merge_leve2(二级品类)。
     * 该表无分区,按 videoid IN (...) 分批拉取,缺失/异常一律返回不含该 id 的 entry,调用方按 null 处理。
     */
    private Map<Long, String> fetchVideoCategoryMap(List<Long> videoIds) {
        Map<Long, String> result = new HashMap<>();
        if (CollectionUtils.isEmpty(videoIds)) {
            return result;
        }
        for (List<Long> partition : Lists.partition(videoIds, ODPS_CATEGORY_CHUNK)) {
            String inClause = partition.stream().map(String::valueOf).collect(Collectors.joining(","));
            String sql = "SELECT videoid, merge_leve2 FROM loghubods.video_merge_tag WHERE videoid IN (" + inClause + ");";
            try {
                List<Record> records = odpsManager.query(sql);
                if (CollectionUtils.isEmpty(records)) {
                    continue;
                }
                for (Record r : records) {
                    Object vid = r.get("videoid");
                    Object cat = r.get("merge_leve2");
                    if (vid == null || cat == null) continue;
                    try {
                        result.put(Long.parseLong(vid.toString()), cat.toString());
                    } catch (NumberFormatException ignore) {
                    }
                }
            } catch (Exception e) {
                log.error("fetchVideoCategoryMap error, chunkSize={}", partition.size(), e);
            }
        }
        log.info("fetchVideoCategoryMap done, input={}, mapped={}", videoIds.size(), result.size());
        return result;
    }

    /**
     * 滚动清理 content_platform_demand_video 表数据，仅保留最近 14 天（含今天）。
     * 即删除 dt < (今天-14) 的所有记录。param 可传 cutoff dt（yyyyMMdd）覆盖默认值，便于回溯/调试。
     */
    @XxlJob("cleanContentPlatformDemandVideoJob")
    public ReturnT<String> cleanContentPlatformDemandVideoJob(String param) {
        String cutoffDt;
        if (StringUtils.hasText(param)) {
            cutoffDt = param;
        } else {
            cutoffDt = DateUtil.getBeforeDayDateString(14, "yyyyMMdd");
        }
        log.info("cleanContentPlatformDemandVideoJob start, cutoffDt={}", cutoffDt);
        try {
            List<String> dts = demandVideoMapperExt.selectDistinctDtsLessThan(cutoffDt);
            log.info("cleanContentPlatformDemandVideoJob found {} dts less than {}", dts.size(), cutoffDt);
            int totalDeleted = 0;
            for (String dt : dts) {
                int rows = demandVideoMapperExt.deleteByDt(dt);
                totalDeleted += rows;
                log.info("cleanContentPlatformDemandVideoJob deleted dt={}, rows={}", dt, rows);
            }
            log.info("cleanContentPlatformDemandVideoJob success, cutoffDt={}, totalDeletedRows={}", cutoffDt, totalDeleted);
        } catch (Exception e) {
            log.error("cleanContentPlatformDemandVideoJob error, cutoffDt={}", cutoffDt, e);
            return ReturnT.FAIL;
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("checkContentPlatformDemandVideoStatusJob")
    public ReturnT<String> checkContentPlatformDemandVideoStatusJob(String param) {
        String dt = demandVideoMapperExt.getMaxDt(null);
        if (!StringUtils.hasText(dt)) {
            log.info("checkContentPlatformDemandVideoStatusJob no data");
            return ReturnT.SUCCESS;
        }
        log.info("checkContentPlatformDemandVideoStatusJob start, dt={}", dt);
        List<ContentPlatformDemandVideo> activeVideos = demandVideoMapperExt.selectActiveVideos(dt);
        if (CollectionUtils.isEmpty(activeVideos)) {
            return ReturnT.SUCCESS;
        }
        Long now = System.currentTimeMillis();
        // 单次任务执行内复用 videoId -> VideoDetail 的查询结果，避免相同 videoId 跨批次重复请求
        Map<Long, VideoDetail> videoDetailCache = new HashMap<>();
        Set<Long> queriedVideoIds = new HashSet<>();

        // 先按全量去重 videoId 分批拉取详情写入缓存
        List<Long> distinctVideoIds = activeVideos.stream()
                .filter(Objects::nonNull)
                .map(ContentPlatformDemandVideo::getVideoId)
                .filter(Objects::nonNull)
                .distinct()
                .collect(Collectors.toList());
        for (List<Long> partition : Lists.partition(distinctVideoIds, 20)) {
            List<Long> uncachedIds = partition.stream()
                    .filter(id -> !queriedVideoIds.contains(id))
                    .collect(Collectors.toList());
            if (CollectionUtils.isEmpty(uncachedIds)) {
                continue;
            }
            Set<Long> ids = new HashSet<>(uncachedIds);
            videoDetailCache.putAll(messageAttachmentService.getVideoDetail(ids));
            // 无论是否命中，都标记为已查询，避免后续再次发起请求
            queriedVideoIds.addAll(ids);
        }

        // 基于缓存判断状态并更新
        for (ContentPlatformDemandVideo video : activeVideos) {
            if (video == null || video.getVideoId() == null) {
                continue;
            }
            VideoDetail videoDetail = videoDetailCache.get(video.getVideoId());
            if (videoDetail == null || videoDetail.getAuditStatus() != 5) {
                demandVideoMapperExt.updateStatusByVideoId(video.getVideoId(), dt, 0, now);
            }
        }
        log.info("checkContentPlatformDemandVideoStatusJob done, dt={}", dt);
        return ReturnT.SUCCESS;
    }
}
