refactor(web_server.py): rewrite to use OOP
This commit is contained in:
@@ -1,171 +1,53 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import Final
|
||||
from __future__ import annotations
|
||||
|
||||
import flask
|
||||
import flask_login
|
||||
import logging as lg
|
||||
from flask import Flask
|
||||
from flask_login import LoginManager
|
||||
|
||||
PASSWORD: Final[str] = "123"
|
||||
|
||||
app = flask.Flask(__name__)
|
||||
app.secret_key = "dildo"
|
||||
|
||||
login_manager = flask_login.LoginManager()
|
||||
login_manager.init_app(app)
|
||||
from judas_server.web.user import load_user
|
||||
|
||||
|
||||
PC_DETAILS = {
|
||||
"PC1": {
|
||||
"id": "PC1",
|
||||
"status": "online",
|
||||
"last_seen": "2023-10-01 12:00:00",
|
||||
},
|
||||
"PC2": {
|
||||
"id": "PC2",
|
||||
"status": "offline",
|
||||
"last_seen": "2023-10-01 11:00:00",
|
||||
},
|
||||
"PC3": {
|
||||
"id": "PC3",
|
||||
"status": "offline",
|
||||
"last_seen": "2023-10-01 11:00:00",
|
||||
},
|
||||
"PC4": {
|
||||
"id": "PC4",
|
||||
"status": "offline",
|
||||
"last_seen": "2023-10-01 11:00:00",
|
||||
},
|
||||
"PC5": {
|
||||
"id": "PC5",
|
||||
"status": "offline",
|
||||
"last_seen": "2023-10-01 11:00:00",
|
||||
},
|
||||
"PC6": {
|
||||
"id": "PC6",
|
||||
"status": "offline",
|
||||
"last_seen": "2023-10-01 11:00:00",
|
||||
},
|
||||
"PC7": {
|
||||
"id": "PC7",
|
||||
"status": "offline",
|
||||
"last_seen": "2023-10-01 11:00:00",
|
||||
},
|
||||
}
|
||||
class JudasWebServer:
|
||||
def __init__(self, secret_key: str) -> None:
|
||||
self.logger: lg.Logger = lg.getLogger(
|
||||
f"{__name__}.{self.__class__.__name__}"
|
||||
)
|
||||
self.logger.debug("Initializing JudasWebServer...")
|
||||
|
||||
self.app: Flask = Flask(
|
||||
__name__, static_folder="static", template_folder="templates"
|
||||
)
|
||||
self.app.secret_key = secret_key
|
||||
|
||||
class User(flask_login.UserMixin):
|
||||
"""Represents a user for authentication purposes."""
|
||||
# hard-code password
|
||||
self.app.config["PASSWORD"] = "123"
|
||||
|
||||
def __init__(self, id: str):
|
||||
super().__init__()
|
||||
self.id = id
|
||||
# extensions
|
||||
self.login_manager: LoginManager = LoginManager()
|
||||
self.app.config["LoginManager"] = self.login_manager
|
||||
|
||||
self.configure_extensions()
|
||||
self.init_routes()
|
||||
|
||||
@login_manager.user_loader
|
||||
def load_user(user_id: str) -> User | None:
|
||||
"""Loads a user by user_id.
|
||||
def configure_extensions(self) -> None:
|
||||
self.logger.debug("Configuring extensions...")
|
||||
self.login_manager.init_app(self.app)
|
||||
self.login_manager.user_loader(load_user)
|
||||
self.login_manager.login_view = "auth.login"
|
||||
|
||||
Args:
|
||||
user_id: The ID of the user.
|
||||
# TODO: add login page
|
||||
|
||||
Returns:
|
||||
The User object if found, else None.
|
||||
"""
|
||||
if user_id == "admin":
|
||||
return User("admin")
|
||||
return None
|
||||
def init_routes(self) -> None:
|
||||
self.logger.debug("Initializing routes...")
|
||||
from judas_server.web.routes import auth_bp, index_bp, panel_bp
|
||||
|
||||
self.app.register_blueprint(index_bp)
|
||||
self.app.register_blueprint(auth_bp)
|
||||
self.app.register_blueprint(panel_bp)
|
||||
|
||||
@app.route("/")
|
||||
def index() -> flask.Response | str:
|
||||
"""Renders the index page with a link to the login page."""
|
||||
return flask.render_template(
|
||||
"index.html", logged=flask_login.current_user.is_authenticated
|
||||
)
|
||||
|
||||
|
||||
@app.route("/login", methods=["GET", "POST"])
|
||||
def login() -> flask.Response | str:
|
||||
"""Handles user login via password form."""
|
||||
if flask.request.method == "POST":
|
||||
password = flask.request.form.get("password", "")
|
||||
if password == PASSWORD:
|
||||
user = User("admin")
|
||||
flask_login.login_user(user)
|
||||
return flask.redirect(flask.url_for("panel"))
|
||||
else:
|
||||
return flask.render_template(
|
||||
"login.html",
|
||||
error="Invalid password. Please try again.",
|
||||
)
|
||||
return flask.render_template("login.html")
|
||||
|
||||
|
||||
@app.route("/logout")
|
||||
@flask_login.login_required
|
||||
def logout() -> str:
|
||||
"""Logs out the current user."""
|
||||
flask_login.logout_user()
|
||||
return flask.redirect(flask.url_for("index"))
|
||||
|
||||
|
||||
@app.route("/panel")
|
||||
@flask_login.login_required
|
||||
def panel() -> str:
|
||||
"""Renders the main panel page with PC details.
|
||||
|
||||
Returns:
|
||||
Rendered HTML template with PC details.
|
||||
"""
|
||||
return flask.render_template(
|
||||
"panel.html",
|
||||
username=flask_login.current_user.id,
|
||||
pcs=PC_DETAILS,
|
||||
)
|
||||
|
||||
|
||||
@app.route("/details/<pc_id>")
|
||||
@flask_login.login_required
|
||||
def details(pc_id: str) -> str:
|
||||
"""Renders the details page for a specific PC.
|
||||
|
||||
Args:
|
||||
pc_id: The ID of the PC to display details for.
|
||||
|
||||
Returns:
|
||||
Rendered HTML template with PC details.
|
||||
"""
|
||||
return flask.render_template(
|
||||
"details.html",
|
||||
username=flask_login.current_user.id,
|
||||
pc=PC_DETAILS[pc_id],
|
||||
)
|
||||
|
||||
|
||||
@app.route("/stream/<pc_id>")
|
||||
@flask_login.login_required
|
||||
def stream(pc_id: str) -> str:
|
||||
"""Renders the stream page for a specific PC.
|
||||
|
||||
Args:
|
||||
pc_id: The ID of the PC to stream from.
|
||||
|
||||
Returns:
|
||||
Rendered HTML template for streaming.
|
||||
"""
|
||||
return flask.render_template(
|
||||
"stream.html",
|
||||
logged=True,
|
||||
username=flask_login.current_user.id,
|
||||
pc=PC_DETAILS[pc_id],
|
||||
)
|
||||
|
||||
|
||||
@app.route("/stream_feed/<pc_id>")
|
||||
@flask_login.login_required
|
||||
def stream_feed(pc_id: str) -> flask.Response:
|
||||
return flask.Response(mimetype="multipart/x-mixed-replace; boundary=frame")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(debug=True)
|
||||
def run(self, host: str = "0.0.0.0", port: int = 5000) -> None:
|
||||
self.logger.info(f"Starting server on {host}:{port}...")
|
||||
self.app.run(host=host, port=port)
|
||||
self.logger.info("Server stopped.")
|
||||
|
||||
Reference in New Issue
Block a user