支持多供应商配置，不传 key 时使用第一个 provider
+ * + * @author zk + */ +@Slf4j +@Component +public class AiChatAbility { + + @Autowired + private AiChatConfig aiChatConfig; + + /** + * 使用默认 provider 发送对话 + *默认取 providers 配置中的第一个
+ */ + public String chat(String systemPrompt, String userMessage) { + String defaultKey = aiChatConfig.getProviders().keySet().stream() + .findFirst() + .orElseThrow(() -> new RuntimeException("未配置任何 AI provider")); + return chat(defaultKey, systemPrompt, userMessage); + } + + /** + * 使用指定 provider 发送对话 + * + * @param providerKey 供应商标识，对应 yml 中 providers 的 key + * @param systemPrompt 系统提示词 + * @param userMessage 用户消息 + * @return AI 返回的文本内容 + */ + public String chat(String providerKey, String systemPrompt, String userMessage) { + AiChatConfig.ProviderConfig config = aiChatConfig.getProviders().get(providerKey); + if (config == null) { + throw new RuntimeException("AI provider 不存在: " + providerKey); + } + + String url = config.getBaseUrl() + "/chat/completions"; + log.info("AI 请求 URL: {}, model: {}", url, config.getModel()); + + Map读取 app.ai.providers，第一个为默认 provider
+ * + * @author zk + */ +@Data +@Component +@ConfigurationProperties(prefix = "app.ai") +public class AiChatConfig { + + /** 供应商配置，key 为供应商标识，第一个为默认 */ + private Map存储爬虫抓取的原始岗位数据，供清洗服务读取并写入业务表
+ * + * @author zk + */ +@Data +@TableName(value = "app_job_data") +public class AppJobData { + + @TableId(type = IdType.AUTO) + private Long id; + + /** 关联爬取任务ID */ + private Long taskCrawlId; + + /** 职位名称 */ + private String jobTitle; + + /** 薪资 */ + private String salary; + + /** 工作地点 */ + private String location; + + /** 公司名称 */ + private String company; + + /** 经验要求 */ + private String experience; + + /** 学历要求 */ + private String education; + + /** 岗位详情(职责+要求+介绍) */ + private String description; + + /** 详情页URL */ + private String detailUrl; + + /** 内容哈希值，用于查重 */ + private String contentHash; + + /** 数据来源 0=官网 1=平台 */ + private Integer sources; + + /** 是否独立URL 0=页内展示 1=独立页面 */ + private Integer isIndependentUrl; + + /** 是否有效 0=无效 1=有效 */ + private Integer isValid; + + /** 有效期 */ + private Instant expireAt; + + /** 验证状态 pending=待验证 checking=验证中 checked=已验证 */ + private String checkStatus; + + /** 清洗状态 0=待清洗 1=清洗中 2=已入库 3=已丢弃 */ + private Integer cleanStatus; + + /** 上次验证时间 */ + private Instant lastCheckAt; + + /** 创建时间 */ + private Instant createdAt; + + /** 更新时间 */ + private Instant updatedAt; +} diff --git a/manager/src/main/java/org/jiayunet/pojo/po/Company.java b/manager/src/main/java/org/jiayunet/pojo/po/Company.java index 8e2b47a..976d8f4 100644 --- a/manager/src/main/java/org/jiayunet/pojo/po/Company.java +++ b/manager/src/main/java/org/jiayunet/pojo/po/Company.java @@ -67,7 +67,7 @@ public class Company { /** 新闻动态（JSON数组） */ private String news; - /** 状态 0=待完善 1=已完善 2=禁用 */ + /** 状态 0=待完善 1=已完善 2=禁用 3=补充中 */ private Integer status; /** 创建时间 */ diff --git a/manager/src/main/java/org/jiayunet/service/DictCacheService.java b/manager/src/main/java/org/jiayunet/service/DictCacheService.java new file mode 100644 index 0000000..e7350b2 --- /dev/null +++ b/manager/src/main/java/org/jiayunet/service/DictCacheService.java @@ -0,0 +1,124 @@ +package org.jiayunet.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import lombok.extern.slf4j.Slf4j; +import org.jiayunet.mapper.ChinaRegionsCodeMapper; +import org.jiayunet.mapper.IndustryMapper; +import org.jiayunet.mapper.JobCategoryMapper; +import org.jiayunet.pojo.po.ChinaRegionsCode; +import org.jiayunet.pojo.po.Industry; +import org.jiayunet.pojo.po.JobCategory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import javax.annotation.PostConstruct; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** + * 字典数据缓存服务 + *启动时加载岗位分类、行业、地区数据到内存，供清洗/推荐等业务使用
+ *依赖：JobCategoryMapper、IndustryMapper、ChinaRegionsCodeMapper
+ *使用表：bg_job_category（全量缓存）、bg_industry（全量缓存）、bg_china_regions_code（市级缓存）
+ * + * @author zk + */ +@Slf4j +@Service +public class DictCacheService { + + @Autowired + private JobCategoryMapper jobCategoryMapper; + + @Autowired + private IndustryMapper industryMapper; + + @Autowired + private ChinaRegionsCodeMapper chinaRegionsCodeMapper; + + private List分类/行业全量加载用于构建父级路径，文本只取叶子节点
+ */ + @PostConstruct + public void refresh() { + log.info("开始加载字典缓存..."); + + jobCategoryList = jobCategoryMapper.selectList(null); + industryList = industryMapper.selectList(null); + + // 只缓存省级+市级地区（provinceCode 为 null 是省，provinceCode 不为 null 且 cityCode 为 null 是市） + regionList = chinaRegionsCodeMapper.selectList( + new LambdaQueryWrapper模糊匹配，如"北京"匹配"北京市"
+ * + * @param cityName 城市名 + * @return region_code，匹配不上返回 null + */ + public String matchRegionCode(String cityName) { + if (cityName == null || cityName.isBlank()) { + return null; + } + String name = cityName.replace("市", "").replace("省", "").trim(); + return regionList.stream() + .filter(r -> r.getName().contains(name) || name.contains(r.getName().replace("市", "").replace("省", ""))) + .map(ChinaRegionsCode::getCode) + .findFirst() + .orElse(null); + } +} diff --git a/manager/src/main/java/org/jiayunet/service/JobCleanService.java b/manager/src/main/java/org/jiayunet/service/JobCleanService.java new file mode 100644 index 0000000..17a6526 --- /dev/null +++ b/manager/src/main/java/org/jiayunet/service/JobCleanService.java @@ -0,0 +1,248 @@ +package org.jiayunet.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +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.AppJobDataMapper; +import org.jiayunet.mapper.JobMapper; +import org.jiayunet.pojo.po.AppJobData; +import org.jiayunet.pojo.po.Job; +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.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * 岗位清洗服务 + *定时从 app_job_data 捞取待清洗数据，调用 AI 清洗后写入业务表
+ *依赖：AiChatAbility（AI调用）、DictCacheService（字典缓存）、JobCleanTransactionService（事务操作）
+ *使用表：app_job_data（读取/更新状态）、bg_job（去重查询）
+ * + * @author zk + */ +@Slf4j +@Service +public class JobCleanService { + + @Autowired + private AiChatAbility aiChatAbility; + + @Autowired + private DictCacheService dictCacheService; + + @Autowired + private JobCleanTransactionService jobCleanTransactionService; + + @Autowired + private AppJobDataMapper appJobDataMapper; + + @Autowired + private JobMapper jobMapper; + + @Value("${app.job-clean.batch-size:20}") + private int batchSize; + + @Value("${app.job-clean.thread-pool-size:5}") + private int threadPoolSize; + + private ExecutorService executorService; + + /** 初始化线程池 */ + @javax.annotation.PostConstruct + public void init() { + executorService = Executors.newFixedThreadPool(threadPoolSize); + } + + /** + * 定时任务A：岗位清洗（每5分钟） + *1. 批量锁定待清洗数据 2. 多线程并发调用AI清洗 3. 写入业务表
+ */ + @Scheduled(cron = "0 */1 * * * ?") + public void cleanJob() { + // 批量锁定：原子操作，clean_status 0→1 + int locked = appJobDataMapper.update(null, + new LambdaUpdateWrapper将超时10分钟仍在清洗中的数据重置为待清洗
+ */ + @Scheduled(cron = "0 */30 * * * ?") + public void recoverZombie() { + int recovered = appJobDataMapper.update(null, + new LambdaUpdateWrapper1. 前置校验 2. 拼prompt调AI 3. 解析结果 4. 写入业务表
+ */ + public void cleanOne(AppJobData data) { + // 1. 前置校验 + if (data.getDescription() == null || data.getDescription().length() < 20) { + jobCleanTransactionService.updateCleanStatus(data.getId(), 3); + return; + } + + // 2. 拼 prompt + String systemPrompt = buildSystemPrompt(); + String userMessage = buildUserMessage(data); + + // 3. 调用 AI + String aiResponse = aiChatAbility.chat(systemPrompt, userMessage); + + // 4. 解析 JSON + try { + // 去掉可能的 markdown 代码块标记 + String json = aiResponse.trim(); + if (json.startsWith("```")) { + json = json.replaceAll("^```\\w*\\n?", "").replaceAll("\\n?```$", "").trim(); + } + + JsonNode root = HttpTool.objectMapper.readTree(json); + + // valid 校验 + if (!root.path("valid").asBoolean(false)) { + jobCleanTransactionService.updateCleanStatus(data.getId(), 3); + return; + } + + // 5. 去重检查 + String sourceId = String.valueOf(data.getId()); + Long existJob = jobMapper.selectCount( + new LambdaQueryWrapper独立出来解决 @Transactional 同类自调用失效问题
+ *依赖：JobMapper、CompanyMapper、JobRegionRelationMapper、AppJobDataMapper
+ *使用表：bg_job（写入）、bg_company（查询/创建）、bg_job_region_relation（写入）、app_job_data（更新状态）
+ * + * @author zk + */ +@Slf4j +@Service +public class JobCleanTransactionService { + + @Autowired + private JobMapper jobMapper; + + @Autowired + private CompanyMapper companyMapper; + + @Autowired + private JobRegionRelationMapper jobRegionRelationMapper; + + @Autowired + private AppJobDataMapper appJobDataMapper; + + /** + * 写入 bg_job + bg_job_region_relation + 更新 clean_status（短事务） + */ + @Transactional(rollbackFor = Exception.class) + public void saveJobData(JsonNode root, AppJobData data, Long companyId, String sourceId, List按 short_name 查询，不存在则创建一条待完善记录
+ */ + public synchronized Long findOrCreateCompany(String shortName) { + Company company = companyMapper.selectOne( + new LambdaQueryWrapper