X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Fjob-queue%2Fhandlers%2Fvideo-import.ts;h=9901b878c08e1fdeb9c1ae7a838317664d7a0035;hb=405c83f9af377a663a4c8e9ad025fd5c10496922;hp=1e5e52b58f5106d3ba99d68f1b9db7afaba99a5a;hpb=e3b4c084cd0935dd93b1d737c4fc88c65ab5bcc1;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts index 1e5e52b58..9901b878c 100644 --- a/server/lib/job-queue/handlers/video-import.ts +++ b/server/lib/job-queue/handlers/video-import.ts @@ -1,31 +1,37 @@ -import * as Bull from 'bull' +import { Job } from 'bullmq' import { move, remove, stat } from 'fs-extra' -import { extname } from 'path' import { retryTransactionWrapper } from '@server/helpers/database-utils' +import { YoutubeDLWrapper } from '@server/helpers/youtube-dl' +import { CONFIG } from '@server/initializers/config' import { isPostImportVideoAccepted } from '@server/lib/moderation' +import { generateWebTorrentVideoFilename } from '@server/lib/paths' import { Hooks } from '@server/lib/plugins/hooks' +import { ServerConfigManager } from '@server/lib/server-config-manager' import { isAbleToUploadVideo } from '@server/lib/user' -import { addOptimizeOrMergeAudioJob } from '@server/lib/video' -import { getVideoFilePath } from '@server/lib/video-paths' +import { buildMoveToObjectStorageJob, buildOptimizeOrMergeAudioJob } from '@server/lib/video' +import { VideoPathManager } from '@server/lib/video-path-manager' +import { buildNextVideoState } from '@server/lib/video-state' import { ThumbnailModel } from '@server/models/video/thumbnail' import { MVideoImportDefault, MVideoImportDefaultFiles, MVideoImportVideo } from '@server/types/models/video/video-import' +import { getLowercaseExtension } from '@shared/core-utils' +import { isAudioFile } from '@shared/extra-utils' import { + ThumbnailType, VideoImportPayload, + VideoImportPreventExceptionResult, + VideoImportState, VideoImportTorrentPayload, VideoImportTorrentPayloadType, VideoImportYoutubeDLPayload, VideoImportYoutubeDLPayloadType, + VideoResolution, VideoState -} from '../../../../shared' -import { VideoImportState } from '../../../../shared/models/videos' -import { ThumbnailType } from '../../../../shared/models/videos/thumbnail.type' -import { getDurationFromVideoFile, getVideoFileFPS, getVideoFileResolution } from '../../../helpers/ffprobe-utils' +} from '@shared/models' +import { ffprobePromise, getVideoStreamDimensionsInfo, getVideoStreamDuration, getVideoStreamFPS } from '../../../helpers/ffmpeg' import { logger } from '../../../helpers/logger' import { getSecureTorrentName } from '../../../helpers/utils' import { createTorrentAndSetInfoHash, downloadWebTorrentVideo } from '../../../helpers/webtorrent' -import { downloadYoutubeDLVideo } from '../../../helpers/youtube-dl' -import { CONFIG } from '../../../initializers/config' -import { VIDEO_IMPORT_TIMEOUT } from '../../../initializers/constants' +import { JOB_TTL } from '../../../initializers/constants' import { sequelizeTypescript } from '../../../initializers/database' import { VideoModel } from '../../../models/video/video' import { VideoFileModel } from '../../../models/video/video-file' @@ -34,12 +40,31 @@ import { MThumbnail } from '../../../types/models/video/thumbnail' import { federateVideoIfNeeded } from '../../activitypub/videos' import { Notifier } from '../../notifier' import { generateVideoMiniature } from '../../thumbnail' +import { JobQueue } from '../job-queue' -async function processVideoImport (job: Bull.Job) { +async function processVideoImport (job: Job): Promise { const payload = job.data as VideoImportPayload - if (payload.type === 'youtube-dl') return processYoutubeDLImport(job, payload) - if (payload.type === 'magnet-uri' || payload.type === 'torrent-file') return processTorrentImport(job, payload) + const videoImport = await getVideoImportOrDie(payload) + if (videoImport.state === VideoImportState.CANCELLED) { + logger.info('Do not process import since it has been cancelled', { payload }) + return { resultType: 'success' } + } + + videoImport.state = VideoImportState.PROCESSING + await videoImport.save() + + try { + if (payload.type === 'youtube-dl') await processYoutubeDLImport(job, videoImport, payload) + if (payload.type === 'magnet-uri' || payload.type === 'torrent-file') await processTorrentImport(job, videoImport, payload) + + return { resultType: 'success' } + } catch (err) { + if (!payload.preventException) throw err + + logger.warn('Catch error in video import to send value to parent job.', { payload, err }) + return { resultType: 'error' } + } } // --------------------------------------------------------------------------- @@ -50,42 +75,40 @@ export { // --------------------------------------------------------------------------- -async function processTorrentImport (job: Bull.Job, payload: VideoImportTorrentPayload) { - logger.info('Processing torrent video import in job %d.', job.id) +async function processTorrentImport (job: Job, videoImport: MVideoImportDefault, payload: VideoImportTorrentPayload) { + logger.info('Processing torrent video import in job %s.', job.id) - const videoImport = await getVideoImportOrDie(payload.videoImportId) + const options = { type: payload.type, videoImportId: payload.videoImportId } - const options = { - type: payload.type, - videoImportId: payload.videoImportId - } const target = { torrentName: videoImport.torrentName ? getSecureTorrentName(videoImport.torrentName) : undefined, - magnetUri: videoImport.magnetUri + uri: videoImport.magnetUri } - return processFile(() => downloadWebTorrentVideo(target, VIDEO_IMPORT_TIMEOUT), videoImport, options) + return processFile(() => downloadWebTorrentVideo(target, JOB_TTL['video-import']), videoImport, options) } -async function processYoutubeDLImport (job: Bull.Job, payload: VideoImportYoutubeDLPayload) { - logger.info('Processing youtubeDL video import in job %d.', job.id) +async function processYoutubeDLImport (job: Job, videoImport: MVideoImportDefault, payload: VideoImportYoutubeDLPayload) { + logger.info('Processing youtubeDL video import in job %s.', job.id) - const videoImport = await getVideoImportOrDie(payload.videoImportId) - const options = { - type: payload.type, - videoImportId: videoImport.id - } + const options = { type: payload.type, videoImportId: videoImport.id } + + const youtubeDL = new YoutubeDLWrapper( + videoImport.targetUrl, + ServerConfigManager.Instance.getEnabledResolutions('vod'), + CONFIG.TRANSCODING.ALWAYS_TRANSCODE_ORIGINAL_RESOLUTION + ) return processFile( - () => downloadYoutubeDLVideo(videoImport.targetUrl, payload.fileExt, VIDEO_IMPORT_TIMEOUT), + () => youtubeDL.downloadVideo(payload.fileExt, JOB_TTL['video-import']), videoImport, options ) } -async function getVideoImportOrDie (videoImportId: number) { - const videoImport = await VideoImportModel.loadAndPopulateVideo(videoImportId) +async function getVideoImportOrDie (payload: VideoImportPayload) { + const videoImport = await VideoImportModel.loadAndPopulateVideo(payload.videoImportId) if (!videoImport || !videoImport.Video) { - throw new Error('Cannot import video %s: the video import or video linked to this import does not exist anymore.') + throw new Error(`Cannot import video ${payload.videoImportId}: the video import or video linked to this import does not exist anymore.`) } return videoImport @@ -97,7 +120,6 @@ type ProcessFileOptions = { } async function processFile (downloader: () => Promise, videoImport: MVideoImportDefault, options: ProcessFileOptions) { let tempVideoPath: string - let videoDestFile: string let videoFile: VideoFileModel try { @@ -111,15 +133,22 @@ async function processFile (downloader: () => Promise, videoImport: MVid throw new Error('The user video quota is exceeded with this video to import.') } - const { videoFileResolution } = await getVideoFileResolution(tempVideoPath) - const fps = await getVideoFileFPS(tempVideoPath) - const duration = await getDurationFromVideoFile(tempVideoPath) + const probe = await ffprobePromise(tempVideoPath) + + const { resolution } = await isAudioFile(tempVideoPath, probe) + ? { resolution: VideoResolution.H_NOVIDEO } + : await getVideoStreamDimensionsInfo(tempVideoPath) + + const fps = await getVideoStreamFPS(tempVideoPath, probe) + const duration = await getVideoStreamDuration(tempVideoPath, probe) // Prepare video file object for creation in database + const fileExt = getLowercaseExtension(tempVideoPath) const videoFileData = { - extname: extname(tempVideoPath), - resolution: videoFileResolution, + extname: fileExt, + resolution, size: stats.size, + filename: generateWebTorrentVideoFilename(resolution, fileExt), fps, videoId: videoImport.videoId } @@ -154,7 +183,7 @@ async function processFile (downloader: () => Promise, videoImport: MVid const videoImportWithFiles: MVideoImportDefaultFiles = Object.assign(videoImport, { Video: videoWithFiles }) // Move file - videoDestFile = getVideoFilePath(videoImportWithFiles.Video, videoFile) + const videoDestFile = VideoPathManager.Instance.getFSVideoFileOutputPath(videoImportWithFiles.Video, videoFile) await move(tempVideoPath, videoDestFile) tempVideoPath = null // This path is not used anymore @@ -162,7 +191,11 @@ async function processFile (downloader: () => Promise, videoImport: MVid let thumbnailModel: MThumbnail let thumbnailSave: object if (!videoImportWithFiles.Video.getMiniature()) { - thumbnailModel = await generateVideoMiniature(videoImportWithFiles.Video, videoFile, ThumbnailType.MINIATURE) + thumbnailModel = await generateVideoMiniature({ + video: videoImportWithFiles.Video, + videoFile, + type: ThumbnailType.MINIATURE + }) thumbnailSave = thumbnailModel.toJSON() } @@ -170,7 +203,11 @@ async function processFile (downloader: () => Promise, videoImport: MVid let previewModel: MThumbnail let previewSave: object if (!videoImportWithFiles.Video.getPreview()) { - previewModel = await generateVideoMiniature(videoImportWithFiles.Video, videoFile, ThumbnailType.PREVIEW) + previewModel = await generateVideoMiniature({ + video: videoImportWithFiles.Video, + videoFile, + type: ThumbnailType.PREVIEW + }) previewSave = previewModel.toJSON() } @@ -191,14 +228,14 @@ async function processFile (downloader: () => Promise, videoImport: MVid // Update video DB object video.duration = duration - video.state = CONFIG.TRANSCODING.ENABLED ? VideoState.TO_TRANSCODE : VideoState.PUBLISHED + video.state = buildNextVideoState(video.state) await video.save({ transaction: t }) if (thumbnailModel) await video.addAndSaveThumbnail(thumbnailModel, t) if (previewModel) await video.addAndSaveThumbnail(previewModel, t) // Now we can federate the video (reload from database, we need more attributes) - const videoForFederation = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t) + const videoForFederation = await VideoModel.loadFull(video.uuid, t) await federateVideoIfNeeded(videoForFederation, true, t) // Update video import object @@ -222,7 +259,7 @@ async function processFile (downloader: () => Promise, videoImport: MVid }) }) - Notifier.Instance.notifyOnFinishedVideoImport(videoImportUpdated, true) + Notifier.Instance.notifyOnFinishedVideoImport({ videoImport: videoImportUpdated, success: true }) if (video.isBlacklisted()) { const videoBlacklist = Object.assign(video.VideoBlacklist, { Video: video }) @@ -232,9 +269,17 @@ async function processFile (downloader: () => Promise, videoImport: MVid Notifier.Instance.notifyOnNewVideoIfNeeded(video) } + if (video.state === VideoState.TO_MOVE_TO_EXTERNAL_STORAGE) { + await JobQueue.Instance.createJob( + await buildMoveToObjectStorageJob({ video: videoImportUpdated.Video, previousVideoState: VideoState.TO_IMPORT }) + ) + } + // Create transcoding jobs? if (video.state === VideoState.TO_TRANSCODE) { - await addOptimizeOrMergeAudioJob(videoImportUpdated.Video, videoFile, videoImport.User) + await JobQueue.Instance.createJob( + await buildOptimizeOrMergeAudioJob({ video: videoImportUpdated.Video, videoFile, user: videoImport.User }) + ) } } catch (err) { @@ -250,7 +295,7 @@ async function processFile (downloader: () => Promise, videoImport: MVid } await videoImport.save() - Notifier.Instance.notifyOnFinishedVideoImport(videoImport, false) + Notifier.Instance.notifyOnFinishedVideoImport({ videoImport, success: false }) throw err }