Coverage for functions \ flipdare \ core \ video_optimizer.py: 86%
64 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
3#
4# This file is part of Flipdare's proprietary software and contains
5# confidential and copyrighted material. Unauthorised copying,
6# modification, distribution, or use of this file is strictly
7# prohibited without prior written permission from Flipdare Pty Ltd.
8#
9# This software includes third-party components licensed under MIT,
10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
11#
14from pathlib import Path
16from google.cloud.storage.bucket import Bucket as StorageBucket
17from flipdare.app_log import LOG
18from flipdare.constants import FFMPEG_COMMAND
19from flipdare.core.storage_file_type import StorageFileType
20from flipdare.util.file_util import FileUtil
21from flipdare.util.firebase_file import FirebaseFile
22from flipdare.util.process_util import ProcessUtil
23from flipdare.backend.app_storage_client import AppStorageClient
26class VideoOptimizer:
28 def __init__(
29 self,
30 bucket: StorageBucket,
31 fire_file: FirebaseFile,
32 width: int,
33 height: int,
34 ) -> None:
35 self._fire_file = fire_file
36 self._width = width
37 self._height = height
38 self._bucket = bucket
40 def build_command(self, local_path: Path, optimized_local_path: Path) -> list[str]:
41 return [
42 FFMPEG_COMMAND,
43 "-i",
44 f"{local_path}",
45 "-c:v",
46 "libx264",
47 "-crf",
48 "28",
49 "-preset",
50 "medium", # "-vf", f"scale={width}:{height}",
51 "-vf",
52 "scale=-1:720", # preserve aspect ratio, scale height to 720p
53 "-c:a",
54 "aac",
55 "-b:a",
56 "128k",
57 f"{optimized_local_path}",
58 ]
60 def create_thumbnail_file(self) -> FirebaseFile | None:
61 # Placeholder for thumbnail generation logic
62 # and is use as a placeholder before video is downloaded
63 ff = self._fire_file
64 high_quality_url = ff.gs_url
65 width = self._width
66 height = self._height
67 bucket = self._bucket
69 LOG().info(f"Generating thumbnail for video: {high_quality_url} in {bucket.name}")
71 ok = AppStorageClient(bucket).download_video_to_local(ff)
72 if not ok:
73 LOG().error(f"Failed to download video for video: {high_quality_url}")
74 return None
76 download_path = ff.local_path
78 created_file = FileUtil.create_upload(ff, StorageFileType.IMAGE)
79 local_path = created_file.local_path
81 LOG().info(f"Generating thumbnail for {download_path} at {local_path}")
82 command = [
83 FFMPEG_COMMAND,
84 "-i",
85 f"{download_path}",
86 "-vf",
87 f"thumbnail,scale={width}:{height}",
88 "-update",
89 "1",
90 "-y",
91 "-frames:v",
92 "1",
93 f"{local_path}",
94 ]
96 return_code, stdout, stderr = ProcessUtil(command).run()
98 # additional check to ensure file was created
99 if not local_path.exists():
100 return_code = -1
102 if return_code == 0:
103 return created_file
105 LOG().error(
106 f"FFmpeg failed to generate thumbnail for video {local_path}: {stderr}\n{stdout}",
107 )
108 return None
110 def optimize_file(self, cleanup: bool = True) -> FirebaseFile | None:
111 bucket = self._bucket
112 ff = self._fire_file
114 store_util = AppStorageClient(bucket)
115 ok = store_util.download_video_to_local(ff)
117 if not ok:
118 LOG().error(f"Failed to download video for optimization: {ff.gs_url}")
119 return None
121 local_path = ff.local_path
122 high_size = local_path.stat().st_size
123 LOG().debug(f"Downloaded video size: {high_size} bytes")
125 ff_optimized = FileUtil.create_upload(ff, StorageFileType.VIDEO)
126 optimized_local_path = ff_optimized.local_path
128 # version: 1.0
129 # ffmpeg -i bunny_2mb.mp4 -c:v libx264 -crf 28 -preset medium -c:a aac -b:a 128k bunny_2mb_optimized_v1.mp4
130 # version: 2 (reduces size further by scaling to 720p)
131 # ffmpeg -i bunny_2mb.mp4 -c:v libx264 -crf 28 -preset medium -vf "scale=-1:720" -c:a aac -b:a 128k bunny_2mb_optimized_v2.mp4
133 command = self.build_command(local_path, optimized_local_path)
134 return_code, stdout, stderr = ProcessUtil(command).run()
135 if cleanup:
136 FileUtil.cleanup_local_files([local_path])
138 if return_code != 0:
139 LOG().error(f"FFmpeg failed to optimize video {local_path}: {stderr}\n{stdout}")
140 return None
142 low_size = optimized_local_path.stat().st_size
143 percent_reduction = ((high_size - low_size) / high_size) * 100
144 LOG().info(
145 f"High quality video size: {high_size} bytes, "
146 f"Optimized video size: {low_size} bytes "
147 f"({percent_reduction:.2f}% reduction)",
148 )
150 return ff_optimized