Mateusz Gruszczyński
114 lines
4.7 KiB
Python
114 lines
4.7 KiB
Python
from flask import Flask, request, url_for
|
|
import click
|
|
import hashlib
|
|
from pathlib import Path
|
|
from .extensions import db, login_manager
|
|
from .models import User, AppSetting
|
|
from .config import Config
|
|
|
|
THEMES = {name: {} for name in ("bootstrap", "flatly", "darkly", "pulse")}
|
|
|
|
|
|
def _static_hashes(app):
|
|
static_root = Path(app.static_folder)
|
|
hashes = {}
|
|
for path in static_root.rglob("*"):
|
|
if path.is_file():
|
|
relative = path.relative_to(static_root).as_posix()
|
|
hashes[relative] = hashlib.md5(path.read_bytes()).hexdigest()
|
|
return hashes
|
|
|
|
def _themes(app):
|
|
return {
|
|
"bootstrap": {"label":"Bootstrap", "url":app.config["THEME_BOOTSTRAP_URL"], "mode":"light"},
|
|
"flatly": {"label":"Flatly", "url":app.config["THEME_FLATLY_URL"], "mode":"light"},
|
|
"darkly": {"label":"Ciemny", "url":app.config["THEME_BOOTSTRAP_URL"], "mode":"dark"},
|
|
"pulse": {"label":"Pulse", "url":app.config["THEME_PULSE_URL"], "mode":"light"},
|
|
}
|
|
|
|
|
|
def create_app(test_config=None):
|
|
app = Flask(__name__)
|
|
app.config.from_object(Config)
|
|
if test_config: app.config.update(test_config)
|
|
db.init_app(app); login_manager.init_app(app)
|
|
|
|
from .database import prepare_database
|
|
prepare_database(app)
|
|
from .auth import auth; from .main import main
|
|
from .api import api
|
|
from .openapi import docs
|
|
app.register_blueprint(auth); app.register_blueprint(main); app.register_blueprint(api); app.register_blueprint(docs)
|
|
|
|
static_hashes = _static_hashes(app)
|
|
|
|
@app.context_processor
|
|
def inject_static_url():
|
|
def static_url(filename):
|
|
return url_for("static", filename=filename, v=static_hashes.get(filename, ""))
|
|
return {"static_url": static_url}
|
|
|
|
@app.after_request
|
|
def set_cache_headers(response):
|
|
if request.endpoint == "static":
|
|
filename = request.view_args.get("filename", "") if request.view_args else ""
|
|
if filename.endswith(".css"):
|
|
response.headers["Cache-Control"] = "public, max-age=1800, immutable"
|
|
else:
|
|
response.headers["Cache-Control"] = "public, max-age=1800, immutable"
|
|
elif response.mimetype == "text/html":
|
|
response.headers["Cache-Control"] = "private, no-store"
|
|
return response
|
|
|
|
@login_manager.user_loader
|
|
def load_user(user_id): return db.session.get(User, int(user_id))
|
|
|
|
@login_manager.request_loader
|
|
def load_user_from_request(req):
|
|
header = req.headers.get("Authorization", "")
|
|
scheme, _, token = header.partition(" ")
|
|
if scheme.lower() != "bearer" or not token.strip():
|
|
return None
|
|
from .api_tokens import load_access_token
|
|
return load_access_token(token.strip())
|
|
|
|
@login_manager.unauthorized_handler
|
|
def unauthorized():
|
|
if request.path.startswith("/api/"):
|
|
return {"ok": False, "message": "Wymagane uwierzytelnienie Bearer lub aktywna sesja"}, 401
|
|
from flask import redirect
|
|
return redirect(url_for("auth.login", next=request.url))
|
|
|
|
@app.context_processor
|
|
def inject_theme():
|
|
themes = _themes(app)
|
|
record = db.session.get(AppSetting, "theme")
|
|
theme_name = record.value if record and record.value in themes else "bootstrap"
|
|
return {"app_theme": themes[theme_name], "app_theme_name": theme_name, "themes": themes,
|
|
"asset_urls": {"bootstrap_js": app.config["BOOTSTRAP_JS_URL"], "choices_css": app.config["CHOICES_CSS_URL"],
|
|
"choices_js": app.config["CHOICES_JS_URL"], "chart_js": app.config["CHART_JS_URL"]}}
|
|
|
|
@app.get("/health")
|
|
def health(): return {"status":"ok"}
|
|
|
|
@app.cli.command("reset-admin-password")
|
|
@click.option("--email", required=True, help="Adres e-mail konta administratora.")
|
|
@click.option("--password", prompt=True, hide_input=True, confirmation_prompt=True,
|
|
help="Nowe hasło. Bez tej opcji polecenie poprosi o hasło bez wyświetlania go.")
|
|
def reset_admin_password(email, password):
|
|
"""Resetuje hasło istniejącego administratora."""
|
|
normalized_email = email.strip().lower()
|
|
user = User.query.filter_by(email=normalized_email).first()
|
|
if not user:
|
|
raise click.ClickException("Nie znaleziono użytkownika o podanym adresie e-mail.")
|
|
if user.role != "admin":
|
|
raise click.ClickException("Wskazany użytkownik nie ma roli administratora.")
|
|
if len(password) < 8:
|
|
raise click.ClickException("Hasło musi mieć co najmniej 8 znaków.")
|
|
user.set_password(password)
|
|
user.active = True
|
|
db.session.commit()
|
|
click.echo(f"Zresetowano hasło administratora: {normalized_email}")
|
|
|
|
return app
|