import os import json import importlib import importlib.util from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate from flask_login import LoginManager from flask_wtf.csrf import CSRFProtect from dotenv import load_dotenv # Load environment variables load_dotenv() # Initialize extensions db = SQLAlchemy() migrate = Migrate() login_manager = LoginManager() csrf = CSRFProtect() def create_app(): app = Flask(__name__) app.config.from_object('app.config.Config') # Initialize core extensions csrf.init_app(app) db.init_app(app) migrate.init_app(app, db) login_manager.init_app(app) login_manager.login_view = 'auth.login' # Register error handlers from .errors import bp as errors_bp app.register_blueprint(errors_bp) # Auto-discover and register plugins plugin_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'plugins')) for plugin in os.listdir(plugin_path): if plugin.endswith('.noload'): print(f"[⏭] Skipping plugin '{plugin}' (marked as .noload)") continue plugin_dir = os.path.join(plugin_path, plugin) if not os.path.isdir(plugin_dir): continue # 1. Register routes route_file = os.path.join(plugin_dir, 'routes.py') if os.path.isfile(route_file): try: spec = importlib.util.spec_from_file_location(f"plugins.{plugin}.routes", route_file) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) if hasattr(mod, 'bp'): app.register_blueprint(mod.bp, strict_slashes=False) except Exception as e: print(f"[⚠️] Failed to load routes from plugin '{plugin}': {e}") # Define paths init_file = os.path.join(plugin_dir, '__init__.py') plugin_json = os.path.join(plugin_dir, 'plugin.json') model_file = os.path.join(plugin_dir, 'models.py') # 2. Register CLI commands and run entry point if os.path.isfile(init_file): try: cli_module = importlib.import_module(f"plugins.{plugin}") if hasattr(cli_module, 'register_cli'): cli_module.register_cli(app) if os.path.isfile(plugin_json): with open(plugin_json, 'r') as f: meta = json.load(f) entry = meta.get('entry_point') if entry and hasattr(cli_module, entry): getattr(cli_module, entry)(app) except Exception as e: print(f"[⚠️] Failed to load CLI for plugin '{plugin}': {e}") ## 3. Auto-load plugin models for migrations #if os.path.isfile(model_file): # try: # spec = importlib.util.spec_from_file_location(f"plugins.{plugin}.models", model_file) # mod = importlib.util.module_from_spec(spec) # spec.loader.exec_module(mod) # except Exception as e: # print(f"[⚠️] Failed to load models from plugin '{plugin}': {e}") @app.context_processor def inject_current_year(): from datetime import datetime return {'current_year': datetime.now().year} return app @login_manager.user_loader def load_user(user_id): from plugins.auth.models import User return User.query.get(int(user_id))