|
|
|
from __future__ import annotations |
|
from datetime import datetime, timedelta, timezone |
|
from typing import TYPE_CHECKING, Any |
|
|
|
from flask import Flask, abort, redirect, request, url_for |
|
from flask_admin import Admin, AdminIndexView, expose, helpers |
|
from flask_admin.consts import ICON_TYPE_FONT_AWESOME |
|
from flask_admin.contrib.sqla import ModelView |
|
from flask_babel import Babel |
|
from flask_caching import Cache |
|
from flask_login import current_user |
|
from flask_security.core import RoleMixin, Security, UserMixin |
|
from flask_security.datastore import SQLAlchemyUserDatastore |
|
from flask_security.utils import hash_password |
|
from flask_sqlalchemy import SQLAlchemy |
|
from sqlalchemy import inspect |
|
from wtforms import PasswordField |
|
|
|
from admin.views.users import UserView as AppUserView |
|
from bot.database.models import UserModel as AppUserModel |
|
|
|
if TYPE_CHECKING: |
|
from werkzeug.wrappers.response import Response |
|
|
|
|
|
app = Flask(__name__) |
|
app.config.from_pyfile("config.py") |
|
db = SQLAlchemy(app) |
|
cache = Cache(app) |
|
babel = Babel(app) |
|
|
|
|
|
roles_admins = db.Table( |
|
"roles_admins", |
|
db.Column("admin_id", db.Integer(), db.ForeignKey("admin.id")), |
|
db.Column("role_id", db.Integer(), db.ForeignKey("role.id")), |
|
) |
|
|
|
|
|
class RoleModel(db.Model, RoleMixin): |
|
__tablename__ = "role" |
|
|
|
id = db.Column(db.Integer(), primary_key=True) |
|
name = db.Column(db.String(80), unique=True) |
|
description = db.Column(db.String(255)) |
|
|
|
def __str__(self) -> str: |
|
return self.name |
|
|
|
|
|
class AdminModel(db.Model, UserMixin): |
|
__tablename__ = "admin" |
|
|
|
id = db.Column(db.Integer, primary_key=True) |
|
first_name = db.Column(db.String(255)) |
|
last_name = db.Column(db.String(255)) |
|
email = db.Column(db.String(255), unique=True, nullable=False) |
|
password = db.Column(db.String(255), nullable=False) |
|
active = db.Column(db.Boolean()) |
|
confirmed_at = db.Column(db.DateTime(), default=datetime.utcnow) |
|
fs_uniquifier = db.Column(db.String(255), unique=True) |
|
roles = db.relationship("RoleModel", secondary=roles_admins, backref=db.backref("admins", lazy="dynamic")) |
|
|
|
def __str__(self) -> str: |
|
return self.email |
|
|
|
|
|
|
|
admin_datastore = SQLAlchemyUserDatastore(db, AdminModel, RoleModel) |
|
security = Security(app, admin_datastore) |
|
|
|
|
|
|
|
class RoleView(ModelView): |
|
can_delete = False |
|
can_edit = False |
|
can_create = False |
|
can_view_details = False |
|
edit_modal = True |
|
create_modal = True |
|
can_export = False |
|
details_modal = True |
|
|
|
def is_accessible(self) -> bool: |
|
if not current_user.is_active or not current_user.is_authenticated: |
|
return False |
|
|
|
return bool(current_user.has_role("superuser")) |
|
|
|
def _handle_view(self, _name: str, **_kwargs: dict) -> Response | None: |
|
"""Override builtin _handle_view in order to redirect users when a view is not accessible.""" |
|
if not self.is_accessible(): |
|
if current_user.is_authenticated: |
|
|
|
abort(403) |
|
else: |
|
|
|
return redirect(url_for("security.login", next=request.url)) |
|
return None |
|
|
|
|
|
class AdminView(RoleView): |
|
can_view_details = True |
|
can_delete = True |
|
can_edit = True |
|
can_export = True |
|
can_create = True |
|
export_types = ["csv", "xlsx", "json", "yaml"] |
|
|
|
column_editable_list = ["email", "first_name", "last_name"] |
|
column_searchable_list = column_editable_list |
|
column_exclude_list = ["password"] |
|
form_excluded_columns = ["confirmed_at"] |
|
column_details_exclude_list = column_exclude_list |
|
column_filters = column_editable_list |
|
form_overrides = {"password": PasswordField} |
|
|
|
|
|
|
|
def get_orders_count() -> int: |
|
return 0 |
|
|
|
|
|
def get_user_count() -> int: |
|
return db.session.query(AppUserModel).count() |
|
|
|
|
|
def get_new_user_count(days_before: int = 1) -> int: |
|
period_start = datetime.now(timezone.utc) - timedelta(days=days_before) |
|
return db.session.query(AppUserModel).filter(AppUserModel.created_at >= period_start).count() |
|
|
|
|
|
class CustomAdminIndexView(AdminIndexView): |
|
@expose("/") |
|
def index(self) -> str: |
|
days_before: int = 1 |
|
period_start = datetime.now(timezone.utc) - timedelta(days=days_before) |
|
order_count = get_orders_count() |
|
user_count = get_user_count() |
|
new_user_count = get_new_user_count(days_before) |
|
new_user_count = get_new_user_count(days_before) |
|
|
|
return self.render( |
|
"admin/index.html", |
|
order_count=order_count, |
|
user_count=user_count, |
|
new_user_count=new_user_count, |
|
period_start=period_start, |
|
default_email=app.config.get("DEFAULT_ADMIN_EMAIL"), |
|
default_password=app.config.get("DEFAULT_ADMIN_PASSWORD"), |
|
) |
|
|
|
|
|
@app.route("/") |
|
def index() -> Response: |
|
return redirect(url_for("admin.index")) |
|
|
|
|
|
|
|
admin = Admin( |
|
app, |
|
name="Telegram Bot", |
|
base_template="my_master.html", |
|
index_view=CustomAdminIndexView( |
|
name="Home", |
|
url="/admin", |
|
menu_icon_type=ICON_TYPE_FONT_AWESOME, |
|
menu_icon_value="fa-home", |
|
), |
|
template_mode="bootstrap4", |
|
) |
|
|
|
admin.add_view( |
|
AppUserView( |
|
AppUserModel, |
|
db.session, |
|
menu_icon_type=ICON_TYPE_FONT_AWESOME, |
|
menu_icon_value="fa-users", |
|
name="Users", |
|
endpoint="users", |
|
), |
|
) |
|
|
|
admin.add_view( |
|
AdminView( |
|
AdminModel, |
|
db.session, |
|
menu_icon_type=ICON_TYPE_FONT_AWESOME, |
|
menu_icon_value="fa-black-tie", |
|
name="Admins", |
|
endpoint="admins", |
|
), |
|
) |
|
admin.add_view( |
|
RoleView( |
|
RoleModel, |
|
db.session, |
|
menu_icon_type=ICON_TYPE_FONT_AWESOME, |
|
menu_icon_value="fa-tags", |
|
name="Roles", |
|
endpoint="roles", |
|
), |
|
) |
|
|
|
|
|
|
|
@security.context_processor |
|
def security_context_processor() -> dict[str, Any]: |
|
return { |
|
"admin_base_template": admin.base_template, |
|
"admin_view": admin.index_view, |
|
"h": helpers, |
|
"get_url": url_for, |
|
} |
|
|
|
|
|
def init_db() -> None: |
|
inspector = inspect(db.engine) |
|
if inspector.has_table("admin") and inspector.has_table("role"): |
|
return |
|
|
|
db.create_all() |
|
|
|
admin_role = RoleModel(name="user", description="does not have access to other administrators") |
|
super_admin_role = RoleModel(name="superuser", description="has access to manage all administrators") |
|
db.session.add(admin_role) |
|
db.session.add(super_admin_role) |
|
db.session.commit() |
|
|
|
admin_datastore.create_user( |
|
first_name="Admin", |
|
email=app.config.get("DEFAULT_ADMIN_EMAIL"), |
|
password=hash_password(str(app.config.get("DEFAULT_ADMIN_PASSWORD"))), |
|
roles=[admin_role, super_admin_role], |
|
) |
|
|
|
db.session.commit() |
|
|
|
return |
|
|
|
|
|
with app.app_context(): |
|
init_db() |
|
|
|
if __name__ == "__main__": |
|
app.run(host=app.config.get("ADMIN_HOST"), port=app.config.get("ADMIN_PORT"), debug=app.config.get("DEBUG")) |
|
|