|
|
@@ -21,10 +21,12 @@ import com.tzld.longarticle.recommend.server.model.dto.DemandSearchArticleRelati
|
|
|
import com.tzld.longarticle.recommend.server.model.dto.UserGroupCountDTO;
|
|
|
import com.tzld.longarticle.recommend.server.model.entity.crawler.AccountAvgInfo;
|
|
|
import com.tzld.longarticle.recommend.server.model.entity.crawler.PublishSortLog;
|
|
|
+import com.tzld.longarticle.recommend.server.model.param.PlanColdStartSortRequest;
|
|
|
import com.tzld.longarticle.recommend.server.model.param.RecommendParam;
|
|
|
import com.tzld.longarticle.recommend.server.model.param.RecommendRequest;
|
|
|
import com.tzld.longarticle.recommend.server.model.vo.ArticleSortResponseData;
|
|
|
import com.tzld.longarticle.recommend.server.model.vo.ArticleSortResponseDataItem;
|
|
|
+import com.tzld.longarticle.recommend.server.model.vo.PlanColdStartSortResponse;
|
|
|
import com.tzld.longarticle.recommend.server.model.vo.RecommendResponse;
|
|
|
import com.tzld.longarticle.recommend.server.model.vo.RecommendWithUserGroupResponse;
|
|
|
import com.tzld.longarticle.recommend.server.remote.pq.VideoArticleMatchService;
|
|
|
@@ -104,6 +106,9 @@ public class RecommendService {
|
|
|
@Value("${spring.profiles.active}")
|
|
|
private String env;
|
|
|
|
|
|
+ @Value("${coldStart.readingAvgThreshold:100}")
|
|
|
+ private int coldStartReadingAvgThreshold;
|
|
|
+
|
|
|
|
|
|
public RecommendResponse recommend(RecommendRequest request) {
|
|
|
long start = System.currentTimeMillis();
|
|
|
@@ -834,6 +839,306 @@ public class RecommendService {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 跨账号冷启池统一排序分配。
|
|
|
+ * 收集 plan 下所有账号的冷启内容,全局排序后按位置阅读均值门控分配到各账号冷启位(3-8)。
|
|
|
+ */
|
|
|
+ public PlanColdStartSortResponse planColdStartSort(PlanColdStartSortRequest request) {
|
|
|
+ long start = System.currentTimeMillis();
|
|
|
+
|
|
|
+ if (request == null || CollectionUtils.isEmpty(request.getAccounts())) {
|
|
|
+ return PlanColdStartSortResponse.fail("accounts不能为空");
|
|
|
+ }
|
|
|
+ if (!StringUtils.hasText(request.getPlanId())) {
|
|
|
+ return PlanColdStartSortResponse.fail("planId不能为空");
|
|
|
+ }
|
|
|
+
|
|
|
+ String coldPool = ContentPoolEnum.accountDemandPoolLevelCold.getContentPool();
|
|
|
+ List<PlanColdStartSortRequest.AccountInfo> accounts = request.getAccounts();
|
|
|
+
|
|
|
+ // Phase 1 & 2: 并行 per-account recall + rank (只处理冷启内容)
|
|
|
+ List<CompletableFuture<AccountColdRankResult>> futures = new ArrayList<>();
|
|
|
+ for (PlanColdStartSortRequest.AccountInfo account : accounts) {
|
|
|
+ futures.add(CompletableFuture.supplyAsync(() -> rankAccountColdContent(request.getPlanId(), account, coldPool)));
|
|
|
+ }
|
|
|
+
|
|
|
+ List<AccountColdRankResult> perAccountResults = futures.stream()
|
|
|
+ .map(CompletableFuture::join)
|
|
|
+ .collect(Collectors.toList());
|
|
|
+
|
|
|
+ // Phase 3: 合并所有账号冷启内容,全局按 score 降序排序
|
|
|
+ List<GlobalColdItem> globalColdList = new ArrayList<>();
|
|
|
+ for (AccountColdRankResult ar : perAccountResults) {
|
|
|
+ if (CollectionUtils.isNotEmpty(ar.coldContents)) {
|
|
|
+ for (Content c : ar.coldContents) {
|
|
|
+ globalColdList.add(new GlobalColdItem(c, ar.accountId, ar.accountName, ar.ghId));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ globalColdList.sort((a, b) -> Double.compare(b.content.getScore(), a.content.getScore()));
|
|
|
+ log.info("planColdStartSort planId:{} globalColdList size:{}", request.getPlanId(), globalColdList.size());
|
|
|
+
|
|
|
+ // Phase 4: 批量查阅读均值 + 位置分配
|
|
|
+ // 收集所有 ghId
|
|
|
+ Set<String> ghIds = accounts.stream()
|
|
|
+ .map(PlanColdStartSortRequest.AccountInfo::getGhId)
|
|
|
+ .filter(StringUtils::hasText)
|
|
|
+ .collect(Collectors.toSet());
|
|
|
+ List<AccountAvgInfo> allAvgInfo = accountAvgInfoRepository.getAllByGhIdInAndStatusEquals(ghIds, 1);
|
|
|
+ Map<String, List<AccountAvgInfo>> avgByGhId = allAvgInfo.stream()
|
|
|
+ .collect(Collectors.groupingBy(AccountAvgInfo::getGhId));
|
|
|
+
|
|
|
+ // 初始化每个账号的冷启位 tracker: accountId -> nextUnfilledPosition (3-8)
|
|
|
+ Map<String, Integer> nextPosition = new LinkedHashMap<>();
|
|
|
+ Map<String, PlanColdStartSortResponse.AccountAllocation> allocationMap = new LinkedHashMap<>();
|
|
|
+ for (PlanColdStartSortRequest.AccountInfo account : accounts) {
|
|
|
+ nextPosition.put(account.getAccountId(), 3);
|
|
|
+ PlanColdStartSortResponse.AccountAllocation alloc = new PlanColdStartSortResponse.AccountAllocation();
|
|
|
+ alloc.setAccountId(account.getAccountId());
|
|
|
+ alloc.setAccountName(account.getAccountName());
|
|
|
+ alloc.setGhId(account.getGhId());
|
|
|
+ alloc.setPositions(new ArrayList<>());
|
|
|
+ allocationMap.put(account.getAccountId(), alloc);
|
|
|
+ }
|
|
|
+
|
|
|
+ int allocatedCount = 0;
|
|
|
+ int skippedCount = 0;
|
|
|
+ List<String> allocatedSourceIds = new ArrayList<>();
|
|
|
+ int totalColdPositions = accounts.size() * 6; // positions 3-8 per account
|
|
|
+ // sourceId → Content 映射,用于后续写入 publish_sort_log
|
|
|
+ Map<String, Content> sourceIdToContent = new HashMap<>();
|
|
|
+
|
|
|
+ for (GlobalColdItem item : globalColdList) {
|
|
|
+ if (allocatedCount >= totalColdPositions) break;
|
|
|
+
|
|
|
+ String accId = item.accountId;
|
|
|
+ String ghId = item.ghId;
|
|
|
+ Integer nextPos = nextPosition.get(accId);
|
|
|
+ if (nextPos == null || nextPos > 8) {
|
|
|
+ continue; // 该账号冷启位已满
|
|
|
+ }
|
|
|
+
|
|
|
+ // 查找该账号下一个阅读均值 <= threshold 的位置
|
|
|
+ Integer assignedPos = null;
|
|
|
+ for (int pos = nextPos; pos <= 8; pos++) {
|
|
|
+ double avgRead = accountIndexAvgViewCountService.getAvgReadCountByDB(
|
|
|
+ avgByGhId.getOrDefault(ghId, Collections.emptyList()), ghId, pos);
|
|
|
+ if (avgRead <= coldStartReadingAvgThreshold) {
|
|
|
+ assignedPos = pos;
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (assignedPos == null) {
|
|
|
+ // 该账号所有剩余位置阅读均值都超过阈值
|
|
|
+ skippedCount++;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 分配
|
|
|
+ PlanColdStartSortResponse.PositionAllocation pa = new PlanColdStartSortResponse.PositionAllocation();
|
|
|
+ pa.setPositionIndex(assignedPos);
|
|
|
+ pa.setSourceId(item.content.getSourceId());
|
|
|
+ pa.setPublishContentId(item.content.getId());
|
|
|
+ pa.setSourceType(item.content.getSourceType());
|
|
|
+ pa.setTitle(item.content.getTitle());
|
|
|
+ pa.setScore(item.content.getScore());
|
|
|
+ pa.setContentPoolType(item.content.getContentPoolType());
|
|
|
+ pa.setExperimentId(item.content.getExperimentId());
|
|
|
+ pa.setSkipped(false);
|
|
|
+
|
|
|
+ double assignedAvgRead = accountIndexAvgViewCountService.getAvgReadCountByDB(
|
|
|
+ avgByGhId.getOrDefault(ghId, Collections.emptyList()), ghId, assignedPos);
|
|
|
+ pa.setPositionAvgRead(assignedAvgRead);
|
|
|
+
|
|
|
+ allocationMap.get(accId).getPositions().add(pa);
|
|
|
+ allocatedSourceIds.add(item.content.getSourceId());
|
|
|
+ sourceIdToContent.put(item.content.getSourceId(), item.content);
|
|
|
+ nextPosition.put(accId, assignedPos + 1);
|
|
|
+ allocatedCount++;
|
|
|
+ }
|
|
|
+
|
|
|
+ // Phase 4.5: 保存排序结果到 publish_sort_log(供后续数据统计晋级流程使用)
|
|
|
+ savePlanColdStartSortLog(accounts, allocationMap, sourceIdToContent, avgByGhId);
|
|
|
+
|
|
|
+ // Phase 4.6: 清理 publish_content_gzh_waiting 中已分配的 sourceId(标记 status=0)
|
|
|
+ if (!allocatedSourceIds.isEmpty()) {
|
|
|
+ long updateTimestamp = System.currentTimeMillis();
|
|
|
+ for (List<String> partition : Lists.partition(allocatedSourceIds, 2000)) {
|
|
|
+ longArticleBaseMapper.batchUpdateContentStatusBySourceIds(partition, 0, updateTimestamp);
|
|
|
+ }
|
|
|
+ log.info("planColdStartSort cleaned {} sourceIds from publish_content_gzh_waiting", allocatedSourceIds.size());
|
|
|
+ }
|
|
|
+
|
|
|
+ // Phase 5: 构建响应
|
|
|
+ List<PlanColdStartSortResponse.AccountAllocation> data = new ArrayList<>(allocationMap.values());
|
|
|
+ long totalCost = System.currentTimeMillis() - start;
|
|
|
+ log.info("planColdStartSort planId:{} accounts:{} allocated:{} skipped:{} totalColdContents:{} cost:{}ms",
|
|
|
+ request.getPlanId(), accounts.size(), allocatedCount, skippedCount, globalColdList.size(), totalCost);
|
|
|
+
|
|
|
+ return PlanColdStartSortResponse.success(data, allocatedSourceIds);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 对单个账号做冷启内容 recall + rank,返回已打分排序的冷启内容列表
|
|
|
+ */
|
|
|
+ private AccountColdRankResult rankAccountColdContent(String planId, PlanColdStartSortRequest.AccountInfo account, String coldPool) {
|
|
|
+ RecommendParam param = new RecommendParam();
|
|
|
+ param.setAccountId(account.getAccountId());
|
|
|
+ param.setAccountName(account.getAccountName());
|
|
|
+ param.setGhId(account.getGhId());
|
|
|
+ param.setPlanId(planId);
|
|
|
+ param.setScene(DEMAND_POOL);
|
|
|
+ param.setStrategy(RankStrategyEnum.DEMAND_POOL_STRATEGY.getStrategy());
|
|
|
+ // 足够大的 publishNum 以确保所有冷启内容都被 rank
|
|
|
+ param.setPublishNum(6);
|
|
|
+
|
|
|
+ // recall
|
|
|
+ RecallResult recallResult = recallService.recall(convertToRecallParam(param));
|
|
|
+
|
|
|
+ // 只取冷启池内容
|
|
|
+ List<Content> coldContents = new ArrayList<>();
|
|
|
+ if (recallResult != null && CollectionUtils.isNotEmpty(recallResult.getData())) {
|
|
|
+ for (RecallResult.RecallData rd : recallResult.getData()) {
|
|
|
+ if (CollectionUtils.isNotEmpty(rd.getContents())) {
|
|
|
+ for (Content c : rd.getContents()) {
|
|
|
+ if (coldPool.equals(c.getContentPoolType())) {
|
|
|
+ coldContents.add(c);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (coldContents.isEmpty()) {
|
|
|
+ return new AccountColdRankResult(account.getAccountId(), account.getAccountName(), account.getGhId(),
|
|
|
+ Collections.emptyList());
|
|
|
+ }
|
|
|
+
|
|
|
+ // rank 冷启内容
|
|
|
+ RankParam rankParam = new RankParam();
|
|
|
+ rankParam.setContents(coldContents);
|
|
|
+ rankParam.setBackup(Collections.emptyList());
|
|
|
+ rankParam.setStrategy(param.getStrategy());
|
|
|
+ rankParam.setGhId(param.getGhId());
|
|
|
+ rankParam.setAccountId(param.getAccountId());
|
|
|
+ rankParam.setAccountName(param.getAccountName());
|
|
|
+ rankParam.setSize(coldContents.size());
|
|
|
+ rankParam.setScene(param.getScene());
|
|
|
+ rankParam.setPlanId(param.getPlanId());
|
|
|
+
|
|
|
+ RankResult rankResult = rankService.rank(rankParam);
|
|
|
+
|
|
|
+ return new AccountColdRankResult(account.getAccountId(), account.getAccountName(), account.getGhId(),
|
|
|
+ rankResult != null ? rankResult.getContents() : Collections.emptyList());
|
|
|
+ }
|
|
|
+
|
|
|
+ private static class AccountColdRankResult {
|
|
|
+ final String accountId;
|
|
|
+ final String accountName;
|
|
|
+ final String ghId;
|
|
|
+ final List<Content> coldContents;
|
|
|
+
|
|
|
+ AccountColdRankResult(String accountId, String accountName, String ghId, List<Content> coldContents) {
|
|
|
+ this.accountId = accountId;
|
|
|
+ this.accountName = accountName;
|
|
|
+ this.ghId = ghId;
|
|
|
+ this.coldContents = coldContents;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private static class GlobalColdItem {
|
|
|
+ final Content content;
|
|
|
+ final String accountId;
|
|
|
+ final String accountName;
|
|
|
+ final String ghId;
|
|
|
+
|
|
|
+ GlobalColdItem(Content content, String accountId, String accountName, String ghId) {
|
|
|
+ this.content = content;
|
|
|
+ this.accountId = accountId;
|
|
|
+ this.accountName = accountName;
|
|
|
+ this.ghId = ghId;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 将跨账号冷启池分配结果写入 publish_sort_log,
|
|
|
+ * 确保后续数据统计晋级流程可以正确读取排序结果。
|
|
|
+ * 写入逻辑与 saveSortLog 保持一致:按 ghId + dateStr 去重 title,记录 position、score、strategy 等。
|
|
|
+ */
|
|
|
+ private void savePlanColdStartSortLog(
|
|
|
+ List<PlanColdStartSortRequest.AccountInfo> accounts,
|
|
|
+ Map<String, PlanColdStartSortResponse.AccountAllocation> allocationMap,
|
|
|
+ Map<String, Content> sourceIdToContent,
|
|
|
+ Map<String, List<AccountAvgInfo>> avgByGhId) {
|
|
|
+ if (!"prod".equals(env)) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ String dateStr = DateUtils.getCurrentDateStr("yyyyMMdd");
|
|
|
+ String strategy = RankStrategyEnum.DEMAND_POOL_STRATEGY.getStrategy();
|
|
|
+ List<PublishSortLog> publishSortLogSaveList = new ArrayList<>();
|
|
|
+
|
|
|
+ for (PlanColdStartSortRequest.AccountInfo account : accounts) {
|
|
|
+ PlanColdStartSortResponse.AccountAllocation alloc = allocationMap.get(account.getAccountId());
|
|
|
+ if (alloc == null || CollectionUtils.isEmpty(alloc.getPositions())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 获取该账号当天已有记录,按 title 去重
|
|
|
+ List<PublishSortLog> hisSortLog = publishSortLogRepository.findByDateStrAndGhId(dateStr, account.getGhId());
|
|
|
+ Set<String> hisSortTitles = hisSortLog != null
|
|
|
+ ? hisSortLog.stream().map(PublishSortLog::getTitle).collect(Collectors.toSet())
|
|
|
+ : Collections.emptySet();
|
|
|
+
|
|
|
+ List<AccountAvgInfo> avgInfoList = avgByGhId.getOrDefault(account.getGhId(), Collections.emptyList());
|
|
|
+
|
|
|
+ for (PlanColdStartSortResponse.PositionAllocation pa : alloc.getPositions()) {
|
|
|
+ if (pa.isSkipped() || !StringUtils.hasText(pa.getSourceId())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ // 同一天同一账号不重复记录相同 title
|
|
|
+ if (hisSortTitles.contains(pa.getTitle())) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ Content content = sourceIdToContent.get(pa.getSourceId());
|
|
|
+ if (content == null) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ PublishSortLog sortLog = new PublishSortLog();
|
|
|
+ sortLog.setDateStr(dateStr);
|
|
|
+ sortLog.setGhId(account.getGhId());
|
|
|
+ sortLog.setAccountName(account.getAccountName());
|
|
|
+ sortLog.setPublishContentId(content.getId());
|
|
|
+ sortLog.setCrawlerChannelContentId(content.getCrawlerChannelContentId());
|
|
|
+ sortLog.setSourceType(content.getSourceType());
|
|
|
+ sortLog.setSourceId(content.getSourceId());
|
|
|
+ sortLog.setTitle(content.getTitle());
|
|
|
+ sortLog.setIndex(pa.getPositionIndex());
|
|
|
+ sortLog.setIndexAvgCount(accountIndexAvgViewCountService.getAvgReadCountByDB(
|
|
|
+ avgInfoList, account.getGhId(), pa.getPositionIndex()));
|
|
|
+ if (CollectionUtils.isNotEmpty(content.getCategory())) {
|
|
|
+ sortLog.setCategory(JSONObject.toJSONString(content.getCategory()));
|
|
|
+ }
|
|
|
+ sortLog.setStrategy(strategy);
|
|
|
+ sortLog.setScore(String.valueOf(content.getScore()));
|
|
|
+ sortLog.setScoreMap(JSONObject.toJSONString(content.getScoreMap()));
|
|
|
+ sortLog.setCreateTimestamp(System.currentTimeMillis());
|
|
|
+ publishSortLogSaveList.add(sortLog);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (CollectionUtils.isNotEmpty(publishSortLogSaveList)) {
|
|
|
+ publishSortLogRepository.saveAll(publishSortLogSaveList);
|
|
|
+ log.info("savePlanColdStartSortLog saved {} records to publish_sort_log", publishSortLogSaveList.size());
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("savePlanColdStartSortLog error", e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* 删除指定计划与发布账号下的 publish_content_gzh_waiting 数据,
|
|
|
* 并清理对应的 ContentPreFilterJob redis 缓存。
|