|
|
@@ -0,0 +1,304 @@
|
|
|
+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"
|
|
|
+ );
|
|
|
+
|
|
|
+ /** 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) {
|
|
|
+ syncReqDailyReport(confId, null, null);
|
|
|
+ syncRepDailyReport(confId, null, null);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 同步 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)
|
|
|
+ .between(confId == null, AdPlanConf::getCreateTime, startTime, endTime)
|
|
|
+ .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, 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, String timeLine, boolean isReqTime) throws Exception {
|
|
|
+ String jobNumber = conf.getJobNumber();
|
|
|
+ TencentAds tencentAds = clientFactory.getTencentAds(jobNumber, Constant.MAIN_ORG_ACCOUNT_ID);
|
|
|
+
|
|
|
+ // 日期范围使用 conf.createTime 的日期
|
|
|
+ LocalDateTime confCreateTime = conf.getCreateTime();
|
|
|
+ String dateStr = confCreateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
|
|
|
+ // + 1天
|
|
|
+ String endStr = confCreateTime.plusDays(1).format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
|
|
|
+ ReportDateRange dateRange = new ReportDateRange();
|
|
|
+ dateRange.setStartDate(dateStr);
|
|
|
+ dateRange.setEndDate(endStr);
|
|
|
+
|
|
|
+ log.info("[广告日报] start={} end={} timeLine={} isReqTime={}", dateStr, 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 || response.getList() == null || response.getList().isEmpty()) {
|
|
|
+ log.warn("[syncBatch] accountId={} 未返回日报数据", accountId);
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 构建更新数据
|
|
|
+ List<AdPlan> toUpdate = new ArrayList<>();
|
|
|
+ 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()));
|
|
|
+ } 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);
|
|
|
+ }
|
|
|
+
|
|
|
+ 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);
|
|
|
+ }
|
|
|
+}
|
pudongliang