Coverage for seedboxsync/core/utils.py: 96%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6# 

7"""A collection of utility functions for SeedboxSync.""" 

8 

9import mimetypes 

10import os 

11from os import PathLike 

12from pathlib import Path 

13from typing import Any 

14from urllib.parse import urlparse 

15from bcoding import bdecode 

16from flask import current_app 

17import puremagic 

18 

19 

20def byte_to_gi(bytes_value: float, suffix: str = "B") -> str: 

21 """ 

22 Convert in human readable units. 

23 

24 Args: 

25 bytes_value (integer): Value not human readable. 

26 suffix (str): Suffix for value given to (default: B). 

27 

28 Returns: 

29 str: human readable value in Gi. 

30 """ 

31 gib = bytes_value / (1024**3) 

32 return f"{gib:.1f}Gi{suffix}" 

33 

34 

35def ensure_dir_exists(path: str | PathLike[str]) -> None: 

36 """ 

37 Ensure the directory ``path`` exists, and if not create it. 

38 

39 Args: 

40 path (str): The filesystem path of a directory. 

41 

42 Raises: 

43 AssertionError: If the directory ``path`` exists, but is not a directory. 

44 

45 """ 

46 path = Path(path).expanduser().resolve() 

47 

48 if path.exists() and not path.is_dir(): 

49 raise AssertionError(f"Path `{path}` exists but is not a directory!") 

50 if not path.exists(): 

51 path.mkdir() 

52 

53 

54def get_torrent_infos(torrent_path: str | PathLike[str]) -> Any | None: 

55 """ 

56 Extracts information from a torrent file. 

57 

58 Args: 

59 torrent_path (str | PathLike[str]): Path to the torrent file. 

60 

61 Returns: 

62 str: Decoded torrent information. 

63 

64 Raises: 

65 Exception: If the file is not a valid torrent. 

66 """ 

67 with Path(torrent_path).open("rb") as torrent: 

68 torrent_info = None 

69 

70 try: 

71 torrent_info = bdecode(torrent.read()) 

72 except Exception: 

73 current_app.logger.exception("Not valid torrent") 

74 finally: 

75 torrent.close() 

76 

77 return torrent_info 

78 

79 

80def is_running_in_docker() -> bool: 

81 """ 

82 Return whether the current process appears to run inside Docker. 

83 

84 Returns: 

85 bool: True if in docker envoronment. 

86 

87 """ 

88 # Test mountinfo 

89 if Path("/proc/self/mountinfo").exists(): 

90 with Path("/proc/self/mountinfo").open() as f: 

91 if "docker" in f.read() or "overlay" in f.read(): 

92 return True 

93 

94 # Test du cgroup 

95 if Path("/proc/1/cgroup").exists(): 

96 with Path("/proc/1/cgroup").open() as f: 

97 lines = f.read() 

98 if "docker" in lines or "kubepods" in lines: 

99 return True 

100 

101 # Test /.dockerenv but not on podman 

102 return Path("/.dockerenv").exists() 

103 

104 

105def get_database_path_from_paths() -> Path: 

106 """ 

107 Find and return an existing writable database file path. 

108 

109 Iterates through a predefined list of standard locations to locate a valid 

110 database file. Returns the first path that exists, is a regular file, and 

111 has write permissions. If no match is found, falls back to the default 

112 user configuration path. 

113 

114 Returns: 

115 Path: The resolved writable database path if found; otherwise, the default 

116 path (~/.config/seedboxsync/seedboxsync.db). 

117 """ 

118 db_paths = [ 

119 Path("~/.config/seedboxsync/seedboxsync.db").expanduser().resolve(), 

120 Path("~/.seedboxsync.db").expanduser().resolve(), 

121 Path("~/.seedboxsync/config/seedboxsync.db").expanduser().resolve(), 

122 Path("/etc/seedboxsync/seedboxsync.db"), 

123 ] 

124 for path in db_paths: 

