根据rootSourceId查询合作方、公众号 写入ad-new redis

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/common/constants/RedisConstant.java

@@ -19,4 +19,6 @@ public interface RedisConstant {
     String EXIST_RESULT_KEY = "exist_result_key_%s";
 
     String SINGLE_VIDEO_LOCK_KEY = "single_video_lock_key_%s";
+
+    String ROOT_SOURCE_CHANNEL_ACCOUNT_KEY_PREFIX = "root-source:channel-account:";
 }

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/component/AdNewRedisClient.java

@@ -0,0 +1,87 @@
+package com.tzld.piaoquan.longarticle.component;
+
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import com.tzld.piaoquan.longarticle.model.dto.RootSourceChannelCache;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static com.tzld.piaoquan.longarticle.common.constants.RedisConstant.ROOT_SOURCE_CHANNEL_ACCOUNT_KEY_PREFIX;
+
+@Component
+@Slf4j
+public class AdNewRedisClient {
+
+    private static final long CACHE_TTL_SECONDS = TimeUnit.HOURS.toSeconds(36);
+
+    @Value("${spring.redis-ad-cache.hostName}")
+    private String hostName;
+
+    @Value("${spring.redis-ad-cache.port:6379}")
+    private int port;
+
+    @Value("${spring.redis-ad-cache.password:}")
+    private String password;
+
+    @Value("${spring.redis-ad-cache.database:0}")
+    private int database;
+
+    private LettuceConnectionFactory connectionFactory;
+
+    private StringRedisTemplate redisTemplate;
+
+    @PostConstruct
+    public void init() {
+        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration(hostName, port);
+        configuration.setDatabase(database);
+        if (StringUtils.isNotBlank(password)) {
+            configuration.setPassword(password);
+        }
+        connectionFactory = new LettuceConnectionFactory(configuration);
+        connectionFactory.afterPropertiesSet();
+        redisTemplate = new StringRedisTemplate(connectionFactory);
+        redisTemplate.afterPropertiesSet();
+    }
+
+    @PreDestroy
+    public void destroy() {
+        if (connectionFactory != null) {
+            connectionFactory.destroy();
+        }
+    }
+
+    public int writeChannels(List<RootSourceChannelCache> channels) {
+        Map<String, String> values = new LinkedHashMap<>();
+        for (RootSourceChannelCache channel : channels) {
+            if (channel != null && StringUtils.isNotBlank(channel.getRootSourceId())) {
+                values.put(ROOT_SOURCE_CHANNEL_ACCOUNT_KEY_PREFIX + channel.getRootSourceId(),
+                        JSONObject.toJSONString(channel, SerializerFeature.WriteMapNullValue));
+            }
+        }
+        if (!values.isEmpty()) {
+            redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
+                values.forEach((key, value) -> connection.setEx(
+                        redisTemplate.getStringSerializer().serialize(key),
+                        CACHE_TTL_SECONDS,
+                        redisTemplate.getStringSerializer().serialize(value)));
+                return null;
+            });
+            log.info("Root source channel cache written to redis, values={}",
+                    JSONObject.toJSONString(values, SerializerFeature.WriteMapNullValue));
+        }
+        return values.size();
+    }
+}

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/controller/JobController.java

@@ -1,6 +1,7 @@
 package com.tzld.piaoquan.longarticle.controller;
 
 import com.tzld.piaoquan.longarticle.job.NewMatchVideoJob;
