Coverage for src/edwh/discover.py: 17%

1import json 

2import re 

3import sys 

4import typing 

5from contextlib import contextmanager 

6from pathlib import Path 

7from typing import TypedDict 

8 

9import humanize 

10from ewok import Context 

11from termcolor import colored, cprint 

12 

13from .helpers import AnyDict, dc_config, dump_set_as_list, noop 

14 

15 

16def indent(text: str, prefix: str = " ") -> str: 

17 return prefix + text 

18 

19 

20def dedent(text: str, prefix: str = " ") -> str: 

21 return text.replace(prefix, "", 1) 

22 

23 

24def terminal_link(url: str, text: str | None = None) -> str: 

25 """ 

26 Creëer een klikbare hyperlink voor de terminal met OSC 8 escape codes. 

27 

28 Args: 

29 url (str): De URL waarnaar de link moet verwijzen 

30 text (str, optional): De tekst die getoond moet worden. 

31 Als None, wordt de URL zelf getoond. 

32 

33 Returns: 

34 str: Een string met embedded ANSI escape codes voor een klikbare link 

35 

36 Voorbeeld: 

37 print(terminal_link('https://github.com', 'Naar GitHub')) 

38 print(terminal_link('https://example.com')) 

39 """ 

40 if text is None: 

41 text = url 

42 

43 # OSC 8 formaat: \033]8;;{url}\033\\{text}\033]8;;\033\\ 

44 escape_mask = "\033]8;;{}\033\\{}\033]8;;\033\\" 

45 

46 return escape_mask.format(url, text) 

47 

48 

49HOST_RE = re.compile(r"`(.*?)`") 

50 

51 

52def strip_host(s: str) -> str: 

53 matches: list[str] = HOST_RE.findall(s.strip()) 

54 return matches[0] 

55 

56 

57class ServiceDict(TypedDict, total=False): 

58 name: str 

59 exposes: list[int] 

60 ports: list[str] 

61 domains: set[str] 

62 

63 

64class ProjectDict(TypedDict, total=False): 

65 name: str 

66 hostingdomain: str 

67 disk_usage_human: str 

68 disk_usage_raw: int 

69 settings: AnyDict 

70 services: list[ServiceDict] 

71 

72 

73class DataDict(TypedDict): 

74 server: str 

75 projects: list[ProjectDict] 

76 

77 

78def get_hosts_for_service(docker_service: AnyDict) -> set[str]: 

79 domains = set() 

80 

81 for label, value in docker_service.get("labels", {}).items(): 

82 if "Host" not in value: 

83 # irrelevant 

84 continue 

85 

86 if "||" in value: 

87 # OR 

88 for host in value.split("||"): 

89 domains.add(strip_host(host)) 

90 else: 

91 # only one 

92 domains.add(strip_host(value)) 

93 

94 return domains 

95 

96 

97class Discover: 

98 i: str 

99 data: DataDict 

100 

101 def __init__( 

102 self, 

103 ctx: Context, 

104 du: bool = False, 

105 exposes: bool = False, 

106 ports: bool = False, 

107 host_labels: bool = True, 

108 short: bool = False, 

109 settings: bool = False, 

110 as_json: bool = False, 

111 ): 

112 self.ctx = ctx 

113 self.du = du 

114 self.exposes = exposes 

115 self.ports = ports 

116 self.host_labels = host_labels 

117 self.short = short 

118 self.settings = settings 

119 self.as_json = as_json 

120 

121 print_fn = noop if as_json else cprint 

122 

123 self.print_fn = typing.cast(typing.Callable[..., None], print_fn) 

124 self.reset() 

125 

126 def reset(self) -> None: 

127 hostname = self.find_hostname() 

128 

129 self.i = "" 

130 self.data = { 

131 "server": hostname, 

132 "projects": [], 

133 } 

134 

135 @contextmanager 

136 def indent(self, prefix: str = " ") -> typing.Generator[None, None, None]: 

137 # context manager 

138 self.i = indent(self.i, prefix) 

139 yield 

140 self.i = dedent(self.i, prefix) 

141 

142 def print(self, *args: typing.Any, **kwargs: typing.Any) -> None: 

143 sep = kwargs.pop("sep", " ") 

144 msg = sep.join([self.i, *args]) 

145 self.print_fn(msg, **kwargs) 

146 

147 def get_hostingdomain_from_env(self) -> str: 

148 if ran := self.ctx.run("cat .env | grep HOSTINGDOMAIN", echo=False, hide=True, warn=True): 

149 hosting_domain = ran.stdout.strip() 

150 else: 

151 hosting_domain = None 

152 

153 return hosting_domain.strip().split("=")[-1] if hosting_domain else "" 

154 

155 def find_compose_files(self) -> list[str]: 

156 try: 

157 ran = self.ctx.run( 

158 "find ./docker-compose.yaml ./docker-compose.yml */docker-compose.yaml */docker-compose.yml", 

159 echo=False, 

160 hide=True, 

161 warn=True, 

162 ) 

163 return ran.stdout.strip().split("\n") 

164 except Exception: 

165 return [] 

166 

167 def find_hostname(self) -> str: 

