"""Flask application factory for hermes-dashboard. Read-only monitoring dashboard for omo + Gitea + kanban. """ import logging import os import time from datetime import datetime from zoneinfo import ZoneInfo from flask import Flask, g, has_request_context, jsonify, request import uuid import config import security from extensions import close_db from routes import api_bp, pages_bp # Logging setup class _RequestIdFilter(logging.Filter): """Inject request_id into LogRecord.""" def filter(self, record: logging.LogRecord) -> bool: if has_request_context(): record.request_id = getattr(g, "request_id", "-") else: record.request_id = "-" return True _log_format = "%(asctime)s [%(process)d] [%(request_id)s] %(levelname)s: %(message)s" _handler = logging.StreamHandler() _handler.setFormatter(logging.Formatter(_log_format)) _handler.addFilter(_RequestIdFilter()) _root = logging.getLogger() _root.setLevel(logging.INFO) if not any(isinstance(h, logging.StreamHandler) and h.formatter for h in _root.handlers): _root.addHandler(_handler) logger = logging.getLogger(__name__) def _format_dt(ts: int | str | None) -> str: """Format unix epoch as Asia/Shanghai local time.""" if not ts: return "" try: epoch = int(ts) dt = datetime.fromtimestamp(epoch, tz=ZoneInfo("Asia/Shanghai")) return dt.strftime("%Y-%m-%d %H:%M:%S") except (ValueError, TypeError, OSError): return "" def _format_rel(ts: int | str | None) -> str: """Format unix epoch as relative time (e.g., '3m ago').""" if not ts: return "" try: epoch = int(ts) now = int(time.time()) diff = now - epoch if diff < 0: return "in the future" if diff < 60: return f"{diff}s ago" if diff < 3600: return f"{diff // 60}m ago" if diff < 86400: return f"{diff // 3600}h ago" return f"{diff // 86400}d ago" except (ValueError, TypeError, OSError): return "" def create_app() -> Flask: """Create and configure the Flask application.""" app = Flask( __name__, template_folder=os.path.join(config.BASE_DIR, "templates"), static_folder=os.path.join(config.BASE_DIR, "static"), ) app.secret_key = config.SECRET_KEY # Jinja filters app.jinja_env.filters["dt"] = _format_dt app.jinja_env.filters["rel"] = _format_rel @app.before_request def _assign_request_id() -> None: g.request_id = uuid.uuid4().hex[:8] @app.after_request def _log_request(response): logger.info( "%s %s -> %s (%s)", request.method, request.path, response.status_code, g.request_id, ) return response @app.route("/healthz") def healthz(): return jsonify({"status": "ok"}), 200 app.teardown_appcontext(close_db) security.init_app(app) app.register_blueprint(api_bp) app.register_blueprint(pages_bp) return app app = create_app() if __name__ == "__main__": app.run(host="0.0.0.0", port=config.PORT, debug=False, threaded=True)