jianghaifeng/FruitServer
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(media): 修正转码任务认领与媒资 id 校验

Browse Source
This commit is contained in:
Zhou Weidong committed 2026-10-22 15:27:51 +08:00
1 parent ad84704d28
commit 5552caa6c1
8 files changed
+200 -101
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/EpisodeVideoSourcePort.java
+2 -2
View File
@@ -9,8 +9,8 @@ import java.util.Optional;
*/
public interface EpisodeVideoSourcePort {
/** 按候选 id 查找分集；候选值可能是分集 id 或 video_asset_id。 */
Optional<EpisodeVideoSource> resolve(Long candidateId);
/** 按视频播放地址查找分集，用于把未回填 video_asset_id 的媒资绑定到分集。 */
Optional<EpisodeVideoSource> resolveByVideoUrl(String videoUrl);
/** 回填分集的 video_asset_id。 */
void link(Long episodeId, Long mediaAssetId);
fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java
+7 -77
View File
@@ -21,9 +21,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Set;
import java.util.UUID;
@@ -37,8 +34,6 @@ public class MediaService {
private static final String STATUS_FAILED = "failed";
private static final String TASK_UPLOADING = "uploading";
private static final String TASK_COMPLETED = "completed";
private static final String BIZ_TYPE_EPISODE = "EPISODE";
private static final String STORAGE_TYPE_LOCAL = "LOCAL";
private static final Set<String> IMAGE_EXTENSIONS =
Set.of(".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp", ".avif");
@@ -210,9 +205,6 @@ public class MediaService {
@Transactional(rollbackFor = Exception.class)
public TranscodeTriggerRespVO triggerTranscode(Long id) {
MediaAsset asset = lockAsset(id);
if (asset == null) {
asset = resolveLegacyAsset(id);
}
if (asset == null) {
throw new ServiceException(ErrorCode.DATA_NOT_FOUND, "媒资不存在");
}
@@ -226,74 +218,6 @@ public class MediaService {
.last("for update"));
}
private MediaAsset resolveLegacyAsset(Long candidateId) {
return episodeVideoSourcePort.resolve(candidateId)
.map(this::resolveOrRegisterAsset)
.orElse(null);
}
private MediaAsset resolveOrRegisterAsset(EpisodeVideoSourcePort.EpisodeVideoSource source) {
String videoUrl = source.videoUrl();
if (videoUrl == null || videoUrl.isBlank()) {
return null;
}
String objectKey = storageService.objectKeyOf(videoUrl);
if (objectKey == null || objectKey.isBlank()) {
return null;
}
MediaAsset asset = mediaAssetMapper.selectOne(Wrappers.<MediaAsset>lambdaQuery()
.and(wrapper -> wrapper.eq(MediaAsset::getSourceUrl, videoUrl)
.or().eq(MediaAsset::getObjectKey, objectKey))
.last("limit 1"));
if (asset == null) {
asset = registerLegacyAsset(source, objectKey);
}
if (asset != null && source.episodeId() != null) {
episodeVideoSourcePort.link(source.episodeId(), asset.getId());
}
return asset;
}
private MediaAsset registerLegacyAsset(
EpisodeVideoSourcePort.EpisodeVideoSource source, String objectKey) {
Path file;
try {
file = storageService.fileFor(objectKey);
} catch (ServiceException ex) {
return null;
}
if (!Files.isRegularFile(file)) {
return null;
}
MediaAsset asset = new MediaAsset();
asset.setBizType(BIZ_TYPE_EPISODE);
asset.setFileName(file.getFileName().toString());
asset.setObjectKey(objectKey);
asset.setSourceUrl(source.videoUrl());
asset.setSizeBytes(sizeOf(file));
asset.setMd5("");
asset.setDurationSec(source.durationSec() == null ? 0 : source.durationSec());
asset.setWidth(0);
asset.setHeight(0);
asset.setStatus(STATUS_UPLOADED);
asset.setHlsUrl("");
asset.setMp4Url(source.videoUrl());
asset.setPosterUrl("");
asset.setTranscodeStatus(TranscodeService.STATUS_QUEUED);
asset.setStorageType(STORAGE_TYPE_LOCAL);
asset.setCreateBy(0L);
mediaAssetMapper.insert(asset);
return asset;
}
private long sizeOf(Path file) {
try {
return Files.size(file);
} catch (IOException ex) {
return 0L;
}
}
private TranscodeTriggerRespVO triggerAsset(MediaAsset asset) {
if (isImage(asset.getFileName())) {
throw new ServiceException(ErrorCode.BAD_REQUEST, "图片媒资无需转码");
@@ -302,7 +226,13 @@ public class MediaService {
if (sourceUrl == null || sourceUrl.isBlank()) {
sourceUrl = asset.getMp4Url();
}
TranscodeTriggerRespVO result = transcodeService.trigger(asset.getId(), sourceUrl);
EpisodeVideoSourcePort.EpisodeVideoSource episode = episodeVideoSourcePort
.resolveByVideoUrl(sourceUrl)
.orElseThrow(() -> new ServiceException(
ErrorCode.BAD_REQUEST, "媒资未绑定分集视频"));
episodeVideoSourcePort.link(episode.episodeId(), asset.getId());
TranscodeTriggerRespVO result = transcodeService.trigger(
asset.getId(), sourceUrl, episode.dramaId(), episode.episodeId());
asset.setStatus(STATUS_TRANSCODING);
asset.setTranscodeStatus(result.getStatus());
mediaAssetMapper.updateById(asset);
fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java
+67 -7
View File
@@ -23,6 +23,8 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
* 转码任务编排。
@@ -42,15 +44,17 @@ public class TranscodeService {
public static final String STATUS_FAILED = "FAILED";
/** 失败重试上限（`backend-design-phase1.md` §7.2：失败重试 3 次并记录原因）。 */
public static final int MAX_RETRY = 3;
public static final int STALE_RUNNING_MINUTES = 10;
public static final int STALE_RUNNING_MINUTES = 5;
private static final int MAX_RUNNING_PROGRESS = 99;
private static final int CLAIM_SKIP_LOG_LIMIT = 20;
private static final String STALE_RUNNING_REASON =
"任务 RUNNING 超过 10 分钟无进度更新，已自动标记失败，请重新转码";
"任务 RUNNING 超过 5 分钟无进度更新，已自动标记失败，请重新转码";
private final TranscodeTaskMapper transcodeTaskMapper;
private final MediaRenditionMapper mediaRenditionMapper;
private final MediaThumbnailMapper mediaThumbnailMapper;
private final ThumbnailRecordService thumbnailRecordService;
private final Set<Long> loggedSkipTaskIds = ConcurrentHashMap.newKeySet();
/** 上传完成后入队；同一媒资重复入队时复用未完成的任务，避免重复转码。 */
@Transactional(rollbackFor = Exception.class)
@@ -82,20 +86,27 @@ public class TranscodeService {
*/
@Transactional(rollbackFor = Exception.class)
public TranscodeTriggerRespVO trigger(Long mediaAssetId, String sourceUrl) {
return trigger(mediaAssetId, sourceUrl, 0L, 0L);
}
/** 手动重转时同时补齐分集绑定，禁止创建 episode_id=0 的不可认领任务。 */
@Transactional(rollbackFor = Exception.class)
public TranscodeTriggerRespVO trigger(
Long mediaAssetId, String sourceUrl, Long dramaId, Long episodeId) {
TranscodeTask latest = findLatestTaskByMediaAssetId(mediaAssetId);
if (latest != null && isActive(latest.getStatus())) {
updateTaskBinding(latest, dramaId, episodeId, sourceUrl);
return response(latest.getId(), latest.getStatus());
}
if (latest != null && isRetryable(latest.getStatus())) {
resetTaskForRetry(latest, sourceUrl);
resetTaskForRetry(latest, dramaId, episodeId, sourceUrl);
return response(latest.getId(), STATUS_QUEUED);
}
Long dramaId = latest == null ? 0L : latest.getDramaId();
Long episodeId = latest == null ? 0L : latest.getEpisodeId();
String retrySource = latest == null || latest.getSourceUrl() == null
|| latest.getSourceUrl().isBlank() ? sourceUrl : latest.getSourceUrl();
Long taskId = enqueue(mediaAssetId, dramaId, episodeId, retrySource);
Long taskId = enqueue(mediaAssetId, nonPositiveToZero(dramaId),
nonPositiveToZero(episodeId), retrySource);
return response(taskId, STATUS_QUEUED);
}
@@ -139,9 +150,13 @@ public class TranscodeService {
/** worker 认领任务。并发下依赖状态条件更新，抢不到返回 null。 */
public TranscodeTask claimNext() {
logSkippedQueuedTasks();
TranscodeTask candidate = transcodeTaskMapper.selectOne(Wrappers.<TranscodeTask>lambdaQuery()
.eq(TranscodeTask::getStatus, STATUS_QUEUED)
.gt(TranscodeTask::getMediaAssetId, 0)
.gt(TranscodeTask::getEpisodeId, 0)
.isNotNull(TranscodeTask::getSourceUrl)
.ne(TranscodeTask::getSourceUrl, "")
.orderByAsc(TranscodeTask::getId)
.last("limit 1"));
if (candidate == null) {
@@ -150,12 +165,48 @@ public class TranscodeService {
int updated = transcodeTaskMapper.update(null, Wrappers.<TranscodeTask>lambdaUpdate()
.eq(TranscodeTask::getId, candidate.getId())
.eq(TranscodeTask::getStatus, STATUS_QUEUED)
.gt(TranscodeTask::getMediaAssetId, 0)
.gt(TranscodeTask::getEpisodeId, 0)
.isNotNull(TranscodeTask::getSourceUrl)
.ne(TranscodeTask::getSourceUrl, "")
.set(TranscodeTask::getStatus, STATUS_RUNNING)
.set(TranscodeTask::getStartedAt, LocalDateTime.now()));
return updated == 1 ? candidate : null;
}
private void logSkippedQueuedTasks() {
List<TranscodeTask> skipped = transcodeTaskMapper.selectList(
Wrappers.<TranscodeTask>lambdaQuery()
.eq(TranscodeTask::getStatus, STATUS_QUEUED)
.and(wrapper -> wrapper
.isNull(TranscodeTask::getMediaAssetId)
.or().le(TranscodeTask::getMediaAssetId, 0)
.or().isNull(TranscodeTask::getEpisodeId)
.or().le(TranscodeTask::getEpisodeId, 0)
.or().isNull(TranscodeTask::getSourceUrl)
.or().eq(TranscodeTask::getSourceUrl, ""))
.orderByAsc(TranscodeTask::getId)
.last("limit " + CLAIM_SKIP_LOG_LIMIT));
for (TranscodeTask task : skipped) {
if (task.getId() != null && loggedSkipTaskIds.add(task.getId())) {
log.warn("跳过转码任务 {}：{}", task.getId(), claimSkipReason(task));
}
}
}
private String claimSkipReason(TranscodeTask task) {
if (task.getMediaAssetId() == null || task.getMediaAssetId() <= 0) {
return "media_asset_id 无效";
}
if (task.getEpisodeId() == null || task.getEpisodeId() <= 0) {
return "episode_id 未绑定";
}
if (task.getSourceUrl() == null || task.getSourceUrl().isBlank()) {
return "source_url 为空";
}
return "未知原因";
}
@Transactional(rollbackFor = Exception.class)
public void succeed(Long taskId, List<MediaRendition> renditions) {
succeed(taskId, renditions, null);
@@ -347,8 +398,11 @@ public class TranscodeService {
|| STATUS_FAILED.equals(status) || "CANCELED".equals(status);
}
private void resetTaskForRetry(TranscodeTask task, String sourceUrl) {
private void resetTaskForRetry(
TranscodeTask task, Long dramaId, Long episodeId, String sourceUrl) {
boolean hasSource = sourceUrl != null && !sourceUrl.isBlank();
boolean hasDrama = dramaId != null && dramaId > 0;
boolean hasEpisode = episodeId != null && episodeId > 0;
transcodeTaskMapper.update(null, Wrappers.<TranscodeTask>lambdaUpdate()
.eq(TranscodeTask::getId, task.getId())
.in(TranscodeTask::getStatus,
@@ -359,9 +413,15 @@ public class TranscodeService {
.set(TranscodeTask::getErrorMsg, "")
.set(TranscodeTask::getStartedAt, null)
.set(TranscodeTask::getFinishedAt, null)
.set(hasDrama, TranscodeTask::getDramaId, dramaId)
.set(hasEpisode, TranscodeTask::getEpisodeId, episodeId)
.set(hasSource, TranscodeTask::getSourceUrl, sourceUrl));
}
private long nonPositiveToZero(Long value) {
return value == null || value <= 0 ? 0L : value;
}
private TranscodeTriggerRespVO response(Long taskId, String status) {
return new TranscodeTriggerRespVO(taskId, status);
}
fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java
+7 -2
View File
@@ -60,8 +60,13 @@ public class FfmpegTranscodeRunner implements TranscodeRunner {
for (int index = 0; index < request.ladder().size(); index++) {
int start = index * TRANSCODE_PROGRESS_END / totalRungs;
int end = (index + 1) * TRANSCODE_PROGRESS_END / totalRungs;
renditions.add(transcodeRung(
request, request.ladder().get(index), source, durationSec, start, end));
LadderRung rung = request.ladder().get(index);
ProducedRendition produced = transcodeRung(
request, rung, source, durationSec, start, end);
renditions.add(produced);
log.info("档位完成 task={} quality={} 实际={}x{} 进度={}%",
request.taskId(), rung.quality(),
produced.width(), produced.height(), end);
}
reportProgress(request.progressListener(), POSTER_PROGRESS_START);
writeMasterPlaylist(request, renditions);
fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java
+2
View File
@@ -75,6 +75,8 @@ public class TranscodeWorker {
if (task == null) {
return false;
}
log.info("开始处理转码任务 task={} asset={} episode={}",
task.getId(), task.getMediaAssetId(), task.getEpisodeId());
process(task);
return true;
}