import logging import os from flask import Flask, redirect, url_for from flask_migrate import Migrate from zoneinfo import ZoneInfo from app.config import DevelopmentConfig, ProductionConfig from app.extensions import db, login_manager, scheduler from app.services.scheduler import SchedulerService def create_app() -> Flask: """ Application factory creating the Flask app, loading config, registering blueprints, initializing extensions, and booting the scheduler. """ app = Flask(__name__) config_name = os.getenv("FLASK_ENV", "development").lower() if config_name == "production": app_config = ProductionConfig() else: app_config = DevelopmentConfig() app.config.from_object(app_config) configure_logging(app) register_extensions(app) register_blueprints(app) register_template_filters(app) enable_scheduler = app.config.get("ENABLE_SCHEDULER", True) and os.getenv("FLASK_SKIP_SCHEDULER") != "1" if enable_scheduler: init_scheduler(app) else: app.logger.info("Scheduler not started (ENABLE_SCHEDULER=%s, FLASK_SKIP_SCHEDULER=%s)", app.config.get("ENABLE_SCHEDULER", True), os.getenv("FLASK_SKIP_SCHEDULER")) @app.route("/") def index(): return redirect(url_for("apis.list_apis")) return app def configure_logging(app: Flask) -> None: log_level = logging.DEBUG if app.debug else logging.INFO logging.basicConfig(level=log_level, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") def register_extensions(app: Flask) -> None: db.init_app(app) login_manager.init_app(app) Migrate(app, db) scheduler.init_app(app) def register_blueprints(app: Flask) -> None: from app.views.auth import auth_bp from app.views.apis import apis_bp from app.views.logs import logs_bp app.register_blueprint(auth_bp) app.register_blueprint(apis_bp) app.register_blueprint(logs_bp) def register_template_filters(app: Flask) -> None: @app.template_filter("cron_human") def cron_human(expr: str) -> str: """ 将常见的 5 字段 cron 表达式转换为简单中文描述,不能完全覆盖所有情况。 """ parts = expr.strip().split() if len(parts) != 5: return expr minute, hour, day, month, dow = parts # 每 N 分钟 if minute.startswith("*/") and hour == "*" and day == "*" and month == "*" and dow == "*": return f"每 {minute[2:]} 分钟" # 整点或固定时间 if minute.isdigit() and hour.isdigit() and day == "*" and month == "*" and dow in ("*", "?"): return f"每天 {hour.zfill(2)}:{minute.zfill(2)}" # 每 N 小时的整点 if minute == "0" and hour.startswith("*/") and day == "*" and month == "*" and dow in ("*", "?"): return f"每 {hour[2:]} 小时整点" # 每月某日 if minute.isdigit() and hour.isdigit() and day.isdigit() and month == "*" and dow in ("*", "?"): return f"每月 {day} 日 {hour.zfill(2)}:{minute.zfill(2)}" # 每周某天 weekday_map = {"0": "周日", "1": "周一", "2": "周二", "3": "周三", "4": "周四", "5": "周五", "6": "周六", "7": "周日"} if minute.isdigit() and hour.isdigit() and day in ("*", "?") and month == "*" and dow not in ("*", "?"): label = weekday_map.get(dow, f"周{dow}") return f"每{label} {hour.zfill(2)}:{minute.zfill(2)}" return expr @app.template_filter("to_cst") def to_cst(dt, fmt: str = "%Y-%m-%d %H:%M:%S"): """ 将 UTC 时间转换为中国标准时间字符串,如果值为空则返回空字符串。 """ if not dt: return "" try: tz = ZoneInfo(app.config.get("SCHEDULER_TIMEZONE", "Asia/Shanghai")) # 如果是 naive datetime,认为是 UTC if dt.tzinfo is None: dt = dt.replace(tzinfo=ZoneInfo("UTC")) return dt.astimezone(tz).strftime(fmt) except Exception: return str(dt) def init_scheduler(app: Flask) -> None: """ Start APScheduler and load enabled jobs from database. """ scheduler.start() with app.app_context(): service = SchedulerService(scheduler) service.load_enabled_jobs()