mygpo/web/logo.py

   1 import os.path
   2 import io
   3 import requests
   4 import hashlib
   5 import socket
   6 import struct
   7
   8 from PIL import Image, ImageDraw
   9
  10 from django.urls import reverse
  11 from django.conf import settings
  12 from django.http import Http404, HttpResponseRedirect
  13 from django.views import View
  14 from django.utils.decorators import method_decorator
  15 from django.views.decorators.http import last_modified
  16 from django.contrib.staticfiles.storage import staticfiles_storage
  17 from django.core.files.storage import FileSystemStorage
  18
  19 from mygpo.utils import file_hash
  20
  21 import logging
  22
  23 logger = logging.getLogger(__name__)
  24
  25 # Use Django's File Storage API to access podcast logos. This could be swapped
  26 # out for another storage implementation (eg for storing to Amazon S3)
  27 # https://docs.djangoproject.com/en/1.11/ref/files/storage/
  28 LOGO_STORAGE = FileSystemStorage(location=settings.MEDIA_ROOT)
  29
  30
  31 def _last_modified(request, size, prefix, filename):
  32
  33     target = os.path.join("logo", str(size), prefix, filename)
  34
  35     try:
  36         return LOGO_STORAGE.get_modified_time(target)
  37
  38     except (FileNotFoundError, NotImplementedError):
  39         return None
  40
  41
  42 class CoverArt(View):
  43     def __init__(self):
  44         self.storage = LOGO_STORAGE
  45
  46     @method_decorator(last_modified(_last_modified))
  47     def get(self, request, size, prefix, filename):
  48
  49         size = int(size)
  50
  51         prefix = get_prefix(filename)
  52         target = self.get_thumbnail_path(size, prefix, filename)
  53         original = self.get_original_path(prefix, filename)
  54
  55         if self.storage.exists(target):
  56             return self.send_file(target)
  57
  58         if not self.storage.exists(original):
  59             logger.warning("Original cover {} not found".format(original))
  60             raise Http404("Cover Art not available" + original)
  61
  62         target_dir = self.get_dir(filename)
  63         try:
  64             fp = self.storage.open(original, "rb")
  65             im = Image.open(fp)
  66             if im.mode not in ("RGB", "RGBA"):
  67                 im = im.convert("RGBA")
  68         except IOError as ioe:
  69             logger.warning("Cover file {} cannot be opened: {}".format(original, ioe))
  70             raise Http404("Cannot open cover file") from ioe
  71
  72         try:
  73             im.thumbnail((size, size), Image.ANTIALIAS)
  74             resized = im
  75         except (struct.error, IOError, IndexError) as ex:
  76             # raised when trying to read an interlaced PNG;
  77             logger.warning("Could not create thumbnail: %s", str(ex))
  78
  79             # we use the original instead
  80             return self.send_file(original)
  81
  82         sio = io.BytesIO()
  83
  84         try:
  85             resized.save(
  86                 sio,
  87                 "JPEG" if im.mode == "RGB" else "PNG",
  88                 optimize=True,
  89                 progression=True,
  90                 quality=80,
  91             )
  92         except IOError as ex:
  93             return self.send_file(original)
  94         finally:
  95             fp.close()
  96
  97         self.storage.save(target, sio)
  98
  99         return self.send_file(target)
 100
 101     @staticmethod
 102     def get_thumbnail_path(size, prefix, filename):
 103         return os.path.join("logo", str(size), prefix, filename)
 104
 105     @staticmethod
 106     def get_dir(filename):
 107         return os.path.dirname(filename)
 108
 109     @staticmethod
 110     def remove_existing_thumbnails(prefix, filename):
 111         dirs, _files = LOGO_STORAGE.listdir("logo")  # TODO: cache list of sizes
 112         for size in dirs:
 113             if size == "original":
 114                 continue
 115
 116             path = os.path.join("logo", size, prefix, filename)
 117             logger.info("Removing {}".format(path))
 118             LOGO_STORAGE.delete(path)
 119
 120     @staticmethod
 121     def get_original_path(prefix, filename):
 122         return os.path.join("logo", "original", prefix, filename)
 123
 124     def send_file(self, filename):
 125         return HttpResponseRedirect(LOGO_STORAGE.url(filename))
 126
 127     @classmethod
 128     def save_podcast_logo(cls, cover_art_url):
 129         if not cover_art_url:
 130             return
 131
 132         try:
 133             image_sha1 = hashlib.sha1(cover_art_url.encode("utf-8")).hexdigest()
 134             prefix = get_prefix(image_sha1)
 135
 136             filename = cls.get_original_path(prefix, image_sha1)
 137             dirname = cls.get_dir(filename)
 138
 139             # get hash of existing file
 140             if LOGO_STORAGE.exists(filename):
 141                 with LOGO_STORAGE.open(filename, "rb") as f:
 142                     old_hash = file_hash(f).digest()
 143             else:
 144                 old_hash = ""
 145
 146             logger.info("Logo {}, saving to {}".format(cover_art_url, filename))
 147
 148             # save new cover art
 149             LOGO_STORAGE.delete(filename)
 150             source = io.BytesIO(requests.get(cover_art_url).content)
 151             LOGO_STORAGE.save(filename, source)
 152
 153             # get hash of new file
 154             with LOGO_STORAGE.open(filename, "rb") as f:
 155                 new_hash = file_hash(f).digest()
 156
 157             # remove thumbnails if cover changed
 158             if old_hash != new_hash:
 159                 logger.info("Removing thumbnails")
 160                 thumbnails = cls.remove_existing_thumbnails(prefix, filename)
 161
 162             return cover_art_url
 163
 164         except (
 165             ValueError,
 166             requests.exceptions.RequestException,
 167             socket.error,
 168             IOError,
 169         ) as e:
 170             logger.warning("Exception while updating podcast logo: %s", str(e))
 171
 172
 173 def get_prefix(filename):
 174     return filename[:3]
 175
 176
 177 def get_logo_url(podcast, size):
 178     """Return the logo URL for the podcast
 179
 180     The logo either comes from the media storage (see CoverArt) or from the
 181     default logos in the static storage.
 182     """
 183
 184     if podcast.logo_url:
 185         filename = hashlib.sha1(podcast.logo_url.encode("utf-8")).hexdigest()
 186         return reverse("logo", args=[size, get_prefix(filename), filename])
 187
 188     else:
 189         filename = "podcast-%d.png" % (hash(podcast.title) % 5,)
 190         return staticfiles_storage.url("logo/{0}".format(filename))