feat(media): 新增缺失转码产物自动补齐
This commit is contained in:
Zhou Weidong authored and trae-solo-agent committed
1 parent
df2da25833
commit
40728e685b
10 files changed
+510
-2
No files matched your search
+151
@@ -0,0 +1,151 @@
|
||||
// Author: fruit.com.cn
|
||||
|
||||
package com.fruitshorts.worker;
|
||||
|
||||
import com.fruitshorts.content.dal.entity.DramaEpisode;
|
||||
import com.fruitshorts.content.dal.mapper.DramaEpisodeMapper;
|
||||
import com.fruitshorts.media.config.MediaTranscodeProfile;
|
||||
import com.fruitshorts.media.config.TranscodeProperties;
|
||||
import com.fruitshorts.media.dal.entity.MediaAsset;
|
||||
import com.fruitshorts.media.dal.entity.TranscodeTask;
|
||||
import com.fruitshorts.media.dal.mapper.MediaAssetMapper;
|
||||
import com.fruitshorts.media.service.MediaTranscodeOutputService;
|
||||
import com.fruitshorts.media.service.MediaTranscodeProfileService;
|
||||
import com.fruitshorts.media.service.TranscodeService;
|
||||
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
/**
|
||||
* 转码补齐巡检：定时找出"有片源但缺少码率产物或缩略图轴"的分集，自动补入转码队列。
|
||||
*
|
||||
* <p>与 worker 同进程、同开关（`worker-enabled`）：在线入口不跑巡检，避免与播放链路抢资源。
|
||||
* 已排队/执行中的分集不重复补队；最近一次任务已 FAILED 的分集不自动重试，
|
||||
* 交给运营在后台处理，避免坏片源反复占用转码资源。
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
@ConditionalOnProperty(name = "fruitshorts.media.transcode.worker-enabled", havingValue = "true")
|
||||
public class TranscodeBackfillScanner {
|
||||
|
||||
private static final int MIN_BATCH_SIZE = 1;
|
||||
private static final int MAX_BATCH_SIZE = 100;
|
||||
private static final long CURSOR_START = 0L;
|
||||
|
||||
private final DramaEpisodeMapper episodeMapper;
|
||||
private final TranscodeService transcodeService;
|
||||
private final MediaAssetMapper mediaAssetMapper;
|
||||
private final MediaTranscodeOutputService outputService;
|
||||
private final MediaTranscodeProfileService profileService;
|
||||
private final TranscodeProperties properties;
|
||||
|
||||
/** 游标：记录上一批处理到的分集 ID，处理完一轮后回到起点。 */
|
||||
private final AtomicLong cursor = new AtomicLong(CURSOR_START);
|
||||
|
||||
/** 只对缺少媒资的分集告警一次，避免每轮巡检刷日志。 */
|
||||
private final Set<Long> loggedMissingAssetEpisodeIds = ConcurrentHashMap.newKeySet();
|
||||
|
||||
@Scheduled(fixedDelayString = "${fruitshorts.media.transcode.backfill-delay-ms:300000}")
|
||||
public void scan() {
|
||||
if (!properties.isBackfillEnabled()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
int triggered = scanOnce();
|
||||
if (triggered > 0) {
|
||||
log.info("转码补齐巡检：本轮补入 {} 个分集", triggered);
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
// 巡检异常不能让调度线程终止，否则后续分集再也补不上
|
||||
log.error("转码补齐巡检异常：{}", ex.getMessage(), ex);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 扫描一批分集并补队。
|
||||
*
|
||||
* @return 本轮实际入队的分集数
|
||||
*/
|
||||
public int scanOnce() {
|
||||
int limit = batchSize();
|
||||
List<DramaEpisode> candidates = episodeMapper.selectTranscodeBackfillCandidates(
|
||||
cursor.get(), limit, framesRequired());
|
||||
if (candidates == null || candidates.isEmpty()) {
|
||||
cursor.set(CURSOR_START);
|
||||
return 0;
|
||||
}
|
||||
cursor.set(candidates.size() < limit
|
||||
? CURSOR_START : candidates.get(candidates.size() - 1).getId());
|
||||
int triggered = 0;
|
||||
for (DramaEpisode episode : candidates) {
|
||||
if (triggerIfNeeded(episode)) {
|
||||
triggered++;
|
||||
}
|
||||
}
|
||||
return triggered;
|
||||
}
|
||||
|
||||
private boolean triggerIfNeeded(DramaEpisode episode) {
|
||||
Long mediaAssetId = resolveMediaAssetId(episode);
|
||||
if (mediaAssetId == null) {
|
||||
warnMissingAsset(episode.getId());
|
||||
return false;
|
||||
}
|
||||
TranscodeTask latest = transcodeService.findLatestTaskByMediaAssetId(mediaAssetId);
|
||||
if (latest != null && isInProgress(latest.getStatus())) {
|
||||
return false;
|
||||
}
|
||||
if (latest != null && TranscodeService.STATUS_FAILED.equals(latest.getStatus())) {
|
||||
return false;
|
||||
}
|
||||
MediaAsset asset = mediaAssetMapper.selectById(mediaAssetId);
|
||||
if (asset == null) {
|
||||
warnMissingAsset(episode.getId());
|
||||
return false;
|
||||
}
|
||||
TranscodeTriggerRespVO result = transcodeService.trigger(
|
||||
mediaAssetId, asset.getSourceUrl(), episode.getDramaId(), episode.getId());
|
||||
log.info("分集 {} 缺少码率产物或缩略图，已自动入队 task={}",
|
||||
episode.getId(), result == null ? null : result.getTaskId());
|
||||
return true;
|
||||
}
|
||||
|
||||
/** 优先用分集已绑定的媒资，老数据没有绑定则按视频地址反查。 */
|
||||
private Long resolveMediaAssetId(DramaEpisode episode) {
|
||||
Long videoAssetId = episode.getVideoAssetId();
|
||||
if (videoAssetId != null && videoAssetId > 0) {
|
||||
return videoAssetId;
|
||||
}
|
||||
return outputService.resolveMediaAssetId(episode.getVideoUrl());
|
||||
}
|
||||
|
||||
private boolean isInProgress(String status) {
|
||||
return TranscodeService.STATUS_QUEUED.equals(status)
|
||||
|| TranscodeService.STATUS_RUNNING.equals(status);
|
||||
}
|
||||
|
||||
/** 关闭雪碧图时不会产出缩略图，此时只看码率产物，避免每轮误判重复转码。 */
|
||||
private boolean framesRequired() {
|
||||
MediaTranscodeProfile profile = profileService.currentProfile();
|
||||
return profile != null && profile.getSprite() != null && profile.getSprite().isEnabled();
|
||||
}
|
||||
|
||||
private int batchSize() {
|
||||
return Math.max(MIN_BATCH_SIZE, Math.min(properties.getBackfillBatchSize(), MAX_BATCH_SIZE));
|
||||
}
|
||||
|
||||
private void warnMissingAsset(Long episodeId) {
|
||||
if (episodeId != null && loggedMissingAssetEpisodeIds.add(episodeId)) {
|
||||
log.warn("分集 {} 找不到对应媒资，跳过自动转码", episodeId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -42,6 +42,10 @@ fruitshorts:
|
||||
# 只有本进程打开转码（§8.5 隔离要求）
|
||||
worker-enabled: true
|
||||
poll-delay-ms: 3000
|
||||
# 自动巡检缺失码率产物/缩略图的分集并补入队
|
||||
backfill-enabled: true
|
||||
backfill-delay-ms: 300000
|
||||
backfill-batch-size: 20
|
||||
ffmpeg-path: ${FFMPEG_PATH:ffmpeg}
|
||||
ffprobe-path: ${FFPROBE_PATH:ffprobe}
|
||||
segment-seconds: 6
|
||||
|
||||
+143
@@ -0,0 +1,143 @@
|
||||
// Author: fruit.com.cn
|
||||
|
||||
package com.fruitshorts.worker;
|
||||
|
||||
import com.fruitshorts.content.dal.entity.DramaEpisode;
|
||||
import com.fruitshorts.content.dal.mapper.DramaEpisodeMapper;
|
||||
import com.fruitshorts.media.config.MediaTranscodeProfile;
|
||||
import com.fruitshorts.media.config.TranscodeProperties;
|
||||
import com.fruitshorts.media.dal.entity.MediaAsset;
|
||||
import com.fruitshorts.media.dal.entity.TranscodeTask;
|
||||
import com.fruitshorts.media.dal.mapper.MediaAssetMapper;
|
||||
import com.fruitshorts.media.service.MediaTranscodeOutputService;
|
||||
import com.fruitshorts.media.service.MediaTranscodeProfileService;
|
||||
import com.fruitshorts.media.service.TranscodeService;
|
||||
import com.fruitshorts.media.vo.TranscodeTriggerRespVO;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyBoolean;
|
||||
import static org.mockito.ArgumentMatchers.anyInt;
|
||||
import static org.mockito.ArgumentMatchers.anyLong;
|
||||
import static org.mockito.BDDMockito.given;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
/**
|
||||
* 缺失转码产物/缩略图的分集自动补队。
|
||||
*/
|
||||
class TranscodeBackfillScannerTest {
|
||||
|
||||
private static final long EPISODE_ID = 1L;
|
||||
private static final long DRAMA_ID = 2L;
|
||||
private static final long ASSET_ID = 9L;
|
||||
private static final long RESOLVED_ASSET_ID = 77L;
|
||||
private static final long TASK_ID = 55L;
|
||||
private static final String SOURCE_URL = "/files/episode/video.mp4";
|
||||
private static final String VIDEO_URL = "/files/episode/video.mp4";
|
||||
|
||||
private DramaEpisodeMapper episodeMapper;
|
||||
private TranscodeService transcodeService;
|
||||
private MediaAssetMapper mediaAssetMapper;
|
||||
private MediaTranscodeOutputService outputService;
|
||||
private MediaTranscodeProfileService profileService;
|
||||
private TranscodeBackfillScanner scanner;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
episodeMapper = mock(DramaEpisodeMapper.class);
|
||||
transcodeService = mock(TranscodeService.class);
|
||||
mediaAssetMapper = mock(MediaAssetMapper.class);
|
||||
outputService = mock(MediaTranscodeOutputService.class);
|
||||
profileService = mock(MediaTranscodeProfileService.class);
|
||||
TranscodeProperties properties = new TranscodeProperties();
|
||||
given(profileService.currentProfile()).willReturn(new MediaTranscodeProfile());
|
||||
scanner = new TranscodeBackfillScanner(episodeMapper, transcodeService,
|
||||
mediaAssetMapper, outputService, profileService, properties);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_enqueue_task_when_episode_missing_rendition() {
|
||||
givenCandidates(episode(ASSET_ID));
|
||||
given(transcodeService.findLatestTaskByMediaAssetId(ASSET_ID)).willReturn(null);
|
||||
given(mediaAssetMapper.selectById(ASSET_ID)).willReturn(asset(ASSET_ID));
|
||||
given(transcodeService.trigger(ASSET_ID, SOURCE_URL, DRAMA_ID, EPISODE_ID))
|
||||
.willReturn(new TranscodeTriggerRespVO(TASK_ID, TranscodeService.STATUS_QUEUED));
|
||||
|
||||
int triggered = scanner.scanOnce();
|
||||
|
||||
assertThat(triggered).isEqualTo(1);
|
||||
verify(transcodeService).trigger(ASSET_ID, SOURCE_URL, DRAMA_ID, EPISODE_ID);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_resolve_asset_by_video_url_when_episode_has_no_asset_id() {
|
||||
givenCandidates(episode(null));
|
||||
given(outputService.resolveMediaAssetId(VIDEO_URL)).willReturn(RESOLVED_ASSET_ID);
|
||||
given(transcodeService.findLatestTaskByMediaAssetId(RESOLVED_ASSET_ID)).willReturn(null);
|
||||
given(mediaAssetMapper.selectById(RESOLVED_ASSET_ID)).willReturn(asset(RESOLVED_ASSET_ID));
|
||||
given(transcodeService.trigger(RESOLVED_ASSET_ID, SOURCE_URL, DRAMA_ID, EPISODE_ID))
|
||||
.willReturn(new TranscodeTriggerRespVO(TASK_ID, TranscodeService.STATUS_QUEUED));
|
||||
|
||||
int triggered = scanner.scanOnce();
|
||||
|
||||
assertThat(triggered).isEqualTo(1);
|
||||
verify(transcodeService).trigger(RESOLVED_ASSET_ID, SOURCE_URL, DRAMA_ID, EPISODE_ID);
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_skip_when_latest_task_failed() {
|
||||
givenCandidates(episode(ASSET_ID));
|
||||
given(transcodeService.findLatestTaskByMediaAssetId(ASSET_ID))
|
||||
.willReturn(task(TranscodeService.STATUS_FAILED));
|
||||
|
||||
int triggered = scanner.scanOnce();
|
||||
|
||||
assertThat(triggered).isZero();
|
||||
verify(transcodeService, never()).trigger(anyLong(), any(), anyLong(), anyLong());
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_skip_when_task_still_running() {
|
||||
givenCandidates(episode(ASSET_ID));
|
||||
given(transcodeService.findLatestTaskByMediaAssetId(ASSET_ID))
|
||||
.willReturn(task(TranscodeService.STATUS_RUNNING));
|
||||
|
||||
int triggered = scanner.scanOnce();
|
||||
|
||||
assertThat(triggered).isZero();
|
||||
verify(transcodeService, never()).trigger(anyLong(), any(), anyLong(), anyLong());
|
||||
}
|
||||
|
||||
private void givenCandidates(DramaEpisode episode) {
|
||||
given(episodeMapper.selectTranscodeBackfillCandidates(anyLong(), anyInt(), anyBoolean()))
|
||||
.willReturn(List.of(episode));
|
||||
}
|
||||
|
||||
private DramaEpisode episode(Long videoAssetId) {
|
||||
DramaEpisode episode = new DramaEpisode();
|
||||
episode.setId(EPISODE_ID);
|
||||
episode.setDramaId(DRAMA_ID);
|
||||
episode.setVideoAssetId(videoAssetId);
|
||||
episode.setVideoUrl(VIDEO_URL);
|
||||
return episode;
|
||||
}
|
||||
|
||||
private MediaAsset asset(Long id) {
|
||||
MediaAsset asset = new MediaAsset();
|
||||
asset.setId(id);
|
||||
asset.setSourceUrl(SOURCE_URL);
|
||||
return asset;
|
||||
}
|
||||
|
||||
private TranscodeTask task(String status) {
|
||||
TranscodeTask task = new TranscodeTask();
|
||||
task.setStatus(status);
|
||||
return task;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user