|
|
@@ -1,41 +1,39 @@
|
|
|
package com.moka.gdtauto.service;
|
|
|
|
|
|
-import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
|
|
+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.Set;
|
|
|
+import java.util.concurrent.TimeUnit;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+import org.springframework.scheduling.annotation.Async;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+
|
|
|
+import com.google.common.collect.Lists;
|
|
|
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.exception.TencentAdsResponseException;
|
|
|
-import com.tencent.ads.model.v3.DailyReportsFilteringStruct;
|
|
|
-import com.tencent.ads.model.v3.FilterOperator;
|
|
|
+import com.tencent.ads.model.v3.DailyReportsGetResponseData;
|
|
|
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 jakarta.validation.constraints.NotNull;
|
|
|
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.concurrent.TimeUnit;
|
|
|
-import java.util.stream.Collectors;
|
|
|
|
|
|
/**
|
|
|
* 广告计划日报表同步服务
|
|
|
* 按广告组级别同步日报数据到 ad_plan 表
|
|
|
+ * 改造为按 accountId 直接查询日报，无需 adgroup_id 过滤
|
|
|
*
|
|
|
* @author moka
|
|
|
* @since 2026-03-10
|
|
|
@@ -48,6 +46,7 @@ public class AdPlanDailyReportService {
|
|
|
private final TencentAdsApiClientFactory clientFactory;
|
|
|
private final AdPlanConfMapper adPlanConfMapper;
|
|
|
private final AdPlanMapper adPlanMapper;
|
|
|
+ private final AdPlanService adPlanService;
|
|
|
|
|
|
/** REQUEST_TIME 口径同步字段 */
|
|
|
private static final List<String> REQ_FIELDS = Arrays.asList(
|
|
|
@@ -58,9 +57,6 @@ public class AdPlanDailyReportService {
|
|
|
private static final List<String> REP_FIELDS = Arrays.asList(
|
|
|
"adgroup_id", "date", "income_val_1", "income_roi_1");
|
|
|
|
|
|
- private static final int PAGE_SIZE = 100;
|
|
|
- private static final int ADGROUP_BATCH_LIMIT = 100;
|
|
|
-
|
|
|
@Async
|
|
|
public void syncDailyReport(Long confId) {
|
|
|
LocalDateTime start = LocalDateTime.now();
|
|
|
@@ -93,229 +89,139 @@ public class AdPlanDailyReportService {
|
|
|
|
|
|
/**
|
|
|
* 执行同步逻辑
|
|
|
+ * 直接按 accountId 查询日报，无需 adgroup_id 过滤
|
|
|
*
|
|
|
- * @param startTime 开始时间
|
|
|
- * @param endTime 结束时间
|
|
|
- * @param timeLine 时间口径
|
|
|
- * @param isReqTime 是否为 REQUEST_TIME 口径（决定更新哪些字段）
|
|
|
+ * @param startTime 开始时间
|
|
|
+ * @param endTime 结束时间
|
|
|
+ * @param timeLine 时间口径
|
|
|
+ * @param isReqTime 是否为 REQUEST_TIME 口径（决定更新哪些字段）
|
|
|
*/
|
|
|
- private void doSyncDailyReport(Long confId, LocalDateTime startTime, LocalDateTime endTime, String timeLine,
|
|
|
- boolean isReqTime) {
|
|
|
- // 1. 分页查询 create_time 在范围内的 adPlanConf
|
|
|
- int confPageNum = 1;
|
|
|
- int totalUpdated = 0;
|
|
|
- Map<Long, Integer> confUpdatedCount = new ConcurrentHashMap<>();
|
|
|
-
|
|
|
- while (true) {
|
|
|
- LambdaQueryWrapper<AdPlanConf> confWrapper = new LambdaQueryWrapper<>();
|
|
|
- confWrapper
|
|
|
- .eq(confId != null, AdPlanConf::getId, confId)
|
|
|
- .eq(AdPlanConf::getDeleted, 0)
|
|
|
- .orderByDesc(AdPlanConf::getId);
|
|
|
- Page<AdPlanConf> confPage = adPlanConfMapper.selectPage(new Page<>(confPageNum, PAGE_SIZE), confWrapper);
|
|
|
- List<AdPlanConf> confList = confPage.getRecords();
|
|
|
-
|
|
|
- if (confList.isEmpty()) {
|
|
|
- break;
|
|
|
- }
|
|
|
-
|
|
|
- // 2. 每个 conf 启动一个线程处理
|
|
|
- List<CompletableFuture<Void>> futures = confList.stream()
|
|
|
- .map(conf -> CompletableFuture.runAsync(() -> {
|
|
|
- try {
|
|
|
- int updated = processConf(conf, startTime, endTime, timeLine, isReqTime);
|
|
|
- confUpdatedCount.put(conf.getId(), updated);
|
|
|
- } catch (Exception e) {
|
|
|
- log.error("[doSyncDailyReport] 处理 confId={} 失败", conf.getId(), e);
|
|
|
- }
|
|
|
- }))
|
|
|
- .collect(Collectors.toList());
|
|
|
-
|
|
|
- // 等待所有线程完成
|
|
|
- CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
|
|
|
-
|
|
|
- if (!confPage.hasNext()) {
|
|
|
- break;
|
|
|
- }
|
|
|
- confPageNum++;
|
|
|
+ private void doSyncDailyReport(@NotNull Long confId, LocalDateTime startTime, LocalDateTime endTime, String timeLine, boolean isReqTime) {
|
|
|
+ // 查询该配置下的所有 adgroup_id 和 account_id
|
|
|
+ List<AdPlan> adPlans = adPlanService.lambdaQuery()
|
|
|
+ .select(AdPlan::getAccountId, AdPlan::getAdgroupId)
|
|
|
+ .gt(AdPlan::getAdgroupId, 0)
|
|
|
+ .eq(AdPlan::getAdPlanConfId, confId)
|
|
|
+ .list();
|
|
|
+ Set<Long> adgroupIdSet = adPlans.stream().map(AdPlan::getAdgroupId).collect(Collectors.toSet());
|
|
|
+ Set<Long> accountIdSet = adPlans.stream().map(AdPlan::getAccountId).collect(Collectors.toSet());
|
|
|
+ if (adgroupIdSet.isEmpty()) {
|
|
|
+ log.warn("[doSyncDailyReport] 没有广告组数据，不进行同步");
|
|
|
+ return;
|
|
|
}
|
|
|
-
|
|
|
- totalUpdated = confUpdatedCount.values().stream().mapToInt(Integer::intValue).sum();
|
|
|
- log.info("[doSyncDailyReport] 同步完成 timeLine={} totalUpdated={}", timeLine, totalUpdated);
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 处理单个 AdPlanConf
|
|
|
- */
|
|
|
- private int processConf(AdPlanConf conf, LocalDateTime startTime, LocalDateTime endTime, String timeLine,
|
|
|
- boolean isReqTime) throws Exception {
|
|
|
- String jobNumber = conf.getJobNumber();
|
|
|
- TencentAds tencentAds = clientFactory.getTencentAds(jobNumber, Constant.MAIN_ORG_ACCOUNT_ID);
|
|
|
-
|
|
|
String startStr = startTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
|
|
|
String endStr = endTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
|
|
|
ReportDateRange dateRange = new ReportDateRange();
|
|
|
dateRange.setStartDate(startStr);
|
|
|
dateRange.setEndDate(endStr);
|
|
|
|
|
|
- log.info("[广告日报] start={} end={} timeLine={} isReqTime={}", startStr, endStr, timeLine, isReqTime);
|
|
|
-
|
|
|
List<String> groupBy = Arrays.asList("adgroup_id", "date");
|
|
|
List<String> fields = isReqTime ? REQ_FIELDS : REP_FIELDS;
|
|
|
|
|
|
- int totalUpdated = 0;
|
|
|
- int planPageNum = 1;
|
|
|
-
|
|
|
- // 3. 分页查询该 conf 下的所有 adPlan
|
|
|
- while (true) {
|
|
|
- LambdaQueryWrapper<AdPlan> planWrapper = new LambdaQueryWrapper<>();
|
|
|
- planWrapper.eq(AdPlan::getAdPlanConfId, conf.getId())
|
|
|
- .isNotNull(AdPlan::getAdgroupId)
|
|
|
- .gt(AdPlan::getAdgroupId, 0)
|
|
|
- .eq(AdPlan::getDeleted, 0)
|
|
|
- .orderByAsc(AdPlan::getId);
|
|
|
- Page<AdPlan> planPage = adPlanMapper.selectPage(new Page<>(planPageNum, PAGE_SIZE), planWrapper);
|
|
|
- List<AdPlan> plans = planPage.getRecords();
|
|
|
-
|
|
|
- if (plans.isEmpty()) {
|
|
|
- break;
|
|
|
- }
|
|
|
-
|
|
|
- // 4. 按 accountId 分组，每 100 个 adgroup_id 一批调用 API
|
|
|
- Map<Long, List<AdPlan>> groupByAccount = plans.stream()
|
|
|
- .filter(p -> p.getAccountId() != null && p.getAdgroupId() != null)
|
|
|
- .collect(Collectors.groupingBy(AdPlan::getAccountId));
|
|
|
-
|
|
|
- for (Map.Entry<Long, List<AdPlan>> entry : groupByAccount.entrySet()) {
|
|
|
- Long accountId = entry.getKey();
|
|
|
- List<AdPlan> plansOfAccount = entry.getValue();
|
|
|
-
|
|
|
- // 按 100 条分批
|
|
|
- int batchCount = (int) Math.ceil((double) plansOfAccount.size() / ADGROUP_BATCH_LIMIT);
|
|
|
- for (int i = 0; i < batchCount; i++) {
|
|
|
- int fromIdx = i * ADGROUP_BATCH_LIMIT;
|
|
|
- int toIdx = Math.min(fromIdx + ADGROUP_BATCH_LIMIT, plansOfAccount.size());
|
|
|
- List<AdPlan> batch = plansOfAccount.subList(fromIdx, toIdx);
|
|
|
-
|
|
|
- try {
|
|
|
- int updated = syncBatch(tencentAds, accountId, batch, dateRange, groupBy, fields, timeLine,
|
|
|
- isReqTime);
|
|
|
- totalUpdated += updated;
|
|
|
- } catch (Exception e) {
|
|
|
- log.error("[processConf] 同步失败 accountId={} confId={} batch={}", accountId, conf.getId(), i, e);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if (!planPage.hasNext()) {
|
|
|
- break;
|
|
|
+ try {
|
|
|
+ TencentAds tencentAds = clientFactory.getTencentAdsWithoutUserToken(Constant.MAIN_ORG_ACCOUNT_ID);
|
|
|
+ for (Long accountId : accountIdSet) {
|
|
|
+ syncBatch(tencentAds, accountId, adgroupIdSet, dateRange, groupBy, fields, timeLine, isReqTime);
|
|
|
}
|
|
|
- planPageNum++;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("[doSyncDailyReport] 同步失败 confId={} startTime={} endTime={} timeLine={} isReqTime={}", confId, startTime, endTime, timeLine, isReqTime, e);
|
|
|
}
|
|
|
-
|
|
|
- return totalUpdated;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * 批量同步单批 adPlan
|
|
|
+ * 批量同步单 accountId 下的日报数据
|
|
|
+ * 支持分页循环获取所有数据，每页数据立即处理并更新
|
|
|
*/
|
|
|
- private int syncBatch(TencentAds tencentAds, Long accountId, List<AdPlan> batch,
|
|
|
- ReportDateRange dateRange, List<String> groupBy, List<String> fields,
|
|
|
- String timeLine, boolean isReqTime) throws Exception {
|
|
|
+ private int syncBatch(TencentAds tencentAds, Long accountId, Set<Long> adgroupIdSet,
|
|
|
+ ReportDateRange dateRange, List<String> groupBy, List<String> fields,
|
|
|
+ String timeLine, boolean isReqTime) throws Exception {
|
|
|
+ // 用于跟踪已处理的 adgroupId，避免重复处理
|
|
|
+ Set<Long> processedAdgroupIds = new java.util.HashSet<>();
|
|
|
+ Long page = 1L;
|
|
|
+ Long pageSize = 2000L;
|
|
|
+ boolean hasMore = true;
|
|
|
+ int totalUpdated = 0;
|
|
|
|
|
|
- // 构建 filtering: adgroup_id IN [...]
|
|
|
- List<String> adgroupIdValues = batch.stream()
|
|
|
- .map(p -> String.valueOf(p.getAdgroupId()))
|
|
|
- .collect(Collectors.toList());
|
|
|
+ // 分页循环获取所有数据，每页立即处理
|
|
|
+ while (hasMore) {
|
|
|
+ DailyReportsGetResponseData response = null;
|
|
|
+ final Long currentPage = page;
|
|
|
+ try {
|
|
|
+ response = tencentAds.dailyReports().dailyReportsGet(
|
|
|
+ "REPORT_LEVEL_ADGROUP",
|
|
|
+ dateRange,
|
|
|
+ groupBy,
|
|
|
+ fields,
|
|
|
+ accountId,
|
|
|
+ null, // filtering 不需要过滤，直接查询该账号下所有广告组
|
|
|
+ null, // orderBy
|
|
|
+ timeLine,
|
|
|
+ currentPage,
|
|
|
+ pageSize,
|
|
|
+ null // modularDetailMode
|
|
|
+ );
|
|
|
+ } catch (TencentAdsResponseException e) {
|
|
|
+ // 11016 Application has reached the request limit per second.
|
|
|
+ // 应用程序已达到每秒的请求限制，请稍后再试
|
|
|
+ // 11017 Application has reached the request limit per minute.
|
|
|
+ // 应用请求已达分钟频次上限，请登陆开发者后台查看当前接口对应分配的频次，并在要求范围内请求接口。
|
|
|
+ if (e.getCode() == 11016) {
|
|
|
+ log.warn("请求达到每秒限制，请稍后再试");
|
|
|
+ TimeUnit.SECONDS.sleep(1);
|
|
|
+ } else if (e.getCode() == 11017) {
|
|
|
+ log.warn("请求达到每分钟限制，请稍后再试");
|
|
|
+ // 计算出还剩多少时间
|
|
|
+ LocalDateTime now = LocalDateTime.now();
|
|
|
+ // the second-of-minute, from 0 to 59
|
|
|
+ int second = now.getSecond();
|
|
|
+ int sleepSeconds = (59 - second);
|
|
|
+ log.warn("请求达到每分钟限制，sleep：{} s", sleepSeconds);
|
|
|
+ TimeUnit.SECONDS.sleep(sleepSeconds);
|
|
|
+ }
|
|
|
+ // 重试当前页
|
|
|
+ continue;
|
|
|
+ }
|
|
|
|
|
|
- DailyReportsFilteringStruct filter = new DailyReportsFilteringStruct();
|
|
|
- filter.setField("adgroup_id");
|
|
|
- filter.setOperator(FilterOperator.IN);
|
|
|
- filter.setValues(adgroupIdValues);
|
|
|
+ if (response == null || response.getList() == null) {
|
|
|
+ log.warn("[syncBatch] accountId={} page={} 未返回日报数据", accountId, currentPage);
|
|
|
+ break;
|
|
|
+ }
|
|
|
|
|
|
- List<DailyReportsFilteringStruct> filtering = new ArrayList<>();
|
|
|
- filtering.add(filter);
|
|
|
+ List<ReportStruct> pageList = response.getList();
|
|
|
+ log.info("[syncBatch] accountId={} page={} 获取 {} 条数据", accountId, currentPage, pageList.size());
|
|
|
|
|
|
- // 调用 API
|
|
|
- DailyReportsGetResponseData response = null;
|
|
|
- try {
|
|
|
- response = tencentAds.dailyReports().dailyReportsGet(
|
|
|
- "REPORT_LEVEL_ADGROUP",
|
|
|
- dateRange,
|
|
|
- groupBy,
|
|
|
- fields,
|
|
|
- accountId,
|
|
|
- filtering,
|
|
|
- null, // orderBy
|
|
|
- timeLine,
|
|
|
- 1L,
|
|
|
- (long) ADGROUP_BATCH_LIMIT,
|
|
|
- null // modularDetailMode
|
|
|
- );
|
|
|
- } catch (TencentAdsResponseException e) {
|
|
|
- // 11016 Application has reached the request limit per second.
|
|
|
- // 应用程序已达到每秒的请求限制，请稍后再试
|
|
|
- // 11017 Application has reached the request limit per minute.
|
|
|
- // 应用请求已达分钟频次上限，请登陆开发者后台查看当前接口对应分配的频次，并在要求范围内请求接口。
|
|
|
- if (e.getCode() == 11016) {
|
|
|
- log.warn("请求达到每秒限制，请稍后再试");
|
|
|
- TimeUnit.SECONDS.sleep(1);
|
|
|
- } else if (e.getCode() == 11017) {
|
|
|
- log.warn("请求达到每分钟限制，请稍后再试");
|
|
|
- // 计算出还剩多少时间
|
|
|
- LocalDateTime now = LocalDateTime.now();
|
|
|
- // the second-of-minute, from 0 to 59
|
|
|
- int second = now.getSecond();
|
|
|
- int sleepSeconds = (59 - second);
|
|
|
- log.warn("请求达到每分钟限制，sleep：{} s", sleepSeconds);
|
|
|
- TimeUnit.SECONDS.sleep(sleepSeconds);
|
|
|
+ // 立即处理当前页数据
|
|
|
+ if (!pageList.isEmpty()) {
|
|
|
+ int updated = processAndUpdatePage(pageList, accountId, adgroupIdSet, processedAdgroupIds, isReqTime);
|
|
|
+ totalUpdated += updated;
|
|
|
}
|
|
|
- syncBatch(tencentAds, accountId, batch, dateRange, groupBy, fields, timeLine, isReqTime);
|
|
|
- }
|
|
|
|
|
|
- if (response == null) {
|
|
|
- log.warn("[syncBatch] accountId={} 未返回日报数据", accountId);
|
|
|
- return 0;
|
|
|
+ // 判断是否有更多数据：当前页数据量小于pageSize表示已到最后一页
|
|
|
+ if (pageList.size() < pageSize) {
|
|
|
+ hasMore = false;
|
|
|
+ } else {
|
|
|
+ page++;
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- // 构建更新数据
|
|
|
- List<AdPlan> toUpdate = new ArrayList<>();
|
|
|
- if (response.getList() != null) {
|
|
|
- for (ReportStruct rs : response.getList()) {
|
|
|
- if (rs.getAdgroupId() == null) {
|
|
|
- continue;
|
|
|
- }
|
|
|
- AdPlan update = new AdPlan();
|
|
|
- update.setAdgroupId(rs.getAdgroupId());
|
|
|
+ // 所有分页完成后，对无报表数据的 adgroupId 填充0值
|
|
|
+ int zeroFilledCount = fillZeroForMissingAdgroups(accountId, adgroupIdSet, processedAdgroupIds, isReqTime);
|
|
|
+ totalUpdated += zeroFilledCount;
|
|
|
|
|
|
- if (isReqTime) {
|
|
|
- // REQUEST_TIME 口径：更新 cost, view_count, valid_click_count, ctr,
|
|
|
- // conversions_cost, req_income_val_1, req_income_roi_1
|
|
|
- update.setCost(rs.getCost());
|
|
|
- update.setViewCount(rs.getViewCount());
|
|
|
- update.setValidClickCount(rs.getValidClickCount());
|
|
|
- update.setCtr(d2b(rs.getCtr()));
|
|
|
- update.setConversionsCost(rs.getConversionsCost());
|
|
|
- update.setReqIncomeVal1(rs.getIncomeVal1());
|
|
|
- update.setReqIncomeRoi1(d2b(rs.getIncomeRoi1()));
|
|
|
- update.setScanFollowUserCost(rs.getScanFollowUserCost());
|
|
|
- } else {
|
|
|
- // REPORTING_TIME 口径：更新 rep_income_val_1, rep_income_roi_1
|
|
|
- update.setRepIncomeVal1(rs.getIncomeVal1());
|
|
|
- update.setRepIncomeRoi1(d2b(rs.getIncomeRoi1()));
|
|
|
- }
|
|
|
- update.setUpdateTime(LocalDateTime.now());
|
|
|
- toUpdate.add(update);
|
|
|
- }
|
|
|
- }
|
|
|
+ log.info("[syncBatch] accountId={} 总共更新 {} 条 timeLine={}", accountId, totalUpdated, timeLine);
|
|
|
+ return totalUpdated;
|
|
|
+ }
|
|
|
|
|
|
- // 无报表数据的，填充0值
|
|
|
- Map<Long, AdPlan> updateMap = toUpdate.stream().collect(Collectors.toMap(AdPlan::getAdgroupId, p -> p));
|
|
|
- batch.forEach(p -> {
|
|
|
- if (!updateMap.containsKey(p.getAdgroupId())) {
|
|
|
+ /**
|
|
|
+ * 对无报表数据的 adgroupId 填充0值
|
|
|
+ */
|
|
|
+ private int fillZeroForMissingAdgroups(Long accountId, Set<Long> adgroupIdSet,
|
|
|
+ Set<Long> processedAdgroupIds, boolean isReqTime) {
|
|
|
+ List<AdPlan> zeroUpdates = new ArrayList<>();
|
|
|
+ adgroupIdSet.forEach(adgroupId -> {
|
|
|
+ if (!processedAdgroupIds.contains(adgroupId)) {
|
|
|
AdPlan update = new AdPlan();
|
|
|
- update.setAdgroupId(p.getAdgroupId());
|
|
|
-
|
|
|
+ update.setAdgroupId(adgroupId);
|
|
|
if (isReqTime) {
|
|
|
update.setCost(0L);
|
|
|
update.setViewCount(0L);
|
|
|
@@ -330,22 +236,79 @@ public class AdPlanDailyReportService {
|
|
|
update.setRepIncomeRoi1(BigDecimal.ZERO);
|
|
|
}
|
|
|
update.setUpdateTime(LocalDateTime.now());
|
|
|
- toUpdate.add(update);
|
|
|
+ zeroUpdates.add(update);
|
|
|
}
|
|
|
});
|
|
|
|
|
|
+ if (zeroUpdates.isEmpty()) {
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 批量更新0值
|
|
|
+ if (isReqTime) {
|
|
|
+ Lists.partition(zeroUpdates, 200).forEach(batch -> adPlanMapper.batchUpdateReqDailyReport(batch));
|
|
|
+ } else {
|
|
|
+ Lists.partition(zeroUpdates, 200).forEach(batch -> adPlanMapper.batchUpdateRepDailyReport(batch));
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("[fillZeroForMissingAdgroups] accountId={} 填充0值 {} 条", accountId, zeroUpdates.size());
|
|
|
+ return zeroUpdates.size();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 处理单页数据并更新数据库
|
|
|
+ */
|
|
|
+ private int processAndUpdatePage(List<ReportStruct> pageList, Long accountId, Set<Long> adgroupIdSet,
|
|
|
+ Set<Long> processedAdgroupIds, boolean isReqTime) {
|
|
|
+ // 构建更新数据
|
|
|
+ List<AdPlan> toUpdate = new ArrayList<>();
|
|
|
+ for (ReportStruct rs : pageList) {
|
|
|
+ if (rs.getAdgroupId() == null) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ Long adgroupId = rs.getAdgroupId();
|
|
|
+ // 跳过不在目标 adgroupIdSet 中的广告组
|
|
|
+ if (!adgroupIdSet.contains(adgroupId)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ // 跳过已处理的 adgroupId
|
|
|
+ if (processedAdgroupIds.contains(adgroupId)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ processedAdgroupIds.add(adgroupId);
|
|
|
+
|
|
|
+ AdPlan update = new AdPlan();
|
|
|
+ update.setAdgroupId(adgroupId);
|
|
|
+ if (isReqTime) {
|
|
|
+ // REQUEST_TIME 口径：更新 cost, view_count, valid_click_count, ctr, conversions_cost, req_income_val_1, req_income_roi_1
|
|
|
+ update.setCost(rs.getCost());
|
|
|
+ update.setViewCount(rs.getViewCount());
|
|
|
+ update.setValidClickCount(rs.getValidClickCount());
|
|
|
+ update.setCtr(d2b(rs.getCtr()));
|
|
|
+ update.setConversionsCost(rs.getConversionsCost());
|
|
|
+ update.setReqIncomeVal1(rs.getIncomeVal1());
|
|
|
+ update.setReqIncomeRoi1(d2b(rs.getIncomeRoi1()));
|
|
|
+ update.setScanFollowUserCost(rs.getScanFollowUserCost());
|
|
|
+ } else {
|
|
|
+ // REPORTING_TIME 口径：更新 rep_income_val_1, rep_income_roi_1
|
|
|
+ update.setRepIncomeVal1(rs.getIncomeVal1());
|
|
|
+ update.setRepIncomeRoi1(d2b(rs.getIncomeRoi1()));
|
|
|
+ }
|
|
|
+ update.setUpdateTime(LocalDateTime.now());
|
|
|
+ toUpdate.add(update);
|
|
|
+ }
|
|
|
+
|
|
|
if (toUpdate.isEmpty()) {
|
|
|
return 0;
|
|
|
}
|
|
|
|
|
|
// 批量更新
|
|
|
if (isReqTime) {
|
|
|
- adPlanMapper.batchUpdateReqDailyReport(toUpdate);
|
|
|
+ Lists.partition(toUpdate, 200).forEach(batch -> adPlanMapper.batchUpdateReqDailyReport(batch));
|
|
|
} else {
|
|
|
- adPlanMapper.batchUpdateRepDailyReport(toUpdate);
|
|
|
+ Lists.partition(toUpdate, 200).forEach(batch -> adPlanMapper.batchUpdateRepDailyReport(batch));
|
|
|
}
|
|
|
|
|
|
- log.info("[syncBatch] accountId={} 批量更新 {} 条 timeLine={}", accountId, toUpdate.size(), timeLine);
|
|
|
return toUpdate.size();
|
|
|
}
|
|
|
|
pudongliang