168 if ran := self.ctx.run("hostname", hide=True): 

169 return ran.stdout.strip() 

170 else: 

171 return "" 

172 

173 def get_disk_usage(self) -> tuple[str, int]: 

174 if ran := self.ctx.run("du -sh . --block-size=1", echo=False, hide=True): 

175 usage_raw = ran.stdout.strip().split("\t")[0] 

176 usage = humanize.naturalsize(usage_raw, binary=True) 

177 self.print(f"Disk usage: {usage}", color="red", attrs=["bold"]) 

178 

179 return usage, int(usage_raw) 

180 else: 

181 raise EnvironmentError("Failed running `du`") 

182 

183 def get_settings(self, folder: str) -> AnyDict | None: 

184 json_flag = "--json" if self.as_json else "" 

185 if ran := self.ctx.run(f"~/.local/bin/edwh settings {json_flag}", echo=False, hide=True): 

186 settings_output = ran.stdout.strip() 

187 else: 

188 settings_output = "" 

189 

190 if self.as_json: 

191 try: 

192 return typing.cast(AnyDict, json.loads(settings_output)) 

193 except json.JSONDecodeError: 

194 print(f"Error loading settings for {self.data['server']}/{folder}", file=sys.stderr) 

195 else: 

196 self.print("Settings:", color="red", attrs=["bold"]) 

197 with self.indent(): 

198 for line in settings_output.split("\n"): 

199 self.print(line) 

200 return None 

201 

202 def process_docker_service(self, name: str, docker_service: AnyDict, hosting_domain: str) -> ServiceDict: 

203 service: ServiceDict = {"name": name} 

204 if not self.short: 

205 self.print(name, color="green") 

206 with self.indent(): 

207 if self.exposes and not self.short: 

208 if _exposes := docker_service.get("expose", []): 

209 self.print(f"Exposes: {', '.join([str(port) for port in _exposes])}", color="red", attrs=["bold"]) 

210 service["exposes"] = _exposes 

211 if self.ports: 

212 if _ports := docker_service.get("ports", []): 

213 self.print( 

214 f"Ports: {', '.join([str(port) for port in _ports]) if _ports else ''}", 

215 color="red", 

216 attrs=["bold"], 

217 ) 

218 service["ports"] = _ports 

219 

220 service["domains"] = set() 

221 if self.host_labels and not self.short: 

222 service["domains"] = get_hosts_for_service(docker_service) 

223 for domain in service["domains"]: 

224 self.print( 

225 terminal_link( 

226 f"https://{domain}", 

227 domain.replace(hosting_domain, colored(hosting_domain, color="dark_grey")), 

228 ), 

229 ) 

230 

231 if service["domains"]: 

232 self.print() 

233 

234 return service 

235 

236 def process_omgeving(self, folder: str) -> ProjectDict | None: 

237 project: ProjectDict = {} 

238 hosting_domain = self.get_hostingdomain_from_env() 

239 self.print( 

240 colored(f"{folder}", color="light_blue"), 

241 colored(f"{hosting_domain}", color="light_yellow"), 

242 ) 

243 

244 project["name"] = folder 

245 project["hostingdomain"] = hosting_domain 

246 

247 with self.indent(): 

248 config = dc_config(self.ctx) 

249 if config is None: 

250 return None 

251 

252 if self.du and not self.short: 

253 usage, usage_raw = self.get_disk_usage() 

254 project["disk_usage_human"] = usage 

255 project["disk_usage_raw"] = usage_raw 

256 

257 if self.settings and (settings := self.get_settings(folder)) and not self.short: 

258 project["settings"] = settings 

259 

260 project["services"] = [] 

261 # if not self.short: 

262 # print(config.get("services", {}).items()) 

263 for name, docker_service in config.get("services", {}).items(): 

264 project["services"].append(self.process_docker_service(name, docker_service, hosting_domain)) 

265 

266 return project 

267 

268 def process_compose_file(self, compose_file_path: Path) -> None: 

269 folder = compose_file_path.parent 

270 with self.ctx.cd(folder): 

271 if project := self.process_omgeving(str(folder)): 

272 self.data["projects"].append(project) 

273 

274 def discover(self) -> None: 

275 self.reset() 

276 

277 self.print(self.data["server"], attrs=["bold"]) 

278 

279 compose_file_paths = self.find_compose_files() 

280 

281 for compose_file in compose_file_paths: 

282 self.process_compose_file(Path(compose_file)) 

283 

284 if self.as_json: 

285 print(json.dumps({"data": self.data}, indent=2, default=dump_set_as_list)) 

286 

287 

288def discover( 

289 ctx: Context, 

290 du: bool = False, 

291 exposes: bool = False, 

292 ports: bool = False, 

293 host_labels: bool = True, 

294 short: bool = False, 

295 settings: bool = False, 

296 as_json: bool = False, 

297) -> None: 

298 d = Discover( 

299 ctx, 

300 du=du, 

301 exposes=exposes, 

302 ports=ports, 

303 host_labels=host_labels, 

304 short=short, 

305 settings=settings, 

306 as_json=as_json, 

307 ) 

308 

309 return d.discover()