134 lines
4.8 KiB
Python
134 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from flask import Flask, abort, jsonify, request, redirect, session
|
|
from flask_login import current_user, logout_user
|
|
|
|
from .config import Config, DATA_DIR, ensure_dirs
|
|
from .extensions import db, login_manager
|
|
from .models import AppSetting, User
|
|
from .migrations import upgrade_schema
|
|
from .prediction import FEATURES, load_model
|
|
from .security import csrf_token, validate_csrf_token
|
|
from .time_utils import current_year_for_timezone, format_datetime_for_timezone
|
|
from .time_utils import utc_now
|
|
|
|
|
|
def create_app(config_object: type[Config] = Config, *, load_model_on_start: bool = True) -> Flask:
|
|
ensure_dirs()
|
|
configure_logging()
|
|
|
|
app = Flask(__name__, template_folder="../templates", static_folder="../static")
|
|
app.config.from_object(config_object)
|
|
|
|
if app.config.get("SECRET_KEY_GENERATED"):
|
|
logging.warning("未设置 SECRET_KEY，已生成临时密钥；服务重启后登录会话将失效。")
|
|
|
|
db.init_app(app)
|
|
login_manager.init_app(app)
|
|
login_manager.login_view = "main.login"
|
|
login_manager.login_message = "请先登录后再访问该页面。"
|
|
|
|
register_app_hooks(app)
|
|
|
|
from .routes import bp
|
|
|
|
app.register_blueprint(bp)
|
|
|
|
with app.app_context():
|
|
upgrade_schema()
|
|
init_admin_user(app)
|
|
|
|
if load_model_on_start:
|
|
try:
|
|
app.config["RSF_MODEL"] = load_model(app.config["FUSION_MODEL_CORE_DIR"])
|
|
logging.info("模型加载成功")
|
|
except Exception as exc:
|
|
app.config["RSF_MODEL"] = None
|
|
logging.exception("模型加载失败: %s", exc)
|
|
|
|
return app
|
|
|
|
|
|
def configure_logging() -> None:
|
|
logging.basicConfig(
|
|
filename=str(DATA_DIR / "app.log"),
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(levelname)s - %(message)s",
|
|
)
|
|
|
|
|
|
def init_admin_user(app: Flask) -> None:
|
|
admin_username = app.config["ADMIN_USERNAME"]
|
|
admin_password = app.config["ADMIN_PASSWORD"]
|
|
admin_email = app.config["ADMIN_EMAIL"]
|
|
if admin_password and admin_email:
|
|
admin = User.query.filter_by(username=admin_username).first()
|
|
if admin is None:
|
|
admin = User(username=admin_username, email=admin_email, is_admin=True, is_active_account=True)
|
|
admin.set_password(admin_password)
|
|
db.session.add(admin)
|
|
else:
|
|
admin.is_admin = True
|
|
if not admin.email:
|
|
admin.email = admin_email
|
|
admin.email_verified_at = utc_now()
|
|
admin.is_active_account = True
|
|
db.session.commit()
|
|
elif not User.query.filter_by(is_admin=True).first():
|
|
logging.warning("未设置 ADMIN_PASSWORD 或 ADMIN_EMAIL，跳过自动创建管理员账号。")
|
|
|
|
default_admin = User.query.filter_by(username="admin", is_admin=True).first()
|
|
if default_admin and default_admin.check_password("admin123"):
|
|
logging.warning("检测到默认管理员密码 admin123，请通过账户安全页立即更新。")
|
|
|
|
|
|
def register_app_hooks(app: Flask) -> None:
|
|
@login_manager.user_loader
|
|
def load_user(user_id: str):
|
|
try:
|
|
return db.session.get(User, int(user_id))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
@app.context_processor
|
|
def inject_helpers():
|
|
def now_year() -> int:
|
|
return current_year_for_timezone(app.config["APP_TIMEZONE"])
|
|
|
|
def format_datetime(value) -> str:
|
|
return format_datetime_for_timezone(value, app.config["APP_TIMEZONE"])
|
|
|
|
return {
|
|
"feature_list": FEATURES,
|
|
"now_year": now_year,
|
|
"format_datetime": format_datetime,
|
|
"csrf_token": csrf_token,
|
|
"allow_registration": AppSetting.get_bool(
|
|
"allow_registration",
|
|
app.config["ALLOW_REGISTRATION"],
|
|
),
|
|
}
|
|
|
|
@app.errorhandler(413)
|
|
def handle_file_too_large(_exc):
|
|
max_mb = app.config["MAX_CONTENT_LENGTH"] // (1024 * 1024)
|
|
if request.path == "/predict":
|
|
return jsonify({"error": f"文件过大，请上传 {max_mb}MB 以内的文件。"}), 413
|
|
return f"文件过大，请上传 {max_mb}MB 以内的文件。", 413
|
|
|
|
@app.before_request
|
|
def protect_csrf():
|
|
if current_user.is_authenticated and session.get("auth_version") != current_user.auth_version:
|
|
logout_user()
|
|
session.clear()
|
|
return redirect("/login")
|
|
if request.method not in {"POST", "PUT", "PATCH", "DELETE"}:
|
|
return None
|
|
if validate_csrf_token():
|
|
return None
|
|
if request.path == "/predict" or request.headers.get("X-Requested-With") == "XMLHttpRequest":
|
|
return jsonify({"error": "CSRF 校验失败，请刷新页面后重试。"}), 400
|
|
abort(400)
|