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

import com.alibaba.fastjson.JSONObject;
import com.aliyun.odps.data.Record;
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
import com.google.common.collect.Lists;
import com.tzld.piaoquan.api.common.enums.contentplatform.AccountStatusEnum;
import com.tzld.piaoquan.api.common.enums.contentplatform.BussinessTypeEnum;
import com.tzld.piaoquan.api.component.AigcApiService;
import com.tzld.piaoquan.api.dao.mapper.contentplatform.*;
import com.tzld.piaoquan.api.dao.mapper.contentplatform.ext.ContentPlatformDataStatMapperExt;
import com.tzld.piaoquan.api.model.po.contentplatform.*;
import com.tzld.piaoquan.api.model.vo.WxAccountDatastatVO;
import com.tzld.piaoquan.api.service.contentplatform.ContentPlatformPlanService;
import com.tzld.piaoquan.growth.common.utils.DateUtil;
import com.tzld.piaoquan.growth.common.utils.OdpsUtil;
import com.tzld.piaoquan.growth.common.utils.RedisUtils;
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 org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Component
public class ContentPlatformDatastatJob {

    @Autowired
    private ContentPlatformGzhDataStatMapper gzhDataStatMapper;

    @Autowired
    private ContentPlatformFwhDataStatMapper fwhDataStatMapper;

    @Autowired
    private ContentPlatformAccountMapper accountMapper;

    @Autowired
    private ContentPlatformGzhAccountMapper gzhAccountMapper;

    @Autowired
    private ContentPlatformQwDataStatMapper qwDataStatMapper;

    @Autowired
    private ContentPlatformQwPlanMapper qwPlanMapper;

    @Autowired
    private ContentPlatformDataStatMapperExt dataStatMapperExt;

    @Autowired
    private AigcApiService aigcApiService;

    @Autowired
    private ContentPlatformPlanService planService;

    @Autowired
    private RedisUtils redisUtils;

    @ApolloJsonValue("${unit.price.min:{\"gzh_autoReply\":0.1,\"fwh_push\":0.1,\"qw\":0.15}}")
    private Map<String, Double> unitPriceMinConfig;

    @ApolloJsonValue("${unit.price.max:{\"gzh_autoReply\":0.2,\"fwh_push\":0.2,\"qw\":0.3}}")
    private Map<String, Double> unitPriceMaxConfig;


