添加公司信息处理逻辑

2026-03-19 15:22:11 +08:00
parent 0fe7266de8
commit baa4b489d3
10 changed files with 498 additions and 64 deletions
@@ -1,8 +1,12 @@
 package org.jiayunet.mapper;

 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
 import org.jiayunet.pojo.po.AppJobData;

+import java.util.List;
+
 /**
 * 爬虫岗位原始数据Mapper
 *
@@ -10,4 +14,11 @@ import org.jiayunet.pojo.po.AppJobData;
 */
@Mapper
 public interface AppJobDataMapper extends CommonMapper<AppJobData> {
+
+    /**
+     * 查询待清洗数据并加行锁（SELECT ... FOR UPDATE）
+     * <p>必须在事务内调用，配合状态更新实现原子锁定</p>
+     */
+    @Select("SELECT * FROM app_job_data WHERE clean_status = 0 AND is_valid = 1 LIMIT #{limit} FOR UPDATE")
+    List<AppJobData> selectForUpdate(@Param("limit") int limit);
 }
@@ -1,8 +1,12 @@
 package org.jiayunet.mapper;

 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
 import org.jiayunet.pojo.po.Company;

+import java.util.List;
+
 /**
 * 公司Mapper
 *
@@ -10,4 +14,11 @@ import org.jiayunet.pojo.po.Company;
 */
@Mapper
 public interface CompanyMapper extends CommonMapper<Company> {
+
+    /**
+     * 查询待完善公司并加行锁（SELECT ... FOR UPDATE）
+     * <p>必须在事务内调用，配合状态更新实现原子锁定</p>
+     */
+    @Select("SELECT * FROM bg_company WHERE status = 0 LIMIT #{limit} FOR UPDATE")
+    List<Company> selectForUpdate(@Param("limit") int limit);
 }
