--- a/api/admin-worker/src/main/java/cn/jojo/cc/worker/task/EnglishBatchPublishTask.java
+++ b/api/admin-worker/src/main/java/cn/jojo/cc/worker/task/EnglishBatchPublishTask.java
@@ -109,10 +109,14 @@
/**
* 失败条目逐条落日志:没有进度接口之后,日志是定位"哪条为什么没发出去"的唯一出口。
*/
private void logFailures(String entity, BatchPublishResultBo batch) {
+ if (batch.getPendingCacheCount() > 0) {
+ log.info("[englishBatchPublish] {} PROD已提交,缓存失效待outbox恢复确认: {} 条",
+ entity, batch.getPendingCacheCount());
+ }
if (batch.getFailures() == null || batch.getFailures().isEmpty()) {
return;
}
for (EnglishTaskFailureBo failure : batch.getFailures()) {
if (failure == null) {
@@ -144,33 +148,36 @@
String raw = StringUtils.trimToNull(jobParam);
if (raw == null) {
return new Param(true, true, 0);
}
if (StringUtils.isNumeric(raw)) {
- return new Param(true, true, Integer.parseInt(raw));
+ return new Param(true, true, positiveLimit(raw));
}
String type = TYPE_ALL;
int limit = 0;
- for (String segment : raw.split(",")) {
+ boolean parsedAny = false;
+ for (String segment : raw.split(",", -1)) {
String pair = StringUtils.trimToEmpty(segment);
if (pair.isEmpty()) {
continue;
}
+ parsedAny = true;
String[] kv = pair.split("=", 2);
String key = StringUtils.trimToEmpty(kv[0]);
String value = kv.length > 1 ? StringUtils.trimToEmpty(kv[1]) : "";
if ("limit".equals(key)) {
if (!StringUtils.isNumeric(value)) {
throw new IllegalArgumentException("limit 必须是正整数:" + pair);
}
- limit = Integer.parseInt(value);
+ limit = positiveLimit(value);
} else if ("type".equals(key)) {
type = value;
} else {
throw new IllegalArgumentException("无法识别的参数:" + pair);
}
}
+ if (!parsedAny) { throw new IllegalArgumentException("参数不包含有效键值"); }
if (TYPE_WORDS.equals(type)) {
return new Param(true, false, limit);
}
if (TYPE_PHRASES.equals(type)) {
return new Param(false, true, limit);
@@ -178,7 +185,14 @@
if (!TYPE_ALL.equals(type)) {
throw new IllegalArgumentException("type 只支持 words / phrases / all:" + type);
}
return new Param(true, true, limit);
}
+ private static int positiveLimit(String value) {
+ if (!value.matches("[1-9][0-9]*")) {
+ throw new IllegalArgumentException("limit 必须是严格正整数");
+ }
+ return Integer.parseInt(value);
+ }
+
}
}
--- /dev/null
+++ b/api/admin-worker/src/main/java/cn/jojo/cc/worker/task/EnglishDictOutboxJobHandler.java
@@ -0,0 +1,21 @@
+package cn.jojo.cc.worker.task;
+
+import cn.jojo.cc.biz.service.english.dict.EnglishDictOutboxBizService;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Component;
+
+/** 部署后必须在 XXL-Job 配置 englishDictOutboxReplay,每分钟执行。 */
+@Component
+public class EnglishDictOutboxJobHandler {
+ @Resource
+ private EnglishDictOutboxBizService service;
+
+ @XxlJob("englishDictOutboxReplay")
+ public ReturnT replay(String ignored) {
+ int failed = service.replay(100);
+ return new ReturnT<>(failed == 0 ? 200 : ReturnT.FAIL_CODE,
+ "本轮投递失败=" + failed + ";待办按 retry_after 自动恢复,broker 确认不等于消费者已失效");
+ }
+}
--- /dev/null
+++ b/api/admin-worker/src/main/java/cn/jojo/cc/worker/task/EnglishSegmentOutputIdentityRecoveryJobHandler.java
@@ -0,0 +1,37 @@
+package cn.jojo.cc.worker.task;
+
+import cn.jojo.cc.biz.service.english.segment.EnglishSegmentOutputIdentityRecoveryBizService;
+import cn.jojo.cc.common.enums.RedisKeyEnum;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import jakarta.annotation.Resource;
+import org.redisson.api.RedissonClient;
+import org.springframework.stereotype.Component;
+import java.util.concurrent.TimeUnit;
+
+/** 必须持续调度的来源恢复作业,双库每轮各50题,真实输出重生成会写OSS,部署后人工启用。 */
+@Component
+public class EnglishSegmentOutputIdentityRecoveryJobHandler {
+ @Resource private cn.jojo.cc.biz.common.config.EnglishDictConfig englishDictConfig;
+ @Resource private EnglishSegmentOutputIdentityRecoveryBizService service;
+ @Resource private RedissonClient redissonClient;
+
+ @XxlJob("englishSegmentOutputIdentityRecover")
+ public ReturnT recover(String ignored) {
+ if (!englishDictConfig.isSegmentEnabled() || !englishDictConfig.isSegmentExecutionEnabled()) {
+ return new ReturnT<>("英文分词已暂停,输出恢复未读游标/取锁/生成/上传");
+ }
+ var lock = redissonClient.getLock(RedisKeyEnum.ENGLISH_SEGMENT_REFRESH_LOCK.buildKey());
+ boolean acquired = false;
+ try {
+ acquired = lock.tryLock(0L, TimeUnit.MILLISECONDS);
+ if (!acquired) { return new ReturnT<>("分词刷新/来源恢复正在执行"); }
+ return new ReturnT<>(service.runOnce(BaseScheduleJobHandler.SYSTEM));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return new ReturnT<>(ReturnT.FAIL_CODE, "interrupted");
+ } finally {
+ if (acquired && lock.isHeldByCurrentThread()) { lock.unlock(); }
+ }
+ }
+}
--- /dev/null
+++ b/api/admin-worker/src/test/java/cn/jojo/cc/worker/task/EnglishSegmentOutputIdentityRecoveryJobHandlerTest.java
@@ -0,0 +1,24 @@
+package cn.jojo.cc.worker.task;
+
+import cn.jojo.cc.biz.common.config.EnglishDictConfig;
+import cn.jojo.cc.biz.service.english.segment.EnglishSegmentOutputIdentityRecoveryBizService;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.redisson.api.RedissonClient;
+import static org.mockito.Mockito.*;
+
+class EnglishSegmentOutputIdentityRecoveryJobHandlerTest {
+ @Test void pausedJobDoesNotAcquireLockOrRegenerateOutputs() {
+ var job = new EnglishSegmentOutputIdentityRecoveryJobHandler();
+ var config = mock(EnglishDictConfig.class);
+ var redis = mock(RedissonClient.class);
+ var service = mock(EnglishSegmentOutputIdentityRecoveryBizService.class);
+ when(config.isSegmentEnabled()).thenReturn(true);
+ when(config.isSegmentExecutionEnabled()).thenReturn(false);
+ ReflectionTestUtils.setField(job, "englishDictConfig", config);
+ ReflectionTestUtils.setField(job, "redissonClient", redis);
+ ReflectionTestUtils.setField(job, "service", service);
+ job.recover(null);
+ verifyNoInteractions(redis, service);
+ }
+}
--- a/api/api-server/src/main/java/cn/jojo/cc/server/service/impl/QuestionServiceImpl.java
+++ b/api/api-server/src/main/java/cn/jojo/cc/server/service/impl/QuestionServiceImpl.java
@@ -1258,17 +1258,18 @@
.stream()
.filter(e -> Objects.nonNull(e) && StringUtils.isNotBlank(e.getQuestionKey()) && StringUtils.isBlank(e.getGameSkinKey()))
.map(StandardQuestionUniqueKeyReq::getQuestionKey)
.collect(Collectors.toSet());
if (CollectionUtils.isNotEmpty(genKeys)) {
- List genStandQuestionList = listGenStandQuestion(genKeys);
+ Map selectedOutputs = new HashMap<>();
+ List genStandQuestionList = listGenStandQuestion(genKeys, selectedOutputs);
if (CollectionUtils.isNotEmpty(genStandQuestionList)) {
buildGenStandQuestionScore(genStandQuestionList, req);
//查询组合题的小题
buildSmallQuestions(genStandQuestionList, req);
//装配英文划词:小题已在上面递归里走过同一个入口,父题与小题的 key 不重叠,不会重复查库
- fillSegments(genStandQuestionList);
+ fillSegments(genStandQuestionList, selectedOutputs);
resultList.addAll(genStandQuestionList);
}
}
return DefaultRpcResult.successWithData(resultList);
}
@@ -1706,21 +1707,22 @@
* 为一批通用标准题装配英文划词序列。
*
* 划词是衍生数据:读库、反序列化、词典解析中的任何异常都不得影响题目本体下发,
* 因此在本层兜底;无分词行或解析失败的题保持 {@code segments=null}。
*/
- private void fillSegments(List questions) {
+ private void fillSegments(List questions,
+ Map selectedOutputs) {
if (CollectionUtils.isEmpty(questions)) {
return;
}
try {
Set keys = questions.stream()
.filter(e -> Objects.nonNull(e) && StringUtils.isNotBlank(e.getQuestionKey()))
.map(StandardQuestionDto::getQuestionKey)
.collect(Collectors.toSet());
Map segmentsMap =
- englishSegmentDeliveryBizService.resolveByQuestionKeys(keys);
+ englishSegmentDeliveryBizService.resolveByQuestionKeys(keys, selectedOutputs);
if (MapUtils.isEmpty(segmentsMap)) {
return;
}
questions.stream()
.filter(e -> Objects.nonNull(e) && StringUtils.isNotBlank(e.getQuestionKey()))
@@ -1735,10 +1737,15 @@
*
* @param keys
* @return
*/
private List listGenStandQuestion(Set keys) {
+ return listGenStandQuestion(keys, new HashMap<>());
+ }
+
+ private List listGenStandQuestion(Set keys,
+ Map selectedOutputs) {
if (CollectionUtils.isEmpty(keys)) {
return Collections.emptyList();
}
boolean redisEnabled = ApiServerConfig.INSTANCE.isRedisEnabled();
//查询题信息
@@ -1835,10 +1842,14 @@
StandardQuestionInfoDto questionInfo = new StandardQuestionInfoDto();
questionData.setQuestionInfo(questionInfo);
questionInfo.setQuestionKey(questionBaseInfo.getQuestionKey());
questionInfo.setQuestionType(questionBaseInfo.getQuestionType());
questionInfo.setData(UriToolsUtil.getPath(contentConfigBo.getOutDataJson()));
+ selectedOutputs.put(questionKey, cn.jojo.cc.biz.service.english.segment.SelectedSegmentOutput.capture(
+ cn.jojo.cc.core.multidatasource.MultiDataSourceContextHolder.isJYDatasource()
+ ? DataBaseTypeEnum.TEACHING : DataBaseTypeEnum.PROD,
+ configId, contentConfigBo, questionInfo.getData()));
questionInfo.setQuestionPublic(
UriToolsUtil.getPath(Optional.ofNullable(questionPublic.get(questionBaseInfo.getQuestionId()))
.map(FormAndDataCacheBo::getOutDataJson).orElse(StringUtils.EMPTY))
);
resultList.add(dto);
--- a/api/api-server/src/test/java/cn/jojo/cc/server/service/impl/QuestionServiceImplSegmentsTest.java
+++ b/api/api-server/src/test/java/cn/jojo/cc/server/service/impl/QuestionServiceImplSegmentsTest.java
@@ -49,25 +49,25 @@
dto.setQuestionKey(key);
return dto;
}
private void fill(List questions) {
- ReflectionTestUtils.invokeMethod(questionService, "fillSegments", questions);
+ ReflectionTestUtils.invokeMethod(questionService, "fillSegments", questions, Collections.emptyMap());
}
@Test
@SuppressWarnings("unchecked")
void keysArePassedThroughUnchanged() {
// 断言的是“原值传递”:装配链路里 cc_question_segment.question_key 与 DTO 的 key 同源,
// 任何补前缀/去前缀的“顺手处理”都会让它查空。
- when(englishSegmentDeliveryBizService.resolveByQuestionKeys(anySet())).thenReturn(Collections.emptyMap());
+ when(englishSegmentDeliveryBizService.resolveByQuestionKeys(anySet(), any())).thenReturn(Collections.emptyMap());
List questions = Arrays.asList(question("JY_q1"), question("q2"));
fill(questions);
ArgumentCaptor captor = ArgumentCaptor.forClass(Set.class);
- verify(englishSegmentDeliveryBizService).resolveByQuestionKeys(captor.capture());
+ verify(englishSegmentDeliveryBizService).resolveByQuestionKeys(captor.capture(), any());
assertEquals(2, captor.getValue().size());
// 原值传递:JY_ 前缀不得被“顺手”规整,否则映射到分词表就查空。
assertTrue(captor.getValue().contains("JY_q1"));
assertTrue(captor.getValue().contains("q2"));
}
@@ -79,11 +79,11 @@
StandardQuestionDto blankKey = question(" ");
EnglishSegmentDto segments = EnglishSegmentDto.builder()
.matches(List.of(EnglishSegmentDto.SegmentMatch.builder()
.text("good").type(EnglishSegmentDto.TYPE_WORD).wordId(7L).build()))
.build();
- when(englishSegmentDeliveryBizService.resolveByQuestionKeys(anySet()))
+ when(englishSegmentDeliveryBizService.resolveByQuestionKeys(anySet(), any()))
.thenReturn(Map.of("q1", segments));
fill(Arrays.asList(hit, miss, blankKey));
assertNotNull(hit.getSegments());
@@ -95,11 +95,11 @@
@Test
void resolverFailureLeavesQuestionsDeliverableWithoutSegments() {
StandardQuestionDto question = question("q1");
question.setQuestionType(30002);
- when(englishSegmentDeliveryBizService.resolveByQuestionKeys(anySet()))
+ when(englishSegmentDeliveryBizService.resolveByQuestionKeys(anySet(), any()))
.thenThrow(new IllegalStateException("segment table down"));
fill(List.of(question));
// 划词是衍生数据:读库异常既不冒泡也不污染题目本体。
--- a/application/src/main/java/cn/jojo/cc/biz/cache/english/EnglishPhraseIndexCacheService.java
+++ b/application/src/main/java/cn/jojo/cc/biz/cache/english/EnglishPhraseIndexCacheService.java
@@ -184,10 +184,21 @@
* 一个动作(一轮分词 / 一次下发解析)只能用一张表,中途换表会产出"前半旧版、后半新版"的混合结果。
*
* @return 数据源不合法,或装载失败且处于退避窗口内时返回 {@code null};
* 调用方必须据此跳过本次动作,绝不能用空表去扫——那会把短语整块漏切成单词,并被剪枝永久固化。
*/
+ /** 写侧仅接收已与远端版本一致的句柄;版本读取或重装失败不落库。 */
+ public PhraseIndexSnapshot freshSnapshotForWrite(DataBaseTypeEnum source) {
+ if (!isSourceValid(source)) { return null; }
+ forceVersionCheck(source);
+ PhraseIndexSnapshot snapshot = readySnapshot(source);
+ Long remote = readVersionQuietly(source);
+ if (snapshot == null || remote == null || snapshot.version() == UNKNOWN_VERSION
+ || snapshot.version() != remote) { return null; }
+ return snapshot;
+ }
+
public PhraseIndexSnapshot readySnapshot(DataBaseTypeEnum source) {
if (!isSourceValid(source)) {
return null;
}
LoadingCache currentCache = cache;
@@ -286,10 +297,25 @@
*
* 必须在数据提交之后调用:版本号先于数据可见地自增,会让其它实例重装到旧数据
* 却记下新版本号,从而永久漏掉这次变更(直到下一次变更)。现有三个调用点
* (短语编辑、发布、导入完成)都在事务提交后,新增调用方须守住这一点。
*/
+ /** outbox 恢复入口:版本写入失败必须保留待办,不能吞掉。 */
+ public void rebuildNowReliable(DataBaseTypeEnum source, String reason) {
+ if (!isSourceValid(source) || redissonHelper == null) {
+ throw new IllegalStateException("phrase index version service unavailable");
+ }
+ Optional version = redissonHelper.incrAndGet(buildVersionKey(source));
+ if (version == null || version.isEmpty()) {
+ throw new IllegalStateException("phrase index version not acknowledged");
+ }
+ LoadingCache currentCache = cache;
+ if (currentCache != null) { currentCache.invalidate(source); }
+ lastKnownSnapshots.remove(source);
+ nextRetryAtRef(source).set(0L);
+ }
+
public void rebuildNow(DataBaseTypeEnum source, String reason) {
if (!isSourceValid(source)) {
return;
}
// 先自增版本号(失败只记日志,绝不能影响本机失效,更不能阻断保存/发布主流程):
--- a/application/src/main/java/cn/jojo/cc/biz/model/bo/english/BatchPublishResultBo.java
+++ b/application/src/main/java/cn/jojo/cc/biz/model/bo/english/BatchPublishResultBo.java
@@ -36,6 +36,8 @@
* 本批失败条目明细:identifier=类型+实体 ID、reason=失败原因;
* 由 XxlJob 入口 {@code englishBatchPublish} 逐条落 warn 日志(不再存进度),
* 日志是定位“哪条为何失败”的唯一出口。
*/
private List failures = new ArrayList<>();
+ /** 已提交PROD但缓存失效待outbox恢复者确认的条目数,包含在successCount中。 */
+ private int pendingCacheCount;
}
--- a/application/src/main/java/cn/jojo/cc/biz/model/bo/english/PublishResultBo.java
+++ b/application/src/main/java/cn/jojo/cc/biz/model/bo/english/PublishResultBo.java
@@ -21,10 +21,13 @@
private Long entityId;
/** PROD 是否已发布 */
private boolean prodPublished;
+ /** PROD 事务已登记可恢复失效待办;broker/消费完成由 outbox 作业及 MQ 日志确认。 */
+ private boolean cacheInvalidationPending;
+
/** JY 发布状态是否已确认回写 */
private boolean jyStatusConfirmed;
/** 发布期间内容被再次编辑 */
private boolean contentChangedDuringPublish;
--- a/application/src/main/java/cn/jojo/cc/biz/model/bo/english/SegmentSnapshot.java
+++ b/application/src/main/java/cn/jojo/cc/biz/model/bo/english/SegmentSnapshot.java
@@ -25,10 +25,15 @@
private static final long serialVersionUID = 1L;
/** 划词块序列,与下发结构的 matches 一一对应 */
private List- matches;
+
+ private String sourcePayloadHash;
+ private Long sourceFormDataId;
+ private Long sourceContentConfigId;
+ private String sourceOutputPayloadHash;
/**
* 单个划词块:只保留还原它所必需的文本与类型。
*/
@Data
--- a/application/src/main/java/cn/jojo/cc/biz/model/cache/FormAndDataCacheBo.java
+++ b/application/src/main/java/cn/jojo/cc/biz/model/cache/FormAndDataCacheBo.java
@@ -55,6 +55,19 @@
* JSON对象
*/
@Schema(title = "out data json obj", description = "out data json obj")
private JSON outDataJsonObj;
+ /** 与outDataJson从同一OutputData行读取,旧缓存缺失即不可采信。 */
+ private Long segmentSourceFormDataId;
+ private String segmentSourcePayloadHash;
+ /** 从同一OutputData.outputData原文计算;Redis默认序列化可丢嵌套null,故保留scalar。 */
+ private String segmentSourceOutputHash;
+
+ /** 保留已有六参调用兼容。 */
+ public FormAndDataCacheBo(Long businessId, Integer businessType, Integer businessTypeClassify,
+ Integer outputDataType, String outDataJson, JSON outDataJsonObj) {
+ this.businessId = businessId; this.businessType = businessType;
+ this.businessTypeClassify = businessTypeClassify; this.outputDataType = outputDataType;
+ this.outDataJson = outDataJson; this.outDataJsonObj = outDataJsonObj;
+ }
}
--- /dev/null
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishAudioCompletionTxService.java
@@ -0,0 +1,48 @@
+package cn.jojo.cc.biz.service.english.dict;
+
+import cn.jojo.cc.biz.cache.english.EnglishDictCacheBoFactory;
+import cn.jojo.cc.common.enums.DataBaseTypeEnum;
+import cn.jojo.cc.core.domain.english.EnglishWordDo;
+import cn.jojo.cc.core.service.english.EnglishWordCoreService;
+import cn.jojo.cc.core.service.english.EnglishPhraseCoreService;
+import cn.jojo.cc.core.service.english.EnglishPhraseVariantCoreService;
+import cn.jojo.cc.core.service.english.EnglishDictOutboxCoreService;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/** 回填成功与教研缓存失效/正式补发布待办原子提交。 */
+@Service
+public class EnglishAudioCompletionTxService {
+ @Resource private EnglishWordCoreService words;
+ @Resource private EnglishPhraseCoreService phrases;
+ @Resource private EnglishPhraseVariantCoreService variants;
+ @Resource private EnglishDictOutboxCoreService outbox;
+
+ @Transactional(rollbackFor = Exception.class)
+ public boolean complete(String table, Long id, Long resourceId) {
+ boolean changed;
+ if ("word".equals(table)) {
+ changed = words.markAudioCompleted(id, resourceId);
+ if (changed) {
+ EnglishWordDo word = words.getById(id, 0L);
+ outbox.enqueue("AUDIO_WORD", id, JSON.toJSONString(
+ EnglishDictCacheBoFactory.wordInvalidateBos(DataBaseTypeEnum.TEACHING, id,
+ null, word.getWordNormalized()), SerializerFeature.WriteClassName));
+ }
+ } else {
+ Long phraseId = "phrase".equals(table) ? id
+ : variants.mapPhraseIdByVariantIds(java.util.List.of(id), 0L).get(id);
+ changed = "phrase".equals(table) ? phrases.markAudioCompleted(id, resourceId)
+ : variants.markAudioCompleted(id, resourceId);
+ if (changed && phraseId != null) {
+ outbox.enqueue("AUDIO_PHRASE", phraseId, JSON.toJSONString(
+ EnglishDictCacheBoFactory.phraseInvalidateBos(DataBaseTypeEnum.TEACHING, phraseId),
+ SerializerFeature.WriteClassName));
+ }
+ }
+ return changed;
+ }
+}
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictImportBizServiceImpl.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictImportBizServiceImpl.java
@@ -348,10 +348,22 @@
}
String fileUri = StringUtils.trimToNull(json.getString("fileUri"));
if (fileUri == null) {
throw BusinessException.paramException("任务参数缺少 fileUri");
}
- return new Param(fileUri, Boolean.TRUE.equals(json.getBoolean("dryRun")),
- json.getInteger("limit"));
+ Integer limit = null;
+ if (json.containsKey("limit")) {
+ Object value = json.get("limit");
+ if (!(value instanceof Integer || value instanceof Long || value instanceof java.math.BigInteger)
+ || !value.toString().matches("[1-9][0-9]*")) {
+ throw BusinessException.paramException("limit 必须是严格正整数");
+ }
+ try {
+ limit = Integer.valueOf(value.toString());
+ } catch (NumberFormatException e) {
+ throw BusinessException.paramException("limit 超出整数范围");
+ }
+ }
+ return new Param(fileUri, Boolean.TRUE.equals(json.getBoolean("dryRun")), limit);
}
}
}
--- /dev/null
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictOutboxBizService.java
@@ -0,0 +1,57 @@
+package cn.jojo.cc.biz.service.english.dict;
+
+import cn.jojo.cc.biz.cache.english.EnglishPhraseIndexCacheService;
+import cn.jojo.cc.biz.service.component.AliyunRocketMQService;
+import cn.jojo.cc.common.enums.CcNormalTagEnum;
+import cn.jojo.cc.common.enums.DataBaseTypeEnum;
+import cn.jojo.cc.common.enums.DataSourceEnum;
+import cn.jojo.cc.core.bo.LoginUserBo;
+import cn.jojo.cc.core.domain.english.EnglishDictOutboxDo;
+import cn.jojo.cc.core.multidatasource.MultiDataSourceUtil;
+import cn.jojo.cc.core.service.english.EnglishDictOutboxCoreService;
+import jakarta.annotation.Resource;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/** 至少一次投递:broker 确认后登记 delivered_time;崩溃窗口允许重复失效,绝不丢待办。 */
+@Service
+public class EnglishDictOutboxBizService {
+ @Resource
+ private EnglishDictOutboxCoreService outbox;
+ @Autowired
+ private AliyunRocketMQService rocketMqProducerService;
+ @Resource
+ private EnglishPhraseIndexCacheService index;
+ @Resource
+ private EnglishPublishBizService publisher;
+
+ public int replay(int limit) {
+ int pending = 0;
+ for (DataSourceEnum source : new DataSourceEnum[]{DataSourceEnum.DEFAULT, DataSourceEnum.PROD}) {
+ pending += MultiDataSourceUtil.switchThen(source, () -> replayCurrentSource(limit));
+ }
+ return pending;
+ }
+
+ private int replayCurrentSource(int limit) {
+ int failed = 0;
+ for (EnglishDictOutboxDo row : outbox.listDue(limit)) {
+ try {
+ rocketMqProducerService.sendN(CcNormalTagEnum.CLEAR_REDIS_COMMON_CACHE,
+ row.getPayload(), "english-dict-outbox-" + row.getId());
+ if ("PHRASE_CACHE".equals(row.getEventType())) {
+ index.rebuildNowReliable(DataBaseTypeEnum.PROD, "outbox-" + row.getId());
+ } else if ("AUDIO_WORD".equals(row.getEventType())) {
+ publisher.publishAudioWord(row.getEntityId(), new LoginUserBo(-1L, "admin"));
+ } else if ("AUDIO_PHRASE".equals(row.getEventType())) {
+ publisher.publishAudioPhrase(row.getEntityId(), new LoginUserBo(-1L, "admin"));
+ }
+ outbox.acknowledge(row.getId());
+ } catch (Exception e) {
+ failed++;
+ outbox.postpone(row, e.getClass().getSimpleName() + ": " + e.getMessage());
+ }
+ }
+ return failed;
+ }
+}
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictPublishProServiceHelper.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictPublishProServiceHelper.java
@@ -49,10 +49,13 @@
*/
@Component
@Slf4j
public class EnglishDictPublishProServiceHelper {
+ @Resource
+ private cn.jojo.cc.core.service.english.EnglishDictOutboxCoreService englishDictOutboxCoreService;
+
private static final EnglishDictPublishSnapshotConvert SNAPSHOT = EnglishDictPublishSnapshotConvert.INSTANCE;
@Resource
private EnglishWordCoreService englishWordCoreService;
@@ -97,10 +100,11 @@
// 在途任务号只属于教研库:学生库不跑生成任务,带上它只会让同行永远显示「生成中」
prodWord.setAudioTaskId("");
prodWord.setAudioTaskTime(0L);
prodWord.setStatus(EnglishPublishStatusEnum.PUBLISHED.getCode());
prodWord.setHasPublished(1);
+ prodWord.setAutoPublishEligible(0);
prodWord.setDeleteTime(0L);
prodWord.fillRequired(operator, now, false);
englishWordCoreService.saveOrUpdate(prodWord);
List writes = new ArrayList<>();
@@ -132,10 +136,16 @@
writes.add(deleted);
}
if (!writes.isEmpty()) {
englishWordTranslationCoreService.saveOrUpdateBatch(writes);
}
+ englishDictOutboxCoreService.enqueue("WORD_CACHE", prodWordId,
+ com.alibaba.fastjson.JSON.toJSONString(
+ cn.jojo.cc.biz.cache.english.EnglishDictCacheBoFactory.wordInvalidateBos(
+ cn.jojo.cc.common.enums.DataBaseTypeEnum.PROD, prodWordId,
+ oldProdNormalized, jyWord.getWordNormalized()),
+ com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName));
return oldProdNormalized;
}
/**
* 发布短语到 PROD:主表 + 变形差量 + 音频副本(原形与每个变体各自的音频),同一 PROD 事务。
@@ -197,10 +207,15 @@
writes.add(deleted);
}
if (!writes.isEmpty()) {
englishPhraseVariantCoreService.saveOrUpdateBatch(writes);
}
+ englishDictOutboxCoreService.enqueue("PHRASE_CACHE", prodPhraseId,
+ com.alibaba.fastjson.JSON.toJSONString(
+ cn.jojo.cc.biz.cache.english.EnglishDictCacheBoFactory.phraseInvalidateBos(
+ cn.jojo.cc.common.enums.DataBaseTypeEnum.PROD, prodPhraseId),
+ com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName));
}
/**
* 公共资源副本写入 PROD,返回 {@code jyResourceId -> prodResourceId}。
*
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictWriteTxService.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishDictWriteTxService.java
@@ -86,10 +86,11 @@
if (!isNew) {
// 更新走非空字段策略,只需带 ID 与要改的列;has_published 必须不赋值:它只能由发布链路 0→1 幂等登记,
// 把锁外快照回写回去会把并发发布刚登记的“曾发布”冲掉(问题:改名与发布并发后删除闸门放行)。
target.setId(existing.getId());
}
+ target.setAutoPublishEligible(0);
target.setWord(StringUtils.trim(bo.getWord()));
target.setWordNormalized(normalized);
target.setPhonetic(bo.getPhonetic());
target.setAudioResourceId(bo.getAudioResourceId());
// 编辑时未提交的字段走非空更新策略、库里保持原值,来源也必须跟着保留:回落成「人工」
@@ -173,10 +174,11 @@
target.setWord(StringUtils.trim(wordText));
target.setWordNormalized(normalized);
target.setPhonetic(phonetic);
target.setAudioResourceId(audioResourceId);
target.setSource(DictSourceEnum.SEGMENT_AUTOFILL.getCode());
+ target.setAutoPublishEligible(1);
target.setStatus(EnglishPublishStatusEnum.UNPUBLISHED.getCode());
target.setHasPublished(0);
// 空壳/缺项由完整性计算自然降级为 INCOMPLETE,不另外判定(Spec §7.5)。
target.setCompleteness(calcWordCompleteness(target, translations));
target.fillRequired(operator, now, true);
@@ -526,11 +528,11 @@
target.setId(existingChild.getId());
target.setCreatorId(existingChild.getCreatorId());
target.setCreateTime(existingChild.getCreateTime());
}
target.setWordId(wordId);
- target.setPos(item.getPos());
+ target.setPos(PartOfSpeechEnum.normalizeKnownCodes(item.getPos()));
target.setTranslation(item.getTranslation());
target.setSortOrder(item.getSortOrder() == null ? index : item.getSortOrder());
target.setDeleteTime(0L);
target.fillRequired(operator, now, item.getId() == null);
writes.add(target);
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishImportFileReader.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishImportFileReader.java
@@ -158,30 +158,36 @@
}
List variants = new ArrayList<>();
List variantNormals = new ArrayList<>();
Set seenVariants = new LinkedHashSet<>();
+ boolean invalidVariant = false;
for (String raw : cell(record, parsed, COLUMN_VARIANTS).split(VARIANT_SEPARATOR)) {
String variant = StringUtils.trimToEmpty(raw);
String variantNormal = TextNormalizer.normalize(variant);
if (StringUtils.isBlank(variantNormal)) {
continue;
}
String variantIllegal = illegalReason(variant);
if (variantIllegal != null || variantNormal.length() > MAX_TEXT_LENGTH) {
result.getInvalid().add(invalid(record.rowNo, phrase, "变形非法:" + variant));
- continue;
+ invalidVariant = true;
+ break;
}
// 与词头同串的变形必须丢弃(come about 的过去式仍是 come about):
// 落进变形子表会撞 Spec §7.2「短语内部变形重复」的写前拒绝,整行导入失败。
if (variantNormal.equals(normalized) || !seenVariants.add(variantNormal)) {
continue;
}
variants.add(variant);
variantNormals.add(variantNormal);
}
+ if (invalidVariant) {
+ firstSeen.remove(normalized);
+ continue;
+ }
PhraseRow row = new PhraseRow();
row.setRowNo(record.rowNo);
row.setPhrase(phrase);
row.setPhraseNormalized(normalized);
row.setTranslation(translation);
@@ -318,51 +324,58 @@
private List readRecords(String content) {
List records = new ArrayList<>();
List cells = new ArrayList<>();
StringBuilder field = new StringBuilder();
boolean inQuotes = false;
+ boolean closedQuote = false;
boolean fieldStarted = false;
int rowNo = 1;
+ int recordStart = 1;
for (int i = 0; i < content.length(); i++) {
char c = content.charAt(i);
if (inQuotes) {
if (c == '"') {
if (i + 1 < content.length() && content.charAt(i + 1) == '"') {
- // "" 转义:落一个引号并跳过配对引号
field.append('"');
i++;
} else {
inQuotes = false;
+ closedQuote = true;
}
} else {
- if (c == '\n') {
- rowNo++;
- }
+ if (c == '\n') { rowNo++; }
field.append(c);
}
continue;
}
- if (c == '"') {
- inQuotes = true;
- fieldStarted = true;
- } else if (c == ',') {
+ if (c == ',') {
cells.add(field.toString());
field.setLength(0);
fieldStarted = false;
- } else if (c == '\r') {
- // CRLF:换行由 \n 统一计数
- } else if (c == '\n') {
- addRecord(records, cells, field, fieldStarted, rowNo);
+ closedQuote = false;
+ } else if (c == '\r' || c == '\n') {
+ if (c == '\r' && i + 1 < content.length() && content.charAt(i + 1) == '\n') { i++; }
+ addRecord(records, cells, field, fieldStarted, recordStart);
cells = new ArrayList<>();
fieldStarted = false;
- rowNo++;
+ closedQuote = false;
+ recordStart = ++rowNo;
+ } else if (c == '"' && !fieldStarted && !closedQuote) {
+ inQuotes = true;
+ fieldStarted = true;
} else {
+ if (c == '"' || closedQuote) {
+ throw BusinessException.paramException("CSV 第 " + recordStart + " 行引号位置非法");
+ }
field.append(c);
fieldStarted = true;
}
}
- addRecord(records, cells, field, fieldStarted, rowNo);
+ if (inQuotes) {
+ throw BusinessException.paramException("CSV 第 " + recordStart + " 行引号未闭合");
+ }
+ addRecord(records, cells, field, fieldStarted, recordStart);
return records;
}
/**
* 收尾一条记录;引号外且无内容的空行直接跳过(清单尾部空行是常态)。
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPhraseBizServiceImpl.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPhraseBizServiceImpl.java
@@ -124,11 +124,11 @@
@Override
public IPage page(EnglishPhrasePageBo bo) {
Page page = new Page<>(bo.getPageNum(), bo.getPageSize());
IPage doPage = englishPhraseCoreService.page(page,
- StringUtils.trimToNull(bo.getPhrase()), bo.getStatus(), 0L);
+ TextNormalizer.normalize(bo.getPhrase()), bo.getStatus(), 0L);
List records = assemble(doPage.getRecords());
Page result = new Page<>(doPage.getCurrent(), doPage.getSize(), doPage.getTotal());
result.setRecords(records);
return result;
}
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPublishBizService.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPublishBizService.java
@@ -9,10 +9,16 @@
*
*
教研库到正式库的写入只允许经过本服务,Controller 不得直接调用发布 helper;
* 自动补录(Task 21)、批量发布(Task 17/18)与全量刷新(Task 22)都复用这四个方法。
*/
public interface EnglishPublishBizService {
+
+ /** 系统音频补发布:实体锁内重新确认无人工草稿。 */
+ PublishResultBo publishAudioWord(Long wordId, LoginUserBo operator);
+ PublishResultBo publishAudioPhrase(Long phraseId, LoginUserBo operator);
+ /** 自动补录恢复:实体锁内确认仍是未编辑的补录草稿。 */
+ PublishResultBo publishAutoFillWord(Long wordId, LoginUserBo operator);
/**
* 发布单个单词到正式库,并回刷引用它的待修复题目。
*
* @param wordId 教研库单词 ID
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPublishBizServiceImpl.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishPublishBizServiceImpl.java
@@ -102,10 +102,48 @@
// 锁内重读同 publishWord(问题12);主表与变体都重新取,避免用锁外旧快照写 PROD。
() -> doPublishPhrase(requirePhrase(phraseId), englishPhraseVariantCoreService.listByPhraseId(
phraseId, 0L), operator));
}
+ @Override
+ public PublishResultBo publishAudioWord(Long wordId, LoginUserBo operator) {
+ EnglishWordDo snapshot = requireWord(wordId);
+ return englishDictWriteLockService.callWithWordLock(wordId, snapshot.getWordNormalized(), () -> {
+ EnglishWordDo current = requireWord(wordId);
+ return EnglishPublishStatusEnum.PUBLISHED.getCode().equals(current.getStatus())
+ && Integer.valueOf(1).equals(current.getHasPublished())
+ ? doPublishWord(current, operator) : new PublishResultBo();
+ });
+ }
+
+ @Override
+ public PublishResultBo publishAudioPhrase(Long phraseId, LoginUserBo operator) {
+ EnglishPhraseDo snapshot = requirePhrase(phraseId);
+ return englishDictWriteLockService.callWithPhraseLocks(phraseId, snapshot.getPhraseNormalized(),
+ variantTextsOf(englishPhraseVariantCoreService.listByPhraseId(phraseId, 0L)), () -> {
+ EnglishPhraseDo current = requirePhrase(phraseId);
+ return EnglishPublishStatusEnum.PUBLISHED.getCode().equals(current.getStatus())
+ && Integer.valueOf(1).equals(current.getHasPublished())
+ ? doPublishPhrase(current, englishPhraseVariantCoreService.listByPhraseId(phraseId, 0L), operator)
+ : new PublishResultBo();
+ });
+ }
+
+ @Override
+ public PublishResultBo publishAutoFillWord(Long wordId, LoginUserBo operator) {
+ EnglishWordDo snapshot = requireWord(wordId);
+ return englishDictWriteLockService.callWithWordLock(wordId, snapshot.getWordNormalized(), () -> {
+ EnglishWordDo current = requireWord(wordId);
+ boolean eligible = cn.jojo.cc.common.enums.english.DictSourceEnum.SEGMENT_AUTOFILL.getCode()
+ .equals(current.getSource())
+ && EnglishPublishStatusEnum.UNPUBLISHED.getCode().equals(current.getStatus())
+ && !Integer.valueOf(1).equals(current.getHasPublished())
+ && Integer.valueOf(1).equals(current.getAutoPublishEligible());
+ return eligible ? doPublishWord(current, operator) : new PublishResultBo();
+ });
+ }
+
/**
* 批次失败条目明细累加(问题08):identifier 携实体类型+ID,rowNo 不适用批量发布场景留空。
*/
private static void addBatchFailure(List failures, String type, Long entityId, String reason) {
EnglishTaskFailureBo item = new EnglishTaskFailureBo();
@@ -152,10 +190,13 @@
try {
PublishResultBo single = englishDictWriteLockService.callWithWordLock(word.getId(), word.getWordNormalized(),
() -> doPublishWord(requireWord(word.getId()), operator));
if (single.isProdPublished()) {
result.setSuccessCount(result.getSuccessCount() + 1);
+ if (single.isCacheInvalidationPending()) {
+ result.setPendingCacheCount(result.getPendingCacheCount() + 1);
+ }
} else {
result.setFailedCount(result.getFailedCount() + 1);
// 未发布成功也要带原因上浮(问题08),否则任务侧只有计数没有条目。
addBatchFailure(result.getFailures(), "WORD", word.getId(),
StringUtils.defaultIfBlank(single.getFailReason(), "PROD 写入未确认"));
@@ -186,10 +227,13 @@
result.setNextCursor(phrase.getId());
try {
PublishResultBo single = publishPhrase(phrase.getId(), operator);
if (single.isProdPublished()) {
result.setSuccessCount(result.getSuccessCount() + 1);
+ if (single.isCacheInvalidationPending()) {
+ result.setPendingCacheCount(result.getPendingCacheCount() + 1);
+ }
} else {
result.setFailedCount(result.getFailedCount() + 1);
addBatchFailure(result.getFailures(), "PHRASE", phrase.getId(),
StringUtils.defaultIfBlank(single.getFailReason(), "PROD 写入未确认"));
}
@@ -220,10 +264,12 @@
// 返回更新前 PROD 的真实旧文本(问题05):改名重发布时旧文本映射必须一并失效,
// 否则 cat 仍反查到本词条而详情已是 dog,读时解析会把错引用下发出去。
String oldProdNormalized = englishDictPublishProServiceHelper.publishWordInPro(
jyWord, jyTranslations, jyResources, operator);
result.setProdPublished(true);
+ result.setCacheInvalidationPending(true);
+ result.setFailReason("PROD 已提交,缓存失效由 outbox 恢复任务确认");
try {
// 发布事实与当前状态分两个生命周期:条件回写只更新当前版本的状态,
// 版本已被推进时单独登记「曾发布」,禁止把 v2 标成已发布,也不丢失历史发布事实(问题12)。
boolean confirmed = englishWordCoreService.updatePublishStatusIfUnmodified(jyWordId,
EnglishPublishStatusEnum.PUBLISHED.getCode(), 1, snapshotUpdateTime,
@@ -249,18 +295,11 @@
}
log.error("单词发布后 JY 状态回写异常:PROD 已提交,JY 保持待发布、运营需重新发布一次, wordId={}",
jyWordId, e);
}
- // PROD 失效由“PROD 已提交”这一事实驱动,不依赖状态回写正常返回(Spec §4.7)。
- try {
- Long prodWordId = CommonService.decrementAndGet(jyWordId);
- invalidateProdWordCache(prodWordId, oldProdNormalized, jyWord.getWordNormalized());
- } catch (Exception e) {
- log.error("单词发布后 PROD 缓存失效失败,JY 状态不回写、运营重发一次即重新触发, wordId={}",
- jyWordId, e);
- }
+ // 失效事件已在 PROD 写入事务内持久化;恢复任务同步发送,失败不丢身份。
// 单词发布不再在请求线程内同步回刷 PROD 题目:词条发布后由下发时读时解析自然点亮,无待修复行;
// 由 09 的定向回刷作业按 normalized_text 扫描消费(Spec §5.7),避免锁 lease 60s 内跑无上限循环。
return result;
}
@@ -273,10 +312,12 @@
List jyResources = loadAliveResources(collectAudioResourceIds(
jyPhrase.getAudioResourceId(), jyVariants));
englishDictPublishProServiceHelper.publishPhraseInPro(jyPhrase, jyVariants, jyResources, operator);
result.setProdPublished(true);
+ result.setCacheInvalidationPending(true);
+ result.setFailReason("PROD 已提交,缓存失效由 outbox 恢复任务确认");
try {
boolean confirmed = englishPhraseCoreService.updatePublishStatusIfUnmodified(jyPhraseId,
EnglishPublishStatusEnum.PUBLISHED.getCode(), 1, snapshotUpdateTime,
operator.getLoginUserId());
result.setJyStatusConfirmed(confirmed);
@@ -298,22 +339,11 @@
}
log.error("短语发布后 JY 状态回写异常:PROD 已提交,JY 保持待发布、运营需重新发布一次, phraseId={}",
jyPhraseId, e);
}
- // PROD 失效与全量重分词通知由“PROD 已提交”驱动,不依赖状态回写正常返回(Spec §4.7):
- // 短语索引不重建、脏标不置上,学生会继续点在旧短语上,且题目分词永不重算。
- try {
- Long prodPhraseId = CommonService.decrementAndGet(jyPhraseId);
- invalidateProdPhraseCache(prodPhraseId);
- // PROD 短语索引只在正式库发布/删除后失效重载,教研库编辑不得触发(Spec §10.5)。
- englishPhraseIndexCacheService.rebuildNow(DataBaseTypeEnum.PROD, "english-phrase-publish");
- // 分词重算不再自动触发:全量重分词只在每日低峰跑一轮,需要提前生效时由运营走按短语重分词接口。
- } catch (Exception e) {
- log.error("短语发布后 PROD 失效/重分词通知失败,JY 状态不回写、运营重发一次即重新触发, phraseId={}",
- jyPhraseId, e);
- }
+ // 短语失效及跨实例索引版本推进同样由持久化 outbox 恢复。
return result;
}
private void invalidateProdWordCache(Long prodWordId, String oldNormalizedWord, String newNormalizedWord) {
rocketMqProducerService.clearRedisCommonCache(EnglishWordDo.class, prodWordId,
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishWordBizServiceImpl.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/EnglishWordBizServiceImpl.java
@@ -111,11 +111,11 @@
@Override
public IPage page(EnglishWordPageBo bo) {
Page page = new Page<>(bo.getPageNum(), bo.getPageSize());
IPage doPage = englishWordCoreService.page(page,
- StringUtils.trimToNull(bo.getWord()), bo.getStatus(), bo.getCompleteness(),
+ TextNormalizer.normalize(bo.getWord()), bo.getStatus(), bo.getCompleteness(),
0L);
List records = assemble(doPage.getRecords());
Page result = new Page<>(doPage.getCurrent(), doPage.getSize(), doPage.getTotal());
result.setRecords(records);
return result;
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/SegmentAutoFillBizServiceImpl.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/SegmentAutoFillBizServiceImpl.java
@@ -62,10 +62,13 @@
@Resource
private EnglishDictConfig englishDictConfig;
@Resource
+ private cn.jojo.cc.core.service.english.EnglishWordCoreService englishWordCoreService;
+
+ @Resource
private BaiduDictClient baiduDictClient;
@Resource
private EnglishDictWriteLockService englishDictWriteLockService;
@@ -136,22 +139,19 @@
return false;
}
}
private void autoFillOne(String normalizedWord, LoginUserBo operator) {
- // 1. 百度:异常/空结果统一按空壳口径(§11.4)。
- BaiduDictResult dict = queryBaiduSafely(normalizedWord);
- // 2. 音频不在补录链路里生成:建行时留空,由音频回填任务按 source=分词自动补录 统一提交。
- // 一条消息里有多个词时,这里不再逐个等合成,避免把整条消息拖成小时级。
-
- // 3. 落库:在单词写入锁内完成,与人工保存/删除互斥(05 冻结口径)。
- // entityId 传 null:补录以新建为主,锁内才能幂等复用旧行,取锁时拿不到稳定实体 ID;
- // 与“同一实体人工编辑”交错是已接受的窄窗口(低频,事务+条件回写兜底)。
- EnglishWordDo saved = englishDictWriteLockService.callWithWordLock(null, normalizedWord,
- () -> englishDictWriteTxService.saveWordForAutoFill(
- normalizedWord, dict == null ? null : dict.getPhonetic(), null,
- mapTranslations(dict), operator));
+ // 文本锁内先认账(含人工删除行);已有行只走幂等恢复,不再调用百度。
+ EnglishWordDo saved = englishDictWriteLockService.callWithWordLock(null, normalizedWord, () -> {
+ EnglishWordDo existing = englishWordCoreService.getByNormalized(normalizedWord, 0L);
+ if (existing == null) { existing = englishWordCoreService.getByNormalizedIgnoreDeleted(normalizedWord); }
+ if (existing != null) { return existing; }
+ BaiduDictResult dict = queryBaiduSafely(normalizedWord);
+ return englishDictWriteTxService.saveWordForAutoFill(normalizedWord,
+ dict == null ? null : dict.getPhonetic(), null, mapTranslations(dict), operator);
+ });
if (isSoftDeleted(saved)) {
// 运营删过这个词(第四轮复审问题12 的姊妹缺口):锁内已按“含已删除行”认过账,既不重建也不发布,
// 也不做完整性提示——那一行已经不在词典里了,再喊“等业务方补全”只会误导。
log.info("单词已被运营删除,本次不重建也不发布, word={}, wordId={}", normalizedWord, saved.getId());
return;
@@ -167,36 +167,32 @@
// 还必须验出身(第四轮复审问题12):“待发布+从未发布”这个状态人工/导入草稿也有,
// 只判状态就等于替运营把他还没审完的词条发到学生端——发布是人工闸门,系统越不过去。
// 本仓词典无“撤回发布”,且 hasPublished 仅发布链路写,所以
// “补录出身 + 待发布 + 从未发布 + 建后未被编辑”只会是补录自己建到一半的行,
// 自动补发仍是设计内恢复。
- // 最后一条不可省:运营手工补全过的行也是“补录出身+待发布+从未发布”,
- // 不排除就会替运营把它发出去、绕过人工闸门。编辑会刷新 update_time,
- // 而发布回写(updatePublishStatusIfUnmodified / registerHasPublished)
- // 只改状态不动时间,所以两列相等 = 建行后没人改过,且不影响发布失败后的重试。
- // 已知残留窗口:本判定与下面 publishWord 之间若运营恰好完成一次保存,被发布的就是他刚保存的
- // 版本,且 hasPublished 置 1 后该词不可再删;窗口只有几行代码,换来的是双方都不再互等超时。
+ // 仅由新建补录置1,任何人工保存立即清零;不再以秒级审计时间相等证明未编辑。
+ // 历史行迁移默认0,需人工发布,禁止依据旧时间戳回填资格。
boolean needPublish = DictSourceEnum.SEGMENT_AUTOFILL.getCode().equals(saved.getSource())
&& EnglishPublishStatusEnum.UNPUBLISHED.getCode().equals(saved.getStatus())
&& !Integer.valueOf(1).equals(saved.getHasPublished())
- && Objects.equals(saved.getCreateTime(), saved.getUpdateTime());
+ && Integer.valueOf(1).equals(saved.getAutoPublishEligible());
if (needPublish) {
rocketMqProducerService.clearRedisCommonCache(EnglishWordDo.class, saved.getId(),
EnglishDictCacheBoFactory.wordInvalidateBos(DataBaseTypeEnum.TEACHING, saved.getId(),
null, saved.getWordNormalized()));
- englishPublishBizService.publishWord(saved.getId(), operator);
+ englishPublishBizService.publishAutoFillWord(saved.getId(), operator);
}
log.info("自动补录单词完成, word={}, wordId={}, hasPublished={}, operatorId={}",
normalizedWord, saved.getId(), saved.getHasPublished(),
operator == null ? null : operator.getLoginUserId());
// PRD(题保存分词+下游对接):必填项有缺失也落库并立即发布,靠日志告知待人工完善;
// 用户裁决 2026-10-27:本期只打日志,不接钉钉通知。判定只看结果行完整性:
// 幂等复用到的旧行若仍是残缺态,同样属于"等业务方补全"的待办,重复提示是可接受的。
if (WordCompletenessEnum.INCOMPLETE.getCode().equals(saved.getCompleteness())) {
log.warn("自动补录词条完整性有缺失,需产品联系业务方手动完善, word={}, wordId={}, phonetic={}, audio={}, translation={}",
normalizedWord, saved.getId(), saved.getPhonetic(), saved.getAudioResourceId(),
- dict == null ? null : dict.getTranslations());
+ saved.getCompleteness());
}
// 第 5 段“双库定向回刷”已整块删除:库里只存划词块的文本与类型(见 SegmentSnapshot),
// ID 在下发时按当前词典现算,词补出来后存量题目自动可点,不需要按词 × 按题重跑分词。
}
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/dict/impl/EnglishAudioBackfillBizServiceImpl.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/dict/impl/EnglishAudioBackfillBizServiceImpl.java
@@ -61,10 +61,13 @@
* 回填后补发布用的系统操作人:音频是机器回填的,发布也必须以系统身份落库。
*/
private static final LoginUserBo SYSTEM_OPERATOR = new LoginUserBo(-1L, "admin");
@Resource
+ private cn.jojo.cc.biz.service.english.dict.EnglishAudioCompletionTxService englishAudioCompletionTxService;
+
+ @Resource
private EnglishDictConfig englishDictConfig;
@Resource
private InternalTtsClient internalTtsClient;
@@ -98,22 +101,43 @@
return result;
}
private void submitPending(int batch, EnglishAudioBackfillResultBo result) {
long createdBefore = nowSeconds() - SUBMIT_COOLDOWN_SECONDS;
- for (EnglishWordDo row : englishWordCoreService.listAudioPendingSubmit(batch, createdBefore)) {
+ for (EnglishWordDo row : pendingWords(batch, createdBefore)) {
submit(ROW_WORD, row.getId(), row.getWord(), result);
}
- for (EnglishPhraseDo row : englishPhraseCoreService.listAudioPendingSubmit(batch, createdBefore)) {
+ for (EnglishPhraseDo row : pendingPhrases(batch, createdBefore)) {
submit(ROW_PHRASE, row.getId(), row.getPhrase(), result);
}
submitPendingVariants(batch, createdBefore, result);
+ }
+
+ private List pendingWords(int batch, long before) {
+ RAtomicLong cursor = redissonClient.getAtomicLong(failKey("cursor:words"));
+ List rows = englishWordCoreService.listAudioPendingSubmitAfterId(batch, before, cursor.get());
+ cursor.set(rows.size() < batch ? 0L : rows.get(rows.size() - 1).getId());
+ return rows;
+ }
+
+ private List pendingPhrases(int batch, long before) {
+ RAtomicLong cursor = redissonClient.getAtomicLong(failKey("cursor:phrases"));
+ List rows = englishPhraseCoreService.listAudioPendingSubmitAfterId(batch, before, cursor.get());
+ cursor.set(rows.size() < batch ? 0L : rows.get(rows.size() - 1).getId());
+ return rows;
+ }
+
+ private List pendingVariants(int batch, long before) {
+ RAtomicLong cursor = redissonClient.getAtomicLong(failKey("cursor:variants"));
+ List rows = englishPhraseVariantCoreService.listAudioPendingSubmitAfterId(batch, before, cursor.get());
+ cursor.set(rows.size() < batch ? 0L : rows.get(rows.size() - 1).getId());
+ return rows;
}
private void submitPendingVariants(int batch, long createdBefore, EnglishAudioBackfillResultBo result) {
List variants =
- englishPhraseVariantCoreService.listAudioPendingSubmit(batch, createdBefore);
+ pendingVariants(batch, createdBefore);
if (variants.isEmpty()) {
return;
}
Set phraseIds = variants.stream()
.map(EnglishPhraseVariantDo::getPhraseId)
@@ -148,11 +172,10 @@
try {
TtsRequest request = new TtsRequest();
request.setText(text);
String taskId = internalTtsClient.submitTask(request);
markSubmitted(table, id, taskId);
- clearFailCount(rowKey);
result.setSubmitted(result.getSubmitted() + 1);
} catch (Exception e) {
increaseFailCount(rowKey);
result.setSubmitFailed(result.getSubmitFailed() + 1);
log.warn("音频任务提交失败, row={}, text={}", rowKey, text, e);
@@ -193,13 +216,13 @@
* 而这些词已经发布过,运营不会再去点一次。只给已发布行补,未发布行不自动发布(发布是人工闸门)。
*/
private void publishReleased(String table, Long id) {
try {
if (ROW_WORD.equals(table)) {
- englishPublishBizService.publishWord(id, SYSTEM_OPERATOR);
+ englishPublishBizService.publishAudioWord(id, SYSTEM_OPERATOR);
} else if (ROW_PHRASE.equals(table)) {
- englishPublishBizService.publishPhrase(id, SYSTEM_OPERATOR);
+ englishPublishBizService.publishAudioPhrase(id, SYSTEM_OPERATOR);
}
} catch (Exception e) {
// 补发布失败不回滚音频回填:音频已经在教研库,运营下次发布或再跑一轮仍会带上;
// 这里只留痕,避免因为一次锁竞争把已成功的回填也标记成失败
log.warn("音频回填后补发布失败, row={}:{}", table, id, e);
@@ -241,10 +264,21 @@
EnglishAudioBackfillResultBo result) {
if (id == null || StringUtils.isBlank(taskId)) {
return false;
}
result.setScanned(result.getScanned() + 1);
+ String rowKey = table + ":" + id;
+ long elapsed = submitTime == null ? Long.MAX_VALUE : Math.max(0L, now - submitTime);
+ if (elapsed >= timeoutSeconds) {
+ // 总生命周期预算适用于查询异常、completed 下载失败等全部分支;取消失败也必须本地收口。
+ try { internalTtsClient.cancelTask(taskId); }
+ catch (Exception e) { log.warn("取消超时音频任务失败, taskId={}", taskId, e); }
+ increaseFailCount(rowKey);
+ markCleared(table, id);
+ result.setFailed(result.getFailed() + 1);
+ return false;
+ }
TtsTaskStatus status;
try {
status = internalTtsClient.queryTask(taskId);
} catch (Exception e) {
// 查状态失败按「还没好」处理:下轮再查,不在这里清任务号
@@ -252,27 +286,19 @@
result.setRunning(result.getRunning() + 1);
return false;
}
if (status == null || status.isRunning()) {
- long elapsed = submitTime == null ? Long.MAX_VALUE : now - submitTime;
- if (elapsed >= timeoutSeconds) {
- // 平台丢任务或长时间不出结果:先取消平台任务,再清本地在途标记,否则该行永远挂着「生成中」
- internalTtsClient.cancelTask(taskId);
- markCleared(table, id);
- result.setFailed(result.getFailed() + 1);
- log.warn("音频任务超时清理, row={}:{}, taskId={}, elapsed={}s", table, id, taskId, elapsed);
- } else {
- result.setRunning(result.getRunning() + 1);
- }
+ result.setRunning(result.getRunning() + 1);
return false;
}
if (status.isCompleted()) {
// 人工优先:在途期间运营可能已上传了音频,这时只清任务号,既不下载也不覆盖
if (currentAudioResourceId != null && currentAudioResourceId > 0) {
markCleared(table, id);
+ clearFailCount(rowKey);
result.setCompleted(result.getCompleted() + 1);
log.info("音频任务完成但该行已有音频,保留人工音频并清任务号, row={}:{}", table, id);
return true;
}
Long resourceId = englishAudioBizService.saveAudio(taskId);
@@ -280,21 +306,23 @@
if (!markCompleted(table, id, resourceId)) {
// 条件更新未命中:写完前的一瞬间人工补了音频(并发窗口),同样只清任务号,不回写
markCleared(table, id);
log.info("音频回填时发现该行已被人工补齐,保留人工音频, row={}:{}", table, id);
}
+ clearFailCount(rowKey);
result.setCompleted(result.getCompleted() + 1);
return true;
}
// 下载或落库失败(未就绪/源文件过期/校验失败):保留任务号,由超时清理最终收口,
// 不在这里清掉——清掉就意味着这个词再也不会自动补音频
result.setFailed(result.getFailed() + 1);
log.warn("音频下载或保存失败,保留任务号待下轮, row={}:{}, taskId={}", table, id, taskId);
return false;
}
- // failed / cancelled:终态失败,清掉在途标记,音频留空(运营仍可手工上传)
+ // 整个尝试失败才累计,成功 submit 不重置计数。
+ increaseFailCount(rowKey);
markCleared(table, id);
result.setFailed(result.getFailed() + 1);
log.warn("音频任务终态失败, row={}:{}, taskId={}, status={}, error={}",
table, id, taskId, status.getStatus(), status.getError());
return false;
@@ -320,17 +348,11 @@
englishPhraseVariantCoreService.markAudioTaskCleared(id);
}
}
private boolean markCompleted(String table, Long id, Long resourceId) {
- if (ROW_WORD.equals(table)) {
- return englishWordCoreService.markAudioCompleted(id, resourceId);
- }
- if (ROW_PHRASE.equals(table)) {
- return englishPhraseCoreService.markAudioCompleted(id, resourceId);
- }
- return englishPhraseVariantCoreService.markAudioCompleted(id, resourceId);
+ return englishAudioCompletionTxService.complete(table, id, resourceId);
}
private boolean isAutoSource(Integer source) {
return DictSourceEnum.IMPORT.getCode().equals(source)
|| DictSourceEnum.SEGMENT_AUTOFILL.getCode().equals(source);
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/segment/DefaultSegmentTextExtractor.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/DefaultSegmentTextExtractor.java
@@ -56,10 +56,13 @@
@Resource
private EnglishDictConfig englishDictConfig;
@Resource
+ private cn.jojo.cc.core.service.question.standard.ContentNormConfigCoreService contentNormConfigCoreService;
+
+ @Resource
private FormDataCoreService formDataCoreService;
@Resource
private OutputDataCoreService outputDataCoreService;
@@ -91,31 +94,54 @@
this.historicalContentResolver = historicalContentResolver;
}
@Override
public Map> extractRichTextsByContentType(QuestionDo question) {
- if (question == null || question.getId() == null || question.getQuestionType() == null) {
- return new LinkedHashMap<>(0);
- }
- if (!currentWhitelist().contains(question.getQuestionType())) {
- return new LinkedHashMap<>(0);
- }
- try {
- String formDataJson = resolveFormDataJson(question);
- if (StringUtils.isNotBlank(formDataJson)) {
- return extractFromFormData(question.getQuestionType(), formDataJson);
- }
- String historicalJson = resolveHistoricalContentJson(question);
- if (StringUtils.isNotBlank(historicalJson)) {
- return extractFromHistoricalContent(historicalJson);
- }
- return new LinkedHashMap<>(0);
- } catch (Exception e) {
- log.error("标准题富文本抽取失败, questionId={}, questionType={}",
- question.getId(), question.getQuestionType(), e);
- throw e;
- }
+ return readRichTextSource(question).richTexts();
+ }
+
+ @Override
+ public RichTextSource readRichTextSource(QuestionDo question) {
+ if (question == null || question.getId() == null || question.getQuestionType() == null
+ || !currentWhitelist().contains(question.getQuestionType())) {
+ return new RichTextSource(new LinkedHashMap<>(), null);
+ }
+ if (formDataResolver != null) {
+ String json = formDataResolver.apply(question);
+ return new RichTextSource(StringUtils.isBlank(json) ? new LinkedHashMap<>()
+ : extractFromFormData(question.getQuestionType(), json), null);
+ }
+ // 真实表单业务键是内容配置ID(DefaultFormDataProcessing.saveContentNormConfigDo)。
+ List configs =
+ contentNormConfigCoreService == null ? List.of() : contentNormConfigCoreService.listByQuestionId(question.getId());
+ Long configId = configs == null || configs.isEmpty() ? question.getId() : configs.get(0).getId();
+ FormDataDo form = formDataCoreService == null ? null : formDataCoreService.get(null, configId,
+ FormDataBusinessTypeEnum.STANDARD_QUESTION.getType(),
+ BusinessTypeClassifyEnum.STANDARD_QUESTION_CONTENT.getType());
+ if (form != null && form.getFormData() != null) {
+ JSON input = (JSON) JSON.parse(JSON.toJSONString(form.getFormData(),
+ com.alibaba.fastjson.serializer.SerializerFeature.WriteMapNullValue));
+ String inputHash = SegmentPayloadIdentity.hash(input);
+ String outputHash = null;
+ if (outputDataCoreService != null) {
+ List generated = outputDataCoreService.listByFormDataIdsAndType(List.of(form.getId()),
+ OutPutDataTypeEnum.GAME.getType(), 0);
+ // 只认同一输入实际生成过的输出;不把当前DB输入hash贴到旧输出上。
+ for (OutputDataDo output : generated) {
+ if (java.util.Objects.equals(inputHash, output.getSegmentSourcePayloadHash())
+ && StringUtils.isNotBlank(output.getOutputData())) {
+ outputHash = SegmentPayloadIdentity.outputHash((JSON) JSON.parse(output.getOutputData()));
+ }
+ }
+ }
+ return new RichTextSource(extractFromFormData(question.getQuestionType(), input.toJSONString()),
+ new SegmentPayloadIdentity(inputHash, form.getId(), form.getBusinessId(), outputHash));
+ }
+ // 没有原输入时仅保留历史匹配能力;无可信来源身份,RPC不会冒认与输出URL一致。
+ String historical = resolveHistoricalContentJson(question);
+ return new RichTextSource(StringUtils.isBlank(historical) ? new LinkedHashMap<>()
+ : extractFromHistoricalContent(historical), null);
}
@Override
public String extractPlainText(String richText) {
if (StringUtils.isBlank(richText)) {
@@ -125,13 +151,11 @@
document.select("formula, img, script, style").forEach(element -> {
element.before(" ");
element.remove();
});
String text = document.text();
- // 只按 IPA 字符区间剔除音标字母:题目富文本不含斜杠包裹的音标,不能用“成对斜杠”猜测音标,
- // 否则会误删 and/or、his/her、日期与 URL 等正文(斜杠本身由分词器当作分隔符)。
- text = text.replaceAll("[\\u02B0-\\u02FF\\u0250-\\u02AF\\u1D00-\\u1D7F]+", " ");
+ // 保留音标token原貌,交给EnglishWords统一排除;先剔音标字母会漏出ASCII残片。
return text.replaceAll("\\s+", " ").trim();
}
private Collection currentWhitelist() {
if (englishDictConfig != null) {
@@ -139,37 +163,10 @@
// 否则与候选题扫描的解读相反,同一份配置会出现两种行为。
return englishDictConfig.getQuestionTypeWhitelist();
}
// 未装配配置对象的只有单测路径,取值来自 setQuestionTypeWhitelist 测试钩子。
return questionTypeWhitelist;
- }
-
- private String resolveFormDataJson(QuestionDo question) {
- if (formDataResolver != null) {
- return formDataResolver.apply(question);
- }
- if (formDataCoreService == null) {
- return null;
- }
- FormDataDo formData = formDataCoreService.get(null, question.getId(),
- FormDataBusinessTypeEnum.STANDARD_QUESTION.getType(),
- BusinessTypeClassifyEnum.STANDARD_QUESTION_CONTENT.getType());
- if (formData == null) {
- return null;
- }
- if (formData.getFormData() != null) {
- return formData.getFormData().toJSONString();
- }
- if (outputDataCoreService == null) {
- return null;
- }
- List outputs = outputDataCoreService.listByFormDataIdsAndType(
- List.of(formData.getId()), OutPutDataTypeEnum.GAME.getType(), 0);
- if (outputs == null || outputs.isEmpty()) {
- return null;
- }
- return outputs.get(0).getOutputData();
}
private String resolveHistoricalContentJson(QuestionDo question) {
if (historicalContentResolver != null) {
return historicalContentResolver.apply(question);
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/segment/EnglishSegmentDeliveryBizService.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/EnglishSegmentDeliveryBizService.java
@@ -65,11 +65,17 @@
* 线程上永远是一个具体库,与本次实际读到的分词行一致。
*
* @param questionKeys 题目的 questionKey 集合,须与库内 {@code question_key} 同形态
* @return questionKey -> 已按当前词典解析过 ID 的划词序列;无分词行或解析失败的 key 不在结果中
*/
+ @org.springframework.transaction.annotation.Transactional(readOnly = true)
public Map resolveByQuestionKeys(Set questionKeys) {
+ return Collections.emptyMap(); // 未传本次输出身份,不从新DB猜测旧响应的版本。
+ }
+
+ public Map resolveByQuestionKeys(Set questionKeys,
+ Map selectedOutputs) {
// 回滚契约(总开关一键止血):关闭后不查库、整批不下发 segments,
// 题目照常返回、C 端划词一律降级为不可点;仅关新增落库不叫止血。
if (!englishDictConfig.isSegmentEnabled()) {
return Collections.emptyMap();
}
@@ -97,11 +103,29 @@
}
rowsByKey.computeIfAbsent(row.getQuestionKey(), key -> new ArrayList<>()).add(row);
}
Map result = new HashMap<>(rowsByKey.size());
for (Map.Entry> entry : rowsByKey.entrySet()) {
- EnglishSegmentDto merged = resolveGroup(source, entry.getValue());
+ EnglishSegmentDto merged;
+ try {
+ SelectedSegmentOutput selected = selectedOutputs == null ? null : selectedOutputs.get(entry.getKey());
+ if (selected == null || !selected.known() || selected.source() != source) { continue; }
+ SegmentPayloadIdentity identity = selected.identity();
+ boolean same = true;
+ for (QuestionSegmentDo row : entry.getValue()) {
+ SegmentSnapshot stored = JSON.parseObject(row.getSegments(), SegmentSnapshot.class);
+ same &= stored != null && java.util.Objects.equals(identity.payloadHash(), stored.getSourcePayloadHash())
+ && java.util.Objects.equals(identity.formDataId(), stored.getSourceFormDataId())
+ && java.util.Objects.equals(identity.contentConfigId(), stored.getSourceContentConfigId())
+ && java.util.Objects.equals(identity.outputPayloadHash(), stored.getSourceOutputPayloadHash());
+ }
+ if (!same) { continue; }
+ merged = resolveGroup(source, entry.getValue());
+ } catch (Exception e) {
+ log.warn("核验本次响应来源身份失败,本题不下发划词, questionKey={}", entry.getKey(), e);
+ continue;
+ }
if (merged != null) {
result.put(entry.getKey(), merged);
}
}
return result;
--- /dev/null
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/EnglishSegmentOutputIdentityRecoveryBizService.java
@@ -0,0 +1,112 @@
+package cn.jojo.cc.biz.service.english.segment;
+
+import cn.jojo.cc.biz.cache.FormAndDataCacheService;
+import cn.jojo.cc.biz.cache.key.FormAndDataCacheKey;
+import cn.jojo.cc.biz.model.bo.CommonCacheBo;
+import cn.jojo.cc.biz.model.cache.FormAndDataCacheBo;
+import cn.jojo.cc.biz.service.question.QuestionBizService;
+import cn.jojo.cc.common.enums.DataBaseTypeEnum;
+import cn.jojo.cc.common.enums.DataSourceEnum;
+import cn.jojo.cc.common.enums.RedisKeyEnum;
+import cn.jojo.cc.common.enums.question.FormDataBusinessTypeEnum;
+import cn.jojo.cc.common.enums.question.BusinessTypeClassifyEnum;
+import cn.jojo.cc.common.enums.question.OutPutDataTypeEnum;
+import cn.jojo.cc.core.bo.LoginUserBo;
+import cn.jojo.cc.core.domain.question.library.QuestionDo;
+import cn.jojo.cc.core.domain.schema.common.data.OutputDataDo;
+import cn.jojo.cc.core.multidatasource.MultiDataSourceUtil;
+import cn.jojo.cc.core.service.question.library.QuestionCoreService;
+import cn.jojo.cc.core.service.schema.OutputDataCoreService;
+import cn.jojo.cc.core.service.english.EnglishDictOutboxCoreService;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RedissonClient;
+import org.springframework.stereotype.Service;
+import java.util.List;
+import java.util.Set;
+import java.util.Objects;
+
+/** 持续来源恢复:旧输出及后续跨库身份失配必须真实重生成,不能给旧URL贴新输入hash。 */
+@Service
+@Slf4j
+public class EnglishSegmentOutputIdentityRecoveryBizService {
+ @Resource private cn.jojo.cc.biz.common.config.EnglishDictConfig englishDictConfig;
+ @Resource private QuestionCoreService questions;
+ @Resource private QuestionBizService questionBizService;
+ @Resource private OutputDataCoreService outputs;
+ @Resource private SegmentContentSnapshotService content;
+ @Resource private QuestionSegmentBizService segments;
+ @Resource private EnglishSegmentSkeletonBizService skeletons;
+ @Resource private EnglishDictOutboxCoreService outbox;
+ @Resource private RedissonClient redissonClient;
+
+ public String runOnce(LoginUserBo operator) {
+ if (paused()) { return "paused: no cursor/read/regeneration/upload"; }
+ int failed = 0;
+ StringBuilder summary = new StringBuilder();
+ for (DataBaseTypeEnum env : new DataBaseTypeEnum[]{DataBaseTypeEnum.TEACHING, DataBaseTypeEnum.PROD}) {
+ DataSourceEnum dataSource = env == DataBaseTypeEnum.PROD ? DataSourceEnum.PROD : DataSourceEnum.DEFAULT;
+ var cursor = redissonClient.getAtomicLong(RedisKeyEnum.ENGLISH_SEGMENT_REFRESH_LOCK.buildKey()
+ + ":output-source-cursor:" + env.name());
+ long after = cursor.get();
+ List batch = MultiDataSourceUtil.switchThen(dataSource,
+ () -> questions.listSegmentCandidatesByIdGt(after, 50, env));
+ for (QuestionDo question : batch) {
+ if (paused()) { return summary.append("paused; cursor retained").toString(); }
+ try {
+ MultiDataSourceUtil.switchThen(dataSource, () -> { recover(question, env, operator); return null; });
+ } catch (Exception e) {
+ failed++;
+ log.error("输出来源恢复失败,保留fail-closed并推进后续题, env={}, id={}", env, question.getId(), e);
+ }
+ cursor.set(question.getId());
+ }
+ boolean finished = batch.size() < 50;
+ if (finished) {
+ cursor.set(0L);
+ log.info("输出身份恢复完整扫描一圈, env={}, lastId={}, 本轮失败={}; 持续Job下一轮重新扫描", env, after, failed);
+ }
+ summary.append(env).append(" scanned=").append(batch.size()).append(" finished=").append(finished).append("; ");
+ }
+ return summary.append("failed=").append(failed).toString();
+ }
+
+ private void recover(QuestionDo question, DataBaseTypeEnum env, LoginUserBo operator) {
+ if (paused()) { return; }
+ SegmentPayloadIdentity identity = content.readSource(question).identity();
+ if (identity == null || !identity.inputKnown()) { throw new IllegalStateException("原FormData不可用,不能伪造输出身份"); }
+ if (!hasMatchingOutput(identity)) {
+ // 现有公开入口按真实schema重生成并上传输出,genOutPutData在生成前记录输入身份。
+ if (paused()) { return; }
+ questionBizService.refreshGenStandQuestion(question.getId(), operator);
+ identity = content.readSource(question).identity();
+ if (identity == null || !identity.inputKnown() || !hasMatchingOutput(identity)) {
+ throw new IllegalStateException("输出未能按当前输入重生成,保持不可点击");
+ }
+ }
+ FormAndDataCacheKey key = new FormAndDataCacheKey(identity.contentConfigId(),
+ FormDataBusinessTypeEnum.STANDARD_QUESTION.getType(),
+ BusinessTypeClassifyEnum.STANDARD_QUESTION_CONTENT.getType(), OutPutDataTypeEnum.GAME.getType());
+ CommonCacheBo evict =
+ CommonCacheBo.builder().bizType(QuestionDo.class)
+ .bizId(question.getId()).ids(Set.of(key)).type(FormAndDataCacheService.class).build();
+ outbox.enqueue("FORM_OUTPUT_CACHE", question.getId(), JSON.toJSONString(List.of(evict), SerializerFeature.WriteClassName));
+ if (env == DataBaseTypeEnum.PROD) {
+ skeletons.ensurePair(cn.jojo.cc.biz.service.component.CommonService.incrementAndGet(question.getId()), question.getId());
+ }
+ segments.refreshOne(question, env, operator);
+ }
+
+ private boolean paused() {
+ return !englishDictConfig.isSegmentEnabled() || !englishDictConfig.isSegmentExecutionEnabled();
+ }
+
+ private boolean hasMatchingOutput(SegmentPayloadIdentity identity) {
+ return outputs.listByFormDataIdsAndType(List.of(identity.formDataId()),
+ OutPutDataTypeEnum.GAME.getType(), 0).stream().anyMatch(output ->
+ Objects.equals(identity.payloadHash(), output.getSegmentSourcePayloadHash())
+ && output.getUrl() != null && !output.getUrl().isBlank());
+ }
+}
--- /dev/null
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/EnglishSegmentSkeletonBizService.java
@@ -0,0 +1,35 @@
+package cn.jojo.cc.biz.service.english.segment;
+
+import cn.jojo.cc.biz.service.component.CommonService;
+import cn.jojo.cc.common.enums.DataSourceEnum;
+import cn.jojo.cc.common.enums.english.SegmentContentTypeEnum;
+import cn.jojo.cc.core.domain.english.QuestionSegmentDo;
+import cn.jojo.cc.core.domain.question.library.QuestionDo;
+import cn.jojo.cc.core.multidatasource.MultiDataSourceUtil;
+import cn.jojo.cc.core.service.question.library.QuestionCoreService;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+
+/** 首次发布不依赖 JY 已算成功:先保留 JY 身份,再显式映射 PROD ID。 */
+@Service
+public class EnglishSegmentSkeletonBizService {
+ @Resource private QuestionCoreService questions;
+ @Resource private QuestionSegmentStateTxService state;
+
+ public void ensurePair(Long teachingId, Long prodId) {
+ QuestionDo prod = MultiDataSourceUtil.switchThen(DataSourceEnum.PROD, () -> questions.getById(prodId));
+ if (prod == null) { return; }
+ Integer type = SegmentContentTypeEnum.contentTypeOfQuestionType(prod.getQuestionType());
+ if (type == null) { return; }
+ QuestionSegmentDo identity = MultiDataSourceUtil.switchThen(DataSourceEnum.DEFAULT,
+ () -> state.reserveTeachingIdentity(teachingId, type));
+ if (identity == null || identity.getId() == null) {
+ throw new IllegalStateException("缺少教研题目,禁止在 PROD 自增创建分词身份");
+ }
+ Long mappedId = CommonService.decrementAndGet(identity.getId());
+ MultiDataSourceUtil.switchThen(DataSourceEnum.PROD, () -> {
+ state.installProdSkeleton(prodId, mappedId, type);
+ return null;
+ });
+ }
+}
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentBizServiceImpl.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentBizServiceImpl.java
@@ -53,11 +53,14 @@
* 登记只会写入失败(strict)或被截断污染唯一键(non-strict)。
*/
private static final int MAX_UNMATCHED_WORD_LENGTH = 99;
@Resource
- private SegmentTextExtractor segmentTextExtractor;
+ private SegmentContentSnapshotService segmentContentSnapshotService;
+
+ @Resource
+ private QuestionSegmentTxService questionSegmentTxService;
/**
* 执行闸门:本类是全部“真算一遍”的入口(保存后异步计算、全量刷、发布后重算),
* 暂停时不得抽正文、查词典、写行。
*/
@@ -95,35 +98,11 @@
*
* 同类型内的多个文本块最终合并成一条匹配序列,不同内容位各自成行;抽取不到文字的类型
* 不会出现在结果里(不产空行,否则该题会被当成“已分词”而不再重试)。
*/
private Map> extractPlainTextsByContentType(QuestionDo question) {
- Map> grouped = new TreeMap<>();
- if (question == null || question.getId() == null) {
- return grouped;
- }
- Map> richTexts =
- segmentTextExtractor.extractRichTextsByContentType(question);
- if (richTexts == null || richTexts.isEmpty()) {
- return grouped;
- }
- for (Map.Entry> entry : richTexts.entrySet()) {
- List plainTexts = new ArrayList<>();
- for (String richText : entry.getValue()) {
- if (StringUtils.isBlank(richText)) {
- continue;
- }
- String plainText = segmentTextExtractor.extractPlainText(richText);
- if (StringUtils.isNotBlank(plainText)) {
- plainTexts.add(plainText);
- }
- }
- if (!plainTexts.isEmpty()) {
- grouped.put(entry.getKey(), plainTexts);
- }
- }
- return grouped;
+ return segmentContentSnapshotService.readTexts(question);
}
private SegmentResult match(QuestionDo question, DataBaseTypeEnum segmentSource,
PhraseIndexSnapshot snapshot, List plainTexts) {
if (question == null || question.getId() == null || segmentSource == null
@@ -200,58 +179,96 @@
}
@Override
public SegmentResult segmentAndSave(QuestionDo question, DataBaseTypeEnum segmentSource,
LoginUserBo operator) {
- return doSegmentAndSave(question, segmentSource, operator, false);
+ return runSegmentAttempt(question, segmentSource, operator, false);
}
@Override
public SegmentResult refreshOne(QuestionDo question, DataBaseTypeEnum segmentSource,
LoginUserBo operator) {
- return doSegmentAndSave(question, segmentSource, operator, true);
+ return runSegmentAttempt(question, segmentSource, operator, true);
}
@Override
public SegmentResult refreshOneAllowSkip(QuestionDo question, DataBaseTypeEnum segmentSource,
LoginUserBo operator) {
- return doSegmentAndSave(question, segmentSource, operator, false);
+ return runSegmentAttempt(question, segmentSource, operator, false);
+ }
+
+ private SegmentResult runSegmentAttempt(QuestionDo question, DataBaseTypeEnum source,
+ LoginUserBo operator, boolean force) {
+ if (question == null || question.getId() == null || !englishDictConfig.isSegmentExecutionEnabled()) {
+ return new SegmentResult();
+ }
+ try {
+ SegmentResult result = questionSegmentTxService.compute(question,
+ latest -> doSegmentAndSave(latest, source, operator, force));
+ return result == null ? new SegmentResult() : result;
+ } catch (SegmentAttemptFailure failure) {
+ // 此时compute代理已回滚并释放题锁,REQUIRES_NEW不会自锁。
+ if (failure.register) {
+ try {
+ questionSegmentTxService.recordFailure(failure.question, failure.md5,
+ failure.types, failureMessage((Exception) failure.getCause()));
+ } catch (Exception recordError) {
+ log.error("分词失败独立登记失败, questionId={}", question.getId(), recordError);
+ }
+ }
+ throw (RuntimeException) failure.getCause();
+ }
+ }
+
+ private static final class SegmentAttemptFailure extends RuntimeException {
+ private final QuestionDo question;
+ private final String md5;
+ private final Set types;
+ private final boolean register;
+ private SegmentAttemptFailure(QuestionDo question, String md5, Set types,
+ boolean register, RuntimeException cause) {
+ super(cause);
+ this.question = question;
+ this.md5 = md5;
+ this.types = new LinkedHashSet<>(types);
+ this.register = register;
+ }
}
private SegmentResult doSegmentAndSave(QuestionDo question, DataBaseTypeEnum segmentSource,
LoginUserBo operator, boolean force) {
if (question == null || question.getId() == null) {
- return new SegmentResult();
- }
- if (!englishDictConfig.isSegmentExecutionEnabled()) {
- // 急停:不抽正文、不装索引、不查词典、不写行,也不登记失败行(暂停不是失败)。
- log.warn("英文分词执行已暂停,跳过单题计算, questionId={}", question.getId());
return new SegmentResult();
}
// 抽取本身失败时按 Spec 约定登记空串 MD5,抽取成功后替换为真实正文 MD5。
String sourceTextMd5 = EMPTY_TEXT_MD5;
// 剪枝命中路径只重算返回值、库里已有同修订同正文的成功结果,不能因本次重算异常把好行降级。
boolean reuseStoredResult = false;
// 本次实际抽取出的内容位必须在 try 外可见:catch 登记失败行时要按“这一版正文属于哪些内容位”,
// 不能拿库里的旧行反推。
Map> textsByType = new TreeMap<>();
try {
- textsByType = extractPlainTextsByContentType(question);
- sourceTextMd5 = MD5Util.creatMd5(flatten(textsByType));
+ SegmentContentSnapshotService.ContentSource contentSource = segmentContentSnapshotService.readSource(question);
+ textsByType = contentSource.texts();
+ sourceTextMd5 = segmentContentSnapshotService.fingerprint(textsByType);
// 一次读出该题全部内容位行:剪枝、修订判定与拒绝分支都要按内容位各看自己那一行。
Map existingByType =
toMapByType(questionSegmentCoreService.listByQuestionId(question.getId()));
- if (!force && canSkipAll(existingByType, textsByType.keySet(), question, sourceTextMd5)) {
+ for (QuestionSegmentDo existing : existingByType.values()) {
+ if (isStale(existing, question.getUpdateTime())) { return new SegmentResult(); }
+ }
+ if (!force && canSkipAll(existingByType, textsByType.keySet(), question, sourceTextMd5)
+ && samePayloadIdentity(existingByType, contentSource.identity())) {
// 剪枝命中:库里已有同正文、同修订、同内容位集合的成功结果。直接回显库存,绝不重算——
// 重算会用“当前词典”得出与库存不一致的 segments,而本分支又不落库(词典变更靠每日全量刷传播)。
reuseStoredResult = true;
return mergeStored(existingByType);
}
// 句柄必须在扫描开始前取定:既保证“整个动作只用一张表”,也避免拿着空表白扫一轮再判能不能落库。
// 旧写法是先扫完、再用全局开关判,而开关会在扫描期间被其它线程翻转,
// 于是“用空表切出的结果”恰好能过库(错块以 NORMAL 落库后被剪枝永久固化)。
- PhraseIndexSnapshot snapshot = phraseIndexCacheService.readySnapshot(segmentSource);
+ PhraseIndexSnapshot snapshot = phraseIndexCacheService.freshSnapshotForWrite(segmentSource);
if (snapshot == null) {
// 索引未就绪(冷启动装载失败、处于退避窗口内)时短语必然全漏:落库会把缺失结果固化成 NORMAL 行,
// 并把短语的成分词写进未命中表污染自动补录,所以本次不落新结果,等下次保存重算。
// 这里有意返回空结果而不是扫描产物:不然挂接门面会拿这批“假未命中词”去投自动补录事件。
log.warn("短语索引未就绪,本次跳过分词, questionId={}, source={}",
@@ -265,10 +282,11 @@
}
Map resultsByType =
matchAllTypesByType(question, segmentSource, snapshot, textsByType);
for (SegmentResult one : resultsByType.values()) {
one.setSourceTextMd5(sourceTextMd5);
+ one.setSourceIdentity(contentSource.identity());
}
for (Map.Entry entry : resultsByType.entrySet()) {
if (!isStale(existingByType.get(entry.getKey()), question.getUpdateTime())) {
continue;
}
@@ -300,14 +318,12 @@
return merged;
} catch (Exception e) {
// 抽取/读库/匹配/落库任一阶段失败都要登记 FAILED;否则首次失败无占位行,重试扫描永远发现不了。
log.error("题目分词处理失败, questionId={}, source={}, sourceTextMd5={}",
question.getId(), segmentSource, sourceTextMd5, e);
- if (!reuseStoredResult) {
- registerFailure(question, sourceTextMd5, textsByType.keySet(), e);
- }
- throw e;
+ RuntimeException cause = e instanceof RuntimeException ? (RuntimeException) e : new IllegalStateException(e);
+ throw new SegmentAttemptFailure(question, sourceTextMd5, textsByType.keySet(), !reuseStoredResult, cause);
}
}
/**
* 逐内容位分别匹配,返回 类型 -> 该类型的匹配结果(按内容位类型升序)。
@@ -351,15 +367,11 @@
}
/**
* 正文指纹按“内容位顺序拼接”计算:与拆分前同口径,避免同题同文因内容位划分变化而反复重算。
*/
- private String flatten(Map> textsByType) {
- List all = new ArrayList<>();
- textsByType.values().forEach(all::addAll);
- return String.join("\n", all);
- }
+
private Map toMapByType(List rows) {
Map map = new TreeMap<>();
if (rows == null) {
return map;
@@ -376,10 +388,22 @@
/**
* 整体剪枝:库里现存内容位集合与本次抽取的类型集完全一致,且每一行都是同正文同修订的成功行。
* 集合不等(如新增了内容位)时必须重算,不能因“已有行可跳过”而漏掉新内容位。
*/
+ private boolean samePayloadIdentity(Map rows, SegmentPayloadIdentity identity) {
+ if (identity == null || !identity.known()) { return true; } // 无新凭据不伪造身份;RPC仍不采信旧缺凭据行。
+ for (QuestionSegmentDo row : rows.values()) {
+ SegmentSnapshot stored = JSON.parseObject(row.getSegments(), SegmentSnapshot.class);
+ if (stored == null || !Objects.equals(identity.payloadHash(), stored.getSourcePayloadHash())
+ || !Objects.equals(identity.formDataId(), stored.getSourceFormDataId())
+ || !Objects.equals(identity.contentConfigId(), stored.getSourceContentConfigId())
+ || !Objects.equals(identity.outputPayloadHash(), stored.getSourceOutputPayloadHash())) { return false; }
+ }
+ return true;
+ }
+
private boolean canSkipAll(Map existingByType, Set incomingTypes,
QuestionDo question, String sourceTextMd5) {
if (existingByType.isEmpty() || !existingByType.keySet().equals(incomingTypes)) {
return false;
}
@@ -428,11 +452,23 @@
segmentDo.setContentType(contentType);
segmentDo.setQuestionKey(question.getQuestionKey());
segmentDo.setSourceTextMd5(result.getSourceTextMd5());
segmentDo.setQuestionRevision(question.getUpdateTime());
// 落库只存文本与类型、不存词条 ID:ID 由下发时按当前词典现算(理由见 SegmentSnapshot 类注释)。
- segmentDo.setSegments(JSON.toJSONString(SegmentSnapshot.from(result.getSegments())));
+ SegmentSnapshot snapshot = SegmentSnapshot.from(result.getSegments());
+ SegmentPayloadIdentity identity = result.getSourceIdentity();
+ if (identity != null && identity.known()) {
+ snapshot.setSourcePayloadHash(identity.payloadHash());
+ snapshot.setSourceFormDataId(identity.formDataId());
+ snapshot.setSourceContentConfigId(identity.contentConfigId());
+ snapshot.setSourceOutputPayloadHash(identity.outputPayloadHash());
+ }
+ String storedJson = JSON.toJSONString(snapshot);
+ if (storedJson.length() > 3000) {
+ throw new IllegalStateException("segments超过VARCHAR3000,保留FAILED而非截断");
+ }
+ segmentDo.setSegments(storedJson);
segmentDo.setSegmentStatus(SegmentStatusEnum.NORMAL.getCode());
segmentDo.setRetryCount(0);
if (operator != null) {
// V3 审计只存操作人 ID(新表无人名列),人名仅存在于登录态/日志。
segmentDo.setCreatorId(operator.getLoginUserId());
@@ -474,35 +510,11 @@
}
/**
* 登记失败行;登记自身异常只记日志,不吞掉也不覆盖原始异常。
*/
- private void registerFailure(QuestionDo question, String sourceTextMd5,
- Set contentTypes, Exception cause) {
- try {
- // 本次已经抽出了内容位:逐行登记失败,成功行与失败行的内容位才能一一对应。
- // 不能再用“库里第一行”反推,旧行可能是上一版题型留下的 MATERIAL,而本次实际是 ANSWER_AREA。
- if (!contentTypes.isEmpty()) {
- for (Integer contentType : contentTypes) {
- questionSegmentCoreService.markFailedByQuestionId(question.getId(), contentType,
- question.getQuestionKey(), sourceTextMd5, question.getUpdateTime(),
- failureMessage(cause));
- }
- return;
- }
- // 抽取阶段就失败,拿不到内容位:按题型兜底。兜底映射与抽取器共用同一份规则;
- // 题型也不认识时沿用历史默认 MATERIAL,至少留下一条可重试的失败行。
- Integer fallbackType = SegmentContentTypeEnum.contentTypeOfQuestionType(question.getQuestionType());
- questionSegmentCoreService.markFailedByQuestionId(question.getId(),
- fallbackType == null ? SegmentContentTypeEnum.MATERIAL.getCode() : fallbackType,
- question.getQuestionKey(), sourceTextMd5, question.getUpdateTime(),
- failureMessage(cause));
- } catch (Exception e) {
- log.error("题目分词失败登记失败, questionId={}, sourceTextMd5={}",
- question.getId(), sourceTextMd5, e);
- }
- }
+
/**
* 异常可能无 message(如 NPE),统一用简单类型名保证 last_error 可诊断且不出现两种前缀口径。
*/
private String failureMessage(Exception cause) {
--- a/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentSaveHookService.java
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentSaveHookService.java
@@ -37,10 +37,13 @@
@Service
@Slf4j
public class QuestionSegmentSaveHookService {
@Resource
+ private QuestionSegmentStateTxService questionSegmentStateTxService;
+
+ @Resource
private EnglishDictConfig englishDictConfig;
@Autowired
private AliyunRocketMQService rocketMqProducerService;
@@ -60,11 +63,11 @@
return;
}
if (question == null || question.getId() == null) {
return;
}
- // 发布本事件本身不开事务、不写库;监听器内部兜住全部异常,
+ // 事务内只登记事件;派生表失败不能污染主事务rollback-only。监听器内部兜住全部异常,
// 不会再有带 @Transactional 的 bean 抛异常污染本次同步事务。
applicationEventPublisher.publishEvent(new QuestionSegmentSaveRequestedEvent(this, question, operator));
}
/**
@@ -79,10 +82,16 @@
if (event == null || event.getQuestion() == null || event.getQuestion().getId() == null) {
return;
}
QuestionDo question = event.getQuestion();
try {
+ questionSegmentStateTxService.invalidateAfterSave(question.getId());
+ } catch (Exception e) {
+ // 主事务已经提交;下发正文指纹门仍阻止旧序列,继续投递计算消息修复。
+ log.warn("提交后分词失信失败,保留题目主体并继续投递计算, questionId={}", question.getId(), e);
+ }
+ try {
EnglishSegmentSyncComputeMessage message = new EnglishSegmentSyncComputeMessage();
message.setQuestionId(question.getId());
message.setOperatorId(event.getOperator() == null ? null : event.getOperator().getLoginUserId());
message.setRequestTime(System.currentTimeMillis());
rocketMqProducerService.sendNAsync(
--- /dev/null
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentStateTxService.java
@@ -0,0 +1,54 @@
+package cn.jojo.cc.biz.service.english.segment;
+
+import cn.jojo.cc.common.enums.english.SegmentContentTypeEnum;
+import cn.jojo.cc.core.domain.english.QuestionSegmentDo;
+import cn.jojo.cc.core.domain.question.library.QuestionDo;
+import cn.jojo.cc.core.service.english.QuestionSegmentCoreService;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.annotation.Propagation;
+
+/** 同库正文锁与派生状态事务;调用方必须先选定数据源。 */
+@Service
+public class QuestionSegmentStateTxService {
+ @Resource private QuestionSegmentCoreService segments;
+
+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
+ public void invalidateAfterSave(Long id) {
+ QuestionDo current = segments.lockQuestion(id);
+ if (current == null) { return; }
+ Integer type = SegmentContentTypeEnum.contentTypeOfQuestionType(current.getQuestionType());
+ if (type != null) {
+ segments.reserveIdentity(id, current.getQuestionKey(), type, current.getUpdateTime());
+ }
+ segments.invalidateQuestion(id, current.getUpdateTime());
+ }
+
+ @Transactional(rollbackFor = Exception.class)
+ public QuestionSegmentDo reserveTeachingIdentity(Long id, Integer type) {
+ QuestionDo current = segments.lockQuestion(id);
+ if (current == null) { return null; }
+ return segments.reserveIdentity(id, current.getQuestionKey(), type, current.getUpdateTime());
+ }
+
+ @Transactional(rollbackFor = Exception.class)
+ public void installProdSkeleton(Long prodId, Long mappedId, Integer type) {
+ QuestionDo current = segments.lockQuestion(prodId);
+ if (current == null) { return; }
+ QuestionSegmentDo row = new QuestionSegmentDo();
+ row.setId(mappedId);
+ row.setQuestionId(prodId);
+ row.setQuestionKey(current.getQuestionKey());
+ row.setContentType(type);
+ row.setQuestionRevision(current.getUpdateTime());
+ segments.insertSkeletonIfAbsent(row);
+ // 重新发布后正文/修订可能改变;先降级,消息迟到前不下发旧 NORMAL。
+ for (QuestionSegmentDo old : segments.listByQuestionId(prodId)) {
+ if (!java.util.Objects.equals(old.getQuestionRevision(), current.getUpdateTime())) {
+ segments.invalidateQuestion(prodId, current.getUpdateTime());
+ break;
+ }
+ }
+ }
+}
--- /dev/null
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/QuestionSegmentTxService.java
@@ -0,0 +1,47 @@
+package cn.jojo.cc.biz.service.english.segment;
+
+import cn.jojo.cc.biz.utils.MD5Util;
+import cn.jojo.cc.common.enums.english.SegmentContentTypeEnum;
+import cn.jojo.cc.core.domain.question.library.QuestionDo;
+import cn.jojo.cc.core.service.english.QuestionSegmentCoreService;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import java.util.Set;
+import java.util.Objects;
+import java.util.function.Function;
+
+/** 独立单库事务:代理退出后先回滚/释放题锁,调用者才允许记录失败。 */
+@Service
+public class QuestionSegmentTxService {
+ @Resource private QuestionSegmentCoreService segments;
+ @Resource private SegmentContentSnapshotService content;
+
+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
+ public T compute(QuestionDo expected, Function action) {
+ QuestionDo current = segments.lockQuestion(expected.getId());
+ if (current == null || !Objects.equals(current.getUpdateTime(), expected.getUpdateTime())) { return null; }
+ return action.apply(current);
+ }
+
+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
+ public void recordFailure(QuestionDo attempted, String md5, Set types, String error) {
+ QuestionDo current = segments.lockQuestion(attempted.getId());
+ if (current == null || !Objects.equals(current.getUpdateTime(), attempted.getUpdateTime())) { return; }
+ // 同审计时间的正文也可能变化:可抽取时用同一正文指纹再拒绝过时失败登记。
+ if (!Objects.equals(md5, MD5Util.creatMd5(""))) {
+ String currentMd5 = content.fingerprint(content.readTexts(current));
+ if (!Objects.equals(md5, currentMd5)) { return; }
+ }
+ Set targets = types;
+ if (targets.isEmpty()) {
+ Integer fallback = SegmentContentTypeEnum.contentTypeOfQuestionType(current.getQuestionType());
+ targets = Set.of(fallback == null ? SegmentContentTypeEnum.MATERIAL.getCode() : fallback);
+ }
+ for (Integer type : targets) {
+ segments.markFailedByQuestionId(current.getId(), type, current.getQuestionKey(), md5,
+ current.getUpdateTime(), error);
+ }
+ }
+}
--- /dev/null
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/SegmentContentSnapshotService.java
@@ -0,0 +1,47 @@
+package cn.jojo.cc.biz.service.english.segment;
+
+import cn.jojo.cc.biz.utils.MD5Util;
+import cn.jojo.cc.core.domain.question.library.QuestionDo;
+import jakarta.annotation.Resource;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Service;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+/** 计算、失败恢复和下发共用的正文/内容位指纹;不拿审计时间代替内容版本。 */
+@Service
+public class SegmentContentSnapshotService {
+ @Resource private SegmentTextExtractor segmentTextExtractor;
+
+ public record ContentSource(Map> texts, SegmentPayloadIdentity identity) { }
+
+ public Map> readTexts(QuestionDo question) { return readSource(question).texts(); }
+
+ public ContentSource readSource(QuestionDo question) {
+ Map> grouped = new TreeMap<>();
+ if (question == null || question.getId() == null) { return new ContentSource(grouped, null); }
+ SegmentTextExtractor.RichTextSource source = segmentTextExtractor.readRichTextSource(question);
+ // 兼容历史自定义抽取器;没有来源凭据时RPC仍fail-closed。
+ Map> richTexts = source == null
+ ? segmentTextExtractor.extractRichTextsByContentType(question) : source.richTexts();
+ if (richTexts == null) { return new ContentSource(grouped, null); }
+ for (Map.Entry> entry : richTexts.entrySet()) {
+ List plain = new ArrayList<>();
+ for (String rich : entry.getValue()) {
+ if (StringUtils.isBlank(rich)) { continue; }
+ String text = segmentTextExtractor.extractPlainText(rich);
+ if (StringUtils.isNotBlank(text)) { plain.add(text); }
+ }
+ if (!plain.isEmpty()) { grouped.put(entry.getKey(), plain); }
+ }
+ return new ContentSource(grouped, source == null ? null : source.identity());
+ }
+
+ public String fingerprint(Map> grouped) {
+ List all = new ArrayList<>();
+ new TreeMap<>(grouped).values().forEach(all::addAll);
+ return MD5Util.creatMd5(String.join("\n", all));
+ }
+}
--- /dev/null
+++ b/application/src/main/java/cn/jojo/cc/biz/service/english/segment/SegmentPayloadIdentity.java
@@ -0,0 +1,40 @@
+package cn.jojo.cc.biz.service.english.segment;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import cn.jojo.cc.biz.utils.MD5Util;
+import java.util.*;
+
+/** 原始FormData生成输入的v1稳定身份;ID独立比较,不把别库ID混入正文hash。 */
+public record SegmentPayloadIdentity(String payloadHash, Long formDataId, Long contentConfigId, String outputPayloadHash) {
+ public SegmentPayloadIdentity(String hash, Long formId, Long configId) { this(hash, formId, configId, null); }
+ public boolean known() { return inputKnown() && outputPayloadHash != null
+ && outputPayloadHash.matches("output-v1:[a-f0-9]{32}"); }
+ public boolean inputKnown() {
+ return payloadHash != null && payloadHash.matches("form-v1:[a-f0-9]{32}")
+ && formDataId != null && contentConfigId != null;
+ }
+ public static String hash(JSON formData) {
+ if (formData == null) { return null; }
+ return "form-v1:" + MD5Util.creatMd5(JSON.toJSONString(canonical(formData),
+ SerializerFeature.MapSortField, SerializerFeature.WriteMapNullValue));
+ }
+ public static String outputHash(JSON output) {
+ if (output == null) { return null; }
+ return "output-v1:" + MD5Util.creatMd5(JSON.toJSONString(canonical(output),
+ SerializerFeature.MapSortField, SerializerFeature.WriteMapNullValue));
+ }
+ private static Object canonical(Object value) {
+ if (value instanceof Map, ?> map) {
+ Map result = new TreeMap<>();
+ map.forEach((k, v) -> result.put(String.valueOf(k), canonical(v)));
+ return result;
+ }
+ if (value instanceof Iterable> list) {
+ List