files/tasks.py

@@ -41,18 +41,18 @@ def make_thumbnails(file_id: int) -> None:
     assert file.validation.is_ok, f'File pk={file_id} is flagged'
 
     source_path = os.path.join(settings.MEDIA_ROOT, file.source.path)
-    thumbnail_field = file.thumbnail
 
     if file.is_video:
+        f_path = os.path.join(settings.MEDIA_ROOT, files.utils.get_thumbnail_upload_to(file.hash))
         # For a video, source of the thumbnail is some frame fetched with ffpeg
-        output_path = os.path.join(settings.MEDIA_ROOT, thumbnail_field.upload_to(file, 'x.png'))
+        files.utils.extract_frame(source_path, f_path)
-        files.utils.extract_frame(source_path, output_path)
+        source_path = f_path
-        source_path = output_path
 
     thumbnails = files.utils.make_thumbnails(source_path, file.hash)
     thumbnail_default_path = thumbnails['l']['path']
 
     update_fields = set()
+    thumbnail_field = file.thumbnail
     if thumbnail_default_path != thumbnail_field.name:
         thumbnail_field.name = thumbnail_default_path
         update_fields.add('thumbnail')

files/tests/test_utils.py

@@ -1,6 +1,11 @@
 from django.test import TestCase
 
-from files.utils import find_path_by_name, find_exact_path, filter_paths_by_ext
+from files.utils import (
+    filter_paths_by_ext,
+    find_exact_path,
+    find_path_by_name,
+    get_thumbnail_upload_to,
+)
 
 
 class UtilsTest(TestCase):
@@ -98,3 +103,12 @@ class UtilsTest(TestCase):
         ]
         paths = filter_paths_by_ext(name_list, '.md')
         self.assertEqual(list(paths), [])
+
+    def test_get_thumbnail_upload_to(self):
+        for file_hash, kwargs, expected in (
+            ('foobar', {}, 'thumbnails/fo/foobar.png'),
+            ('deadbeaf', {'width': None, 'height': None}, 'thumbnails/de/deadbeaf.png'),
+            ('deadbeaf', {'width': 640, 'height': 360}, 'thumbnails/de/deadbeaf_640x360.png'),
+        ):
+            with self.subTest(file_hash=file_hash, kwargs=kwargs):
+                self.assertEqual(get_thumbnail_upload_to(file_hash, **kwargs), expected)

files/utils.py

@@ -202,21 +202,23 @@ def delete_thumbnails(file_metadata: dict) -> None:
         delete_file_in_storage(path)
 
 
-def get_base_path(file_path: str) -> str:
+def get_thumbnail_upload_to(file_hash: str, width: int = None, height: int = None) -> str:
-    """Return the given path without its extension."""
+    """Return a full media path of a thumbnail.
-    base_path, _ = os.path.splitext(file_path)
-    return base_path
 
-
+    Optionally, append thumbnail dimensions to the file name.
-def get_thumbnail_upload_to(file_hash: str):
+    """
     prefix = 'thumbnails/'
     _hash = file_hash.split(':')[-1]
-    extension = f'.{THUMBNAIL_FORMAT.lower()}'
+    thumbnail_ext = THUMBNAIL_FORMAT.lower()
-    path = Path(prefix, _hash[:2], _hash).with_suffix(extension)
+    if thumbnail_ext == 'jpeg':
-    return path
+        thumbnail_ext = 'jpg'
+    suffix = f'.{thumbnail_ext}'
+    size_suffix = f'_{width}x{height}' if width and height else ''
+    path = Path(prefix, _hash[:2], f'{_hash}{size_suffix}').with_suffix(suffix)
+    return str(path)
 
 
-def _resize(image: Image, size: tuple, output, output_format: str = 'PNG', **output_params):
+def resize_image(image: Image, size: tuple, output, output_format: str = 'PNG', **output_params):
     """Resize a models.ImageField to a given size and write it into output file."""
     source_image = image.convert('RGBA' if output_format == 'PNG' else 'RGB')
     source_image.thumbnail(size, Image.LANCZOS)
@@ -229,18 +231,14 @@ def make_thumbnails(file_path: str, file_hash: str, output_format: str = THUMBNA
     Resulting thumbnail paths a derived from the given file hash and thumbnail sizes.
     Return a dict of size keys to output paths of generated thumbnail images.
     """
-    thumbnail_ext = output_format.lower()
-    if thumbnail_ext == 'jpeg':
-        thumbnail_ext = 'jpg'
-    base_path = get_base_path(get_thumbnail_upload_to(file_hash))
     thumbnails = {}
     for size_key, (w, h) in THUMBNAIL_SIZES.items():
-        output_path = f'{base_path}_{w}x{h}.{thumbnail_ext}'
+        output_path = get_thumbnail_upload_to(file_hash, width=w, height=h)
         size = (w, h)
         with tempfile.TemporaryFile() as f:
             logger.info('Resizing %s to %s (%s)', file_path, size, output_format)
             image = Image.open(file_path)
-            _resize(
+            resize_image(
                 image,
                 size,
                 f,
@@ -259,17 +257,14 @@ def make_thumbnails(file_path: str, file_hash: str, output_format: str = THUMBNA
     return thumbnails
 
 
-def extract_frame(source_path: str, output_path: str):
+def extract_frame(source_path: str, output_path: str, at_time: str = '00:00:00.01'):
     """Extract a single frame of a video at a given path, write it to the given output path."""
     try:
         ffmpeg = (
             FFmpeg()
             .option('y')
             .input(source_path)
-            .output(
+            .output(output_path, {'ss': at_time, 'frames:v': 1, 'update': 'true'})
-                output_path,
-                {'ss': '00:00:00.01', 'frames:v': 1, 'update': 'true'},
-            )
         )
         output_dir = os.path.dirname(output_path)
         if not os.path.isdir(output_dir):