@@ -0,0 +1,785 @@
package com.hotwj.platform.system.maintenance.service.impl ;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper ;
import com.baomidou.mybatisplus.core.toolkit.Wrappers ;
import com.hotwj.platform.config.mediaResource.domain.HotMediaResource ;
import com.hotwj.platform.config.mediaResource.mapper.HotMediaResourceMapper ;
import com.hotwj.platform.system.maintenance.config.MediaResourceMetadataRepairProperties ;
import com.hotwj.platform.system.maintenance.domain.bo.MediaResourceMetadataRepairRequest ;
import com.hotwj.platform.system.maintenance.domain.vo.MediaResourceMetadataRepairSubmitVo ;
import com.hotwj.platform.system.maintenance.domain.vo.MediaResourceMetadataRepairTaskStatusVo ;
import com.hotwj.platform.system.maintenance.service.IMediaResourceMetadataRepairService ;
import lombok.RequiredArgsConstructor ;
import lombok.extern.slf4j.Slf4j ;
import org.dromara.common.core.exception.ServiceException ;
import org.dromara.common.core.service.OssService ;
import org.dromara.common.core.utils.StringUtils ;
import org.dromara.common.redis.utils.RedisUtils ;
import org.dromara.common.satoken.utils.LoginHelper ;
import org.springframework.stereotype.Service ;
import java.io.IOException ;
import java.io.InputStream ;
import java.io.Serial ;
import java.net.URI ;
import java.net.URISyntaxException ;
import java.net.http.HttpClient ;
import java.net.http.HttpRequest ;
import java.net.http.HttpResponse ;
import java.nio.charset.StandardCharsets ;
import java.nio.file.DirectoryStream ;
import java.nio.file.Files ;
import java.nio.file.Path ;
import java.time.Duration ;
import java.util.ArrayList ;
import java.util.List ;
import java.util.Locale ;
import java.util.UUID ;
import java.util.concurrent.ScheduledExecutorService ;
import java.util.regex.Matcher ;
import java.util.regex.Pattern ;
/**
* 媒体资源元数据修复服务实现
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class MediaResourceMetadataRepairServiceImpl implements IMediaResourceMetadataRepairService {
private static final String STATUS_PENDING = " PENDING " ;
private static final String STATUS_RUNNING = " RUNNING " ;
private static final String STATUS_SUCCESS = " SUCCESS " ;
private static final String STATUS_PARTIAL_SUCCESS = " PARTIAL_SUCCESS " ;
private static final String STATUS_FAILED = " FAILED " ;
private static final Pattern PURE_NUMBER = Pattern . compile ( " ^ \\ d+$ " ) ;
private static final Pattern KEY_URI_PATTERN = Pattern . compile ( " URI= \" ([^ \" ]+) \" " ) ;
private static final Pattern FFMPEG_DURATION_PATTERN = Pattern . compile ( " Duration: ( \\ d{2}):( \\ d{2}):( \\ d{2}(?: \\ . \\ d+)?) " ) ;
private final HotMediaResourceMapper hotMediaResourceMapper ;
private final OssService ossService ;
private final ScheduledExecutorService scheduledExecutorService ;
private final MediaResourceMetadataRepairProperties properties ;
@Override
public MediaResourceMetadataRepairSubmitVo submitRepairTask ( MediaResourceMetadataRepairRequest request ) {
MediaResourceMetadataRepairRequest normalized = normalizeRequest ( request ) ;
String taskId = UUID . randomUUID ( ) . toString ( ) . replace ( " - " , " " ) ;
MediaResourceMetadataRepairTaskStatusVo taskStatus = new MediaResourceMetadataRepairTaskStatusVo ( ) ;
taskStatus . setTaskId ( taskId ) ;
taskStatus . setStatus ( STATUS_PENDING ) ;
taskStatus . setMessage ( " 任务已提交,等待执行 " ) ;
taskStatus . setOperatorName ( resolveOperatorName ( ) ) ;
taskStatus . setRequestSummary ( buildRequestSummary ( normalized ) ) ;
saveTask ( taskStatus ) ;
scheduledExecutorService . execute ( ( ) - > runTask ( taskId , normalized ) ) ;
return new MediaResourceMetadataRepairSubmitVo ( taskId , Boolean . TRUE , " 媒体资源元数据修复任务已提交 " ) ;
}
@Override
public MediaResourceMetadataRepairTaskStatusVo getTaskStatus ( String taskId ) {
MediaResourceMetadataRepairTaskStatusVo taskStatus = loadTask ( taskId ) ;
if ( taskStatus = = null ) {
throw new ServiceException ( " 任务不存在或已过期 " ) ;
}
return taskStatus ;
}
private void runTask ( String taskId , MediaResourceMetadataRepairRequest request ) {
MediaResourceMetadataRepairTaskStatusVo taskStatus = requireTask ( taskId ) ;
long taskStart = System . currentTimeMillis ( ) ;
Path taskWorkDir = getTaskWorkDir ( taskId ) ;
try {
taskStatus . setStatus ( STATUS_RUNNING ) ;
taskStatus . setStartTime ( taskStart ) ;
taskStatus . setMessage ( " 任务执行中 " ) ;
long total = countMatchingResources ( request ) ;
taskStatus . setTotal ( ( int ) Math . min ( total , request . getLimit ( ) ) ) ;
saveTask ( taskStatus ) ;
if ( taskStatus . getTotal ( ) < = 0 ) {
taskStatus . setStatus ( STATUS_SUCCESS ) ;
taskStatus . setEndTime ( System . currentTimeMillis ( ) ) ;
taskStatus . setDurationMillis ( taskStatus . getEndTime ( ) - taskStart ) ;
taskStatus . setMessage ( " 未找到符合条件的媒体资源 " ) ;
saveTask ( taskStatus ) ;
return ;
}
long lastId = 0L ;
while ( taskStatus . getProcessedCount ( ) < taskStatus . getTotal ( ) ) {
int fetchSize = Math . min ( request . getBatchSize ( ) , taskStatus . getTotal ( ) - taskStatus . getProcessedCount ( ) ) ;
List < HotMediaResource > resources = fetchResources ( request , lastId , fetchSize ) ;
if ( resources . isEmpty ( ) ) {
break ;
}
for ( HotMediaResource resource : resources ) {
lastId = resource . getId ( ) ;
try {
ProcessResult result = processResource ( taskId , request , resource ) ;
if ( result . type = = ProcessResultType . SUCCESS ) {
taskStatus . setSuccessCount ( taskStatus . getSuccessCount ( ) + 1 ) ;
taskStatus . getUpdatedIds ( ) . add ( resource . getId ( ) ) ;
} else if ( result . type = = ProcessResultType . SKIP ) {
taskStatus . setSkipCount ( taskStatus . getSkipCount ( ) + 1 ) ;
} else {
recordFailure ( taskStatus , resource , result . resolvedUrl , result . message ) ;
}
} catch ( Exception ex ) {
log . error ( " 媒体资源元数据修复失败, resourceId={} " , resource . getId ( ) , ex ) ;
recordFailure ( taskStatus , resource , null , abbreviate ( ex . getMessage ( ) , 300 ) ) ;
} finally {
taskStatus . setProcessedCount ( taskStatus . getProcessedCount ( ) + 1 ) ;
saveTask ( taskStatus ) ;
}
if ( taskStatus . getProcessedCount ( ) > = taskStatus . getTotal ( ) ) {
break ;
}
}
}
taskStatus . setEndTime ( System . currentTimeMillis ( ) ) ;
taskStatus . setDurationMillis ( taskStatus . getEndTime ( ) - taskStart ) ;
taskStatus . setStatus ( determineFinalStatus ( taskStatus ) ) ;
taskStatus . setMessage ( buildFinishMessage ( taskStatus ) ) ;
saveTask ( taskStatus ) ;
} catch ( Exception ex ) {
log . error ( " 媒体资源元数据修复任务执行失败, taskId={} " , taskId , ex ) ;
taskStatus . setStatus ( STATUS_FAILED ) ;
taskStatus . setEndTime ( System . currentTimeMillis ( ) ) ;
taskStatus . setDurationMillis ( taskStatus . getEndTime ( ) - taskStart ) ;
taskStatus . setMessage ( " 任务执行失败: " + abbreviate ( ex . getMessage ( ) , 300 ) ) ;
saveTask ( taskStatus ) ;
} finally {
deleteQuietly ( taskWorkDir ) ;
}
}
private ProcessResult processResource ( String taskId , MediaResourceMetadataRepairRequest request , HotMediaResource resource ) throws IOException , InterruptedException {
ResolvedUrl resolvedUrl = resolveDownloadUrl ( resource . getFileUrl ( ) ) ;
if ( resolvedUrl = = null | | StringUtils . isBlank ( resolvedUrl . url ) ) {
String reason = " 未解析到可用下载地址 " ;
if ( Boolean . TRUE . equals ( request . getSkipUnresolvedUrl ( ) ) ) {
return ProcessResult . skip ( reason , null ) ;
}
return ProcessResult . fail ( reason , null ) ;
}
Path resourceDir = getResourceWorkDir ( taskId , resource . getId ( ) ) ;
try {
Files . createDirectories ( resourceDir ) ;
DownloadedResource downloadedResource = resolvedUrl . hls
? downloadHlsResource ( resolvedUrl . url , resourceDir )
: downloadFileResource ( resolvedUrl . url , resourceDir ) ;
Long sizeBytes = downloadedResource . totalSizeBytes ;
Long durationSeconds = probeDuration ( downloadedResource . probePath , downloadedResource . extinfDuration ) ;
HotMediaResource update = new HotMediaResource ( ) ;
update . setId ( resource . getId ( ) ) ;
boolean hasUpdate = false ;
if ( Boolean . TRUE . equals ( request . getRepairSize ( ) ) ) {
if ( sizeBytes = = null | | sizeBytes < = 0 ) {
return ProcessResult . fail ( " 未获取到有效资源大小 " , resolvedUrl . url ) ;
}
if ( Boolean . TRUE . equals ( request . getOverwriteExisting ( ) ) | | isInvalid ( resource . getSizeBytes ( ) ) ) {
update . setSizeBytes ( sizeBytes ) ;
hasUpdate = true ;
}
}
if ( Boolean . TRUE . equals ( request . getRepairDuration ( ) ) ) {
if ( durationSeconds = = null | | durationSeconds < = 0 ) {
return ProcessResult . fail ( " 未获取到有效媒体时长 " , resolvedUrl . url ) ;
}
if ( Boolean . TRUE . equals ( request . getOverwriteExisting ( ) ) | | isInvalid ( resource . getDurationSeconds ( ) ) ) {
update . setDurationSeconds ( durationSeconds ) ;
hasUpdate = true ;
}
}
if ( ! hasUpdate ) {
return ProcessResult . skip ( " 未命中可覆盖字段 " , resolvedUrl . url ) ;
}
if ( hotMediaResourceMapper . updateById ( update ) < = 0 ) {
return ProcessResult . fail ( " 数据库更新失败 " , resolvedUrl . url ) ;
}
return ProcessResult . success ( " 修复成功 " , resolvedUrl . url ) ;
} finally {
deleteQuietly ( resourceDir ) ;
}
}
private ResolvedUrl resolveDownloadUrl ( String fileUrl ) {
String candidate = firstToken ( fileUrl ) ;
if ( StringUtils . isBlank ( candidate ) ) {
return null ;
}
String resolvedUrl = candidate ;
if ( PURE_NUMBER . matcher ( candidate ) . matches ( ) ) {
resolvedUrl = firstToken ( ossService . selectUrlByIds ( candidate ) ) ;
}
if ( StringUtils . isBlank ( resolvedUrl ) ) {
return null ;
}
URI uri = toHttpUri ( resolvedUrl ) ;
if ( ! properties . isHostAllowed ( uri . getHost ( ) ) ) {
throw new ServiceException ( " 下载地址不在允许的域名白名单中: " + uri . getHost ( ) ) ;
}
String path = StringUtils . defaultString ( uri . getPath ( ) ) . toLowerCase ( Locale . ROOT ) ;
return new ResolvedUrl ( uri . toString ( ) , path . endsWith ( " .m3u8 " ) ) ;
}
private DownloadedResource downloadFileResource ( String url , Path resourceDir ) throws IOException , InterruptedException {
URI uri = toHttpUri ( url ) ;
String extension = extensionFromPath ( uri . getPath ( ) , " .bin " ) ;
Path targetFile = resourceDir . resolve ( " resource " + extension ) ;
long size = downloadToFile ( uri , targetFile , properties . getMaxFileSizeBytes ( ) ) ;
if ( size < = 0 ) {
throw new ServiceException ( " 下载文件大小为0 " ) ;
}
return new DownloadedResource ( targetFile , size , null ) ;
}
private DownloadedResource downloadHlsResource ( String url , Path resourceDir ) throws IOException , InterruptedException {
HlsContext context = new HlsContext ( ) ;
HlsPlaylistResult playlistResult = localizePlaylist ( toHttpUri ( url ) , resourceDir . resolve ( " root.m3u8 " ) , context ) ;
if ( context . totalSize < = 0 ) {
throw new ServiceException ( " HLS 下载结果为空 " ) ;
}
return new DownloadedResource ( playlistResult . probePath , context . totalSize , playlistResult . extinfDuration ) ;
}
private HlsPlaylistResult localizePlaylist ( URI playlistUri , Path localPlaylistPath , HlsContext context ) throws IOException , InterruptedException {
byte [ ] contentBytes = downloadToBytes ( playlistUri , properties . getMaxPlaylistSizeBytes ( ) ) ;
context . addBytes ( contentBytes . length ) ;
Files . createDirectories ( localPlaylistPath . getParent ( ) ) ;
String content = new String ( contentBytes , StandardCharsets . UTF_8 ) ;
if ( content . contains ( " #EXT-X-STREAM-INF " ) ) {
Files . write ( localPlaylistPath , contentBytes ) ;
URI variantUri = findFirstVariantUri ( playlistUri , content ) ;
if ( variantUri = = null ) {
throw new ServiceException ( " 未找到可用的 HLS 变体播放列表 " ) ;
}
return localizePlaylist ( variantUri , localPlaylistPath . getParent ( ) . resolve ( " variant_0.m3u8 " ) , context ) ;
}
List < String > outputLines = new ArrayList < > ( ) ;
long extinfDuration = 0L ;
String [ ] lines = content . split ( " \\ r? \\ n " , - 1 ) ;
for ( String line : lines ) {
String trimmed = line . trim ( ) ;
if ( trimmed . startsWith ( " #EXTINF: " ) ) {
extinfDuration + = parseExtinfDuration ( trimmed ) ;
outputLines . add ( line ) ;
continue ;
}
if ( trimmed . startsWith ( " #EXT-X-KEY " ) | | trimmed . startsWith ( " #EXT-X-MAP " ) ) {
outputLines . add ( rewriteAttributeUriLine ( playlistUri , line , localPlaylistPath . getParent ( ) , context ) ) ;
continue ;
}
if ( trimmed . isEmpty ( ) | | trimmed . startsWith ( " # " ) ) {
outputLines . add ( line ) ;
continue ;
}
URI resourceUri = playlistUri . resolve ( trimmed ) ;
String lowerPath = StringUtils . defaultString ( resourceUri . getPath ( ) ) . toLowerCase ( Locale . ROOT ) ;
if ( lowerPath . endsWith ( " .m3u8 " ) ) {
String childName = " nested_ " + context . nextFileIndex ( ) + " .m3u8 " ;
localizePlaylist ( resourceUri , localPlaylistPath . getParent ( ) . resolve ( childName ) , context ) ;
outputLines . add ( childName ) ;
} else {
String fileName = buildHlsAssetName ( resourceUri , " segment_ " + context . nextFileIndex ( ) ) ;
Path target = localPlaylistPath . getParent ( ) . resolve ( fileName ) ;
downloadToFile ( resourceUri , target , properties . getMaxHlsTotalSizeBytes ( ) ) ;
context . addSegment ( ) ;
context . addBytes ( Files . size ( target ) ) ;
outputLines . add ( fileName ) ;
}
}
Files . writeString ( localPlaylistPath , String . join ( System . lineSeparator ( ) , outputLines ) , StandardCharsets . UTF_8 ) ;
return new HlsPlaylistResult ( localPlaylistPath , extinfDuration > 0 ? extinfDuration : null ) ;
}
private String rewriteAttributeUriLine ( URI baseUri , String line , Path localDir , HlsContext context ) throws IOException , InterruptedException {
Matcher matcher = KEY_URI_PATTERN . matcher ( line ) ;
if ( ! matcher . find ( ) ) {
return line ;
}
URI assetUri = baseUri . resolve ( matcher . group ( 1 ) ) ;
String fileName = buildHlsAssetName ( assetUri , " asset_ " + context . nextFileIndex ( ) ) ;
Path target = localDir . resolve ( fileName ) ;
long fileSize = downloadToFile ( assetUri , target , properties . getMaxHlsTotalSizeBytes ( ) ) ;
context . addBytes ( fileSize ) ;
return matcher . replaceFirst ( " URI= \" " + fileName . replace ( " \\ " , " / " ) + " \" " ) ;
}
private URI findFirstVariantUri ( URI playlistUri , String content ) {
String [ ] lines = content . split ( " \\ r? \\ n " ) ;
for ( int i = 0 ; i < lines . length ; i + + ) {
if ( ! lines [ i ] . trim ( ) . startsWith ( " #EXT-X-STREAM-INF " ) ) {
continue ;
}
for ( int j = i + 1 ; j < lines . length ; j + + ) {
String candidate = lines [ j ] . trim ( ) ;
if ( candidate . isEmpty ( ) ) {
continue ;
}
if ( candidate . startsWith ( " # " ) ) {
break ;
}
return playlistUri . resolve ( candidate ) ;
}
}
return null ;
}
private Long probeDuration ( Path probePath , Long extinfDuration ) throws IOException , InterruptedException {
Long duration = probeByFfprobe ( probePath ) ;
if ( duration ! = null & & duration > 0 ) {
return duration ;
}
duration = probeByFfmpeg ( probePath ) ;
if ( duration ! = null & & duration > 0 ) {
return duration ;
}
return extinfDuration ;
}
private Long probeByFfprobe ( Path probePath ) throws IOException , InterruptedException {
List < String > command = List . of (
properties . getFfprobeCommand ( ) ,
" -v " , " error " ,
" -show_entries " , " format=duration " ,
" -of " , " default=noprint_wrappers=1:nokey=1 " ,
probePath . toAbsolutePath ( ) . toString ( )
) ;
CommandResult commandResult = runCommand ( command , Duration . ofSeconds ( properties . getReadTimeoutSeconds ( ) ) ) ;
if ( commandResult . exitCode ! = 0 ) {
return null ;
}
String output = StringUtils . trim ( commandResult . output ) ;
if ( StringUtils . isBlank ( output ) ) {
return null ;
}
try {
double seconds = Double . parseDouble ( output ) ;
long rounded = Math . round ( seconds ) ;
return rounded > 0 ? rounded : null ;
} catch ( NumberFormatException ignored ) {
return null ;
}
}
private Long probeByFfmpeg ( Path probePath ) throws IOException , InterruptedException {
List < String > command = List . of (
properties . getFfmpegCommand ( ) ,
" -i " ,
probePath . toAbsolutePath ( ) . toString ( )
) ;
CommandResult commandResult = runCommand ( command , Duration . ofSeconds ( properties . getReadTimeoutSeconds ( ) ) ) ;
Matcher matcher = FFMPEG_DURATION_PATTERN . matcher ( commandResult . output ) ;
if ( ! matcher . find ( ) ) {
return null ;
}
int hours = Integer . parseInt ( matcher . group ( 1 ) ) ;
int minutes = Integer . parseInt ( matcher . group ( 2 ) ) ;
double seconds = Double . parseDouble ( matcher . group ( 3 ) ) ;
long duration = Math . round ( hours * 3600L + minutes * 60L + seconds ) ;
return duration > 0 ? duration : null ;
}
private CommandResult runCommand ( List < String > command , Duration timeout ) throws IOException , InterruptedException {
Process process = new ProcessBuilder ( command )
. redirectErrorStream ( true )
. start ( ) ;
boolean finished = process . waitFor ( timeout . toMillis ( ) , java . util . concurrent . TimeUnit . MILLISECONDS ) ;
if ( ! finished ) {
process . destroyForcibly ( ) ;
throw new ServiceException ( " 媒体探测命令执行超时: " + command . get ( 0 ) ) ;
}
String output ;
try ( InputStream inputStream = process . getInputStream ( ) ) {
output = new String ( inputStream . readAllBytes ( ) , StandardCharsets . UTF_8 ) ;
}
return new CommandResult ( process . exitValue ( ) , output ) ;
}
private long countMatchingResources ( MediaResourceMetadataRepairRequest request ) {
return hotMediaResourceMapper . selectCount ( buildBaseQuery ( request ) ) ;
}
private List < HotMediaResource > fetchResources ( MediaResourceMetadataRepairRequest request , long lastId , int fetchSize ) {
LambdaQueryWrapper < HotMediaResource > queryWrapper = buildBaseQuery ( request ) ;
queryWrapper . gt ( lastId > 0 , HotMediaResource : : getId , lastId ) ;
queryWrapper . orderByAsc ( HotMediaResource : : getId ) ;
queryWrapper . last ( " limit " + fetchSize ) ;
return hotMediaResourceMapper . selectList ( queryWrapper ) ;
}
private LambdaQueryWrapper < HotMediaResource > buildBaseQuery ( MediaResourceMetadataRepairRequest request ) {
LambdaQueryWrapper < HotMediaResource > queryWrapper = Wrappers . lambdaQuery ( ) ;
queryWrapper . eq ( request . getCompanyId ( ) ! = null , HotMediaResource : : getCompanyId , request . getCompanyId ( ) ) ;
queryWrapper . eq ( request . getMediaType ( ) ! = null , HotMediaResource : : getMediaType , request . getMediaType ( ) ) ;
queryWrapper . eq ( StringUtils . isNotBlank ( request . getCategoryCode ( ) ) , HotMediaResource : : getCategoryCode , request . getCategoryCode ( ) ) ;
queryWrapper . in ( request . getResourceIds ( ) ! = null & & ! request . getResourceIds ( ) . isEmpty ( ) , HotMediaResource : : getId , request . getResourceIds ( ) ) ;
if ( Boolean . TRUE . equals ( request . getOnlyInvalid ( ) ) ) {
if ( Boolean . TRUE . equals ( request . getRepairSize ( ) ) & & Boolean . TRUE . equals ( request . getRepairDuration ( ) ) ) {
queryWrapper . and ( wrapper - > wrapper
. nested ( size - > size . isNull ( HotMediaResource : : getSizeBytes ) . or ( ) . le ( HotMediaResource : : getSizeBytes , 0 ) )
. or ( )
. nested ( duration - > duration . isNull ( HotMediaResource : : getDurationSeconds ) . or ( ) . le ( HotMediaResource : : getDurationSeconds , 0 ) )
) ;
} else if ( Boolean . TRUE . equals ( request . getRepairSize ( ) ) ) {
queryWrapper . and ( wrapper - > wrapper . isNull ( HotMediaResource : : getSizeBytes ) . or ( ) . le ( HotMediaResource : : getSizeBytes , 0 ) ) ;
} else if ( Boolean . TRUE . equals ( request . getRepairDuration ( ) ) ) {
queryWrapper . and ( wrapper - > wrapper . isNull ( HotMediaResource : : getDurationSeconds ) . or ( ) . le ( HotMediaResource : : getDurationSeconds , 0 ) ) ;
}
}
return queryWrapper ;
}
private MediaResourceMetadataRepairRequest normalizeRequest ( MediaResourceMetadataRepairRequest request ) {
if ( request = = null ) {
throw new ServiceException ( " 请求参数不能为空 " ) ;
}
request . setRepairSize ( request . getRepairSize ( ) = = null ? Boolean . TRUE : request . getRepairSize ( ) ) ;
request . setRepairDuration ( request . getRepairDuration ( ) = = null ? Boolean . TRUE : request . getRepairDuration ( ) ) ;
if ( ! Boolean . TRUE . equals ( request . getRepairSize ( ) ) & & ! Boolean . TRUE . equals ( request . getRepairDuration ( ) ) ) {
throw new ServiceException ( " 至少需要开启一个修复项 " ) ;
}
request . setOverwriteExisting ( request . getOverwriteExisting ( ) = = null ? Boolean . FALSE : request . getOverwriteExisting ( ) ) ;
request . setOnlyInvalid ( request . getOnlyInvalid ( ) = = null ? Boolean . TRUE : request . getOnlyInvalid ( ) ) ;
request . setSkipUnresolvedUrl ( request . getSkipUnresolvedUrl ( ) = = null ? Boolean . TRUE : request . getSkipUnresolvedUrl ( ) ) ;
int limit = request . getLimit ( ) = = null | | request . getLimit ( ) < = 0 ? properties . getDefaultLimit ( ) : request . getLimit ( ) ;
request . setLimit ( Math . min ( limit , properties . getMaxLimit ( ) ) ) ;
int batchSize = request . getBatchSize ( ) = = null | | request . getBatchSize ( ) < = 0 ? properties . getDefaultBatchSize ( ) : request . getBatchSize ( ) ;
request . setBatchSize ( Math . min ( batchSize , properties . getMaxBatchSize ( ) ) ) ;
return request ;
}
private void saveTask ( MediaResourceMetadataRepairTaskStatusVo taskStatus ) {
RedisUtils . setCacheObject ( properties . getTaskKeyPrefix ( ) + taskStatus . getTaskId ( ) , taskStatus , properties . taskExpireDuration ( ) ) ;
}
private MediaResourceMetadataRepairTaskStatusVo loadTask ( String taskId ) {
if ( StringUtils . isBlank ( taskId ) ) {
return null ;
}
return RedisUtils . getCacheObject ( properties . getTaskKeyPrefix ( ) + taskId ) ;
}
private MediaResourceMetadataRepairTaskStatusVo requireTask ( String taskId ) {
MediaResourceMetadataRepairTaskStatusVo taskStatus = loadTask ( taskId ) ;
if ( taskStatus = = null ) {
throw new ServiceException ( " 任务不存在或已过期 " ) ;
}
return taskStatus ;
}
private void recordFailure ( MediaResourceMetadataRepairTaskStatusVo taskStatus , HotMediaResource resource , String resolvedUrl , String reason ) {
taskStatus . setFailCount ( taskStatus . getFailCount ( ) + 1 ) ;
if ( taskStatus . getErrorSamples ( ) . size ( ) > = properties . getMaxErrorSamples ( ) ) {
return ;
}
MediaResourceMetadataRepairTaskStatusVo . ErrorSample errorSample = new MediaResourceMetadataRepairTaskStatusVo . ErrorSample ( ) ;
errorSample . setResourceId ( resource . getId ( ) ) ;
errorSample . setFileUrl ( resource . getFileUrl ( ) ) ;
errorSample . setResolvedUrl ( resolvedUrl ) ;
errorSample . setReason ( reason ) ;
taskStatus . getErrorSamples ( ) . add ( errorSample ) ;
}
private String determineFinalStatus ( MediaResourceMetadataRepairTaskStatusVo taskStatus ) {
if ( taskStatus . getFailCount ( ) > 0 & & taskStatus . getSuccessCount ( ) > 0 ) {
return STATUS_PARTIAL_SUCCESS ;
}
if ( taskStatus . getFailCount ( ) > 0 & & taskStatus . getSuccessCount ( ) < = 0 ) {
return STATUS_FAILED ;
}
return STATUS_SUCCESS ;
}
private String buildFinishMessage ( MediaResourceMetadataRepairTaskStatusVo taskStatus ) {
return String . format (
Locale . ROOT ,
" 任务完成:成功 %d, 跳过 %d, 失败 %d " ,
taskStatus . getSuccessCount ( ) ,
taskStatus . getSkipCount ( ) ,
taskStatus . getFailCount ( )
) ;
}
private String buildRequestSummary ( MediaResourceMetadataRepairRequest request ) {
List < String > parts = new ArrayList < > ( ) ;
parts . add ( " mediaType= " + ( request . getMediaType ( ) = = null ? " ALL " : request . getMediaType ( ) ) ) ;
parts . add ( " categoryCode= " + StringUtils . defaultIfBlank ( request . getCategoryCode ( ) , " ALL " ) ) ;
parts . add ( " companyId= " + ( request . getCompanyId ( ) = = null ? " ALL " : request . getCompanyId ( ) ) ) ;
parts . add ( " repairSize= " + request . getRepairSize ( ) ) ;
parts . add ( " repairDuration= " + request . getRepairDuration ( ) ) ;
parts . add ( " overwriteExisting= " + request . getOverwriteExisting ( ) ) ;
parts . add ( " onlyInvalid= " + request . getOnlyInvalid ( ) ) ;
parts . add ( " limit= " + request . getLimit ( ) ) ;
parts . add ( " batchSize= " + request . getBatchSize ( ) ) ;
if ( request . getResourceIds ( ) ! = null & & ! request . getResourceIds ( ) . isEmpty ( ) ) {
parts . add ( " resourceIds= " + request . getResourceIds ( ) . size ( ) ) ;
}
return String . join ( " , " , parts ) ;
}
private String resolveOperatorName ( ) {
try {
return LoginHelper . getUsername ( ) ;
} catch ( Exception ignored ) {
return " unknown " ;
}
}
private Path getTaskWorkDir ( String taskId ) {
return Path . of ( properties . getWorkDir ( ) , taskId ) ;
}
private Path getResourceWorkDir ( String taskId , Long resourceId ) {
return getTaskWorkDir ( taskId ) . resolve ( resourceId + " _ " + System . currentTimeMillis ( ) ) ;
}
private long downloadToFile ( URI uri , Path targetFile , long maxSizeBytes ) throws IOException , InterruptedException {
HttpResponse < InputStream > response = httpClient ( ) . send ( buildRequest ( uri ) , HttpResponse . BodyHandlers . ofInputStream ( ) ) ;
if ( response . statusCode ( ) < 200 | | response . statusCode ( ) > = 300 ) {
throw new ServiceException ( " 下载失败, HTTP状态码: " + response . statusCode ( ) ) ;
}
Files . createDirectories ( targetFile . getParent ( ) ) ;
long total = 0L ;
try ( InputStream inputStream = response . body ( ) ) {
byte [ ] buffer = new byte [ 8192 ] ;
int read ;
try ( var outputStream = Files . newOutputStream ( targetFile ) ) {
while ( ( read = inputStream . read ( buffer ) ) ! = - 1 ) {
total + = read ;
if ( maxSizeBytes > 0 & & total > maxSizeBytes ) {
throw new ServiceException ( " 下载文件超过大小限制 " ) ;
}
outputStream . write ( buffer , 0 , read ) ;
}
}
}
return total ;
}
private byte [ ] downloadToBytes ( URI uri , long maxSizeBytes ) throws IOException , InterruptedException {
HttpResponse < InputStream > response = httpClient ( ) . send ( buildRequest ( uri ) , HttpResponse . BodyHandlers . ofInputStream ( ) ) ;
if ( response . statusCode ( ) < 200 | | response . statusCode ( ) > = 300 ) {
throw new ServiceException ( " 下载失败, HTTP状态码: " + response . statusCode ( ) ) ;
}
try ( InputStream inputStream = response . body ( ) ) {
byte [ ] buffer = new byte [ 8192 ] ;
int read ;
int total = 0 ;
var outputStream = new java . io . ByteArrayOutputStream ( ) ;
while ( ( read = inputStream . read ( buffer ) ) ! = - 1 ) {
total + = read ;
if ( maxSizeBytes > 0 & & total > maxSizeBytes ) {
throw new ServiceException ( " 下载内容超过大小限制 " ) ;
}
outputStream . write ( buffer , 0 , read ) ;
}
return outputStream . toByteArray ( ) ;
}
}
private HttpRequest buildRequest ( URI uri ) {
return HttpRequest . newBuilder ( uri )
. GET ( )
. timeout ( Duration . ofSeconds ( properties . getReadTimeoutSeconds ( ) ) )
. header ( " User-Agent " , " hotwj-media-resource-repair " )
. build ( ) ;
}
private HttpClient httpClient ( ) {
return HttpClient . newBuilder ( )
. followRedirects ( HttpClient . Redirect . NORMAL )
. connectTimeout ( Duration . ofSeconds ( properties . getConnectTimeoutSeconds ( ) ) )
. build ( ) ;
}
private URI toHttpUri ( String url ) {
try {
URI uri = new URI ( StringUtils . trim ( url ) ) ;
String scheme = StringUtils . defaultString ( uri . getScheme ( ) ) . toLowerCase ( Locale . ROOT ) ;
if ( ! " http " . equals ( scheme ) & & ! " https " . equals ( scheme ) ) {
throw new ServiceException ( " 仅支持 http/https 下载地址 " ) ;
}
return uri ;
} catch ( URISyntaxException ex ) {
throw new ServiceException ( " URL 不合法: " + abbreviate ( url , 200 ) ) ;
}
}
private String firstToken ( String raw ) {
if ( StringUtils . isBlank ( raw ) ) {
return null ;
}
String [ ] values = raw . split ( " [, \\ r \\ n] " ) ;
for ( String value : values ) {
if ( StringUtils . isNotBlank ( value ) ) {
return value . trim ( ) ;
}
}
return null ;
}
private boolean isInvalid ( Long value ) {
return value = = null | | value < = 0 ;
}
private String buildHlsAssetName ( URI uri , String defaultName ) {
return defaultName + extensionFromPath ( uri . getPath ( ) , " .bin " ) ;
}
private String extensionFromPath ( String path , String defaultExtension ) {
if ( StringUtils . isBlank ( path ) ) {
return defaultExtension ;
}
int index = path . lastIndexOf ( '.' ) ;
if ( index < 0 | | index = = path . length ( ) - 1 ) {
return defaultExtension ;
}
String extension = path . substring ( index ) ;
return extension . length ( ) > 12 ? defaultExtension : extension ;
}
private long parseExtinfDuration ( String line ) {
String value = line . substring ( " #EXTINF: " . length ( ) ) ;
int commaIndex = value . indexOf ( ',' ) ;
if ( commaIndex > = 0 ) {
value = value . substring ( 0 , commaIndex ) ;
}
try {
return Math . round ( Double . parseDouble ( value . trim ( ) ) ) ;
} catch ( NumberFormatException ignored ) {
return 0L ;
}
}
private String abbreviate ( String value , int maxLength ) {
if ( value = = null ) {
return " " ;
}
String normalized = value . replaceAll ( " \\ s+ " , " " ) . trim ( ) ;
if ( normalized . length ( ) < = maxLength ) {
return normalized ;
}
return normalized . substring ( 0 , Math . max ( maxLength - 3 , 0 ) ) + " ... " ;
}
private void deleteQuietly ( Path path ) {
if ( path = = null | | ! Files . exists ( path ) ) {
return ;
}
try {
if ( Files . isDirectory ( path ) ) {
try ( DirectoryStream < Path > stream = Files . newDirectoryStream ( path ) ) {
for ( Path child : stream ) {
deleteQuietly ( child ) ;
}
}
}
Files . deleteIfExists ( path ) ;
} catch ( Exception ex ) {
log . warn ( " 删除临时文件失败: {} " , path , ex ) ;
}
}
private enum ProcessResultType {
SUCCESS ,
SKIP ,
FAIL
}
private static class ProcessResult {
private final ProcessResultType type ;
private final String message ;
private final String resolvedUrl ;
private ProcessResult ( ProcessResultType type , String message , String resolvedUrl ) {
this . type = type ;
this . message = message ;
this . resolvedUrl = resolvedUrl ;
}
private static ProcessResult success ( String message , String resolvedUrl ) {
return new ProcessResult ( ProcessResultType . SUCCESS , message , resolvedUrl ) ;
}
private static ProcessResult skip ( String message , String resolvedUrl ) {
return new ProcessResult ( ProcessResultType . SKIP , message , resolvedUrl ) ;
}
private static ProcessResult fail ( String message , String resolvedUrl ) {
return new ProcessResult ( ProcessResultType . FAIL , message , resolvedUrl ) ;
}
}
private static class ResolvedUrl {
private final String url ;
private final boolean hls ;
private ResolvedUrl ( String url , boolean hls ) {
this . url = url ;
this . hls = hls ;
}
}
private static class DownloadedResource {
private final Path probePath ;
private final Long totalSizeBytes ;
private final Long extinfDuration ;
private DownloadedResource ( Path probePath , Long totalSizeBytes , Long extinfDuration ) {
this . probePath = probePath ;
this . totalSizeBytes = totalSizeBytes ;
this . extinfDuration = extinfDuration ;
}
}
private static class HlsPlaylistResult {
private final Path probePath ;
private final Long extinfDuration ;
private HlsPlaylistResult ( Path probePath , Long extinfDuration ) {
this . probePath = probePath ;
this . extinfDuration = extinfDuration ;
}
}
private static class CommandResult {
private final int exitCode ;
private final String output ;
private CommandResult ( int exitCode , String output ) {
this . exitCode = exitCode ;
this . output = output ;
}
}
private class HlsContext implements java . io . Serializable {
@Serial
private static final long serialVersionUID = 1L ;
private long totalSize = 0L ;
private int segmentCount = 0 ;
private int fileIndex = 0 ;
private void addBytes ( long size ) {
totalSize + = size ;
if ( properties . getMaxHlsTotalSizeBytes ( ) ! = null & & properties . getMaxHlsTotalSizeBytes ( ) > 0 & & totalSize > properties . getMaxHlsTotalSizeBytes ( ) ) {
throw new ServiceException ( " HLS 总下载大小超过限制 " ) ;
}
}
private void addSegment ( ) {
segmentCount + + ;
if ( properties . getMaxHlsSegments ( ) ! = null & & properties . getMaxHlsSegments ( ) > 0 & & segmentCount > properties . getMaxHlsSegments ( ) ) {
throw new ServiceException ( " HLS 分片数超过限制 " ) ;
}
}
private int nextFileIndex ( ) {
fileIndex + + ;
return fileIndex ;
}
}
}