src/education/models.py

   1 import json
   2 from django.conf import settings
   3 from django.db import models
   4 from requests_oauthlib import OAuth2Session
   5
   6
   7 YOUTUBE_SCOPE = [
   8     'https://www.googleapis.com/auth/youtube',
   9 ]
  10 YOUTUBE_AUTH_URL = 'https://accounts.google.com/o/oauth2/auth'
  11 YOUTUBE_TOKEN_URL = 'https://oauth2.googleapis.com/token'
  12
  13
  14 class Course(models.Model):
  15     slug = models.SlugField()
  16     title = models.CharField(max_length=255)
  17
  18     def __str__(self):
  19         return self.slug
  20
  21
  22 class Track(models.Model):
  23     course = models.ForeignKey(Course, models.CASCADE)
  24
  25
  26 class Tag(models.Model):
  27     name = models.CharField(max_length=255)
  28
  29
  30 class Item(models.Model):
  31     course = models.ForeignKey(Course, models.CASCADE)
  32     title = models.CharField(max_length=255, blank=True)
  33     tags = models.ManyToManyField(Tag, blank=True)
  34     youtube_id = models.CharField(max_length=255, blank=True)
  35     order = models.IntegerField(default=0, blank=True)
  36
  37     class Meta:
  38         ordering = ('order',)
  39
  40     def __str__(self):
  41         return self.title
  42
  43
  44 class YPlaylist(models.Model):
  45     course = models.ForeignKey(Course, models.CASCADE)
  46     youtube_id = models.CharField(max_length=255, blank=True)
  47
  48     def save(self):
  49         super().save()
  50         self.download()
  51
  52     def download(self):
  53         response = YouTubeToken.objects.first().call(
  54             "GET",
  55             "https://www.googleapis.com/youtube/v3/playlistItems",
  56             params={
  57                 'part': 'snippet',
  58                 'playlistId': self.youtube_id,
  59                 'maxResults': 50,
  60             },
  61         )
  62         data = response.json()
  63         for item in data['items']:
  64             self.course.item_set.update_or_create(
  65                 youtube_id=item['snippet']['resourceId']['videoId'],
  66                 defaults={
  67                     'title': item['snippet']['title'],
  68                     'order': item['snippet']['position'],
  69                 }
  70             )
  71
  72
  73
  74 class YouTubeToken(models.Model):
  75     token = models.TextField()
  76
  77     def token_updater(self, token):
  78         self.token = json.dumps(token)
  79         self.save()
  80
  81     def get_session(self):
  82         return OAuth2Session(
  83             client_id=settings.YOUTUBE_CLIENT_ID,
  84             auto_refresh_url=YOUTUBE_TOKEN_URL,
  85             token=json.loads(self.token),
  86             auto_refresh_kwargs={'client_id':settings.YOUTUBE_CLIENT_ID,'client_secret':settings.YOUTUBE_CLIENT_SECRET},
  87             token_updater=self.token_updater
  88         )
  89
  90     def call(self, method, url, params=None, json=None, data=None, resumable_file_path=None):
  91         params = params or {}
  92         if resumable_file_path:
  93             params['uploadType'] = 'resumable'
  94             file_size = os.stat(resumable_file_path).st_size
  95
  96         session = self.get_session()
  97         response = session.request(
  98             method=method,
  99             url=url,
 100             json=json,
 101             data=data,
 102             params=params,
 103             headers={
 104                 'X-Upload-Content-Length': str(file_size),
 105                 'x-upload-content-type': 'application/octet-stream',
 106             } if resumable_file_path else {}
 107         )
 108         response.raise_for_status()
 109         return response
 110