Coverage for functions \ flipdare \ util \ file_util.py: 91%

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 

14import uuid 

15from pathlib import Path 

16from flipdare.app_log import LOG 

17from flipdare.constants import DOWNLOAD_FILE_DIR, IS_TRACE 

18from flipdare.core.storage_file_type import StorageFileType 

19from flipdare.util.firebase_file import FirebaseFile 

20 

21 

22class FileUtil: 

23 

24 def __init__(self) -> None: 

25 pass 

26 

27 @staticmethod 

28 def sanitize_bucket_name(bucket_name: str) -> str: 

29 bucket_name = bucket_name.removesuffix("/") 

30 return bucket_name.removeprefix("gs://") 

31 

32 @staticmethod 

33 def bucket_name_from_gs_url(gs_url: str) -> str: 

34 if not gs_url.startswith("gs://"): 

35 raise ValueError(f"gs_url '{gs_url}' is not a valid Gs URL.") 

36 

37 bucket_name = gs_url.split("/")[2] 

38 return FileUtil.sanitize_bucket_name(bucket_name) 

39 

40 @staticmethod 

41 def remote_path_from_gs_url(gs_url: str) -> str: 

42 bucket_name = FileUtil.bucket_name_from_gs_url(gs_url) 

43 return gs_url.replace(f"gs://{bucket_name}/", "", 1) 

44 

45 @staticmethod 

46 def cleanup_local_files(file_paths: list[Path]) -> None: 

47 from flipdare.app_log import LOG 

48 

49 for file_path in file_paths: 

50 try: 

51 file_path.unlink() 

52 if IS_TRACE: 

53 LOG().trace(f"Deleted local file: {file_path}") 

54 except Exception as e: 

55 LOG().warning(f"Failed to remove local file '{file_path}': {e}") 

56 

57 @staticmethod 

58 def create_file( 

59 gs_url: str, 

60 uid: str, 

61 file_type: StorageFileType, 

62 download_dir: Path = DOWNLOAD_FILE_DIR, 

63 ) -> FirebaseFile: 

64 if not gs_url.startswith("gs://"): 

65 raise ValueError(f"gs_url '{gs_url}' is not a valid Gs URL.") 

66 

67 if not download_dir.exists(): 

68 download_dir.mkdir(parents=True, exist_ok=True) 

69 

70 bucket_name = FileUtil.bucket_name_from_gs_url(gs_url) 

71 # create everything, so its hardcoded 

72 prefix = str(uuid.uuid4().hex[:8]) 

73 base_name = gs_url.rsplit("/", maxsplit=1)[-1] 

74 name = base_name 

75 local_path: Path = download_dir / f"{uid}_{prefix}_{base_name}" 

76 remote_path = FileUtil.remote_path_from_gs_url(gs_url) 

77 

78 return FirebaseFile( 

79 gs_url=gs_url, 

80 uid=uid, 

81 file_type=file_type, 

82 bucket_name=bucket_name, 

83 download_dir=download_dir, 

84 name=name, 

85 local_path=local_path, 

86 remote_path=remote_path, 

87 ) 

88 

89 @staticmethod 

90 def create_upload(sfs: FirebaseFile, file_type: StorageFileType) -> FirebaseFile: 

91 old_name = sfs.name 

92 root = Path(old_name).stem 

93 

94 new_ext = "." + file_type.extension 

95 new_name = f"{root}_{file_type.prefix}{new_ext}" 

96 new_remote_path = f"{file_type.location}/{new_name}" 

97 new_gs_url = f"gs://{sfs.bucket_name}/{new_remote_path}" 

98 

99 local_path = sfs.local_path.with_name(new_name) 

100 

101 if IS_TRACE: 

102 LOG().trace( 

103 f"New Extension={new_ext}\n" 

104 f"New Name={new_name}\n" 

105 f"New Remote Path={new_remote_path}\n" 

106 f"New Gs URL={new_gs_url}\n" 

107 f"Original Name={old_name}\n" 

108 f"Original Remote Path={sfs.remote_path}\n" 

109 f"Original Gs URL={sfs.gs_url}\n" 

110 f"File Type={file_type} prefix={file_type.prefix} extension={file_type.extension}", 

111 ) 

112 

113 return FirebaseFile( 

114 gs_url=new_gs_url, 

115 uid=sfs.uid, 

116 file_type=file_type, 

117 bucket_name=sfs.bucket_name, 

118 download_dir=sfs.download_dir, 

119 name=new_name, 

120 local_path=local_path, 

121 remote_path=new_remote_path, 

122 )