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, page_token=None):
  53         params = {
  54                 'part': 'snippet',
  55                 'playlistId': self.youtube_id,
  56                 'maxResults': 50,
  57             }
  58         if page_token:
  59             params['pageToken'] = page_token
  60         response = YouTubeToken.objects.first().call(
  61             "GET",
  62             "https://www.googleapis.com/youtube/v3/playlistItems",
  63             params=params
  64         )
  65         data = response.json()
  66         for item in data['items']:
  67             self.course.item_set.update_or_create(
  68                 youtube_id=item['snippet']['resourceId']['videoId'],
  69                 defaults={
  70                     'title': item['snippet']['title'],
  71                     'order': item['snippet']['position'],
  72                 }
  73             )
  74         if data.get('nextPageToken'):
  75             self.download(page_token=data['nextPageToken'])
  76
  77
  78
  79 class YouTubeToken(models.Model):
  80     token = models.TextField()
  81
  82     def token_updater(self, token):
  83         self.token = json.dumps(token)
  84         self.save()
  85
  86     def get_session(self):
  87         return OAuth2Session(
  88             client_id=settings.YOUTUBE_CLIENT_ID,
  89             auto_refresh_url=YOUTUBE_TOKEN_URL,
  90             token=json.loads(self.token),
  91             auto_refresh_kwargs={'client_id':settings.YOUTUBE_CLIENT_ID,'client_secret':settings.YOUTUBE_CLIENT_SECRET},
  92             token_updater=self.token_updater
  93         )
  94
  95     def call(self, method, url, params=None, json=None, data=None, resumable_file_path=None):
  96         params = params or {}
  97         if resumable_file_path:
  98             params['uploadType'] = 'resumable'
  99             file_size = os.stat(resumable_file_path).st_size
 100
 101         session = self.get_session()
 102         response = session.request(
 103             method=method,
 104             url=url,
 105             json=json,
 106             data=data,
 107             params=params,
 108             headers={
 109                 'X-Upload-Content-Length': str(file_size),
 110                 'x-upload-content-type': 'application/octet-stream',
 111             } if resumable_file_path else {}
 112         )
 113         response.raise_for_status()
 114         return response
 115