需求内容池 获取账号内容对应过滤

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/mapper/longArticle/LongArticleBaseMapper.java

@@ -158,4 +158,9 @@ public interface LongArticleBaseMapper {
                                              @Param("status") Integer status,
                                              @Param("updateTimestamp") Long updateTimestamp);
 
+    /**
+     * 通过 experiment_id 查询 demand_search_queue 获取账号名称
+     */
+    List<DemandSearchArticleRelationDTO> getAccountByExperimentIds(@Param("list") List<String> experimentIds);
+
 }

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/repository/aigc/PublishAccountRepository.java

@@ -16,4 +16,6 @@ public interface PublishAccountRepository extends JpaRepository<PublishAccount,
     List<PublishAccount> getByIdIn(List<String> publishAccountIds);
 
     PublishAccount getById(String publishAccountId);
+
+    List<PublishAccount> findByNameIn(List<String> names);
 }

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/repository/longArticle/DemandPoolPromotionBindRepository.java

@@ -4,6 +4,8 @@ import com.tzld.longarticle.recommend.server.model.entity.longArticle.DemandPool
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.stereotype.Repository;
 
+import java.util.List;
+
 /**
  * @author dyp
  */
@@ -11,4 +13,6 @@ import org.springframework.stereotype.Repository;
 public interface DemandPoolPromotionBindRepository extends JpaRepository<DemandPoolPromotionBind, Long> {
 
     DemandPoolPromotionBind getByPromotionChannelContentIdAndAccountId(String promotionChannelContentId, String accountId);
+
+    List<DemandPoolPromotionBind> findByPromotionChannelContentIdInAndAccountIdIn(List<String> promotionChannelContentIds, List<String> accountIds);
 }

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/XxlJobService.java

@@ -40,6 +40,7 @@ import com.tzld.longarticle.recommend.server.remote.aigc.AIGCProduceContentAudit
 import com.tzld.longarticle.recommend.server.remote.aigc.AIGCPublishContentDiscardService;
 import com.tzld.longarticle.recommend.server.remote.aigc.AIGCWaitingPublishContentService;
 import com.tzld.longarticle.recommend.server.repository.aigc.ProducePlanRepository;
+import com.tzld.longarticle.recommend.server.repository.aigc.PublishAccountRepository;
 import com.tzld.longarticle.recommend.server.repository.aigc.PublishPlanRepository;
 import com.tzld.longarticle.recommend.server.repository.aigc.PublishPlanSettingRepository;
 import com.tzld.longarticle.recommend.server.repository.crawler.GetOffVideoCrawlerRepository;
