majorly overhauled the design to now use docker-compose and postgres
This commit is contained in:
151
app/utils.py
151
app/utils.py
@@ -1,101 +1,131 @@
|
||||
import os
|
||||
import logging
|
||||
|
||||
from flask import Flask, request, session
|
||||
from flask import request, session, redirect, url_for, current_app
|
||||
from typing import cast
|
||||
from itsdangerous import BadSignature, URLSafeSerializer
|
||||
from xkcdpass import xkcd_password
|
||||
from functools import wraps, lru_cache
|
||||
|
||||
from .models import Deck, InfoBanner, Token, db, Course
|
||||
|
||||
# Create a module-level logger per Flask/Python standards
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def init_password_generator():
|
||||
wordfile = xkcd_password.locate_wordfile()
|
||||
global words
|
||||
words = xkcd_password.generate_wordlist(wordfile=wordfile, min_length=3, max_length=6)
|
||||
words = xkcd_password.generate_wordlist(
|
||||
wordfile=wordfile, min_length=3, max_length=6
|
||||
)
|
||||
logger.info("Password generator initialized.")
|
||||
|
||||
def init_serializer(app: Flask):
|
||||
|
||||
def init_serializer(app):
|
||||
global serializer
|
||||
serializer = URLSafeSerializer(app.config['SECRET_KEY'])
|
||||
serializer = URLSafeSerializer(app.config["SECRET_KEY"])
|
||||
|
||||
|
||||
def get_active_banner() -> InfoBanner | None:
|
||||
return db.session.execute(db.select(InfoBanner).where(InfoBanner.is_active)).scalar_one_or_none()
|
||||
return db.session.execute(
|
||||
db.select(InfoBanner).where(InfoBanner.is_active)
|
||||
).scalar_one_or_none()
|
||||
|
||||
def get_available_decks(app: Flask, course: Course):
|
||||
slides_dir = os.path.join(app.config['SLIDES_DIR'], course.folder)
|
||||
|
||||
def get_available_decks(course: Course):
|
||||
slides_dir = os.path.join(current_app.config["SLIDES_DIR"], course.folder)
|
||||
try:
|
||||
deck_names = [name for name in os.listdir(slides_dir) if os.path.isdir(os.path.join(slides_dir, name))]
|
||||
deck_names = [
|
||||
name
|
||||
for name in os.listdir(slides_dir)
|
||||
if os.path.isdir(os.path.join(slides_dir, name))
|
||||
]
|
||||
deck_names.sort()
|
||||
return deck_names
|
||||
except FileNotFoundError:
|
||||
# Log this error as it indicates a configuration or filesystem issue
|
||||
logger.error(f"Slides directory not found for course {course.folder}: {slides_dir}")
|
||||
logger.error(
|
||||
f"Slides directory not found for course {course.folder}: {slides_dir}"
|
||||
)
|
||||
return []
|
||||
|
||||
|
||||
def generate_username() -> str:
|
||||
return cast(str, xkcd_password.generate_xkcdpassword(wordlist=words, numwords=3, delimiter="-"))
|
||||
return cast(
|
||||
str,
|
||||
xkcd_password.generate_xkcdpassword(wordlist=words, numwords=3, delimiter="-"),
|
||||
)
|
||||
|
||||
|
||||
def generate_password() -> str:
|
||||
return cast(str, xkcd_password.generate_xkcdpassword(wordlist=words, numwords=2, delimiter="-"))
|
||||
return cast(
|
||||
str,
|
||||
xkcd_password.generate_xkcdpassword(wordlist=words, numwords=2, delimiter="-"),
|
||||
)
|
||||
|
||||
|
||||
def get_passwords(course: Course):
|
||||
return [password for deck in course.decks for password in deck.passwords]
|
||||
return [password for deck in course.decks for password in deck.passwords]
|
||||
|
||||
|
||||
def generate_token() -> Token:
|
||||
max_tries = 100
|
||||
for _ in range(max_tries):
|
||||
user_name = generate_username()
|
||||
|
||||
if not db.session.execute(db.select(Token).where(Token.name == user_name)).scalar():
|
||||
if not db.session.execute(
|
||||
db.select(Token).where(Token.name == user_name)
|
||||
).scalar():
|
||||
break
|
||||
else:
|
||||
# Critical error: Application cannot register new users
|
||||
logger.critical(f"Failed to generate unique username after {max_tries} tries.")
|
||||
raise RuntimeError(f"Unique username could not be generated after {max_tries} tries")
|
||||
raise RuntimeError(
|
||||
f"Unique username could not be generated after {max_tries} tries"
|
||||
)
|
||||
return create_token(user_name)
|
||||
|
||||
def create_token(user_name : str) -> Token:
|
||||
|
||||
def create_token(user_name: str) -> Token:
|
||||
token = Token(user_name)
|
||||
db.session.add(token)
|
||||
db.session.commit()
|
||||
logger.info(f"Created new token for user: {user_name}")
|
||||
return token
|
||||
|
||||
|
||||
def serialize_token(token: Token) -> str:
|
||||
return serializer.dumps(token.id)
|
||||
|
||||
def get_token() -> (Token | None):
|
||||
token_cookie = request.cookies.get('access_token')
|
||||
|
||||
def get_token() -> Token | None:
|
||||
token_cookie = request.cookies.get("access_token")
|
||||
if not token_cookie:
|
||||
return None
|
||||
try:
|
||||
id = serializer.loads(token_cookie)
|
||||
token = db.session.get(Token, id)
|
||||
return token
|
||||
return token
|
||||
except BadSignature:
|
||||
# Warning: Client sent a tampered or invalid cookie
|
||||
logger.warning("Invalid signature for token cookie encountered.")
|
||||
return None
|
||||
|
||||
def retrieve_token(username : str):
|
||||
return db.session.execute(db.select(Token).filter_by(name=username)).scalar_one_or_none()
|
||||
|
||||
def get_unlocked_decks(token: (Token|None)):
|
||||
def retrieve_token(username: str):
|
||||
return db.session.execute(
|
||||
db.select(Token).filter_by(name=username)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def get_unlocked_decks(token: (Token | None)):
|
||||
if token is None:
|
||||
return []
|
||||
decks = [deck for deck in token.decks if deck.course_id == token.active_course_id]
|
||||
decks.sort(key=lambda deck: deck.deck)
|
||||
return decks
|
||||
|
||||
def is_deck_unlocked(deck: str):
|
||||
token = get_token()
|
||||
if token:
|
||||
return token.is_unlocked(deck)
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def update_token(decks: list[Deck]):
|
||||
"""
|
||||
@@ -111,8 +141,71 @@ def update_token(decks: list[Deck]):
|
||||
token.decks = list(set(decks + token.decks))
|
||||
db.session.commit()
|
||||
|
||||
def get_course(id: int) -> (Course | None):
|
||||
|
||||
def get_course(id: int) -> Course | None:
|
||||
return db.session.get(Course, id)
|
||||
|
||||
|
||||
def get_all_courses() -> list[Course]:
|
||||
return db.session.execute(db.select(Course).order_by(Course.id)).scalars().all()
|
||||
|
||||
|
||||
def token_required(f):
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
token = get_token()
|
||||
|
||||
if not token:
|
||||
logger.debug("User has no token. Redirecting to set_token.")
|
||||
|
||||
# --- Capture the target URL ---
|
||||
# request.url captures the full URL including query parameters
|
||||
# (e.g., /dashboard?section=homework)
|
||||
session["next_url"] = request.url
|
||||
|
||||
return redirect(url_for("main.set_token"))
|
||||
kwargs["token"] = token
|
||||
return f(*args, **kwargs)
|
||||
|
||||
return decorated_function
|
||||
|
||||
|
||||
# Cache the last 1000 lookups per worker.
|
||||
@lru_cache(maxsize=1000)
|
||||
def get_cached_deck_info(token_id, course_folder, deck_slug):
|
||||
"""
|
||||
Returns a dictionary of primitive types for route authorization and routing.
|
||||
Never return raw SQLAlchemy models from an lru_cache!
|
||||
"""
|
||||
# 1. Fetch Token
|
||||
token = db.session.get(Token, token_id)
|
||||
if not token:
|
||||
return {"auth": False, "course_exists": False, "deck_exists": False}
|
||||
|
||||
# 2. Fetch Course
|
||||
course = db.session.execute(
|
||||
db.select(Course).where(Course.folder == course_folder)
|
||||
).scalar()
|
||||
|
||||
if not course:
|
||||
return {"auth": False, "course_exists": False, "deck_exists": False}
|
||||
|
||||
# 3. Fetch Deck
|
||||
deck_obj = db.session.execute(
|
||||
db.select(Deck).where(Deck.deck == deck_slug, Deck.course_id == course.id)
|
||||
).scalar()
|
||||
|
||||
if not deck_obj:
|
||||
return {"auth": False, "course_exists": True, "deck_exists": False}
|
||||
|
||||
# 4. Check Authorization
|
||||
is_unlocked = deck_slug in [d.deck for d in token.decks]
|
||||
|
||||
# Return only primitive data (bools, strings) so the cache is thread-safe!
|
||||
return {
|
||||
"auth": is_unlocked,
|
||||
"course_exists": True,
|
||||
"deck_exists": True,
|
||||
"deck_name": deck_obj.name,
|
||||
"index_file": deck_obj.index_file
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user