    @XxlJob("syncContentPlatformGzhDatastatJob")
    public ReturnT<String> syncContentPlatformGzhDatastatJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Long now = System.currentTimeMillis();
        // 公众号自动回复数据统计
        String sql = String.format(
                "SELECT distinct first_level.channel_shortname, first_level.subchannel, first_level.first_uv, fission.split_uv, fission.裂变arpu " +
                        "FROM loghubods.out_channel_mid_first_total first_level " +
                        "left join loghubods.out_channel_mid_split_total fission " +
                        "on first_level.channel_shortname = fission.channel_shortname and first_level.subchannel = fission.subchannel " +
                        "and first_level.dt = fission.dt and first_level.type = fission.type and first_level.tag = fission.tag " +
                        "WHERE first_level.dt = %s and first_level.type = '公众号即时回复' and first_level.tag = '分投放渠道客户分账号去重';", dt);
        List<Record> dataList = OdpsUtil.getOdpsData(sql);
        List<ContentPlatformAccount> accountList = getAllAccount();
        Map<String, ContentPlatformAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformAccount::getChannel, account -> account));
        // 所有公众号
        List<ContentPlatformGzhAccount> gzhAccountList = getAllGzhAccount();
        Map<Long, Map<String, ContentPlatformGzhAccount>> gzhAccountMap = gzhAccountList.stream()
                .collect(Collectors.groupingBy(ContentPlatformGzhAccount::getCreateAccountId,
                        Collectors.toMap(ContentPlatformGzhAccount::getGhId, account -> account)));
        List<String> accountExternalIds = gzhAccountList.stream().map(ContentPlatformGzhAccount::getExternalId)
                .collect(Collectors.toList());
        String dateStr = dt.substring(0, 4) + "-" + dt.substring(4, 6) + "-" + dt.substring(6, 8);
        List<WxAccountDatastatVO> wxAccountDatastatVOList = aigcApiService.getWxAccountDatastat(dateStr, accountExternalIds);
        Map<String, WxAccountDatastatVO> wxAccountDatastatMap = wxAccountDatastatVOList.stream()
                .collect(Collectors.toMap(WxAccountDatastatVO::getAccountId, wxAccountDatastatVO -> wxAccountDatastatVO));
        if (CollectionUtils.isNotEmpty(dataList)) {
            List<ContentPlatformGzhDataStat> saveList = new ArrayList<>();
            BigDecimal sumScore = BigDecimal.ZERO;
            for (Record record : dataList) {
                String channel = (String) record.get(0);
                String ghId = (String) record.get(1);
                Integer firstLevelCount = parseInteger(record.get(2));
                Integer fissionCount = parseInteger(record.get(3));
                Double fissionArpu = parseDouble(record.get(4));
                if (fissionArpu > 0.3) {
                    fissionArpu = 0.3;
                }
                if ("SUM".equals(channel)) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() * 10 / firstLevelCount)
                            .setScale(2, RoundingMode.HALF_UP);
                    fissionArpu = BigDecimal.valueOf(fissionArpu).setScale(2, RoundingMode.HALF_UP).doubleValue();
                    sumScore = fissionRate.add(new BigDecimal(fissionArpu));
                    continue;
                }
                ContentPlatformAccount account = accountMap.get(channel);
                if (Objects.isNull(account)) {
                    continue;
                }
                Map<String, ContentPlatformGzhAccount> gzhAccounts = gzhAccountMap.get(account.getId());
                if (Objects.isNull(gzhAccounts)) {
                    continue;
                }

                ContentPlatformGzhAccount gzhAccount = gzhAccounts.get(ghId);
                if (Objects.isNull(gzhAccount)) {
                    continue;
                }
                ContentPlatformGzhDataStat item = new ContentPlatformGzhDataStat();
                item.setDateStr(dt);
                item.setAccountId(gzhAccount.getId());
                item.setFirstLevelCount(firstLevelCount);
                WxAccountDatastatVO wxAccountDatastatVO = wxAccountDatastatMap.get(gzhAccount.getExternalId());
                if (Objects.nonNull(wxAccountDatastatVO)) {
                    item.setFansIncreaseCount(wxAccountDatastatVO.getFansIncreaseCount());
                }

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() * 10 / firstLevelCount)
                            .setScale(2, RoundingMode.HALF_UP);
                    fissionArpu = BigDecimal.valueOf(fissionArpu).setScale(2, RoundingMode.HALF_UP).doubleValue();
                    BigDecimal totalScore = fissionRate.add(new BigDecimal(fissionArpu));
                    item.setScore(totalScore.doubleValue());
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            for (ContentPlatformGzhDataStat item : saveList) {
                if (item.getFirstLevelCount() < 10) {
                    item.setScore(0.0);
                    continue;
                }
                if (Objects.nonNull(item.getScore())) {
                    BigDecimal score = BigDecimal.valueOf(item.getScore());
                    item.setScore(score.divide(sumScore, 3, RoundingMode.HALF_UP).multiply(BigDecimal.valueOf(10)).doubleValue());
                }
            }
            if (CollectionUtils.isNotEmpty(saveList)) {
                dataStatMapperExt.deleteGzhDatastat(dt);
                dataStatMapperExt.batchInsertGzhDatastat(saveList);
            }
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformGzhVideoDatastatJob")
    public ReturnT<String> syncContentPlatformGzhVideoDatastatJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        // 公众号自动回复分视频数据统计
        String sql = String.format("SELECT ghid, video_id, video_title, first_uv, split0_uv " +
                "FROM loghubods.auto_reply_per_video WHERE dt = %s;", dt);
        List<Record> dataList = OdpsUtil.getOdpsData(sql);
        // 所有公众号
        List<ContentPlatformGzhAccount> accountList = getAllGzhAccount();
        Map<String, ContentPlatformGzhAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformGzhAccount::getGhId, account -> account));
        List<String> ghIds = accountList.stream().map(ContentPlatformGzhAccount::getGhId).collect(Collectors.toList());
        Long now = System.currentTimeMillis();
        if (CollectionUtils.isNotEmpty(dataList)) {
            List<ContentPlatformGzhVideoDataStat> saveList = new ArrayList<>();
            for (Record record : dataList) {
                ContentPlatformGzhVideoDataStat item = new ContentPlatformGzhVideoDataStat();
                String ghId = (String) record.get(0);
                Long videoId = parseLong(record.get(1));
                String title = (String) record.get(2);
                int firstLevelCount = Integer.parseInt((String) record.get(3));
                Integer fissionCount = parseInteger(record.get(4));
                item.setDateStr(dt);
                if (!ghIds.contains(ghId) || videoId == 0 || firstLevelCount == 0) {
                    continue;
                }
                ContentPlatformGzhAccount gzhAccount = accountMap.get(ghId);
                item.setAccountId(gzhAccount.getId());
                item.setVideoId(videoId);
                item.setTitle(title);
                item.setFirstLevelCount(firstLevelCount);

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal num = BigDecimal.valueOf(fissionCount.doubleValue() * 10 / firstLevelCount);
                    BigDecimal rounded = num.setScale(2, RoundingMode.HALF_UP);
                    item.setScore(rounded.doubleValue());
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            if (CollectionUtils.isNotEmpty(saveList)) {
                dataStatMapperExt.deleteGzhVideoDatastat(dt);
                dataStatMapperExt.batchInsertGzhVideoDatastat(saveList);
            }
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformGzhDatastatTotalJob")
    public ReturnT<String> syncContentPlatformGzhDatastatTotalJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Long now = System.currentTimeMillis();
        // 公众号自动回复数据统计
        String sql = String.format(
                "SELECT distinct first_level.channel_shortname, first_level.first_uv, fission.split_uv, fission.裂变arpu, price.arpu " +
                        "FROM loghubods.out_channel_mid_first_total first_level " +
                        "left join loghubods.out_channel_mid_split_total fission " +
                        "on first_level.channel_shortname = fission.channel_shortname and first_level.dt = fission.dt " +
                        "and first_level.type = fission.type and first_level.tag = fission.tag " +
                        "left join loghubods.wecom_cooperation_dynamic_unit_price price " +
                        "on first_level.channel_shortname = price.channel_shortname and first_level.dt = price.dt " +
                        "and first_level.type = price.type and first_level.tag = price.tag " +
                        "WHERE first_level.dt = %s and first_level.type = '公众号即时回复' and first_level.tag = '投放渠道内去重' ;", dt);
        List<Record> dataList = OdpsUtil.getOdpsData(sql);
        // 所有公众号
        List<ContentPlatformAccount> accountList = getAllAccount();
        Map<String, ContentPlatformAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformAccount::getChannel, account -> account));
        List<ContentPlatformGzhAccount> gzhAccountList = getAllGzhAccount();
        Map<Long, List<ContentPlatformGzhAccount>> gzhAccountMap = gzhAccountList.stream()
                .collect(Collectors.groupingBy(ContentPlatformGzhAccount::getCreateAccountId));
        List<String> accountExternalIds = gzhAccountList.stream().map(ContentPlatformGzhAccount::getExternalId)
                .collect(Collectors.toList());
        String dateStr = dt.substring(0, 4) + "-" + dt.substring(4, 6) + "-" + dt.substring(6, 8);
        List<WxAccountDatastatVO> wxAccountDatastatVOList = aigcApiService.getWxAccountDatastat(dateStr, accountExternalIds);
        Map<String, WxAccountDatastatVO> wxAccountDatastatMap = wxAccountDatastatVOList.stream()
                .collect(Collectors.toMap(WxAccountDatastatVO::getAccountId, wxAccountDatastatVO -> wxAccountDatastatVO));
        if (CollectionUtils.isNotEmpty(dataList)) {
            BigDecimal sumScore = BigDecimal.ZERO;
            List<ContentPlatformGzhDataStatTotal> saveList = new ArrayList<>();
            for (Record record : dataList) {
                String channel = record.getString(0);
                Integer firstLevelCount = parseInteger(record.get(1));
                Integer fissionCount = parseInteger(record.get(2));
                Double fissionArpu = parseDouble(record.get(3));
                if (fissionArpu > 0.3) {
                    fissionArpu = 0.3;
                }
                Double arpu = parseDouble(record.get(4));
                if ("SUM".equals(channel)) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() * 10 / firstLevelCount)
                            .setScale(2, RoundingMode.HALF_UP);
                    fissionArpu = BigDecimal.valueOf(fissionArpu).setScale(2, RoundingMode.HALF_UP).doubleValue();
                    sumScore = fissionRate.add(new BigDecimal(fissionArpu));
                    continue;
                }
                ContentPlatformAccount account = accountMap.get(channel);
                if (Objects.isNull(account)) {
                    continue;
                }
                List<ContentPlatformGzhAccount> gzhAccounts = gzhAccountMap.get(account.getId());
                int fansIncreaseCount = 0;
                if (CollectionUtils.isNotEmpty(gzhAccounts)) {
                    for (ContentPlatformGzhAccount gzhAccount : gzhAccounts) {
                        WxAccountDatastatVO wxAccountDatastatVO = wxAccountDatastatMap.get(gzhAccount.getExternalId());
                        if (Objects.nonNull(wxAccountDatastatVO)) {
                            fansIncreaseCount += wxAccountDatastatVO.getFansIncreaseCount();
                        }
                    }
                }
                ContentPlatformGzhDataStatTotal item = new ContentPlatformGzhDataStatTotal();
                item.setDateStr(dt);
                item.setChannel(channel);
                item.setFirstLevelCount(firstLevelCount);
                item.setFansIncreaseCount(fansIncreaseCount);

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() * 10 / firstLevelCount)
                            .setScale(2, RoundingMode.HALF_UP);
                    fissionArpu = BigDecimal.valueOf(fissionArpu).setScale(2, RoundingMode.HALF_UP).doubleValue();
                    BigDecimal totalScore = fissionRate.add(new BigDecimal(fissionArpu));
                    item.setScore(totalScore.doubleValue());

                    BigDecimal unitPrice = getUnitPrice(account.getPrice(), fissionRate, arpu, BussinessTypeEnum.GZH_AUTO_REPLY);
                    if (Objects.nonNull(unitPrice)) {
                        item.setUnitPrice(unitPrice.doubleValue());
                        BigDecimal settlementAmount = unitPrice.multiply(new BigDecimal(firstLevelCount)).setScale(2, RoundingMode.HALF_UP);
                        item.setSettlementAmount(settlementAmount.doubleValue());
                    }
                } else {
                    item.setScore(0.0);
                    item.setUnitPrice(0.1);
                    item.setSettlementAmount(0.0);
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            for (ContentPlatformGzhDataStatTotal item : saveList) {
                if (item.getFirstLevelCount() < 25) {
                    item.setScore(0.0);
                    continue;
                }
                if (Objects.nonNull(item.getScore())) {
                    BigDecimal score = BigDecimal.valueOf(item.getScore());
                    item.setScore(score.divide(sumScore, 3, RoundingMode.HALF_UP).multiply(BigDecimal.valueOf(10)).doubleValue());
                }
            }
            if (CollectionUtils.isNotEmpty(saveList)) {
                dataStatMapperExt.deleteGzhDatastatTotal(dt);
                dataStatMapperExt.batchInsertGzhDatastatTotal(saveList);
            }
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformFwhDatastatJob")
    public ReturnT<String> syncContentPlatformFwhDatastatJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Long now = System.currentTimeMillis();
        // 公众号自动回复数据统计
        String sql = String.format(
                "SELECT distinct first_level.subchannel, first_level.first_uv, fission.split_uv " +
                        "FROM loghubods.out_channel_mid_first_total first_level " +
                        "left join loghubods.out_channel_mid_split_total fission " +
                        "on first_level.subchannel = fission.subchannel and first_level.dt = fission.dt " +
                        "and first_level.type = fission.type and first_level.tag = fission.tag " +
                        "WHERE first_level.dt = %s and first_level.type = '服务号代运营' and first_level.tag = '分投放渠道客户分账号去重' ;", dt);
        List<Record> dataList = OdpsUtil.getOdpsData(sql);
        // 所有公众号
        List<ContentPlatformGzhAccount> accountList = getAllGzhAccount();
        Map<String, ContentPlatformGzhAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformGzhAccount::getGhId, account -> account));
        List<String> ghIds = accountList.stream().map(ContentPlatformGzhAccount::getGhId).collect(Collectors.toList());
        List<String> accountExternalIds = accountList.stream().map(ContentPlatformGzhAccount::getExternalId)
                .collect(Collectors.toList());
        String dateStr = dt.substring(0, 4) + "-" + dt.substring(4, 6) + "-" + dt.substring(6, 8);
        List<WxAccountDatastatVO> wxAccountDatastatVOList = aigcApiService.getWxAccountDatastat(dateStr, accountExternalIds);
        Map<String, WxAccountDatastatVO> wxAccountDatastatMap = wxAccountDatastatVOList.stream()
                .collect(Collectors.toMap(WxAccountDatastatVO::getAccountId, wxAccountDatastatVO -> wxAccountDatastatVO));
        if (CollectionUtils.isNotEmpty(dataList)) {
            List<ContentPlatformFwhDataStat> saveList = new ArrayList<>();
            for (Record record : dataList) {
                ContentPlatformFwhDataStat item = new ContentPlatformFwhDataStat();
                String ghId = (String) record.get(0);
                int firstLevelCount = Integer.parseInt((String) record.get(1));
                Integer fissionCount = parseInteger(record.get(2));
                item.setDateStr(dt);
                if (!ghIds.contains(ghId)) {
                    continue;
                }
                ContentPlatformGzhAccount gzhAccount = accountMap.get(ghId);
                item.setAccountId(gzhAccount.getId());
                item.setFirstLevelCount(firstLevelCount);
                WxAccountDatastatVO wxAccountDatastatVO = wxAccountDatastatMap.get(gzhAccount.getExternalId());
                if (Objects.nonNull(wxAccountDatastatVO)) {
                    item.setFansIncreaseCount(wxAccountDatastatVO.getFansIncreaseCount());
                }

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal num = BigDecimal.valueOf(fissionCount.doubleValue() * 10 / firstLevelCount);
                    BigDecimal rounded = num.setScale(2, RoundingMode.HALF_UP);
                    item.setScore(rounded.doubleValue());
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            if (CollectionUtils.isNotEmpty(saveList)) {
                dataStatMapperExt.deleteFwhDatastat(dt);
                dataStatMapperExt.batchInsertFwhDatastat(saveList);
            }
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformFwhDatastatTotalJob")
    public ReturnT<String> syncContentPlatformFwhDatastatTotalJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Long now = System.currentTimeMillis();
        // 公众号自动回复数据统计
        String sql = String.format(
                "SELECT distinct first_level.channel_shortname, first_level.first_uv, fission.split_uv, price.arpu " +
                        "FROM loghubods.out_channel_mid_first_total first_level " +
                        "left join loghubods.out_channel_mid_split_total fission " +
                        "on first_level.channel_shortname = fission.channel_shortname and first_level.dt = fission.dt " +
                        "and first_level.type = fission.type and first_level.tag = fission.tag " +
                        "left join loghubods.wecom_cooperation_dynamic_unit_price price " +
                        "on first_level.channel_shortname = price.channel_shortname and first_level.dt = price.dt " +
                        "and first_level.type = price.type and first_level.tag = price.tag " +
                        "WHERE first_level.dt = %s and first_level.type = '服务号代运营' and first_level.tag = '投放渠道内去重' ;", dt);
        List<Record> dataList = OdpsUtil.getOdpsData(sql);
        // 所有公众号
        List<ContentPlatformAccount> accountList = getAllAccount();
        Map<String, ContentPlatformAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformAccount::getChannel, account -> account));
        List<ContentPlatformGzhAccount> gzhAccountList = getAllGzhAccount();
        Map<Long, List<ContentPlatformGzhAccount>> gzhAccountMap = gzhAccountList.stream()
                .collect(Collectors.groupingBy(ContentPlatformGzhAccount::getCreateAccountId));
        List<String> accountExternalIds = gzhAccountList.stream().map(ContentPlatformGzhAccount::getExternalId)
                .collect(Collectors.toList());
        String dateStr = dt.substring(0, 4) + "-" + dt.substring(4, 6) + "-" + dt.substring(6, 8);
        List<WxAccountDatastatVO> wxAccountDatastatVOList = aigcApiService.getWxAccountDatastat(dateStr, accountExternalIds);
        Map<String, WxAccountDatastatVO> wxAccountDatastatMap = wxAccountDatastatVOList.stream()
                .collect(Collectors.toMap(WxAccountDatastatVO::getAccountId, wxAccountDatastatVO -> wxAccountDatastatVO));
        if (CollectionUtils.isNotEmpty(dataList)) {
            List<ContentPlatformFwhDataStatTotal> saveList = new ArrayList<>();
            for (Record record : dataList) {
                ContentPlatformFwhDataStatTotal item = new ContentPlatformFwhDataStatTotal();
                String channel = record.getString(0);
                int firstLevelCount = Integer.parseInt((String) record.get(1));
                Integer fissionCount = parseInteger(record.get(2));
                Double arpu = parseDouble(record.get(3));
                item.setDateStr(dt);
                item.setChannel(channel);
                item.setFirstLevelCount(firstLevelCount);
                ContentPlatformAccount account = accountMap.get(channel);
                if (Objects.isNull(account)) {
                    continue;
                }
                List<ContentPlatformGzhAccount> gzhAccounts = gzhAccountMap.get(account.getId());
                int fansIncreaseCount = 0;
                if (CollectionUtils.isNotEmpty(gzhAccounts)) {
                    for (ContentPlatformGzhAccount gzhAccount : gzhAccounts) {
                        WxAccountDatastatVO wxAccountDatastatVO = wxAccountDatastatMap.get(gzhAccount.getExternalId());
                        if (Objects.nonNull(wxAccountDatastatVO)) {
                            fansIncreaseCount += wxAccountDatastatVO.getFansIncreaseCount();
                        }
                    }
                }
                item.setFansIncreaseCount(fansIncreaseCount);

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() / firstLevelCount);
                    BigDecimal rounded = fissionRate.multiply(new BigDecimal(10)).setScale(2, RoundingMode.HALF_UP);
                    item.setScore(rounded.doubleValue());
                    BigDecimal unitPrice = getUnitPrice(account.getPrice(), fissionRate, arpu, BussinessTypeEnum.FWH_PUSH);
                    if (Objects.nonNull(unitPrice)) {
                        item.setUnitPrice(unitPrice.doubleValue());
                        BigDecimal settlementAmount = unitPrice.multiply(new BigDecimal(firstLevelCount)).setScale(2, RoundingMode.HALF_UP);
                        item.setSettlementAmount(settlementAmount.doubleValue());
                    }
                } else {
                    item.setScore(0.0);
                    item.setUnitPrice(0.1);
                    item.setSettlementAmount(0.0);
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            if (CollectionUtils.isNotEmpty(saveList)) {
                dataStatMapperExt.deleteFwhDatastatTotal(dt);
                dataStatMapperExt.batchInsertFwhDatastatTotal(saveList);
            }
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformGzhPushDatastatJob")
    public ReturnT<String> syncContentPlatformGzhPushDatastatJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Long now = System.currentTimeMillis();
        // 公众号自动回复数据统计
        String sql = String.format(
                "SELECT distinct first_level.subchannel, first_level.first_uv, fission.split_uv " +
                        "FROM loghubods.out_channel_mid_first_total first_level " +
                        "left join loghubods.out_channel_mid_split_total fission " +
                        "on first_level.subchannel = fission.subchannel and first_level.dt = fission.dt " +
                        "and first_level.type = fission.type and first_level.tag = fission.tag " +
                        "WHERE first_level.dt = %s and first_level.type = '公众号推送' and first_level.tag = '分投放渠道客户分账号去重' ;", dt);
        List<Record> dataList = OdpsUtil.getOdpsData(sql);
        // 所有公众号
        List<ContentPlatformGzhAccount> accountList = getAllGzhAccount();
        Map<String, ContentPlatformGzhAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformGzhAccount::getGhId, account -> account));
        List<String> ghIds = accountList.stream().map(ContentPlatformGzhAccount::getGhId).collect(Collectors.toList());
        List<String> accountExternalIds = accountList.stream().map(ContentPlatformGzhAccount::getExternalId)
                .collect(Collectors.toList());
        String dateStr = dt.substring(0, 4) + "-" + dt.substring(4, 6) + "-" + dt.substring(6, 8);
        List<WxAccountDatastatVO> wxAccountDatastatVOList = aigcApiService.getWxAccountDatastat(dateStr, accountExternalIds);
        Map<String, WxAccountDatastatVO> wxAccountDatastatMap = wxAccountDatastatVOList.stream()
                .collect(Collectors.toMap(WxAccountDatastatVO::getAccountId, wxAccountDatastatVO -> wxAccountDatastatVO));
        if (CollectionUtils.isNotEmpty(dataList)) {
            List<ContentPlatformGzhPushDataStat> saveList = new ArrayList<>();
            for (Record record : dataList) {
                ContentPlatformGzhPushDataStat item = new ContentPlatformGzhPushDataStat();
                String ghId = (String) record.get(0);
                int firstLevelCount = Integer.parseInt((String) record.get(1));
                Integer fissionCount = parseInteger(record.get(2));
                item.setDateStr(dt);
                if (!ghIds.contains(ghId)) {
                    continue;
                }
                ContentPlatformGzhAccount gzhAccount = accountMap.get(ghId);
                item.setAccountId(gzhAccount.getId());
                item.setFirstLevelCount(firstLevelCount);
                WxAccountDatastatVO wxAccountDatastatVO = wxAccountDatastatMap.get(gzhAccount.getExternalId());
                if (Objects.nonNull(wxAccountDatastatVO)) {
                    item.setFansIncreaseCount(wxAccountDatastatVO.getFansIncreaseCount());
                }

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal num = BigDecimal.valueOf(fissionCount.doubleValue() * 10 / firstLevelCount);
                    BigDecimal rounded = num.setScale(2, RoundingMode.HALF_UP);
                    item.setScore(rounded.doubleValue());
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            if (CollectionUtils.isNotEmpty(saveList)) {
                dataStatMapperExt.deleteGzhPushDatastat(dt);
                dataStatMapperExt.batchInsertGzhPushDatastat(saveList);
            }
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformGzhPushDatastatTotalJob")
    public ReturnT<String> syncContentPlatformGzhPushDatastatTotalJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Long now = System.currentTimeMillis();
        // 公众号自动回复数据统计
        String sql = String.format(
                "SELECT distinct first_level.channel_shortname, first_level.first_uv, fission.split_uv, price.arpu " +
                        "FROM loghubods.out_channel_mid_first_total first_level " +
                        "left join loghubods.out_channel_mid_split_total fission " +
                        "on first_level.channel_shortname = fission.channel_shortname and first_level.dt = fission.dt " +
                        "and first_level.type = fission.type and first_level.tag = fission.tag " +
                        "left join loghubods.wecom_cooperation_dynamic_unit_price price " +
                        "on first_level.channel_shortname = price.channel_shortname and first_level.dt = price.dt " +
                        "and first_level.type = price.type and first_level.tag = price.tag " +
                        "WHERE first_level.dt = %s and first_level.type = '公众号推送' and first_level.tag = '投放渠道内去重' ;", dt);
        List<Record> dataList = OdpsUtil.getOdpsData(sql);
        // 所有公众号
        List<ContentPlatformAccount> accountList = getAllAccount();
        Map<String, ContentPlatformAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformAccount::getChannel, account -> account));
        List<ContentPlatformGzhAccount> gzhAccountList = getAllGzhAccount();
        Map<Long, List<ContentPlatformGzhAccount>> gzhAccountMap = gzhAccountList.stream()
                .collect(Collectors.groupingBy(ContentPlatformGzhAccount::getCreateAccountId));
        List<String> accountExternalIds = gzhAccountList.stream().map(ContentPlatformGzhAccount::getExternalId)
                .collect(Collectors.toList());
        String dateStr = dt.substring(0, 4) + "-" + dt.substring(4, 6) + "-" + dt.substring(6, 8);
        List<WxAccountDatastatVO> wxAccountDatastatVOList = aigcApiService.getWxAccountDatastat(dateStr, accountExternalIds);
        Map<String, WxAccountDatastatVO> wxAccountDatastatMap = wxAccountDatastatVOList.stream()
                .collect(Collectors.toMap(WxAccountDatastatVO::getAccountId, wxAccountDatastatVO -> wxAccountDatastatVO));
        if (CollectionUtils.isNotEmpty(dataList)) {
            List<ContentPlatformGzhPushDataStatTotal> saveList = new ArrayList<>();
            for (Record record : dataList) {
                ContentPlatformGzhPushDataStatTotal item = new ContentPlatformGzhPushDataStatTotal();
                String channel = record.getString(0);
                int firstLevelCount = Integer.parseInt((String) record.get(1));
                Integer fissionCount = parseInteger(record.get(2));
                Double arpu = parseDouble(record.get(3));
                item.setDateStr(dt);
                item.setChannel(channel);
                item.setFirstLevelCount(firstLevelCount);
                ContentPlatformAccount account = accountMap.get(channel);
                if (Objects.isNull(account)) {
                    continue;
                }
                List<ContentPlatformGzhAccount> gzhAccounts = gzhAccountMap.get(account.getId());
                int fansIncreaseCount = 0;
                if (CollectionUtils.isNotEmpty(gzhAccounts)) {
                    for (ContentPlatformGzhAccount gzhAccount : gzhAccounts) {
                        WxAccountDatastatVO wxAccountDatastatVO = wxAccountDatastatMap.get(gzhAccount.getExternalId());
                        if (Objects.nonNull(wxAccountDatastatVO)) {
                            fansIncreaseCount += wxAccountDatastatVO.getFansIncreaseCount();
                        }
                    }
                }
                item.setFansIncreaseCount(fansIncreaseCount);

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() / firstLevelCount);
                    BigDecimal rounded = fissionRate.multiply(new BigDecimal(10)).setScale(2, RoundingMode.HALF_UP);
                    item.setScore(rounded.doubleValue());
                    BigDecimal unitPrice = getUnitPrice(account.getPrice(), fissionRate, arpu, BussinessTypeEnum.GZH_PUSH);
                    if (Objects.nonNull(unitPrice)) {
                        item.setUnitPrice(unitPrice.doubleValue());
                        BigDecimal settlementAmount = unitPrice.multiply(new BigDecimal(firstLevelCount)).setScale(2, RoundingMode.HALF_UP);
                        item.setSettlementAmount(settlementAmount.doubleValue());
                    }
                } else {
                    item.setScore(0.0);
                    item.setUnitPrice(0.1);
                    item.setSettlementAmount(0.0);
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            if (CollectionUtils.isNotEmpty(saveList)) {
                dataStatMapperExt.deleteGzhPushDatastatTotal(dt);
                dataStatMapperExt.batchInsertGzhPushDatastatTotal(saveList);
            }
        }
        return ReturnT.SUCCESS;
    }

    private List<ContentPlatformAccount> getAllAccount() {
        ContentPlatformAccountExample example = new ContentPlatformAccountExample();
        example.createCriteria().andStatusEqualTo(AccountStatusEnum.NORMAL.getVal());
        return accountMapper.selectByExample(example);
    }

    private List<ContentPlatformGzhAccount> getAllGzhAccount() {
        ContentPlatformGzhAccountExample example = new ContentPlatformGzhAccountExample();
        example.createCriteria().andStatusEqualTo(AccountStatusEnum.NORMAL.getVal());
        return gzhAccountMapper.selectByExample(example);
    }

    @XxlJob("syncContentPlatformQwDatastatJob")
    public ReturnT<String> syncContentPlatformQwDatastatJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Long now = System.currentTimeMillis();

        // 1. 一次性查询两张 ODPS 表，收集 rootSourceId → firstLevelCount
        // qw_out 优先，同一个 rootSourceId 在 qw_out 中存在则跳过 qw_out2
        Map<String, Integer> outFirstLevelMap = new HashMap<>();
        Map<String, Integer> out2FirstLevelMap = new HashMap<>();
        int odpsPageSize = 5000;

        int odpsPageNum = 1;
        while (true) {
            Integer offset = (odpsPageNum - 1) * odpsPageSize;
            String outSql = String.format("SELECT rootsourceid, 首层访问人数 " +
                    "FROM loghubods.qw_out_touliu_behavior_detail " +
                    "WHERE dt=%s and 首层访问人数 > 0 limit %s,%s;", dt, offset, odpsPageSize);
            List<Record> outDataList = OdpsUtil.getOdpsData(outSql);
            if (CollectionUtils.isEmpty(outDataList)) {
                break;
            }
            for (Record record : outDataList) {
                outFirstLevelMap.put((String) record.get(0), Integer.parseInt((String) record.get(1)));
            }
            if (outDataList.size() < odpsPageSize) {
                break;
            }
            odpsPageNum++;
        }

        odpsPageNum = 1;
        while (true) {
            Integer offset = (odpsPageNum - 1) * odpsPageSize;
            String out2Sql = String.format("SELECT rootsourceid, 首层访问人数 " +
                    "FROM loghubods.qw_out2_touliu_behavior_detail " +
                    "WHERE dt=%s and 首层访问人数 > 0 limit %s,%s;", dt, offset, odpsPageSize);
            List<Record> outDataList = OdpsUtil.getOdpsData(out2Sql);
            if (CollectionUtils.isEmpty(outDataList)) {
                break;
            }
            for (Record record : outDataList) {
                String rootSourceId = (String) record.get(0);
                // qw_out 优先，已存在的跳过
                if (!outFirstLevelMap.containsKey(rootSourceId)) {
                    out2FirstLevelMap.put(rootSourceId, Integer.parseInt((String) record.get(1)));
                }
            }
            if (outDataList.size() < odpsPageSize) {
                break;
            }
            odpsPageNum++;
        }

        if (outFirstLevelMap.isEmpty() && out2FirstLevelMap.isEmpty()) {
            return ReturnT.SUCCESS;
        }

        // 2. 根据 ODPS 查到的 rootSourceId，只查询匹配的 qwPlan（不再查询全量 qwPlan）
        Set<String> allRootSourceIds = new HashSet<>();
        allRootSourceIds.addAll(outFirstLevelMap.keySet());
        allRootSourceIds.addAll(out2FirstLevelMap.keySet());
        List<String> rootSourceIdList = new ArrayList<>(allRootSourceIds);
        List<List<String>> rootSourceIdBatches = Lists.partition(rootSourceIdList, 5000);

        boolean firstBatch = true;
        for (List<String> batchRootSourceIds : rootSourceIdBatches) {
            List<ContentPlatformQwPlan> qwPlanList = planService.getQwPlanListByRootSourceIds(batchRootSourceIds);
            if (CollectionUtils.isEmpty(qwPlanList)) {
                continue;
            }
            // 构建 plan 相关 map
            Map<Long, ContentPlatformQwPlan> planMap = new HashMap<>();
            Map<String, Long> rootSourceIdToPlanId = new HashMap<>();
            for (ContentPlatformQwPlan plan : qwPlanList) {
                planMap.put(plan.getId(), plan);
                rootSourceIdToPlanId.put(plan.getRootSourceId(), plan.getId());
            }
            // 查询 planVideo
            List<Long> planIds = new ArrayList<>(planMap.keySet());
            List<ContentPlatformQwPlanVideo> planVideoList = planService.getQwPlanVideoList(planIds);
            Map<Long, Long> planVideoMap = planVideoList.stream()
                    .collect(Collectors.toMap(ContentPlatformQwPlanVideo::getPlanId, ContentPlatformQwPlanVideo::getVideoId));
            // 查询 videoScore
            List<Long> videoIds = planVideoList.stream().map(ContentPlatformQwPlanVideo::getVideoId).collect(Collectors.toList());
            Map<Long, Double> videoScoreMap = new HashMap<>();
            if (CollectionUtils.isNotEmpty(videoIds)) {
                List<List<Long>> videoIdPartitions = Lists.partition(videoIds, 2000);
                for (List<Long> partition : videoIdPartitions) {
                    List<ContentPlatformVideoAgg> videoList = planService.getVideoContentAggListByVideoIds(partition);
                    for (ContentPlatformVideoAgg video : videoList) {
                        videoScoreMap.putIfAbsent(video.getVideoId(), video.getScore());
                    }
                }
            }
            // 遍历当前批次的 rootSourceId，从 ODPS map 中匹配数据
            List<ContentPlatformQwDataStat> saveList = new ArrayList<>();
            for (String rootSourceId : batchRootSourceIds) {
                Long planId = rootSourceIdToPlanId.get(rootSourceId);
                if (planId == null) {
                    continue;
                }
                // qw_out 优先，其次 qw_out2
                Integer firstLevelCount = outFirstLevelMap.get(rootSourceId);
                boolean isFromOut = (firstLevelCount != null);
                if (!isFromOut) {
                    firstLevelCount = out2FirstLevelMap.get(rootSourceId);
                }
                if (firstLevelCount == null) {
                    continue;
                }
                ContentPlatformQwDataStat item = new ContentPlatformQwDataStat();
                item.setDateStr(dt);
                item.setRootSourceId(rootSourceId);
                item.setFirstLevelCount(firstLevelCount);
                if (isFromOut) {
                    ContentPlatformQwPlan qwPlan = planMap.get(planId);
                    item.setSubChannel(StringUtils.hasText(qwPlan.getSubChannel()) ? qwPlan.getSubChannel() : "未知");
                }
                Long videoId = planVideoMap.get(planId);
                Double score = videoScoreMap.get(videoId);
                if (Objects.nonNull(score)) {
                    item.setScore(BigDecimal.valueOf(score).setScale(2, RoundingMode.HALF_UP).doubleValue());
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            // 入库：第一批先删再插，后续批次直接追加
            if (CollectionUtils.isNotEmpty(saveList)) {
                if (firstBatch) {
                    dataStatMapperExt.deleteQwDatastat(dt);
                    firstBatch = false;
                }
                dataStatMapperExt.batchInsertQwDatastat(saveList);
            }
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformQwDatastatTotalJob")
    public ReturnT<String> syncContentPlatformQwDatastatTotalJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Map<String, ContentPlatformQwDataStatTotal> saveMap = new HashMap<>();
        String outSql = String.format(
                "SELECT distinct first_level.channel_shortname, first_level.first_uv, fission.split_uv, price.arpu " +
                        "FROM loghubods.out_channel_mid_first_total first_level " +
                        "left join loghubods.out_channel_mid_split_total fission " +
                        "on first_level.channel_shortname = fission.channel_shortname and first_level.dt = fission.dt " +
                        "and first_level.type = fission.type and first_level.tag = fission.tag " +
                        "left join loghubods.wecom_cooperation_dynamic_unit_price price " +
                        "on first_level.channel_shortname = price.channel_shortname and first_level.dt = price.dt " +
                        "and first_level.type = price.type and first_level.tag = price.tag " +
                        "WHERE first_level.dt = %s and first_level.type = '企微外部' and first_level.tag = '投放渠道内去重' ;", dt);
        List<Record> outDataList = OdpsUtil.getOdpsData(outSql);
        List<ContentPlatformAccount> accountList = getAllAccount();
        Map<String, ContentPlatformAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformAccount::getChannel, account -> account));
        Long now = System.currentTimeMillis();
        if (CollectionUtils.isNotEmpty(outDataList)) {
            for (Record record : outDataList) {
                String channel = (String) record.get(0);
                int firstLevelCount = Integer.parseInt((String) record.get(1));
                Integer fissionCount = parseInteger(record.get(2));
                Double arpu = parseDouble(record.get(3));
                if (firstLevelCount == 0) {
                    continue;
                }
                ContentPlatformQwDataStatTotal item = saveMap.getOrDefault(channel, new ContentPlatformQwDataStatTotal());
                item.setDateStr(dt);
                item.setChannel(channel);
                if (item.getFirstLevelCount() == null) {
                    item.setFirstLevelCount(0);
                }
                ContentPlatformAccount account = accountMap.get(channel);
                if (Objects.isNull(account)) {
                    account = new ContentPlatformAccount();
                    account.setPrice("{\n" +
                            "\"qiweihezuo_price_mode\": \"static\",\n" +
                            "\"qiweihezuo_dynamic_times\":3,\n" +
                            "\"qiweihezuo_static_price\":0.12,\n" +
                            "\"fuwuhaodaiyunying_price_mode\": \"dynamic\",\n" +
                            "\"fuwuhaodaiyunying_dynamic_times\":3,\n" +
                            "\"fuwuhaodaiyunying_static_price\":0.12,\n" +
                            "\"gongzhonghaozidonghuifu_price_mode\": \"static\",\n" +
                            "\"gongzhonghaozidonghuifu_dynamic_times\":3,\n" +
                            "\"gongzhonghaozidonghuifu_static_price\":0.12\n" +
                            "}");
                }

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() / firstLevelCount);
                    BigDecimal rounded = fissionRate.multiply(new BigDecimal(10)).setScale(2, RoundingMode.HALF_UP);
                    item.setScore(rounded.doubleValue());
                    BigDecimal unitPrice = getUnitPrice(account.getPrice(), fissionRate, arpu, BussinessTypeEnum.QW);
                    if (Objects.nonNull(unitPrice)) {
                        item.setUnitPrice(unitPrice.doubleValue());
                        BigDecimal settlementAmount = unitPrice.multiply(new BigDecimal(firstLevelCount)).setScale(2, RoundingMode.HALF_UP);
                        item.setSettlementAmount(settlementAmount.doubleValue());
                    }
                } else {
                    item.setScore(0.0);
                    item.setUnitPrice(0.1);
                    item.setSettlementAmount(0.0);
                }
                item.setFirstLevelCount(item.getFirstLevelCount() + firstLevelCount);
                item.setCreateTimestamp(now);
                saveMap.put(channel, item);
            }
        }
        List<ContentPlatformQwDataStatTotal> saveList = new ArrayList<>(saveMap.values());
        if (CollectionUtils.isNotEmpty(saveList)) {
            dataStatMapperExt.deleteQwDatastatTotal(dt);
            dataStatMapperExt.batchInsertQwDatastatTotal(saveList);
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformQwDatastatReplyTotalJob")
    public ReturnT<String> syncContentPlatformQwDatastatReplyTotalJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        Map<String, ContentPlatformQwDataStatReplyTotal> saveMap = new HashMap<>();
        String outSql = String.format("SELECT * FROM loghubods.qw_out_touliu_behavior_total WHERE dt=%s and 推送方式 = '自动回复';", dt);
        List<Record> outDataList = OdpsUtil.getOdpsData(outSql);
        Long now = System.currentTimeMillis();
        if (CollectionUtils.isNotEmpty(outDataList)) {
            for (Record record : outDataList) {
                String channel = (String) record.get(33);
                int firstLevelCount = Integer.parseInt((String) record.get(6));
                if (firstLevelCount == 0) {
                    continue;
                }
                ContentPlatformQwDataStatReplyTotal item = saveMap.getOrDefault(channel, new ContentPlatformQwDataStatReplyTotal());
                item.setDateStr(dt);
                item.setChannel(channel);
                if (item.getFirstLevelCount() == null) {
                    item.setFirstLevelCount(0);
                }
                item.setFirstLevelCount(item.getFirstLevelCount() + firstLevelCount);
                item.setCreateTimestamp(now);
                saveMap.put(channel, item);
            }
        }
        String out2Sql = String.format("SELECT * FROM loghubods.qw_out2_touliu_behavior_total WHERE dt=%s and 推送方式 = '自动回复';", dt);
        List<Record> out2DataList = OdpsUtil.getOdpsData(out2Sql);
        if (CollectionUtils.isNotEmpty(out2DataList)) {
            List<String> out2Channels = new ArrayList<>();
            for (Record record : out2DataList) {
                String channel = (String) record.get(33);
                int firstLevelCount = Integer.parseInt((String) record.get(6));
                if (firstLevelCount == 0) {
                    continue;
                }
                if (saveMap.containsKey(channel) && !out2Channels.contains(channel)) {
                    saveMap.remove(channel);
                }
                ContentPlatformQwDataStatReplyTotal item = saveMap.getOrDefault(channel, new ContentPlatformQwDataStatReplyTotal());
                item.setDateStr(dt);
                item.setChannel(channel);
                if (item.getFirstLevelCount() == null) {
                    item.setFirstLevelCount(0);
                }
                item.setFirstLevelCount(item.getFirstLevelCount() + firstLevelCount);
                item.setCreateTimestamp(now);
                out2Channels.add(channel);
                saveMap.put(channel, item);
            }
        }
        List<ContentPlatformQwDataStatReplyTotal> saveList = new ArrayList<>(saveMap.values());
        if (CollectionUtils.isNotEmpty(saveList)) {
            List<String> channels = saveList.stream().map(ContentPlatformQwDataStatReplyTotal::getChannel).distinct()
                    .collect(Collectors.toList());
            List<ContentPlatformQwDataStatReplyTotal> channelScoreList = dataStatMapperExt.getQwTotalDatastatReplyScoreList(dt, channels);
            Map<String, Double> channelScoreMap = channelScoreList.stream()
                    .filter(o -> Objects.nonNull(o.getScore()))
                    .collect(Collectors.toMap(ContentPlatformQwDataStatReplyTotal::getChannel,
                            ContentPlatformQwDataStatReplyTotal::getScore, (a, b) -> a));
            for (ContentPlatformQwDataStatReplyTotal item : saveList) {
                item.setScore(channelScoreMap.get(item.getChannel()));
            }
            dataStatMapperExt.deleteQwDatastatReplyTotal(dt);
            dataStatMapperExt.batchInsertQwDatastatReplyTotal(saveList);
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("syncContentPlatformQwDatastatSubChannelJob")
    public ReturnT<String> syncContentPlatformQwDatastatSubChannelJob(String param) {
        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
        if (StringUtils.hasText(param)) {
            dt = param;
        }
        List<ContentPlatformQwDataStatSubChannel> saveList = new ArrayList<>();
        String outSql = String.format(
                "SELECT distinct first_level.channel_shortname, first_level.subchannel, first_level.first_uv, fission.split_uv, price.arpu " +
                        "FROM loghubods.out_channel_mid_first_total first_level " +
                        "left join loghubods.out_channel_mid_split_total fission " +
                        "on first_level.channel_shortname = fission.channel_shortname and first_level.subchannel = fission.subchannel " +
                        "and first_level.dt = fission.dt and first_level.type = fission.type and first_level.tag = fission.tag " +
                        "left join loghubods.wecom_cooperation_dynamic_unit_price price " +
                        "on first_level.channel_shortname = price.channel_shortname and first_level.subchannel = price.subchannel " +
                        "and first_level.dt = price.dt and first_level.type = price.type and first_level.tag = price.tag " +
                        "WHERE first_level.dt = %s and first_level.type = '企微外部' and first_level.tag = '分投放渠道客户分账号去重' ;", dt);
        List<Record> outDataList = OdpsUtil.getOdpsData(outSql);
        List<ContentPlatformAccount> accountList = getAllAccount();
        Map<String, ContentPlatformAccount> accountMap = accountList.stream()
                .collect(Collectors.toMap(ContentPlatformAccount::getChannel, account -> account));
        List<ContentPlatformQwPlan> qwPlanList = getSubChannelQwPlanList();
        Map<Long, List<ContentPlatformQwPlan>> qwPlanMap = qwPlanList.stream().collect(
                Collectors.groupingBy(ContentPlatformQwPlan::getCreateAccountId));
        Long now = System.currentTimeMillis();
        if (CollectionUtils.isNotEmpty(outDataList)) {
            List<Record> noSubChannelRecordList = new ArrayList<>();
            for (Record record : outDataList) {
                String channel = (String) record.get(0);
                String subChannel = (String) record.get(1);
                int firstLevelCount = Integer.parseInt((String) record.get(2));
                Integer fissionCount = parseInteger(record.get(3));
                Double arpu = parseDouble(record.get(4));
                if (firstLevelCount == 0) {
                    continue;
                }
                ContentPlatformQwDataStatSubChannel item = new ContentPlatformQwDataStatSubChannel();
                item.setDateStr(dt);
                ContentPlatformAccount account = accountMap.get(channel);
                if (Objects.isNull(account)) {
                    continue;
                }
                List<ContentPlatformQwPlan> qwPlans = qwPlanMap.getOrDefault(account.getId(), new ArrayList<>());
                List<String> subChannelList = qwPlans.stream().map(ContentPlatformQwPlan::getSubChannel).distinct()
                        .collect(Collectors.toList());
                if (CollectionUtils.isEmpty(qwPlans) || !subChannelList.contains(subChannel)) {
                    noSubChannelRecordList.add(record);
                    continue;
                }
                item.setCreateAccountId(account.getId());
                item.setSubChannel(subChannel);
                item.setFirstLevelCount(firstLevelCount);

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() / firstLevelCount);
                    BigDecimal rounded = fissionRate.multiply(new BigDecimal(10)).setScale(2, RoundingMode.HALF_UP);
                    item.setScore(rounded.doubleValue());
                    BigDecimal unitPrice = getUnitPrice(account.getPrice(), fissionRate, arpu, BussinessTypeEnum.QW);
                    if (Objects.nonNull(unitPrice)) {
                        item.setUnitPrice(unitPrice.doubleValue());
                        BigDecimal settlementAmount = unitPrice.multiply(new BigDecimal(firstLevelCount)).setScale(2, RoundingMode.HALF_UP);
                        item.setSettlementAmount(settlementAmount.doubleValue());
                    }
                } else {
                    item.setScore(0.0);
                    item.setUnitPrice(0.1);
                    item.setSettlementAmount(0.0);
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
            for (Record record : noSubChannelRecordList) {
                String channel = (String) record.get(0);
                int firstLevelCount = Integer.parseInt((String) record.get(2));
                Integer fissionCount = parseInteger(record.get(3));
                Double arpu = parseDouble(record.get(4));
                ContentPlatformQwDataStatSubChannel item = new ContentPlatformQwDataStatSubChannel();
                item.setDateStr(dt);
                ContentPlatformAccount account = accountMap.get(channel);
                if (Objects.isNull(account)) {
                    continue;
                }
                item.setCreateAccountId(account.getId());
                item.setSubChannel("未知");
                item.setFirstLevelCount(firstLevelCount);

                if (fissionCount > 0 && firstLevelCount > 0) {
                    BigDecimal fissionRate = BigDecimal.valueOf(fissionCount.doubleValue() / firstLevelCount);
                    BigDecimal rounded = fissionRate.multiply(new BigDecimal(10)).setScale(2, RoundingMode.HALF_UP);
                    item.setScore(rounded.doubleValue());
                    BigDecimal unitPrice = getUnitPrice(account.getPrice(), fissionRate, arpu, BussinessTypeEnum.QW);
                    if (Objects.nonNull(unitPrice)) {
                        item.setUnitPrice(unitPrice.doubleValue());
                        BigDecimal settlementAmount = unitPrice.multiply(new BigDecimal(firstLevelCount)).setScale(2, RoundingMode.HALF_UP);
                        item.setSettlementAmount(settlementAmount.doubleValue());
                    }
                } else {
                    item.setScore(0.0);
                    item.setUnitPrice(0.1);
                    item.setSettlementAmount(0.0);
                }
                item.setCreateTimestamp(now);
                saveList.add(item);
            }
        }
        if (CollectionUtils.isNotEmpty(saveList)) {
            dataStatMapperExt.deleteQwDatastatSubChannel(dt);
            dataStatMapperExt.batchInsertQwDatastatSubChannel(saveList);
        }
        return ReturnT.SUCCESS;
    }

    private List<ContentPlatformQwPlan> getSubChannelQwPlanList() {
        ContentPlatformQwPlanExample example = new ContentPlatformQwPlanExample();
        example.createCriteria().andSubChannelIsNotNull();
        return qwPlanMapper.selectByExample(example);
    }

    private Integer parseInteger(Object value) {
        try {
            return Integer.parseInt((String) value);
        } catch (NumberFormatException e) {
            return 0;
        }
    }

    private Long parseLong(Object value) {
        try {
            return Long.parseLong((String) value);
        } catch (NumberFormatException e) {
            return 0L;
        }
    }

    private Double parseDouble(Object value) {
        try {
            return Double.parseDouble((String) value);
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    private Double getUnitPriceMin(BussinessTypeEnum typeEnum) {
        return unitPriceMinConfig.get(typeEnum.getVal());
    }

    private Double getUnitPriceMax(BussinessTypeEnum typeEnum) {
        return unitPriceMaxConfig.get(typeEnum.getVal());
    }

    private BigDecimal getUnitPrice(String price, BigDecimal fissionRate, Double arpu, BussinessTypeEnum typeEnum) {
        JSONObject priceJSON;
        try {
            priceJSON = JSONObject.parseObject(price);
        } catch (Exception e) {
            return null;
        }
        String priceModeKey = typeEnum.getPrefix() + "_price_mode";
        String staticPriceKey = typeEnum.getPrefix() + "_static_price";
        String dynamicTimesKey = typeEnum.getPrefix() + "_dynamic_times";
        String priceMode = priceJSON.getString(priceModeKey);
        Double staticPrice = priceJSON.getDouble(staticPriceKey);
        Double dynamicTimes = priceJSON.getDouble(dynamicTimesKey);
        if ("static".equals(priceMode)) {
            if (Objects.isNull(staticPrice) || staticPrice == 0) {
                return null;
            }
            return new BigDecimal(staticPrice);
        } else if ("dynamic".equals(priceMode)) {
            if (Objects.isNull(fissionRate) || Objects.isNull(arpu) || Objects.isNull(dynamicTimes) || dynamicTimes == 0) {
                return null;
            }
            return fissionRate.multiply(new BigDecimal(arpu))
                    .multiply(BigDecimal.valueOf(dynamicTimes))
                    .setScale(2, RoundingMode.HALF_UP)
                    .min(BigDecimal.valueOf(getUnitPriceMax(typeEnum)))
                    .max(BigDecimal.valueOf(getUnitPriceMin(typeEnum)));
        }
        return null;
    }

    @XxlJob("syncQwHzAccountLayerTypeJob")
    public ReturnT<String> syncQwHzAccountLayerTypeJob(String param) {
        String sql = "SELECT partner_channel, layer_type FROM loghubods.feishu_wechat_qw_hz_account_base;";
        List<Record> dataList = OdpsUtil.getOdpsData(sql);
        if (CollectionUtils.isEmpty(dataList)) {
            log.info("syncQwHzAccountLayerTypeJob no data found");
            return ReturnT.SUCCESS;
        }
        for (Record record : dataList) {
            String partnerChannel = (String) record.get(0);
            String layerType = (String) record.get(1);
            if (StringUtils.hasText(partnerChannel) && StringUtils.hasText(layerType)) {
                redisUtils.set("cp:qw_hz_layer_type:" + partnerChannel, layerType, RedisUtils.DEFAULT_EXPIRE_TIME);
            }
        }
        log.info("syncQwHzAccountLayerTypeJob done, size: {}", dataList.size());
        return ReturnT.SUCCESS;
    }
}
