import os import re from datetime import datetime from typing import List, cast import flask_login from flask_socketio import join_room import pytz from flask import ( Blueprint, abort, current_app, # Used for logging flash, make_response, redirect, render_template, request, send_from_directory, session, url_for, ) from . import socketio from .auth import check_password, user_loader from .forms import DeckForm, InfoBannerForm, LoginForm, PasswordForm from .models import Course, Deck, InfoBanner, Password, db from .utils import ( create_token, generate_password, get_active_banner, get_all_courses, get_available_decks, get_passwords, get_token, get_unlocked_decks, is_deck_unlocked, retrieve_token, serialize_token, update_token, get_course ) main = Blueprint('main', __name__) slidev_deck_pattern = re.compile(r"^\/slides\/[^\/]+\/[^\/]+\/(?:\d+|index\.html)(?:\?.*)?$") slidev_all_pattern = re.compile(r"^\/slides\/([^\/]+)\/([^\/]+)(?:\/presenter)?\/(\d+)(?:\?.*)?$") @socketio.on("connect") def on_connect(): if flask_login.current_user.is_authenticated: join_room("instructor") current_app.logger.info(f"Instructor connected to socket: {flask_login.current_user.id}") @main.after_request def inject_confused_button(response): if (slidev_deck_pattern.match(request.path) or request.path == '/') and response.content_type.startswith('text/html') and not flask_login.current_user.is_authenticated : if response.direct_passthrough: response.direct_passthrough = False body = response.get_data(as_text=True) insert_at = body.find('') if insert_at != -1: button_html = """ """ body = body[:insert_at] + button_html + body[insert_at:] response.set_data(body) return response @main.route('/') def index(): token = get_token() if not token: current_app.logger.debug("User has no token. Redirecting to set_token.") return redirect(url_for("main.set_token")) else: current_app.logger.debug(f"User {token.name} is viewing index with {len(token.decks)} decks unlocked.") return render_template('index.html', active_banner=get_active_banner(), decks=get_unlocked_decks(token), user_name=token.name, active_course_name=token.active_course.name, courses=get_all_courses()) def set_cookie(token): next_url = session.pop('next_url', None) if not next_url: current_app.logger.debug("next_url not present redirecting to index") next_url = url_for("main.index") resp = make_response(redirect(next_url)) resp.set_cookie( 'access_token', serialize_token(token), httponly=True, samesite='Lax', path='/' ) return resp @main.route('/set_token', methods=['POST', 'GET']) def set_token(): """ Handles. GET: Renders the registration form. POST: Processes the form submission. """ if request.method == 'POST': # Get form data username = request.form.get('username') # Simple validation if not username: current_app.logger.debug("Login failed: No username supplied") flash('Username is required.', 'error') return render_template('set_token.html') token = retrieve_token(username) if not token: flash(f'The username "{username}" does not exist.', 'error') current_app.logger.warning(f"Login failed: Username '{username}' not found.") return render_template('set_token.html') current_app.logger.info(f"User logged in: {username}") return set_cookie(token) return render_template('set_token.html') @main.route('/register', methods=['POST', 'GET']) def register(): """ Handles user registration. GET: Renders the registration form. POST: Processes the form submission. """ if request.method == 'POST': # Get form data user_name = request.form.get('username') acknowledge = request.form.get('acknowledge') # Simple validation if not user_name: flash('Username is required.', 'error') current_app.logger.debug("Username is required for registering") return render_template('register.html') if retrieve_token(user_name): flash(f'The username "{user_name}" is already taken.', 'error') current_app.logger.debug(f"Username {user_name} is already registered") return render_template('register.html') if not acknowledge: flash('You must acknowledge the important information to proceed.', 'error') current_app.logger.debug("User did not acknowledge register information") return render_template('register.html') # If all validation passes, "register" the user token = create_token(user_name) current_app.logger.info(f"New user registered: {user_name}") return set_cookie(token) # If GET request, render the form return render_template('register.html') @main.route('/logout', methods=['POST']) def logout(): token = get_token() user_name = token.name if token else "Unknown" current_app.logger.info(f"User logout: {user_name}") resp = make_response(redirect(url_for('main.index'))) # Redirect to a safe page, e.g., index or login resp.set_cookie('access_token', '', expires=0) # Clears the cookie return resp @main.route('/switch_course/', methods=['POST']) def switch_course(new_course: int): token = get_token() if not token: return redirect(url_for('main.set_token')) switched = get_course(new_course) if switched: token.active_course = switched db.session.commit() current_app.logger.info(f"User {token.name} switched to course: {switched.name}") else: current_app.logger.warning(f"User {token.name} attempted to switch to invalid course ID: {new_course}") return redirect(url_for("main.index")) @main.route('/admin/login', methods=['POST', 'GET']) def admin_login(): form = LoginForm() if form.validate_on_submit(): assert isinstance(form.user_name.data, str) assert isinstance(form.password.data, str) user_name: str = form.user_name.data password: str = form.password.data user = user_loader(user_name) if user and check_password(user, password): flask_login.login_user(user) current_app.logger.info(f"Admin login successful: {user_name}") next = request.args.get('next') if next is None: return redirect(url_for('main.index')) if not next.startswith('/admin'): return abort(400) return redirect(next) else: current_app.logger.warning(f"Failed admin login attempt: {user_name}") flash('Incorrect username or password. Please try again.', 'error') return render_template("admin_login.html", form=form) def process_access_code(access_code: str): token = get_token() if not token: # remember where the user wanted to go session['next_url'] = url_for('main.access_get', access_code=access_code) current_app.logger.debug("Access code used without session. Redirecting to login.") return redirect(url_for('main.set_token')) password = db.session.execute( db.select(Password) .where(Password.value == access_code) ).scalar_one_or_none() if password is not None: if not password.is_expired(): update_token(password.decks) current_app.logger.info(f"User {token.name} used valid access code.") else: current_app.logger.info(f"User {token.name} used expired access code.") flash('Password has expired. Please obtain a new one.', 'expired') else: current_app.logger.warning(f"User {token.name} used invalid access code: {access_code}") flash('Password is not valid. Please check the code and try again.', 'error') return redirect(url_for('main.index')) @main.route('/access', methods=['POST']) def access_post(): access_code = request.form.get('password') if not access_code: return redirect(url_for("main.index")) return process_access_code(access_code) # Route for direct link/QR code access (GET) @main.route('/access/', methods=['GET']) def access_get(access_code: str): return process_access_code(access_code) @main.route('/confused', methods=['POST']) def confused(): socketio.emit( "confused", { "time": datetime.utcnow().isoformat() }, to="instructor" ) return ("", 204) def log_404(reason: str, deck: str, course: str, path: str): """ Helper function to log 404 errors with the user's identity. Uses current_app.logger to ensure logs are handled by the app's config. """ token = get_token() user_identity = f"User: {token.name} (ID: {token.id})" if token else "User: Anonymous" current_app.logger.warning(f"404 Not Found [{reason}] | {user_identity} | Resource: {course}/{deck}/{path}") @main.route('/slides///') @main.route('/slides///') def serve_deck(deck, course_folder, path='index.html'): course = db.session.execute(db.select(Course).where(Course.folder==course_folder)).scalar() if not course: log_404("Course not found", deck, course_folder, path) abort(404) assert isinstance(course, Course) deck_obj = db.session.execute(db.select(Deck).where(Deck.deck == deck, Deck.course_id == course.id)).scalar() if not deck_obj: log_404("Deck not in database", deck, course_folder, path) abort(404) assert isinstance(deck_obj, Deck) if not is_deck_unlocked(deck_obj.deck): # We generally do not log redirects for locked content as 404s, # but you could log a warning here if desired. current_app.logger.info(f"Access denied (locked) for user {get_token().name if get_token() else 'Anon'} -> {deck_obj.name}") return redirect(url_for('main.index')) slides_dir = os.path.join(current_app.config['SLIDES_DIR'], course.folder) deck_path = os.path.join(slides_dir, deck_obj.deck) if not os.path.exists(deck_path): log_404("Deck directory missing on disk", deck, course_folder, path) return f"Deck '{deck_obj.name}' not found.", 404 if slidev_all_pattern.match(request.path): current_app.logger.info(f"User reloaded in deck {deck_obj.name}") return send_from_directory(deck_path, deck_obj.index_file) requested_file = os.path.join(deck_path, path) if os.path.isfile(requested_file): if path == deck_obj.index_file: current_app.logger.info(f"User {get_token().name if get_token() else 'Anon'} accessed deck {deck_obj.name}") return send_from_directory(deck_path, path) log_404("File missing within deck", deck, course_folder, path) abort(404) @main.route('/exports///') def serve_export(deck, course_folder, path): course = db.session.execute(db.select(Course).where(Course.folder==course_folder)).scalar() if not course: log_404("Export: Course not found", deck, course_folder, path) abort(404) assert isinstance(course, Course) deck_obj = db.session.execute(db.select(Deck).where(Deck.deck==deck, Deck.course_id == course.id)).scalar() if not deck_obj: log_404("Export: Deck not in database", deck, course_folder, path) abort(404) assert isinstance(deck_obj, Deck) if not is_deck_unlocked(deck_obj.deck): return redirect(url_for('main.index')) exports_dir = os.path.join(current_app.config['EXPORTS_DIR'], course.folder) deck_path = os.path.join(exports_dir, deck_obj.deck) requested_file = os.path.join(deck_path, path) if os.path.isfile(requested_file): current_app.logger.info(f"User {get_token().name if get_token() else 'Anon'} exported deck {deck_obj.name}") return send_from_directory(deck_path, path) log_404("Export file missing", deck, course_folder, path) abort(404) @main.route('/admin', methods=['GET', 'POST']) @flask_login.login_required def admin(): active_course = get_course(flask_login.current_user.active_course_id) if active_course is None: current_app.logger.error(f"Admin user {flask_login.current_user.id} has no active_course_id.") raise Exception deck_form = DeckForm() password_form = PasswordForm() if request.method == 'GET': password_form.value.data = generate_password() deck_form.deck.choices = get_available_decks(current_app, active_course) decks = active_course.decks password_form.decks.choices = [(deck.id, deck.name) for deck in decks] if request.method == 'POST': if "submit_deck" in request.form and deck_form.validate(): deck = Deck( name=cast(str,deck_form.name.data), deck=deck_form.deck.data, course=active_course, description=cast(str,deck_form.description.data), ) db.session.add(deck) db.session.commit() current_app.logger.info(f"Admin created deck: {deck.name}") if "submit_password" in request.form: if password_form.validate(): value: str = password_form.value.data expires_at: datetime | None = password_form.expires_at.data if expires_at: tz = pytz.timezone('Europe/Berlin') utc = pytz.utc expires_at = tz.localize(expires_at).astimezone(utc) password = Password( value=value, expires_at=expires_at ) decks = db.session.execute(db.select(Deck).filter(Deck.id.in_(cast(List[Deck], password_form.decks.data)))).scalars() password.decks.extend(decks) db.session.add(password) db.session.commit() current_app.logger.info(f"Admin created password: {value} (Expires: {expires_at})") else: # Form was submitted but validation failed for field, errors in password_form.errors.items(): for error in errors: current_app.logger.warning(f"Admin form validation error ({field}): {error}") return redirect(url_for('main.admin')) return render_template("admin.html", deck_form=deck_form, password_form=password_form, decks=decks, passwords=get_passwords(active_course), courses=get_all_courses(), active_course_name=active_course.name) @main.route('/admin/switch_course/', methods=['POST']) @flask_login.login_required def switch_course_admin(new_course: int): user = flask_login.current_user if get_course(new_course): user.active_course_id = new_course current_app.logger.info(f"Admin {user.id} switched to course ID {new_course}") return redirect(url_for("main.admin")) @main.route('/admin/banner', methods=['POST', 'GET']) @flask_login.login_required def banner(): banner_form = InfoBannerForm() if banner_form.validate_on_submit(): assert isinstance(banner_form.content.data, str) content : str = banner_form.content.data banner = InfoBanner(content=content) db.session.add(banner) db.session.commit() current_app.logger.info(f"Admin added banner: {content[:30]}...") return render_template("banner.html", banner_form=banner_form, banners=db.session.execute(db.select(InfoBanner)).scalars()) @main.route('/admin/decks//delete', methods=['POST']) @flask_login.login_required def delete_deck(deck_id: int): deck = Deck.query.get_or_404(deck_id) current_app.logger.info(f"Admin deleted deck: {deck.name}") db.session.delete(deck) db.session.commit() return redirect(url_for('main.admin')) @main.route('/admin/passwords//delete', methods=['POST']) @flask_login.login_required def delete_password(password_id: int): password = Password.query.get_or_404(password_id) current_app.logger.info(f"Admin deleted password ID {password_id}") db.session.delete(password) db.session.commit() return redirect(url_for('main.admin')) @main.route('/admin/banner/show/', methods=['POST']) @flask_login.login_required def set_active_banner(banner_id): banner = InfoBanner.query.get(banner_id) if banner: # ensures that only one banner is active at all times active_banner = get_active_banner() if active_banner: active_banner.is_active = False banner.is_active = True db.session.commit() current_app.logger.info(f"Admin activated banner ID {banner_id}") return redirect(url_for('main.banner')) @main.route('/admin/banner/hide', methods=['POST']) @flask_login.login_required def hide_active_banner(): banner = get_active_banner() if banner: banner.is_active = False db.session.commit() current_app.logger.info("Admin hid active banner") return redirect(url_for('main.banner')) @main.route('/admin/confused') @flask_login.login_required def confused_notifications(): return render_template("confused.html")