Coverage for src / server_list / cli / webui.py: 79%

1#!/usr/bin/env python3 

2""" 

3Flask application to serve the Server List React app at /server-list 

4 

5Usage: 

6 server-list [-c CONFIG] [-p PORT] [-D] 

7 

8Options: 

9 -c CONFIG : CONFIG を設定ファイルとして読み込んで実行します。[default: config.yaml] 

10 -p PORT : WEB サーバのポートを指定します。[default: 5000] 

11 -D : デバッグモードで動作します。 

12""" 

13 

14from __future__ import annotations 

15 

16import atexit 

17import logging 

18import os 

19import signal 

20from pathlib import Path 

21from typing import TYPE_CHECKING 

22from urllib.parse import unquote 

23 

24import flask 

25import flask_cors 

26import my_lib.webapp.base 

27import my_lib.webapp.config 

28import my_lib.webapp.event 

29import my_lib.webapp.util 

30 

31from server_list.spec import cpu_benchmark, data_collector, db 

32from server_list.spec.data_collector import start_collector, stop_collector 

33from server_list.spec.ogp import ( 

34 generate_machine_page_ogp, 

35 generate_top_page_ogp, 

36 inject_ogp_into_html, 

37) 

38from server_list.spec.webapi.config import config_api 

39from server_list.spec.webapi.cpu import cpu_api 

40from server_list.spec.webapi.power import power_api 

41from server_list.spec.webapi.storage import storage_api 

42from server_list.spec.webapi.ups import ups_api 

43from server_list.spec.webapi.uptime import uptime_api 

44from server_list.spec.webapi.vm import vm_api 

45 

46if TYPE_CHECKING: 

47 from server_list.config import Config 

48 

49URL_PREFIX = "/server-list" 

50 

51 

52def term() -> None: 

53 """Terminate the application gracefully.""" 

54 logging.info("Terminating application...") 

55 stop_collector() 

56 logging.info("Application terminated.") 

57 

58 

59def sig_handler(num: int, frame) -> None: # noqa: ARG001 

60 """Handle signals for graceful shutdown.""" 

61 logging.warning("Received signal %d", num) 

62 

63 if num in (signal.SIGTERM, signal.SIGINT): 

64 term() 

65 

66 

67def create_app( 

68 webapp_config: my_lib.webapp.config.WebappConfig, 

69 config: Config | None = None, 

70) -> flask.Flask: 

71 my_lib.webapp.config.URL_PREFIX = URL_PREFIX 

72 my_lib.webapp.config.init(webapp_config) 

73 

74 # Initialize paths from config 

75 if config: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 db.init_from_config(config) 

77 

78 app = flask.Flask("server-list") 

79 

80 # NOTE: アクセスログは無効にする 

81 logging.getLogger("werkzeug").setLevel(logging.ERROR) 

82 

83 flask_cors.CORS(app) 

84 

85 # Register API blueprints 

86 app.register_blueprint(cpu_api, url_prefix=f"{URL_PREFIX}/api") 

87 app.register_blueprint(config_api, url_prefix=f"{URL_PREFIX}/api") 

88 app.register_blueprint(vm_api, url_prefix=f"{URL_PREFIX}/api") 

89 app.register_blueprint(uptime_api, url_prefix=f"{URL_PREFIX}/api") 

90 app.register_blueprint(power_api, url_prefix=f"{URL_PREFIX}/api") 

91 app.register_blueprint(storage_api, url_prefix=f"{URL_PREFIX}/api") 

92 app.register_blueprint(ups_api, url_prefix=f"{URL_PREFIX}/api") 

93 

94 # Register webapp blueprints 

95 app.register_blueprint(my_lib.webapp.base.blueprint_default) 

96 app.register_blueprint(my_lib.webapp.base.blueprint, url_prefix=URL_PREFIX) 

97 app.register_blueprint(my_lib.webapp.event.blueprint, url_prefix=URL_PREFIX) 

98 app.register_blueprint(my_lib.webapp.util.blueprint, url_prefix=URL_PREFIX) 

99 

100 def get_base_url() -> str: 

101 """Get base URL from request.""" 

102 return flask.request.url_root.rstrip("/") 

103 

104 def serve_html_with_ogp(ogp_tags: str) -> flask.Response: 

105 """Serve index.html with OGP tags injected.""" 

106 static_dir = my_lib.webapp.config.STATIC_DIR_PATH 

107 if static_dir is None: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true

108 return flask.Response("Static directory not configured", status=500) 

109 index_path = Path(static_dir) / "index.html" 