@@ -67,7 +67,7 @@ public class Company {
    /** 新闻动态（JSON数组） */
    private String news;

-    /** 状态 0=待完善 1=已完善 2=禁用 3=补充中 */
+    /** 状态 0=待完善 1=已完善 2=禁用 3=补充中 4=补充失败 */
    private Integer status;

    /** 创建时间 */
@@ -0,0 +1,184 @@
+package org.jiayunet.service;
+
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.extern.slf4j.Slf4j;
+import org.jiayunet.ai.AiChatAbility;
+import org.jiayunet.mapper.CompanyMapper;
+import org.jiayunet.pojo.po.Company;
+import org.jiayunet.tool.HttpTool;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * 公司数据补充服务
+ * <p>定时从 bg_company 捞取待完善数据，调用 AI 补充公司信息</p>
+ * <p>依赖：AiChatAbility（AI调用）、DictCacheService（行业列表/地区匹配）、CompanyCleanTransactionService（事务操作）</p>
+ * <p>使用表：bg_company（读取/更新）</p>
+ *
+ * @author zk
+ */
+@Slf4j
+@Service
+public class CompanyCleanService {
+
+    @Autowired
+    private AiChatAbility aiChatAbility;
+
+    @Autowired
+    private DictCacheService dictCacheService;
+
+    @Autowired
+    private CompanyCleanTransactionService companyCleanTransactionService;
+
+    @Autowired
+    private CompanyMapper companyMapper;
+
+    @Value("${app.company-clean.batch-size:10}")
+    private int batchSize;
+
+    @Value("${app.company-clean.thread-pool-size:3}")
+    private int threadPoolSize;
+
+    private ExecutorService executorService;
+
+    /** 初始化线程池 */
+    @javax.annotation.PostConstruct
+    public void init() {
+        executorService = Executors.newFixedThreadPool(threadPoolSize);
+    }
+
+    /**
+     * 任务C：公司数据补充（每小时）
+     * <p>1. 批量锁定待完善公司 2. 多线程并发调用AI补充 3. 回填数据</p>
+     */
+    @Scheduled(cron = "0 */1 * * * ?")
+    public void cleanCompany() {
+        List<Company> companyList = companyCleanTransactionService.lockBatch(batchSize);
+        if (companyList.isEmpty()) {
+            return;
+        }
+        log.info("公司补充：锁定{}条数据", companyList.size());
+
+        // 多线程并发处理
+        for (Company company : companyList) {
+            executorService.submit(() -> {
+                try {
+                    cleanOne(company);
+                } catch (Exception e) {
+                    log.error("公司补充异常, id={}, shortName={}", company.getId(), company.getShortName(), e);
+                }
+            });
+        }
+    }
+
+    /**
+     * 任务D：公司僵尸恢复（每小时，与任务C错开30分钟）
+     * <p>将超时10分钟仍在补充中的数据重置为待完善</p>
+     */
+    @Scheduled(cron = "0 30 */1 * * ?")
+    public void recoverZombie() {
+        int recovered = companyMapper.update(null,
+                new LambdaUpdateWrapper<Company>()
+                        .set(Company::getStatus, 0)
+                        .eq(Company::getStatus, 3)
+                        .lt(Company::getUpdateTime, Instant.now().minusSeconds(600)));
+
+        if (recovered > 0) {
+            log.info("公司僵尸恢复：重置{}条数据", recovered);
+        }
+    }
+
+    /**
+     * 补充单条公司数据
+     * <p>1. 拼prompt调AI 2. 解析结果 3. 回填数据</p>
+     */
+    public void cleanOne(Company company) {
+        log.info("公司补充开始, id={}, shortName={}", company.getId(), company.getShortName());
+
+        String systemPrompt = buildSystemPrompt();
+        String userMessage = buildUserMessage(company.getShortName());
+
+        String aiResponse = aiChatAbility.chat(systemPrompt, userMessage);
+
+        try {
+            // 去掉可能的 markdown 代码块标记
+            String json = aiResponse.trim();
+            if (json.startsWith("```")) {
+                json = json.replaceAll("^```\\w*\\n?", "").replaceAll("\\n?```$", "").trim();
+            }
+            // 清除控制字符（Tab等），保留换行符
+            json = json.replaceAll("[\\x00-\\x08\\x0B\\x0C\\x0E-\\x1F]", "");
+
+            JsonNode root = HttpTool.objectMapper.readTree(json);
+
+            // valid 校验：AI不认识该公司
+            if (!root.path("valid").asBoolean(false)) {
+                log.info("公司补充：AI不认识该公司, id={}, shortName={}", company.getId(), company.getShortName());
+                companyCleanTransactionService.updateCompanyStatus(company.getId(), 4);
+                return;
+            }
+
+            // 回填数据
+            companyCleanTransactionService.saveCompanyData(root, company);
+            log.info("公司补充完成, id={}, shortName={}", company.getId(), company.getShortName());
+
+        } catch (Exception e) {
+            log.error("公司AI返回解析失败, id={}, shortName={}, response={}",
+                    company.getId(), company.getShortName(), aiResponse, e);
+            // 保持 status=3，由僵尸恢复任务重置
+        }
+    }
+
+    /** 构建系统提示词 */
+    private String buildSystemPrompt() {
+        return """
+                你是一个企业信息补充助手。根据提供的公司简称，补充该公司的详细信息。
+                
+                返回JSON格式要求：
+                {
+                  "valid": true/false,
+                  "name": "公司全称",
+                  "city": "总部所在城市，精确到市",
+                  "companyType": "企业类型",
+                  "industryId": 行业ID,
+                  "tags": ["公司标签，最多5个"],
+                  "summary": "一句话简介，100字以内",
+                  "description": "公司详细描述，500字以内",
+                  "foundedYear": "成立年份",
+                  "address": "总部/注册地址",
+                  "scale": "企业规模",
+                  "website": "官网地址",
+                  "financingStage": "融资状态",
+                  "latestValuation": "最新估值",
+                  "news": ["相关新闻，最多3条，每条50字以内"]
+                }
+                
+                规则：
+                1. 如果不认识该公司，返回 {"valid": false}
+                2. name 根据公司简称推断完整的企业注册名称
+                3. companyType 取值：上市企业、独角兽、国企、央企、民营企业、外资企业、合资企业、事业单位、其他
+                4. industryId 必须从给定行业列表中选择，不确定则null
+                5. scale 取值：少于50人、50-150人、150-500人、500-1000人、1000-5000人、5000-10000人、10000人以上
+                6. tags 体现公司核心业务特征，最多5个
+                7. news 基于你的知识提供该公司最新的3条相关新闻，每条50字以内
+                8. latestValuation 知道就提供，不知道则null
+                9. 不确定的字段返回null，不要编造
+                10. 字符串值中不允许出现Tab、换行等控制字符
+                11. 只返回JSON，不要其他内容
+                """;
+    }
+
+    /** 构建用户消息 */
+    private String buildUserMessage(String shortName) {
+        return "【公司简称】\n" + shortName +
+                "\n\n【行业列表】\n" + dictCacheService.getIndustryText();
+    }
+}
@@ -0,0 +1,160 @@
+package org.jiayunet.service;
+
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.extern.slf4j.Slf4j;
+import org.jiayunet.mapper.CompanyMapper;
+import org.jiayunet.pojo.po.Company;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.List;
+
+/**
+ * 公司数据补充事务服务
+ * <p>独立出来解决 @Transactional 同类自调用失效问题</p>
+ * <p>依赖：CompanyMapper、DictCacheService（地区匹配、行业校验）</p>
+ * <p>使用表：bg_company（更新）</p>
+ *
+ * @author zk
+ */
+@Slf4j
+@Service
+public class CompanyCleanTransactionService {
+
+    @Autowired
+    private CompanyMapper companyMapper;
+
+    @Autowired
+    private DictCacheService dictCacheService;
+
+    /**
+     * 回填公司数据（事务）
+     * <p>1. 解析AI返回的各字段 2. 地区匹配 3. 行业校验 4. 更新bg_company</p>
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void saveCompanyData(JsonNode root, Company company) {
+        // name：AI根据shortName推断的全称
+        String name = root.path("name").asText(null);
+        if (name != null && !name.isBlank()) {
+            company.setName(name);
+        }
+
+        // regionCode：AI返回城市名，Java侧匹配
+        String city = root.path("city").asText(null);
+        if (city != null && !city.isBlank()) {
+            String regionCode = dictCacheService.matchRegionCode(city);
+            company.setRegionCode(regionCode);
+        }
+
+        // companyType
+        String companyType = root.path("companyType").asText(null);
+        if (companyType != null && !"null".equals(companyType)) {
+            company.setCompanyType(companyType);
+        }
+
+        // industryId：校验是否存在于行业列表中
+        long industryId = root.path("industryId").asLong(0);
+        company.setIndustryId(industryId > 0 ? industryId : null);
+
+        // tags：JSON数组
+        JsonNode tagsNode = root.path("tags");
+        if (tagsNode.isArray() && !tagsNode.isEmpty()) {
+            company.setTags(tagsNode.toString());
+        }
+
+        // summary
+        String summary = root.path("summary").asText(null);
+        if (summary != null && !"null".equals(summary)) {
+            company.setSummary(summary);
+        }
+
+        // description
+        String description = root.path("description").asText(null);
+        if (description != null && !"null".equals(description)) {
+            company.setDescription(description);
+        }
+
+        // foundedYear
+        String foundedYear = root.path("foundedYear").asText(null);
+        if (foundedYear != null && !"null".equals(foundedYear)) {
+            company.setFoundedYear(foundedYear);
+        }
+
+        // address
+        String address = root.path("address").asText(null);
+        if (address != null && !"null".equals(address)) {
+            company.setAddress(address);
+        }
+
+        // scale
+        String scale = root.path("scale").asText(null);
+        if (scale != null && !"null".equals(scale)) {
+            company.setScale(scale);
+        }
+
+        // website
+        String website = root.path("website").asText(null);
+        if (website != null && !"null".equals(website)) {
+            company.setWebsite(website);
+        }
+
+        // financingStage
+        String financingStage = root.path("financingStage").asText(null);
+        if (financingStage != null && !"null".equals(financingStage)) {
+            company.setFinancingStage(financingStage);
+        }
+
+        // latestValuation
+        String latestValuation = root.path("latestValuation").asText(null);
+        if (latestValuation != null && !"null".equals(latestValuation)) {
+            company.setLatestValuation(latestValuation);
+        }
+
+        // news：JSON数组
+        JsonNode newsNode = root.path("news");
+        if (newsNode.isArray() && !newsNode.isEmpty()) {
+            company.setNews(newsNode.toString());
+        }
+
+        // 更新状态和时间
+        company.setStatus(1);
+        company.setUpdateTime(Instant.now());
+
+        companyMapper.updateById(company);
+    }
+
+    /** 更新公司状态 */
+    public void updateCompanyStatus(Long id, int status) {
+        companyMapper.update(null,
+                new LambdaUpdateWrapper<Company>()
+                        .set(Company::getStatus, status)
+                        .set(Company::getUpdateTime, Instant.now())
+                        .eq(Company::getId, id));
+    }
+
+    /**
+     * 原子锁定一批待完善公司（事务内 SELECT FOR UPDATE + UPDATE 状态）
+     * <p>行锁保证并发安全，其他线程会阻塞直到事务提交</p>
+     *
+     * @return 锁定成功的公司列表，可能为空
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public List<Company> lockBatch(int batchSize) {
+        List<Company> companyList = companyMapper.selectForUpdate(batchSize);
+        if (companyList.isEmpty()) {
+            return companyList;
+        }
+
+        List<Long> ids = companyList.stream().map(Company::getId).toList();
+        companyMapper.update(null,
+                new LambdaUpdateWrapper<Company>()
+                        .set(Company::getStatus, 3)
+                        .set(Company::getUpdateTime, Instant.now())
+                        .in(Company::getId, ids));
+
+        return companyList;
+    }
+}
@@ -69,25 +69,11 @@ public class JobCleanService {
     */
    @Scheduled(cron = "0 */5 * * * ?")
    public void cleanJob() {
-        // 批量锁定：原子操作，clean_status 0→1
-        int locked = appJobDataMapper.update(null,
-                new LambdaUpdateWrapper<AppJobData>()
-                        .set(AppJobData::getCleanStatus, 1)
-                        .eq(AppJobData::getCleanStatus, 0)
-                        .eq(AppJobData::getIsValid, 1)
-                        .last("LIMIT " + batchSize));
-
-        if (locked == 0) {
+        List<AppJobData> dataList = jobCleanTransactionService.lockBatch(batchSize);
+        if (dataList.isEmpty()) {
            return;
        }
-        log.info("岗位清洗：锁定{}条数据", locked);
-
-        // 查出刚锁定的数据
-        List<AppJobData> dataList = appJobDataMapper.selectList(
-                new LambdaQueryWrapper<AppJobData>()
-                        .eq(AppJobData::getCleanStatus, 1)
-                        .eq(AppJobData::getIsValid, 1)
-                        .last("LIMIT " + batchSize));
+        log.info("岗位清洗：锁定{}条数据", dataList.size());

        // 多线程并发处理
        for (AppJobData data : dataList) {
@@ -143,4 +143,26 @@ public class JobCleanTransactionService {
                        .set(AppJobData::getCleanStatus, status)
                        .eq(AppJobData::getId, id));
    }
+
+    /**
+     * 原子锁定一批待清洗数据（事务内 SELECT FOR UPDATE + UPDATE 状态）
+     * <p>行锁保证并发安全，其他线程会阻塞直到事务提交</p>
+     *
+     * @return 锁定成功的数据列表，可能为空
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public List<AppJobData> lockBatch(int batchSize) {
+        List<AppJobData> dataList = appJobDataMapper.selectForUpdate(batchSize);
+        if (dataList.isEmpty()) {
+            return dataList;
+        }
+
+        List<Long> ids = dataList.stream().map(AppJobData::getId).toList();
+        appJobDataMapper.update(null,
+                new LambdaUpdateWrapper<AppJobData>()
+                        .set(AppJobData::getCleanStatus, 1)
+                        .in(AppJobData::getId, ids));
+
+        return dataList;
+    }
 }