src/youtube/models.py

   1 import io
   2 from os import unlink
   3 from tempfile import NamedTemporaryFile
   4 from django.db import models
   5 from django.utils.translation import gettext_lazy as _
   6 from django.template import Template, Context
   7 import requests
   8 from apiclient import youtube_call
   9 from .utils import (
  10     concat_audio,
  11     concat_videos,
  12     cut_video,
  13     get_duration,
  14     get_framerate,
  15     mux,
  16     standardize_audio,
  17     standardize_video,
  18     video_from_image,
  19 )
  20 from .thumbnail import create_thumbnail
  21
  22
  23 YOUTUBE_TITLE_LIMIT = 100
  24
  25
  26 class YouTube(models.Model):
  27     name = models.CharField(max_length=255)
  28     title_template = models.CharField(max_length=1024, blank=True)
  29     description_template = models.TextField(blank=True)
  30     category = models.IntegerField(null=True, blank=True, choices=[
  31         (27, 'Edukacja'),
  32     ])
  33     intro_flac = models.FileField(upload_to='youtube/intro_flac', blank=True)
  34     outro_flac = models.FileField(upload_to='youtube/outro_flac', blank=True)
  35     loop_card = models.FileField(upload_to='youtube/card', blank=True)
  36     loop_video = models.FileField(upload_to='youtube/loop_video', blank=True)
  37     privacy_status = models.CharField(max_length=16, choices=[
  38         ('public', _('public')),
  39         ('unlisted', _('unlisted')),
  40         ('private', _('private')),
  41     ])
  42     genres = models.CharField(max_length=2048, blank=True)
  43
  44     class Meta:
  45         verbose_name = _("YouTube configuration")
  46         verbose_name_plural = _("YouTube configurations")
  47
  48     def __str__(self):
  49         return self.name
  50
  51     def get_context(self, audiobook):
  52         return Context(dict(
  53             audiobook=audiobook,
  54         ))
  55
  56     def get_description(self, audiobook):
  57         return Template(self.description_template).render(self.get_context(audiobook))
  58
  59     def get_title(self, audiobook):
  60         return Template(self.title_template).render(self.get_context(audiobook))[:YOUTUBE_TITLE_LIMIT]
  61
  62     def get_data(self, audiobook):
  63         return dict(
  64             snippet=dict(
  65                 title=self.get_title(audiobook),
  66                 description=self.get_description(audiobook),
  67                 categoryId=self.category,
  68                 defaultLanguage='pl',
  69                 defaultAudioLanguage='pl',
  70             ),
  71             status=dict(
  72                 privacyStatus=self.privacy_status,
  73             ),
  74         )
  75
  76     def publish(self, audiobook, path):
  77         data = self.get_data(audiobook)
  78         part = ",".join(data.keys())
  79
  80         response = youtube_call(
  81             "POST",
  82             "https://www.googleapis.com/upload/youtube/v3/videos",
  83             params={'part': part},
  84             json=data,
  85             resumable_file_path=path,
  86         )
  87         data = response.json()
  88         audiobook.youtube_id = data['id']
  89         audiobook.save(update_fields=['youtube_id'])
  90
  91         self.update_thumbnail(audiobook)
  92         return response
  93
  94     def update_data(self, audiobook):
  95         data = self.get_data(audiobook)
  96         data['id'] = audiobook.youtube_id
  97         part = ",".join(data.keys())
  98         youtube_call(
  99             "PUT",
 100             "https://www.googleapis.com/youtube/v3/videos",
 101             params={"part": part},
 102             json=data
 103         )
 104
 105     def prepare_file(self, input_paths, output_path=None):
 106         audio = self.prepare_audio(input_paths)
 107         duration = self.get_duration(input_paths)
 108         video = self.prepare_video(duration)
 109         output = mux([video, audio], output_path=output_path)
 110         unlink(audio)
 111         unlink(video)
 112         return output
 113
 114     def get_duration(self, input_paths):
 115         d = 0
 116         for input_path in input_paths:
 117             d += get_duration(input_path)
 118         if self.intro_flac:
 119             d += get_duration(self.intro_flac.path)
 120         if self.outro_flac:
 121             d += get_duration(self.outro_flac.path)
 122         return d
 123
 124     def prepare_audio(self, input_paths):
 125         files = []
 126         if self.intro_flac:
 127             files.append(standardize_audio(self.intro_flac.path))
 128         for input_path in input_paths:
 129             files.append(standardize_audio(input_path, cache=False))
 130         if self.outro_flac:
 131             files.append(standardize_audio(self.outro_flac.path))
 132         output = concat_audio(files)
 133         for d in files:
 134             unlink(d)
 135         return output
 136
 137     def prepare_video(self, duration):
 138         concat = []
 139         outro = []
 140         delete = []
 141
 142         if self.loop_video:
 143             fps = get_framerate(self.loop_video.path)
 144             loop_video = standardize_video(self.loop_video.path)
 145         else:
 146             fps = 25
 147
 148         loop_duration = duration
 149         for card in self.card_set.filter(duration__gt=0):
 150             loop_duration -= card.duration
 151             card_video = video_from_image(
 152                 card.image.path, card.duration, fps=fps
 153             )
 154             (concat if card.order < 0 else outro).append(card_video)
 155             delete.append(card_video)
 156
 157         if self.loop_video:
 158             loop_video_duration = get_duration(loop_video)
 159             times_loop = int(loop_duration // loop_video_duration)
 160
 161             leftover_duration = loop_duration % loop_video_duration
 162             leftover = cut_video(loop_video, leftover_duration)
 163             concat.extend([loop_video] * times_loop + [leftover])
 164             delete.append(leftover)
 165         else:
 166             leftover = video_from_image(self.loop_card.path, loop_duration)
 167             concat.append(video_from_image(self.loop_card.path, loop_duration, fps=fps))
 168             delete.append(leftover)
 169         concat.extend(outro)
 170
 171         output = concat_videos(concat)
 172         for p in delete:
 173             unlink(p)
 174         unlink(loop_video)
 175         return output
 176
 177     # tags
 178     # license
 179     # selfDeclaredMadeForKids
 180
 181     def update_thumbnail(self, audiobook):
 182         thumbnail = self.prepare_thumbnail(audiobook)
 183         if thumbnail is not None:
 184             response = youtube_call(
 185                 "POST",
 186                 "https://www.googleapis.com/upload/youtube/v3/thumbnails/set",
 187                 params={'videoId': audiobook.youtube_id},
 188                 data=thumbnail.getvalue(),
 189             )
 190
 191     def prepare_thumbnail(self, audiobook):
 192         for thumbnail_template in ThumbnailTemplate.objects.filter(is_active=True).order_by('order'):
 193             if not thumbnail_template.is_for_audiobook(audiobook):
 194                 continue
 195             thumbnail = thumbnail_template.generate(audiobook)
 196             if thumbnail is not None:
 197                 return thumbnail
 198
 199
 200 class Card(models.Model):
 201     youtube = models.ForeignKey(YouTube, models.CASCADE)
 202     order = models.SmallIntegerField()
 203     image = models.FileField(upload_to='youtube/card')
 204     duration = models.FloatField()
 205
 206     class Meta:
 207         ordering = ('order', )
 208
 209
 210 class Font(models.Model):
 211     name = models.CharField(max_length=255, unique=True)
 212     truetype = models.FileField(upload_to='youtube/font')
 213
 214     def __str__(self):
 215         return self.name
 216
 217
 218 class ThumbnailTemplate(models.Model):
 219     youtube = models.ForeignKey(YouTube, models.CASCADE)
 220     order = models.SmallIntegerField()
 221     is_active = models.BooleanField()
 222     background = models.FileField(upload_to='youtube/thumbnail')
 223     definition = models.TextField()
 224     authors = models.CharField(max_length=255, blank=True)
 225     epochs = models.CharField(max_length=255, blank=True)
 226     kinds = models.CharField(max_length=255, blank=True)
 227     genres = models.CharField(max_length=255, blank=True)
 228     collections = models.CharField(max_length=255, blank=True)
 229
 230     class Meta:
 231         ordering = ('order', )
 232
 233     def generate(self, audiobook):
 234         try:
 235             img = create_thumbnail(
 236                 self.background.path,
 237                 self.definition,
 238                 {
 239                     "author": ', '.join((a['name'] for a in audiobook.book['authors'])),
 240                     "title": audiobook.book['title'],
 241                     "part": (audiobook.youtube_volume or audiobook.part_name).strip(),
 242                 },
 243                 lambda name: Font.objects.get(name=name).truetype.path
 244             )
 245         except Exception as e:
 246             print(e)
 247             return
 248         else:
 249             buf = io.BytesIO()
 250             img.save(buf, format='PNG')
 251             return buf
 252
 253     def is_for_audiobook(self, audiobook):
 254         for category in 'authors', 'epochs', 'kinds', 'genres':
 255             if getattr(self, category):
 256                 book_slugs = set([g['slug'] for g in audiobook.book[category]])
 257                 template_slugs = set([g.strip() for g in getattr(self, category).split(',')])
 258                 if not book_slugs.intersection(template_slugs):
 259                     return False
 260
 261         if self.collections:
 262             template_collections = set([g.strip() for g in self.collections.split(',')])
 263             in_any = False
 264             for collection in template_collections:
 265                 apidata = requests.get(
 266                     f'https://wolnelektury.pl/api/collections/{collection}/'
 267                 ).json()
 268                 for book in apidata['books']:
 269                     if book['slug'] == audiobook.slug:
 270                         in_any = True
 271                         break
 272                 if in_any:
 273                     break
 274             if not in_any:
 275                 return False
 276
 277         return True