110 if not index_path.exists(): 110 ↛ 113line 110 didn't jump to line 113 because the condition on line 110 was always true

111 return flask.send_from_directory(static_dir, "index.html") 

112 

113 html_content = index_path.read_text(encoding="utf-8") 

114 modified_html = inject_ogp_into_html(html_content, ogp_tags) 

115 return flask.Response(modified_html, mimetype="text/html") 

116 

117 # Top page with OGP 

118 @app.route(f"{URL_PREFIX}/") 

119 def index_with_ogp(): 

120 ogp_tags = generate_top_page_ogp(get_base_url(), config) 

121 return serve_html_with_ogp(ogp_tags) 

122 

123 # SPA fallback route with machine-specific OGP 

124 @app.route(f"{URL_PREFIX}/machine/<path:machine_name>") 

125 def machine_page_with_ogp(machine_name: str): 

126 decoded_name = unquote(machine_name) 

127 ogp_tags = generate_machine_page_ogp( 

128 get_base_url(), 

129 decoded_name, 

130 config, 

131 db.IMAGE_DIR, 

132 ) 

133 return serve_html_with_ogp(ogp_tags) 

134 

135 # UPS page route (SPA) 

136 @app.route(f"{URL_PREFIX}/ups") 

137 def ups_page(): 

138 return serve_html_with_ogp("") 

139 

140 # Serve server model images 

141 @app.route(f"{URL_PREFIX}/api/img/<path:filename>") 

142 def serve_image(filename): 

143 return flask.send_from_directory(db.IMAGE_DIR, filename) 

144 

145 # Initialize databases (required for API to work) 

146 cpu_benchmark.init_db() 

147 data_collector.init_db() 

148 

149 # Start background workers 

150 # - In debug mode with reloader: WERKZEUG_RUN_MAIN == "true" in the main worker process 

151 # - In non-debug mode: WERKZEUG_RUN_MAIN is not set 

152 werkzeug_run_main = os.environ.get("WERKZEUG_RUN_MAIN") 

153 if werkzeug_run_main == "true" or werkzeug_run_main is None: 153 ↛ 157line 153 didn't jump to line 157 because the condition on line 153 was always true

154 start_collector() 

155 atexit.register(stop_collector) 

156 

157 my_lib.webapp.config.show_handler_list(app) 

158 

159 return app 

160 

161 

162def main() -> None: 

163 import pathlib 

164 import sys 

165 import traceback 

166 

167 import docopt 

168 import my_lib.logger 

169 import my_lib.webapp.config 

170 

171 from server_list.config import Config 

172 

173 assert __doc__ is not None 

174 args = docopt.docopt(__doc__) 

175 

176 config_file = args["-c"] 

177 port = int(args["-p"]) 

178 debug_mode = args["-D"] 

179 

180 my_lib.logger.init("server-list", level=logging.DEBUG if debug_mode else logging.INFO) 

181 

182 logging.info("Starting server-list webui...") 

183 logging.info("Config file: %s", config_file) 

184 logging.info("Port: %s, Debug: %s", port, debug_mode) 

185 

186 # Use cwd-relative schema path (works for both source and Docker) 

187 schema_path = pathlib.Path("schema/config.schema") 

188 if not schema_path.exists(): 188 ↛ 190line 188 didn't jump to line 190 because the condition on line 188 was never true

189 # Fall back to db.CONFIG_SCHEMA_PATH for source tree 

190 schema_path = db.CONFIG_SCHEMA_PATH 

191 logging.info("Schema path: %s (exists: %s)", schema_path, schema_path.exists()) 

192 

193 try: 

194 config = Config.load(pathlib.Path(config_file), schema_path) 

195 logging.info("Config loaded successfully, %d machines defined", len(config.machine)) 

196 

197 webapp_config = my_lib.webapp.config.WebappConfig.parse({ 

198 "static_dir_path": config.webapp.static_dir_path, 

199 }) 

200 

201 app = create_app(webapp_config, config=config) 

202 

203 app.config["CONFIG"] = config 

204 

205 signal.signal(signal.SIGTERM, sig_handler) 

206 

207 app.run(host="0.0.0.0", port=port, debug=debug_mode) # noqa: S104 

208 except KeyboardInterrupt: 

209 logging.info("Received KeyboardInterrupt, shutting down...") 

210 sig_handler(signal.SIGINT, None) 

211 except Exception: 

212 logging.exception("Fatal error during application startup") 

213 traceback.print_exc() 

214 sys.exit(1) 

215 

216 

217if __name__ == "__main__": 

218 main()