server/lib/job-queue/handlers/video-import.ts

   1 import * as Bull from 'bull'
   2 import { logger } from '../../../helpers/logger'
   3 import { downloadYoutubeDLVideo } from '../../../helpers/youtube-dl'
   4 import { VideoImportModel } from '../../../models/video/video-import'
   5 import { VideoImportState } from '../../../../shared/models/videos'
   6 import { getDurationFromVideoFile, getVideoFileFPS, getVideoFileResolution } from '../../../helpers/ffmpeg-utils'
   7 import { extname, join } from 'path'
   8 import { VideoFileModel } from '../../../models/video/video-file'
   9 import { renamePromise, statPromise, unlinkPromise } from '../../../helpers/core-utils'
  10 import { CONFIG, sequelizeTypescript } from '../../../initializers'
  11 import { doRequestAndSaveToFile } from '../../../helpers/requests'
  12 import { VideoState } from '../../../../shared'
  13 import { JobQueue } from '../index'
  14 import { federateVideoIfNeeded } from '../../activitypub'
  15
  16 export type VideoImportPayload = {
  17   type: 'youtube-dl'
  18   videoImportId: number
  19   thumbnailUrl: string
  20   downloadThumbnail: boolean
  21   downloadPreview: boolean
  22 }
  23
  24 async function processVideoImport (job: Bull.Job) {
  25   const payload = job.data as VideoImportPayload
  26   logger.info('Processing video import in job %d.', job.id)
  27
  28   const videoImport = await VideoImportModel.loadAndPopulateVideo(payload.videoImportId)
  29   if (!videoImport) throw new Error('Cannot import video %s: the video import entry does not exist anymore.')
  30
  31   let tempVideoPath: string
  32   try {
  33     // Download video from youtubeDL
  34     tempVideoPath = await downloadYoutubeDLVideo(videoImport.targetUrl)
  35
  36     // Get information about this video
  37     const { videoFileResolution } = await getVideoFileResolution(tempVideoPath)
  38     const fps = await getVideoFileFPS(tempVideoPath + 's')
  39     const stats = await statPromise(tempVideoPath)
  40     const duration = await getDurationFromVideoFile(tempVideoPath)
  41
  42     // Create video file object in database
  43     const videoFileData = {
  44       extname: extname(tempVideoPath),
  45       resolution: videoFileResolution,
  46       size: stats.size,
  47       fps,
  48       videoId: videoImport.videoId
  49     }
  50     const videoFile = new VideoFileModel(videoFileData)
  51
  52     // Move file
  53     const destination = join(CONFIG.STORAGE.VIDEOS_DIR, videoImport.Video.getVideoFilename(videoFile))
  54     await renamePromise(tempVideoPath, destination)
  55
  56     // Process thumbnail
  57     if (payload.downloadThumbnail) {
  58       if (payload.thumbnailUrl) {
  59         const destThumbnailPath = join(CONFIG.STORAGE.THUMBNAILS_DIR, videoImport.Video.getThumbnailName())
  60         await doRequestAndSaveToFile({ method: 'GET', uri: payload.thumbnailUrl }, destThumbnailPath)
  61       } else {
  62         await videoImport.Video.createThumbnail(videoFile)
  63       }
  64     }
  65
  66     // Process preview
  67     if (payload.downloadPreview) {
  68       if (payload.thumbnailUrl) {
  69         const destPreviewPath = join(CONFIG.STORAGE.PREVIEWS_DIR, videoImport.Video.getPreviewName())
  70         await doRequestAndSaveToFile({ method: 'GET', uri: payload.thumbnailUrl }, destPreviewPath)
  71       } else {
  72         await videoImport.Video.createPreview(videoFile)
  73       }
  74     }
  75
  76     // Create torrent
  77     await videoImport.Video.createTorrentAndSetInfoHash(videoFile)
  78
  79     const videoImportUpdated: VideoImportModel = await sequelizeTypescript.transaction(async t => {
  80       await videoFile.save({ transaction: t })
  81
  82       // Update video DB object
  83       videoImport.Video.duration = duration
  84       videoImport.Video.state = CONFIG.TRANSCODING.ENABLED ? VideoState.TO_TRANSCODE : VideoState.PUBLISHED
  85       const videoUpdated = await videoImport.Video.save({ transaction: t })
  86
  87       // Now we can federate the video
  88       await federateVideoIfNeeded(videoImport.Video, true, t)
  89
  90       // Update video import object
  91       videoImport.state = VideoImportState.SUCCESS
  92       const videoImportUpdated = await videoImport.save({ transaction: t })
  93
  94       logger.info('Video %s imported.', videoImport.targetUrl)
  95
  96       videoImportUpdated.Video = videoUpdated
  97       return videoImportUpdated
  98     })
  99
 100     // Create transcoding jobs?
 101     if (videoImportUpdated.Video.state === VideoState.TO_TRANSCODE) {
 102       // Put uuid because we don't have id auto incremented for now
 103       const dataInput = {
 104         videoUUID: videoImportUpdated.Video.uuid,
 105         isNewVideo: true
 106       }
 107
 108       await JobQueue.Instance.createJob({ type: 'video-file', payload: dataInput })
 109     }
 110
 111   } catch (err) {
 112     try {
 113       if (tempVideoPath) await unlinkPromise(tempVideoPath)
 114     } catch (errUnlink) {
 115       logger.error('Cannot cleanup files after a video import error.', { err: errUnlink })
 116     }
 117
 118     videoImport.state = VideoImportState.FAILED
 119     await videoImport.save()
 120
 121     throw err
 122   }
 123 }
 124
 125 // ---------------------------------------------------------------------------
 126
 127 export {
 128   processVideoImport
 129 }