@@ -153,6 +154,9 @@ public class XxlJobService {
     @Autowired
     private AIGCPublishContentDiscardService aigcPublishContentDiscardService;
 
+    @Autowired
+    private PublishAccountRepository publishAccountRepository;
+
     @Autowired
     private RedisUtil redisUtil;
 
@@ -1436,12 +1440,73 @@ public class XxlJobService {
 
     /**
      * 来源1：查询 demand_search_article_relation（status=2 拉取详情成功 + channel_content_id 非空）
-     * 直接返回 experiment_id + channel_content_id 对
+     * 返回 experiment_id + channel_content_id + account_id 三元组。
+     * <p>
+     * account_id 获取链路：
+     * experiment_id → demand_search_queue.account → publish_account.name → publish_account.id
      */
     private List<DemandSearchArticleRelationDTO> fetchSearchRelations() {
         List<DemandSearchArticleRelationDTO> searchRelations = longArticleBaseMapper.getDemandSearchArticleRelations();
         log.info("syncDemandProduceContentRelation source1(search) found {} relations",
                 searchRelations != null ? searchRelations.size() : 0);
+
+        if (CollectionUtils.isEmpty(searchRelations)) {
+            return searchRelations;
+        }
+
+        // 提取所有 experiment_id
+        List<String> experimentIds = searchRelations.stream()
+                .map(DemandSearchArticleRelationDTO::getExperimentId)
+                .filter(StringUtils::hasText)
+                .distinct()
+                .collect(Collectors.toList());
+
+        if (CollectionUtils.isEmpty(experimentIds)) {
+            return searchRelations;
+        }
+
+        // 1. experiment_id → account（账号名称），从 demand_search_queue 查询
+        Map<String, String> expIdToAccountName = new HashMap<>();
+        for (List<String> partition : Lists.partition(experimentIds, 500)) {
+            List<DemandSearchArticleRelationDTO> queueResults = longArticleBaseMapper.getAccountByExperimentIds(partition);
+            if (CollectionUtils.isNotEmpty(queueResults)) {
+                for (DemandSearchArticleRelationDTO r : queueResults) {
+                    if (StringUtils.hasText(r.getExperimentId()) && StringUtils.hasText(r.getAccountId())) {
+                        expIdToAccountName.put(r.getExperimentId(), r.getAccountId());
+                    }
+                }
+            }
+        }
+        log.info("syncDemandProduceContentRelation source1 found {} experiment_id -> account_name mappings",
+                expIdToAccountName.size());
+
+        // 2. account名称 → account_id，从 publish_account 查询
+        List<String> accountNames = new ArrayList<>(new HashSet<>(expIdToAccountName.values()));
+        Map<String, String> accountNameToId = new HashMap<>();
+        for (List<String> partition : Lists.partition(accountNames, 500)) {
+            List<PublishAccount> accounts = publishAccountRepository.findByNameIn(partition);
+            if (CollectionUtils.isNotEmpty(accounts)) {
+                for (PublishAccount account : accounts) {
+                    if (StringUtils.hasText(account.getName()) && StringUtils.hasText(account.getId())) {
+                        accountNameToId.put(account.getName(), account.getId());
+                    }
+                }
+            }
+        }
+        log.info("syncDemandProduceContentRelation source1 found {} account_name -> account_id mappings",
+                accountNameToId.size());
+
+        // 3. 回填 accountId
+        for (DemandSearchArticleRelationDTO relation : searchRelations) {
+            String accountName = expIdToAccountName.get(relation.getExperimentId());
+            if (StringUtils.hasText(accountName)) {
+                String accountId = accountNameToId.get(accountName);
+                if (StringUtils.hasText(accountId)) {
+                    relation.setAccountId(accountId);
+                }
+            }
+        }
+
         return searchRelations;
     }
 

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/recommend/FilterContentByAccountService.java

@@ -0,0 +1,142 @@
+package com.tzld.longarticle.recommend.server.service.recommend;
+
+import com.tzld.longarticle.recommend.server.mapper.longArticle.LongArticleBaseMapper;
+import com.tzld.longarticle.recommend.server.model.dto.DemandSearchArticleRelationDTO;
+import com.tzld.longarticle.recommend.server.model.entity.aigc.PublishAccount;
+import com.tzld.longarticle.recommend.server.model.entity.longArticle.DemandPoolPromotionBind;
+import com.tzld.longarticle.recommend.server.repository.aigc.PublishAccountRepository;
+import com.tzld.longarticle.recommend.server.repository.longArticle.DemandPoolPromotionBindRepository;
+import com.google.common.collect.Lists;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * 内容是否属于任一账号的过滤查询服务
+ *
+ * @author dyp
+ */
+@Service
+@Slf4j
+public class FilterContentByAccountService {
+
+    @Autowired
+    private LongArticleBaseMapper longArticleBaseMapper;
+
+    @Autowired
+    private DemandPoolPromotionBindRepository demandPoolPromotionBindRepository;
+
+    @Autowired
+    private PublishAccountRepository publishAccountRepository;
+
+    /**
+     * 根据 channelContentId 和账号ID查询内容是否属于任一账号。
+     * 判断标准：
+     * 1. 通过 channelContentId → experimentId → demand_search_queue.account → publish_account.id 判断所属账号
+     * 2. 通过 channelContentId 查询 demand_pool_promotion_bind，判断内容与所属账号
+     * 任一条件满足即视为通过。
+     *
+     * @param channelContentIds 内容渠道ID列表
+     * @param accountIds        账号ID列表
+     * @return 不在任一账号下的 channelContentId 列表（需要被过滤的内容）
+     */
+    public List<String> filterContentByAccount(List<String> channelContentIds, List<String> accountIds) {
+        if (CollectionUtils.isEmpty(channelContentIds) || CollectionUtils.isEmpty(accountIds)) {
+            return new ArrayList<>();
+        }
+
+        Set<String> accountIdSet = new HashSet<>(accountIds);
+        Set<String> passedChannelContentIds = new HashSet<>();
+
+        // 检查1：channelContentId → experimentId → demand_search_queue.account → publish_account.id
+        List<DemandSearchArticleRelationDTO> demandRelations =
+                longArticleBaseMapper.getDemandExperimentIdByChannelContentIds(channelContentIds);
+
+        if (!CollectionUtils.isEmpty(demandRelations)) {
+            // 构建 channelContentId → experimentId 映射
+            Map<String, String> ccIdToExpId = demandRelations.stream()
+                    .filter(d -> StringUtils.isNotBlank(d.getChannelContentId())
+                            && StringUtils.isNotBlank(d.getExperimentId()))
+                    .collect(Collectors.toMap(
+                            DemandSearchArticleRelationDTO::getChannelContentId,
+                            DemandSearchArticleRelationDTO::getExperimentId,
+                            (v1, v2) -> v1));
+
+            if (!ccIdToExpId.isEmpty()) {
+                // experiment_id → account（账号名称），从 demand_search_queue 查询
+                List<String> experimentIds = new ArrayList<>(new HashSet<>(ccIdToExpId.values()));
+                Map<String, String> expIdToAccountName = new HashMap<>();
+                for (List<String> partition : Lists.partition(experimentIds, 500)) {
+                    List<DemandSearchArticleRelationDTO> queueResults =
+                            longArticleBaseMapper.getAccountByExperimentIds(partition);
+                    if (!CollectionUtils.isEmpty(queueResults)) {
+                        for (DemandSearchArticleRelationDTO r : queueResults) {
+                            if (StringUtils.isNotBlank(r.getExperimentId()) && StringUtils.isNotBlank(r.getAccountId())) {
+                                expIdToAccountName.put(r.getExperimentId(), r.getAccountId());
+                            }
+                        }
+                    }
+                }
+
+                if (!expIdToAccountName.isEmpty()) {
+                    // account名称 → account_id，从 publish_account 查询
+                    List<String> accountNames = new ArrayList<>(new HashSet<>(expIdToAccountName.values()));
+                    Map<String, String> accountNameToId = new HashMap<>();
+                    for (List<String> partition : Lists.partition(accountNames, 500)) {
+                        List<PublishAccount> accounts = publishAccountRepository.findByNameIn(partition);
+                        if (!CollectionUtils.isEmpty(accounts)) {
+                            for (PublishAccount account : accounts) {
+                                if (StringUtils.isNotBlank(account.getName()) && StringUtils.isNotBlank(account.getId())) {
+                                    accountNameToId.put(account.getName(), account.getId());
+                                }
+                            }
+                        }
+                    }
+
+                    // 遍历 channelContentId → experimentId → accountName → accountId，判断是否属于目标账号
+                    for (Map.Entry<String, String> entry : ccIdToExpId.entrySet()) {
+                        String accountName = expIdToAccountName.get(entry.getValue());
+                        if (StringUtils.isNotBlank(accountName)) {
+                            String accountId = accountNameToId.get(accountName);
+                            if (StringUtils.isNotBlank(accountId) && accountIdSet.contains(accountId)) {
+                                passedChannelContentIds.add(entry.getKey());
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        // 检查2：通过 channelContentId 查询 demand_pool_promotion_bind，判断内容与所属账号
+        List<String> remainingIds = channelContentIds.stream()
+                .filter(id -> !passedChannelContentIds.contains(id))
+                .collect(Collectors.toList());
+
+        if (!CollectionUtils.isEmpty(remainingIds)) {
+            List<DemandPoolPromotionBind> binds = demandPoolPromotionBindRepository
+                    .findByPromotionChannelContentIdInAndAccountIdIn(remainingIds, accountIds);
+
+            if (!CollectionUtils.isEmpty(binds)) {
+                binds.stream()
+                        .map(DemandPoolPromotionBind::getPromotionChannelContentId)
+                        .filter(StringUtils::isNotBlank)
+                        .forEach(passedChannelContentIds::add);
+            }
+        }
+
+        // 返回未通过任一检查的 channelContentId（需要被过滤的内容）
+        List<String> filteredIds = channelContentIds.stream()
+                .filter(id -> !passedChannelContentIds.contains(id))
+                .collect(Collectors.toList());
+
+        log.info("filterContentByAccount, total:{}, passed:{}, filtered:{}",
+                channelContentIds.size(), passedChannelContentIds.size(), filteredIds.size());
+
+        return filteredIds;
+    }
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/web/recommend/FilterContentByAccountController.java

@@ -0,0 +1,37 @@
+package com.tzld.longarticle.recommend.server.web.recommend;
+
+import com.tzld.longarticle.recommend.server.common.response.CommonResponse;
+import com.tzld.longarticle.recommend.server.service.recommend.FilterContentByAccountService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/**
+ * 内容账号归属过滤接口
+ *
+ * @author dyp
+ */
+@RestController
+@Slf4j
+public class FilterContentByAccountController {
+
+    @Autowired
+    private FilterContentByAccountService filterContentByAccountService;
+
+    @PostMapping("/filterContentByAccount")
+    public CommonResponse<List<String>> filterContentByAccount(@RequestBody FilterContentByAccountRequest request) {
+        List<String> channelContentIds = request.getChannelContentIds();
+        List<String> accountIds = request.getAccountIds();
+
+        log.info("filterContentByAccount, channelContentIds size:{}, accountIds size:{}",
+                channelContentIds != null ? channelContentIds.size() : 0,
+                accountIds != null ? accountIds.size() : 0);
+
+        List<String> filteredIds = filterContentByAccountService.filterContentByAccount(channelContentIds, accountIds);
+        return CommonResponse.success(filteredIds);
+    }
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/web/recommend/FilterContentByAccountRequest.java

@@ -0,0 +1,13 @@
+package com.tzld.longarticle.recommend.server.web.recommend;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.List;
+
+@Getter
+@Setter
+public class FilterContentByAccountRequest {
+    private List<String> channelContentIds;
+    private List<String> accountIds;
+}

long-article-recommend-service/src/main/resources/mapper/longArticle/LongArticleBaseMapper.xml

@@ -586,13 +586,15 @@
 
     <select id="getDemandPoolPromotionByLevel"
             resultType="com.tzld.longarticle.recommend.server.model.dto.DemandSearchArticleRelationDTO">
-        select channel_content_id      as channelContentId,
-               root_produce_content_id as contentId,
-               account_id              as accountId
-        from demand_pool_promotion_source
-        where level = #{level}
-          and status = #{status}
-          and deleted = #{deleted}
+        select dps.channel_content_id      as channelContentId,
+               dps.root_produce_content_id as contentId,
+               coalesce(dpb.account_id, dps.account_id) as accountId
+        from demand_pool_promotion_source dps
+        left join demand_pool_promotion_bind dpb
+          on dps.channel_content_id = dpb.promotion_channel_content_id
+        where dps.level = #{level}
+          and dps.status = #{status}
+          and dps.deleted = #{deleted}
     </select>
 
     <select id="getDemandExperimentIdByContentIds"
@@ -609,6 +611,17 @@
         </if>
     </select>
 
+    <select id="getAccountByExperimentIds"
+            resultType="com.tzld.longarticle.recommend.server.model.dto.DemandSearchArticleRelationDTO">
+        select experiment_id as experimentId,
+               account       as accountId
+        from demand_search_queue
+        where experiment_id in
+        <foreach collection="list" item="item" open="(" close=")" separator=",">
+            #{item}
+        </foreach>
+    </select>
+
     <select id="getByPlanIdAndAccountIdsAndContentPoolType"
             resultType="com.tzld.longarticle.recommend.server.model.entity.longArticle.PublishContentGzhWaiting">
         select * from publish_content_gzh_waiting