Coverage for functions \ flipdare \ service \ processor \ _processor_mixin.py: 75%

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# 

12 

13 

14from pathlib import Path 

15 

16from google.cloud.storage.bucket import Bucket as StorageBucket 

17from flipdare.app_globals import truncate_string 

18from flipdare.app_log import LOG 

19from flipdare.constants import DOWNLOAD_FILE_DIR, IS_DEBUG 

20from flipdare.result.app_result import AppResult 

21from flipdare.core.hash_generator import HashGenerator 

22from flipdare.core.storage_file_type import StorageFileType 

23from flipdare.core.video_optimizer import VideoOptimizer 

24from flipdare.generated import ImageModel, StoredFileModel, VideoModel 

25from flipdare.util.file_util import FileUtil 

26from flipdare.util.firebase_file import FirebaseFile 

27from flipdare.backend.app_storage_client import AppStorageClient 

28 

29 

30class ProcessorMixin: 

31 

32 def __init__(self, bucket: StorageBucket, local_path: Path = DOWNLOAD_FILE_DIR) -> None: 

33 self._local_path = local_path 

34 self.bucket = bucket 

35 

36 @property 

37 def local_path(self) -> Path: 

38 return self._local_path 

39 

40 def optimize_video(self, video_model: VideoModel) -> StoredFileModel | None: 

41 # Placeholder for video optimization logic 

42 # ffmpeg -i input.mp4 -c:v libx264 -crf 23 -preset medium output.mp4 

43 # ffmpeg -i input.mp4 -c:v libx264 -crf 23 -preset medium -vf scale=1080:-1 output.mp4 

44 # base resolution on mobile devices is 9x16 or 1080x1920 

45 # or 16x9 for landscape 

46 if video_model.low is not None: 

47 LOG().debug( 

48 f"VideoModel: {video_model.source.url} already has low quality version. " 

49 "Skipping optimization.", 

50 ) 

51 return None 

52 

53 bucket = self.bucket 

54 high_gs_url = video_model.source.url 

55 uid = video_model.source.uid 

56 w = video_model.w 

57 h = video_model.h 

58 

59 optimizer = VideoOptimizer( 

60 bucket=bucket, 

61 fire_file=FileUtil.create_file( 

62 gs_url=high_gs_url, 

63 uid=uid, 

64 file_type=StorageFileType.VIDEO, 

65 ), 

66 width=w, 

67 height=h, 

68 ) 

69 ff = optimizer.optimize_file() 

70 if ff is None: 

71 LOG().error(f"Failed to optimize video for video: {high_gs_url}") 

72 return None 

73 

74 result = self.upload_file(ff) 

75 if not result.is_ok: 

76 LOG().error(f"Failed to upload optimized video for video: {high_gs_url}") 

77 return None 

78 

79 optimized_model = result.generated 

80 if optimized_model is None: 

81 LOG().error(f"Uploaded optimized video model is None for video: {high_gs_url}") 

82 return None 

83 

84 return optimized_model 

85 

86 def generate_thumbnail( 

87 self, 

88 uid: str, 

89 video_gs_url: str, 

90 width: int, 

91 height: int, 

92 ) -> ImageModel | None: 

93 # Placeholder for thumbnail generation logic 

94 # and is use as a placeholder before video is downloaded 

95 # 

96 if IS_DEBUG: 

97 LOG().debug(f"Generating thumbnail for video: {video_gs_url}") 

98 

99 bucket = self.bucket 

100 video_file = FileUtil.create_file( 

101 gs_url=video_gs_url, 

102 uid=uid, 

103 file_type=StorageFileType.VIDEO, 

104 ) 

105 ok = AppStorageClient(bucket).download_video_to_local(video_file) 

106 if not ok: 

107 LOG().error(f"Failed to download video for video: {video_gs_url}") 

108 return None 

109 

110 optimizer = VideoOptimizer(bucket=bucket, fire_file=video_file, width=width, height=height) 

111 thumbnail_file = optimizer.create_thumbnail_file() 

112 if thumbnail_file is None: 

113 LOG().error(f"Failed to generate thumbnail for video: {video_gs_url}") 

114 return None 

115 

116 result = self.upload_file(thumbnail_file) 

117 

118 # cleanup temporary files 

119 try: 

120 Path(thumbnail_file.local_path).unlink() 

121 Path(video_file.local_path).unlink() 

122 except Exception as e: 

123 msg = ( 

124 f"Failed to cleanup temporary files for thumbnail of content: {video_gs_url}: {e}" 

125 ) 

126 LOG().warning(msg) 

127 

128 if not result.is_ok: 

129 LOG().error(f"Failed to upload thumbnail for video: {video_gs_url}") 

130 return None 

131 

132 stored_model = result.generated 

133 if stored_model is None: 

134 LOG().error(f"Uploaded thumbnail model is None for video: {video_gs_url}") 

135 return None 

136 

137 return ImageModel( 

138 w=width, 

139 h=height, 

140 source=StoredFileModel( 

141 uid=stored_model.uid, 

142 file_size=stored_model.file_size, 

143 url=stored_model.url, 

144 ), 

145 ) 

146 

147 def get_image_hash(self, image_model: ImageModel, width: int, height: int) -> str | None: 

148 hash_: str | None 

149 uid = image_model.source.uid 

150 url = image_model.source.url 

151 

152 hash_util = HashGenerator.from_bucket( 

153 bucket=self.bucket, 

154 url=url, 

155 width=width, 

156 height=height, 

157 ) 

158 hash_ = hash_util.generate_hash() 

159 if hash_ is None: 

160 LOG().warning(f"Failed to generate hash for StoredFileModel: {uid} : {url}") 

161 return None 

162 

163 if IS_DEBUG: 

164 LOG().debug(f"Updating hash for StoredFileModel: {uid} to {truncate_string(hash_)}") 

165 

166 return hash_ 

167 

168 def upload_file(self, ff: FirebaseFile) -> AppResult[StoredFileModel]: 

169 bucket = self.bucket 

170 result = AppResult[StoredFileModel](task_name="UploadFile") 

171 remote_path = ff.remote_path 

172 local_path = ff.local_path 

173 gs_url = ff.gs_url 

174 

175 try: 

176 if IS_DEBUG: 

177 msg = f"Uploading file from {local_path} to {gs_url} ({remote_path})" 

178 LOG().debug(msg) 

179 

180 blob = bucket.blob(remote_path) 

181 blob.upload_from_filename(local_path) 

182 

183 file_size = blob.size 

184 if file_size is None: 

185 LOG().warning(f"Uploaded blob size is None for thumbnail of content: {gs_url}") 

186 return result 

187 

188 stored_model = StoredFileModel(file_size=file_size, url=gs_url, uid=ff.uid) 

189 

190 if IS_DEBUG: 

191 msg = ( 

192 f"Uploaded thumbnail from {local_path} to {remote_path} for content: {gs_url}" 

193 ) 

194 LOG().debug(msg) 

195 

196 result.generated = stored_model 

197 return result 

198 except Exception as e: 

199 LOG().error(f"Failed to upload thumbnail for content: {gs_url}: {e}") 

200 return result