125 if path.exists() and path.is_file() and os.access(path, os.W_OK): 

126 return path 

127 return db_paths[0] 

128 

129 

130def get_web_healthcheck_url() -> str: 

131 """ 

132 Return the URL used to check the local Flask application. 

133 

134 Returns: 

135 str: The healthcheck URL. 

136 """ 

137 explicit_url = os.getenv("HEALTHCHECK_URL") 

138 if explicit_url: 

139 return explicit_url.rstrip("/") + "/healthcheck" 

140 

141 bind = os.getenv("BIND") 

142 if bind: 

143 return _healthcheck_url_from_bind(bind) 

144 

145 port = 8000 if is_running_in_docker() else 5000 

146 return f"http://127.0.0.1:{port}/healthcheck" 

147 

148 

149def get_mime_type_from_file(filename: str) -> tuple[str, str, str]: 

150 """ 

151 Detect the MIME type and extension of a file. 

152 

153 Args: 

154 filename: Name to the local file to analyze 

155 Returns: 

156 tuple[str, str, str]: (mime_type, mime_extension, confidence) 

157 """ 

158 # Initialize local file path 

159 local_filepath = Path(current_app.seedboxsync_config.get("local_download_path", "")).expanduser().resolve() / filename # type: ignore[attr-defined] 

160 

161 # Use first magic_file 

162 try: 

163 current_app.logger.debug(f"Attempting MIME detection via puremagic header analysis for: {local_filepath}") 

164 results = puremagic.magic_file(local_filepath) 

165 if results: 

166 match = results[0] 

167 

168 mime_extension = match.extension.lstrip(".") 

169 mime_type = match.mime_type 

170 mime_confidence = f"puremagic (confidence: {match.confidence})" 

171 current_app.logger.debug(f"Successfully detected MIME via puremagic: type={mime_type}, ext={mime_extension}, confidence={mime_confidence}") 

172 

173 return mime_type, mime_extension, mime_confidence 

174 

175 except (FileNotFoundError, puremagic.PureError): 

176 pass 

177 

178 # Fallback with mimetypes 

179 current_app.logger.debug(f"Attempting MIME detection fallback via mimetypes for: {local_filepath}") 

180 mime_type, _ = mimetypes.guess_type(local_filepath) 

181 

182 if mime_type: 

183 extension = mimetypes.guess_extension(mime_type) or "" 

184 mime_extension = extension.lstrip(".") 

185 mime_confidence = "mimetypes (path_fallback)" 

186 

187 current_app.logger.debug(f"MIME detection completed using fallback: type={mime_type}, ext={mime_extension}, confidence={mime_confidence}") 

188 

189 return mime_type, mime_extension, "mimetypes (path_fallback)" 

190 

191 # Last resort when nothing can be detected. 

192 parts = str(local_filepath).rsplit(".", 1) if local_filepath else [] 

193 mime_extension = parts[1].lower() if len(parts) > 1 else "unknown" 

194 

195 return "application/octet-stream", mime_extension, "unknown" 

196 

197 

198def _healthcheck_url_from_bind(bind: str) -> str: 

199 """ 

200 Build a local healthcheck URL from a Gunicorn bind value. 

201 

202 Returns: 

203 str: The healthcheck URL from BIND. 

204 """ 

205 bind = bind.strip() 

206 

207 if bind.startswith("unix:"): 

208 raise ValueError("A Unix socket bind cannot be checked with a standard HTTP URL.") 

209 

210 # urlparse requires a scheme to correctly parse host and port. 

211 parsed = urlparse(f"//{bind}") 

212 

213 if parsed.port is None: 

214 raise ValueError(f"Invalid BIND value: {bind}") 

215 

216 host = parsed.hostname or "127.0.0.1" 

217 

218 if host in {"0.0.0.0", "::", "[::]"}: 

219 host = "127.0.0.1" 

220 

221 return f"http://{host}:{parsed.port}/healthcheck"