+import com.tzld.piaoquan.longarticle.job.RootSourceChannelSyncJob;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -13,6 +14,8 @@ import org.springframework.web.bind.annotation.RestController;
 public class JobController {
     @Autowired
     NewMatchVideoJob newMatchVideoJob;
+    @Autowired
+    RootSourceChannelSyncJob rootSourceChannelSyncJob;
 
     @GetMapping("/matchCrawlerVideo")
     public void matchCrawlerVideo() {
@@ -28,6 +31,16 @@ public class JobController {
     public void vectorMatchVideoJob(String flowPoolLevel) {
         newMatchVideoJob.vectorMatchVideoJob(flowPoolLevel);
     }
+
+    @GetMapping("/syncRootSourceChannelToRedis")
+    public void syncRootSourceChannelToRedis(String flowPoolLevel) {
+        rootSourceChannelSyncJob.syncRootSourceChannelToRedis(flowPoolLevel);
+    }
+
+    @GetMapping("/syncTodayRootSourceChannelToRedis")
+    public void syncTodayRootSourceChannelToRedis(String flowPoolLevel) {
+        rootSourceChannelSyncJob.syncTodayRootSourceChannelToRedis(flowPoolLevel);
+    }
 }
 
 

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/dao/mapper/growth/GrowthRootSourceChannelMapper.java

@@ -0,0 +1,27 @@
+package com.tzld.piaoquan.longarticle.dao.mapper.growth;
+
+import com.tzld.piaoquan.longarticle.model.dto.RootSourceChannelCache;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.Date;
+import java.util.List;
+
+public interface GrowthRootSourceChannelMapper {
+
+    List<RootSourceChannelCache> selectCgiReplyChannels(@Param("lastId") long lastId,
+                                                         @Param("lastCreateTime") Date lastCreateTime,
+                                                         @Param("limit") int limit);
+
+    List<RootSourceChannelCache> selectQwPlanChannels(@Param("lastId") long lastId,
+                                                       @Param("lastCreateTimestamp") long lastCreateTimestamp,
+                                                       @Param("limit") int limit);
+
+    /**
+     * 查询公众号推送和服务号推送计划的视频，两个业务线共用一条分页查询。
+     */
+    List<RootSourceChannelCache> selectGzhPushChannels(@Param("lastId") long lastId,
+                                                        @Param("lastCreateTimestamp") long lastCreateTimestamp,
+                                                        @Param("limit") int limit);
+
+    List<RootSourceChannelCache> selectAccountsByGhIds(@Param("ghIds") List<String> ghIds);
+}

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/dao/mapper/longarticle/LongArticleRootSourceChannelMapper.java

@@ -0,0 +1,13 @@
+package com.tzld.piaoquan.longarticle.dao.mapper.longarticle;
+
+import com.tzld.piaoquan.longarticle.model.dto.RootSourceChannelCache;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+public interface LongArticleRootSourceChannelMapper {
+
+    List<RootSourceChannelCache> selectRootSourceChannels(@Param("lastRootSourceId") String lastRootSourceId,
+                                                           @Param("lastRequestTime") long lastRequestTime,
+                                                           @Param("limit") int limit);
+}

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/job/RootSourceChannelSyncJob.java

@@ -0,0 +1,48 @@
+package com.tzld.piaoquan.longarticle.job;
+
+import com.tzld.piaoquan.longarticle.service.local.RootSourceChannelSyncService;
+import com.tzld.piaoquan.longarticle.service.local.RootSourceChannelSyncService.SyncResult;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class RootSourceChannelSyncJob {
+
+    private final RootSourceChannelSyncService syncService;
+
+    public RootSourceChannelSyncJob(RootSourceChannelSyncService syncService) {
+        this.syncService = syncService;
+    }
+
+    @XxlJob("syncRootSourceChannelToRedisJob")
+    public ReturnT<String> syncRootSourceChannelToRedis(String param) {
+        try {
+            SyncResult result = syncService.sync();
+            return successResult("同步完成", result);
+        } catch (Exception e) {
+            log.error("Sync root source channels to redis failed, param={}", param, e);
+            return new ReturnT<>(ReturnT.FAIL_CODE, "同步失败：" + e.getMessage());
+        }
+    }
+
+    @XxlJob("syncTodayRootSourceChannelToRedisJob")
+    public ReturnT<String> syncTodayRootSourceChannelToRedis(String param) {
+        try {
+            return successResult("当日增量同步完成", syncService.syncToday());
+        } catch (Exception e) {
+            log.error("Sync today's root source channels to redis failed, param={}", param, e);
+            return new ReturnT<>(ReturnT.FAIL_CODE, "当日增量同步失败：" + e.getMessage());
+        }
+    }
+
+    private ReturnT<String> successResult(String prefix, SyncResult result) {
+        return new ReturnT<>(ReturnT.SUCCESS_CODE,
+                String.format("%s，长文%d条，即转%d条，企微%d条，公众号/服务号推送%d条，共%d条",
+                        prefix, result.getLongArticleCount(), result.getCgiReplyCount(), result.getQwPlanCount(),
+                        result.getGzhPushCount(),
+                        result.getTotalCount()));
+    }
+}

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/model/dto/RootSourceChannelCache.java

@@ -0,0 +1,28 @@
+package com.tzld.piaoquan.longarticle.model.dto;
+
+import com.alibaba.fastjson.annotation.JSONField;
+import lombok.Data;
+
+import java.util.Date;
+
+@Data
+public class RootSourceChannelCache {
+
+    @JSONField(serialize = false)
+    private Long sourceId;
+
+    @JSONField(serialize = false)
+    private Long sourceTimestamp;
+
+    @JSONField(serialize = false)
+    private Date sourceTime;
+
+    private String rootSourceId;
+
+    private String channel;
+
+    @JSONField(serialize = false)
+    private String accountId;
+
+    private String accountName;
+}

long-article-server/src/main/java/com/tzld/piaoquan/longarticle/service/local/RootSourceChannelSyncService.java

@@ -0,0 +1,196 @@
+package com.tzld.piaoquan.longarticle.service.local;
+
+import com.tzld.piaoquan.longarticle.component.AdNewRedisClient;
+import com.tzld.piaoquan.longarticle.dao.mapper.growth.GrowthRootSourceChannelMapper;
+import com.tzld.piaoquan.longarticle.dao.mapper.longarticle.LongArticleRootSourceChannelMapper;
+import com.tzld.piaoquan.longarticle.model.dto.RootSourceChannelCache;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.Calendar;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Service
+public class RootSourceChannelSyncService {
+
+    private static final int PAGE_SIZE = 500;
+
+    @Autowired
+    LongArticleRootSourceChannelMapper longArticleMapper;
+    @Autowired
+    GrowthRootSourceChannelMapper growthMapper;
+    @Autowired
+    AdNewRedisClient redisClient;
+
+    public SyncResult sync() {
+        long now = System.currentTimeMillis();
+        long longArticleStartTime = getCutoffTime(now, Calendar.MONTH, -6);
+        long cgiReplyStartTime = getCutoffTime(now, Calendar.YEAR, -1);
+        long gzhPushStartTime = getCutoffTime(now, Calendar.MONTH, -6);
+        log.info("Start syncing root source channels, longArticleStartTime={}, cgiReplyStartTime={}, gzhPushStartTime={}",
+                new Date(longArticleStartTime), new Date(cgiReplyStartTime), new Date(gzhPushStartTime));
+
+        int longArticleCount = syncLongArticles(longArticleStartTime);
+        int cgiReplyCount = syncCgiReplies(cgiReplyStartTime);
+        int qwPlanCount = syncQwPlans(0L);
+        int gzhPushCount = syncGzhPushChannels(gzhPushStartTime);
+
+        SyncResult result = new SyncResult(longArticleCount, cgiReplyCount, qwPlanCount, gzhPushCount);
+        log.info("Root source channel sync completed: {}", result);
+        return result;
+    }
+
+    /**
+     * 每小时增量同步，只扫描当天创建的数据。
+     */
+    public SyncResult syncToday() {
+        Calendar today = Calendar.getInstance();
+        today.set(Calendar.HOUR_OF_DAY, 0);
+        today.set(Calendar.MINUTE, 0);
+        today.set(Calendar.SECOND, 0);
+        today.set(Calendar.MILLISECOND, 0);
+        long startTime = today.getTimeInMillis();
+        log.info("Start syncing today's root source channels, startTime={}", new Date(startTime));
+
+        int longArticleCount = syncLongArticles(startTime);
+        int cgiReplyCount = syncCgiReplies(startTime);
+        int qwPlanCount = syncQwPlans(startTime);
+        int gzhPushCount = syncGzhPushChannels(startTime);
+        SyncResult result = new SyncResult(longArticleCount, cgiReplyCount, qwPlanCount, gzhPushCount);
+        log.info("Today's root source channel sync completed: {}", result);
+        return result;
+    }
+
+    private long getCutoffTime(long now, int calendarField, int amount) {
+        Calendar calendar = Calendar.getInstance();
+        calendar.setTimeInMillis(now);
+        calendar.add(calendarField, amount);
+        return calendar.getTimeInMillis();
+    }
+
+    private int syncLongArticles(long startTime) {
+        int count = 0;
+        String lastRootSourceId = "";
+        long lastRequestTime = Math.max(-1L, TimeUnit.MILLISECONDS.toSeconds(startTime) - 1L);
+        while (true) {
+            List<RootSourceChannelCache> page = longArticleMapper.selectRootSourceChannels(
+                    lastRootSourceId, lastRequestTime, PAGE_SIZE);
+            if (page.isEmpty()) {
+                return count;
+            }
+            enrichLongArticleChannels(page);
+            count += redisClient.writeChannels(page);
+            RootSourceChannelCache last = page.get(page.size() - 1);
+            lastRootSourceId = last.getRootSourceId();
+            lastRequestTime = last.getSourceTimestamp();
+            if (page.size() < PAGE_SIZE) {
+                return count;
+            }
+        }
+    }
+
+    private void enrichLongArticleChannels(List<RootSourceChannelCache> page) {
+        List<String> ghIds = page.stream()
+                .map(RootSourceChannelCache::getAccountId)
+                .filter(id -> id != null && !id.isEmpty())
+                .distinct()
+                .collect(Collectors.toList());
+        if (ghIds.isEmpty()) {
+            return;
+        }
+        Map<String, RootSourceChannelCache> accountMap = new HashMap<>();
+        for (RootSourceChannelCache account : growthMapper.selectAccountsByGhIds(ghIds)) {
+            accountMap.put(account.getAccountId(), account);
+        }
+        for (RootSourceChannelCache channel : page) {
+            RootSourceChannelCache account = accountMap.get(channel.getAccountId());
+            if (account != null) {
+                channel.setChannel(account.getChannel());
+                if (account.getAccountName() != null && !account.getAccountName().isEmpty()) {
+                    channel.setAccountName(account.getAccountName());
+                }
+            }
+        }
+    }
+
+    private int syncCgiReplies(long startTime) {
+        int count = 0;
+        long lastId = 0L;
+        Date lastCreateTime = new Date(Math.max(-1L, startTime - 1L));
+        while (true) {
+            List<RootSourceChannelCache> page = growthMapper.selectCgiReplyChannels(lastId, lastCreateTime, PAGE_SIZE);
+            if (page.isEmpty()) {
+                return count;
+            }
+            count += redisClient.writeChannels(page);
+            RootSourceChannelCache last = page.get(page.size() - 1);
+            lastId = last.getSourceId();
+            lastCreateTime = last.getSourceTime();
+            if (page.size() < PAGE_SIZE) {
+                return count;
+            }
+        }
+    }
+
+    private int syncQwPlans(long startTime) {
+        int count = 0;
+        long lastId = 0L;
+        long lastCreateTimestamp = Math.max(-1L, startTime - 1L);
+        while (true) {
+            List<RootSourceChannelCache> page = growthMapper.selectQwPlanChannels(
+                    lastId, lastCreateTimestamp, PAGE_SIZE);
+            if (page.isEmpty()) {
+                return count;
+            }
+            count += redisClient.writeChannels(page);
+            RootSourceChannelCache last = page.get(page.size() - 1);
+            lastId = last.getSourceId();
+            lastCreateTimestamp = last.getSourceTimestamp();
+            if (page.size() < PAGE_SIZE) {
+                return count;
+            }
+        }
+    }
+
+    private int syncGzhPushChannels(long startTime) {
+        int count = 0;
+        long lastId = 0L;
+        long lastCreateTimestamp = Math.max(-1L, startTime - 1L);
+        while (true) {
+            List<RootSourceChannelCache> page = growthMapper.selectGzhPushChannels(
+                    lastId, lastCreateTimestamp, PAGE_SIZE);
+            if (page.isEmpty()) {
+                return count;
+            }
+            count += redisClient.writeChannels(page);
+            RootSourceChannelCache last = page.get(page.size() - 1);
+            lastId = last.getSourceId();
+            lastCreateTimestamp = last.getSourceTimestamp();
+            if (page.size() < PAGE_SIZE) {
+                return count;
+            }
+        }
+    }
+
+    @Data
+    @AllArgsConstructor
+    public static class SyncResult {
+        private int longArticleCount;
+        private int cgiReplyCount;
+        private int qwPlanCount;
+        private int gzhPushCount;
+
+        public int getTotalCount() {
+            return longArticleCount + cgiReplyCount + qwPlanCount + gzhPushCount;
+        }
+    }
+}

long-article-server/src/main/resources/application-dev.properties

@@ -9,10 +9,15 @@ aigc.datasource.password=cyber#crawler_2023
 aigc.datasource.url=jdbc:mysql://rm-t4na9qj85v7790tf84o.mysql.singapore.rds.aliyuncs.com:3306/aigc-admin-prod?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&useSSL=false&allowMultiQueries=true
 
 spring.redis.database=2
-spring.redis.host=r-bp154bpw97gptefiqkpd.redis.rds.aliyuncs.com
+spring.redis.host=r-bp1ikkpsbco7c26v23pd.redis.rds.aliyuncs.com
 spring.redis.port=6379
 spring.redis.password=Qingqu2019
 
+spring.redis-ad-cache.hostName=r-bp1ikkpsbco7c26v23pd.redis.rds.aliyuncs.com
+spring.redis-ad-cache.port=6379
+spring.redis-ad-cache.password=Qingqu2019
+spring.redis-ad-cache.database=0
+
 apollo.meta: https://apolloconfig-internal.piaoquantv.com
 
 xxl.job.admin.addresses=http://xxl-job-internal.piaoquantv.com/xxl-job-admin

long-article-server/src/main/resources/application-prod.properties

@@ -17,10 +17,15 @@ growth.datasource.password=crawler123456@
 growth.datasource.url=jdbc:mysql://rm-bp17q95335a99272b.mysql.rds.aliyuncs.com:3306/growth?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&useSSL=false&allowMultiQueries=true
 
 spring.redis.database=2
-spring.redis.host=r-bp154bpw97gptefiqkpd.redis.rds.aliyuncs.com
+spring.redis.host=r-bp1ikkpsbco7c26v23pd.redis.rds.aliyuncs.com
 spring.redis.port=6379
 spring.redis.password=Qingqu2019
 
+spring.redis-ad-cache.hostName=r-bp1wxi22yugkp044xa.redis.rds.aliyuncs.com
+spring.redis-ad-cache.port=6379
+spring.redis-ad-cache.password=Wqsd@2019
+spring.redis-ad-cache.database=0
+
 apollo.meta: https://apolloconfig-internal.piaoquantv.com
 
 xxl.job.admin.addresses=http://xxl-job-internal.piaoquantv.com/xxl-job-admin

long-article-server/src/main/resources/application-test.properties

@@ -17,10 +17,15 @@ growth.datasource.password=crawler123456@
 growth.datasource.url=jdbc:mysql://rm-bp17q95335a99272b.mysql.rds.aliyuncs.com:3306/growth?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&useSSL=false&allowMultiQueries=true
 
 spring.redis.database=2
-spring.redis.host=r-bp154bpw97gptefiqkpd.redis.rds.aliyuncs.com
+spring.redis.host=r-bp1ikkpsbco7c26v23pd.redis.rds.aliyuncs.com
 spring.redis.port=6379
 spring.redis.password=Qingqu2019
 
+spring.redis-ad-cache.hostName=r-bp1ikkpsbco7c26v23pd.redis.rds.aliyuncs.com
+spring.redis-ad-cache.port=6379
+spring.redis-ad-cache.password=Qingqu2019
+spring.redis-ad-cache.database=0
+
 apollo.meta: https://apolloconfig-internal.piaoquantv.com
 
 xxl.job.admin.addresses=http://test-xxl-job-internal.piaoquantv.com/xxl-job-admin

long-article-server/src/main/resources/mapper/growth/GrowthRootSourceChannelMapper.xml

@@ -0,0 +1,119 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.tzld.piaoquan.longarticle.dao.mapper.growth.GrowthRootSourceChannelMapper">
+
+    <resultMap id="RootSourceChannelMap" type="com.tzld.piaoquan.longarticle.model.dto.RootSourceChannelCache">
+        <result column="source_id" property="sourceId"/>
+        <result column="source_timestamp" property="sourceTimestamp"/>
+        <result column="source_time" property="sourceTime"/>
+        <result column="root_source_id" property="rootSourceId"/>
+        <result column="channel" property="channel"/>
+        <result column="account_id" property="accountId"/>
+        <result column="account_name" property="accountName"/>
+    </resultMap>
+
+    <select id="selectCgiReplyChannels" resultMap="RootSourceChannelMap">
+        select bucket.id as source_id,
+               bucket.create_time as source_time,
+               bucket.root_source_id,
+               partner.channel,
+               bucket.gh_id as account_id,
+               gh_detail.gh_name as account_name
+        from cgi_reply_bucket_data bucket
+        left join content_platform_gzh_account gzh_account
+               on gzh_account.id = (
+                   select max(latest_account.id)
+                   from content_platform_gzh_account latest_account
+                   where latest_account.gh_id = bucket.gh_id
+                     and latest_account.status = 1
+               )
+        left join content_platform_account partner
+               on partner.id = gzh_account.create_account_id
+              and partner.status = 1 and partner.type = 1
+        left join gh_detail
+               on gh_detail.id = (
+                   select max(latest_gh_detail.id)
+                   from gh_detail latest_gh_detail
+                   where latest_gh_detail.gh_id = bucket.gh_id
+                     and latest_gh_detail.is_delete = 0
+               )
+        where bucket.root_source_id is not null
+          and bucket.root_source_id != ''
+          and bucket.is_delete = 0
+          and bucket.create_time is not null
+          and (bucket.create_time > #{lastCreateTime}
+               or (bucket.create_time = #{lastCreateTime} and bucket.id > #{lastId}))
+        order by bucket.create_time, bucket.id
+        limit #{limit}
+    </select>
+
+    <select id="selectQwPlanChannels" resultMap="RootSourceChannelMap">
+        select plan.id as source_id,
+               plan.create_timestamp as source_timestamp,
+               plan.root_source_id,
+               partner.channel,
+               null as account_id,
+               null as account_name
+        from content_platform_qw_plan plan
+        left join content_platform_account partner
+               on partner.id = plan.create_account_id
+              and partner.status = 1 and partner.type = 1
+        where plan.root_source_id is not null
+          and plan.root_source_id != ''
+          and plan.status = 1
+          and plan.create_timestamp is not null
+          and (plan.create_timestamp > #{lastCreateTimestamp}
+               or (plan.create_timestamp = #{lastCreateTimestamp} and plan.id > #{lastId}))
+        order by plan.create_timestamp, plan.id
+        limit #{limit}
+    </select>
+
+    <!-- 公众号推送(type=2)和服务号推送(type=1)共用查询，按创建时间和视频ID稳定分页。 -->
+    <select id="selectGzhPushChannels" resultMap="RootSourceChannelMap">
+        select video.id as source_id,
+               video.create_timestamp as source_timestamp,
+               video.root_source_id,
+               partner.channel,
+               gzh_account.name as account_name
+        from content_platform_gzh_plan_video video
+        join content_platform_gzh_plan plan
+             on plan.id = video.plan_id
+        left join content_platform_gzh_account gzh_account
+               on gzh_account.id = plan.account_id
+              and gzh_account.status = 1
+        left join content_platform_account partner
+               on partner.id = video.create_account_id
+              and partner.status = 1 and partner.type = 1
+        where video.root_source_id is not null
+          and video.root_source_id != ''
+          and video.create_timestamp is not null
+          and plan.type in (1, 2)
+          and plan.status = 1
+          and video.status = 1
+          and (video.create_timestamp > #{lastCreateTimestamp}
+               or (video.create_timestamp = #{lastCreateTimestamp} and video.id > #{lastId}))
+        order by video.create_timestamp, video.id
+        limit #{limit}
+    </select>
+
+    <select id="selectAccountsByGhIds" resultMap="RootSourceChannelMap">
+        select gzh_account.gh_id as account_id,
+               gzh_account.name as account_name,
+               partner.channel
+        from content_platform_gzh_account gzh_account
+        join (
+            select gh_id, max(id) as latest_id
+            from content_platform_gzh_account
+            where status = 1
+              and gh_id in
+              <foreach collection="ghIds" item="ghId" open="(" separator="," close=")">
+                  #{ghId}
+              </foreach>
+            group by gh_id
+        ) latest_account on latest_account.latest_id = gzh_account.id
+        left join content_platform_account partner
+               on partner.id = gzh_account.create_account_id
+              and partner.status = 1 and partner.type = 1
+    </select>
+
+</mapper>

long-article-server/src/main/resources/mapper/longarticle/LongArticleRootSourceChannelMapper.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.tzld.piaoquan.longarticle.dao.mapper.longarticle.LongArticleRootSourceChannelMapper">
+
+    <resultMap id="RootSourceChannelMap" type="com.tzld.piaoquan.longarticle.model.dto.RootSourceChannelCache">
+        <result column="source_timestamp" property="sourceTimestamp"/>
+        <result column="root_source_id" property="rootSourceId"/>
+        <result column="channel" property="channel"/>
+        <result column="account_id" property="accountId"/>
+        <result column="account_name" property="accountName"/>
+    </resultMap>
+
+    <select id="selectRootSourceChannels" resultMap="RootSourceChannelMap">
+        select request_time as source_timestamp,
+               root_source_id,
+               null as channel,
+               gh_id as account_id,
+               account_name
+        from long_articles_root_source_id
+        where root_source_id is not null
+          and root_source_id != ''
+          and request_time is not null
+          and (request_time > #{lastRequestTime}
+               or (request_time = #{lastRequestTime} and root_source_id > #{lastRootSourceId}))
+        order by request_time, root_source_id
+        limit #{limit}
+    </select>
+
+</mapper>

long-article-server/src/test/java/com/tzld/piaoquan/longarticle/service/local/RootSourceChannelSyncServiceTest.java

@@ -0,0 +1,178 @@
+package com.tzld.piaoquan.longarticle.service.local;
+
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import com.tzld.piaoquan.longarticle.component.AdNewRedisClient;
+import com.tzld.piaoquan.longarticle.dao.mapper.growth.GrowthRootSourceChannelMapper;
+import com.tzld.piaoquan.longarticle.dao.mapper.longarticle.LongArticleRootSourceChannelMapper;
+import com.tzld.piaoquan.longarticle.model.dto.RootSourceChannelCache;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class RootSourceChannelSyncServiceTest {
+
+    @Test
+    void syncReadsAllFourBusinessLines() {
+        LongArticleRootSourceChannelMapper longArticleMapper = mock(LongArticleRootSourceChannelMapper.class);
+        GrowthRootSourceChannelMapper growthMapper = mock(GrowthRootSourceChannelMapper.class);
+        AdNewRedisClient redisClient = mock(AdNewRedisClient.class);
+        when(longArticleMapper.selectRootSourceChannels(anyString(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectCgiReplyChannels(anyLong(), any(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectQwPlanChannels(anyLong(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectGzhPushChannels(anyLong(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+
+        RootSourceChannelSyncService.SyncResult result =
+                newService(longArticleMapper, growthMapper, redisClient).sync();
+
+        assertEquals(0, result.getTotalCount());
+        verify(longArticleMapper).selectRootSourceChannels(anyString(), anyLong(), anyInt());
+        verify(growthMapper).selectCgiReplyChannels(anyLong(), any(), anyInt());
+        verify(growthMapper).selectQwPlanChannels(anyLong(), anyLong(), anyInt());
+        verify(growthMapper).selectGzhPushChannels(anyLong(), anyLong(), anyInt());
+    }
+
+    @Test
+    void sourceFailureFailsTheWholeJob() {
+        LongArticleRootSourceChannelMapper longArticleMapper = mock(LongArticleRootSourceChannelMapper.class);
+        GrowthRootSourceChannelMapper growthMapper = mock(GrowthRootSourceChannelMapper.class);
+        AdNewRedisClient redisClient = mock(AdNewRedisClient.class);
+        when(longArticleMapper.selectRootSourceChannels(anyString(), anyLong(), anyInt()))
+                .thenThrow(new IllegalStateException("database unavailable"));
+
+        RootSourceChannelSyncService service = newService(longArticleMapper, growthMapper, redisClient);
+
+        assertThrows(IllegalStateException.class, service::sync);
+    }
+
+    @Test
+    void todaySyncReadsAllFourBusinessLines() {
+        LongArticleRootSourceChannelMapper longArticleMapper = mock(LongArticleRootSourceChannelMapper.class);
+        GrowthRootSourceChannelMapper growthMapper = mock(GrowthRootSourceChannelMapper.class);
+        AdNewRedisClient redisClient = mock(AdNewRedisClient.class);
+        when(longArticleMapper.selectRootSourceChannels(anyString(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectCgiReplyChannels(anyLong(), any(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectQwPlanChannels(anyLong(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectGzhPushChannels(anyLong(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+
+        RootSourceChannelSyncService.SyncResult result =
+                newService(longArticleMapper, growthMapper, redisClient).syncToday();
+
+        assertEquals(0, result.getTotalCount());
+        verify(longArticleMapper).selectRootSourceChannels(anyString(), anyLong(), anyInt());
+        verify(growthMapper).selectCgiReplyChannels(anyLong(), any(), anyInt());
+        verify(growthMapper).selectQwPlanChannels(anyLong(), anyLong(), anyInt());
+        verify(growthMapper).selectGzhPushChannels(anyLong(), anyLong(), anyInt());
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    void longArticleIsEnrichedAndSerializedWithCommonFieldsOnly() {
+        LongArticleRootSourceChannelMapper longArticleMapper = mock(LongArticleRootSourceChannelMapper.class);
+        GrowthRootSourceChannelMapper growthMapper = mock(GrowthRootSourceChannelMapper.class);
+        AdNewRedisClient redisClient = mock(AdNewRedisClient.class);
+
+        RootSourceChannelCache channel = new RootSourceChannelCache();
+        channel.setRootSourceId("longArticles_test");
+        channel.setAccountId("gh_test");
+        channel.setAccountName("旧名称");
+        channel.setSourceTimestamp(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()));
+        RootSourceChannelCache account = new RootSourceChannelCache();
+        account.setAccountId("gh_test");
+        account.setAccountName("公众号名称");
+        account.setChannel("partner_channel");
+
+        when(longArticleMapper.selectRootSourceChannels(anyString(), anyLong(), anyInt()))
+                .thenReturn(Collections.singletonList(channel));
+        when(growthMapper.selectAccountsByGhIds(any())).thenReturn(Collections.singletonList(account));
+        when(growthMapper.selectCgiReplyChannels(anyLong(), any(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectQwPlanChannels(anyLong(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectGzhPushChannels(anyLong(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(redisClient.writeChannels(any())).thenReturn(1);
+
+        newService(longArticleMapper, growthMapper, redisClient).sync();
+
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(redisClient).writeChannels(captor.capture());
+        RootSourceChannelCache enriched = (RootSourceChannelCache) captor.getValue().get(0);
+        String json = JSONObject.toJSONString(enriched, SerializerFeature.WriteMapNullValue);
+        JSONObject value = JSONObject.parseObject(json);
+        assertEquals("partner_channel", value.getString("channel"));
+        assertEquals("公众号名称", value.getString("accountName"));
+        assertTrue(value.containsKey("rootSourceId"));
+        assertFalse(value.containsKey("accountId"));
+        assertFalse(value.containsKey("sourceTimestamp"));
+    }
+
+    @Test
+    void gzhPushChannelsUseCommonRedisFields() {
+        LongArticleRootSourceChannelMapper longArticleMapper = mock(LongArticleRootSourceChannelMapper.class);
+        GrowthRootSourceChannelMapper growthMapper = mock(GrowthRootSourceChannelMapper.class);
+        AdNewRedisClient redisClient = mock(AdNewRedisClient.class);
+        RootSourceChannelCache channel = new RootSourceChannelCache();
+        channel.setSourceId(101L);
+        channel.setSourceTimestamp(System.currentTimeMillis());
+        channel.setRootSourceId("gzh_push_test");
+        channel.setChannel("partner_channel");
+        channel.setAccountName("公众号名称");
+
+        when(longArticleMapper.selectRootSourceChannels(anyString(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectCgiReplyChannels(anyLong(), any(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectQwPlanChannels(anyLong(), anyLong(), anyInt()))
+                .thenReturn(Collections.emptyList());
+        when(growthMapper.selectGzhPushChannels(anyLong(), anyLong(), anyInt()))
+                .thenReturn(Collections.singletonList(channel), Collections.emptyList());
+        when(redisClient.writeChannels(any())).thenReturn(1);
+
+        RootSourceChannelSyncService.SyncResult result =
+                newService(longArticleMapper, growthMapper, redisClient).sync();
+
+        assertEquals(1, result.getGzhPushCount());
+        assertEquals(1, result.getTotalCount());
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(redisClient).writeChannels(captor.capture());
+        RootSourceChannelCache cached = (RootSourceChannelCache) captor.getValue().get(0);
+        JSONObject value = JSONObject.parseObject(JSONObject.toJSONString(cached, SerializerFeature.WriteMapNullValue));
+        assertEquals("partner_channel", value.getString("channel"));
+        assertEquals("公众号名称", value.getString("accountName"));
+        assertFalse(value.containsKey("sourceId"));
+    }
+
+    private RootSourceChannelSyncService newService(LongArticleRootSourceChannelMapper longArticleMapper,
+                                                     GrowthRootSourceChannelMapper growthMapper,
+                                                     AdNewRedisClient redisClient) {
+        RootSourceChannelSyncService service = new RootSourceChannelSyncService();
+        service.longArticleMapper = longArticleMapper;
+        service.growthMapper = growthMapper;
+        service.redisClient = redisClient;
+        return service;
+    }
+}