engineering/puzzle.unturf.com
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

use pyramid tm or have a bad time.

Browse source 
modified:   app.py
	modified:   requirements.txt
	modified:   templates/daily_leaderboard.html.j2
This commit is contained in:
Russell Ballestrini 2024-12-30 13:29:53 -05:00
parent 7629e3affd
commit de2a836f87
3 changed files with 219 additions and 205 deletions
Download patch file Download diff file Expand all files Collapse all files

235
app.py
View file

@ -32,11 +32,22 @@ from sqlalchemy import (
func,
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.orm import (
sessionmaker,
scoped_session,
relationship,
)
from waitress import serve
# Import renderers for templates
from pyramid.renderers import render_to_response
import sqlalchemy
import logging
import zope.sqlalchemy # Importing zope.sqlalchemy for transaction management
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
logging.getLogger("transaction").setLevel(logging.INFO)
################################################################################
# Database Setup
@ -47,7 +58,7 @@ Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
email = Column(String, unique=True, nullable=True) # guests don't have an email.
email = Column(String, unique=True, nullable=True) # guests don't have an email
username = Column(String, unique=True, nullable=False) # user-chosen handle
code_hash = Column(String, nullable=True) # bcrypt hash of code
code_expires = Column(DateTime, nullable=True) # time limit for code
@ -58,14 +69,14 @@ class User(Base):
def get_mime_type(extension):
extension = extension.lower()
if extension == 'jpg' or extension == 'jpeg':
return 'image/jpeg'
elif extension == 'png':
return 'image/png'
elif extension == 'gif':
return 'image/gif'
if extension == "jpg" or extension == "jpeg":
return "image/jpeg"
elif extension == "png":
return "image/png"
elif extension == "gif":
return "image/gif"
else:
return 'application/octet-stream'
return "application/octet-stream"
class Puzzle(Base):
@ -76,14 +87,21 @@ class Puzzle(Base):
image_b64 = Column(String, nullable=False)
image_extension = Column(String, nullable=False)
# Remove the nullable=False constraint as we'll generate these after creation
initial_state_json = Column(String)
solution_state_json = Column(String)
title = Column(String, default="Daily Puzzle")
is_visible = Column(Boolean, default=True)
def __init__(self, date, size, image_b64, image_extension, title="Daily Puzzle", is_visible=True):
def __init__(
self,
date,
size,
image_b64,
image_extension,
title="Daily Puzzle",
is_visible=True,
):
self.date = date
self.size = size
self.image_b64 = image_b64
@ -121,7 +139,7 @@ class Puzzle(Base):
# Update currentIndex after shuffling
for idx, piece in enumerate(pieces):
piece['currentIndex'] = idx
piece["currentIndex"] = idx
# Save states to the database
self.initial_state_json = json.dumps(pieces)
@ -131,7 +149,7 @@ class Puzzle(Base):
"uuid": piece["uuid"],
"correctIndex": piece["correctIndex"],
"currentIndex": piece["correctIndex"],
"rotation": 0
"rotation": 0,
}
for piece in pieces
]
@ -193,10 +211,26 @@ class Attempt(Base):
DB_URL = "sqlite:///puzzle_game.db"
engine = create_engine(DB_URL, echo=False)
Session = sessionmaker(bind=engine)
engine = create_engine(
DB_URL,
echo=False,
connect_args={
"check_same_thread": False
}, # Important for SQLite in multi-threaded apps
poolclass=sqlalchemy.pool.StaticPool, # Use StaticPool for SQLite
)
# Use scoped_session to manage sessions in a thread-safe way
SessionFactory = sessionmaker(bind=engine)
DBSession = scoped_session(SessionFactory)
# Register the session with zope.sqlalchemy
zope.sqlalchemy.register(DBSession)
Base.metadata.bind = engine
Base.metadata.create_all(engine)
################################################################################
# Helper Functions
################################################################################
@ -236,7 +270,7 @@ def send_email(to_email, subject, body):
def admin_required(view_func):
def wrapper(request):
user = get_current_user(request)
user = request.user
if not user or not user.is_admin:
raise HTTPForbidden("You must be an admin to access this page.")
return view_func(request)
@ -274,7 +308,7 @@ def get_current_user(request):
is_verified=False,
)
s.add(guest_user)
s.commit()
s.flush() # Flush to assign an ID
# Store the user ID in the session
request.session["user_id"] = guest_user.id
@ -282,10 +316,15 @@ def get_current_user(request):
return guest_user
def get_ordered_attempts(s, puzzle_id, sort_method='time'):
def get_ordered_attempts(s, puzzle_id, sort_method="time"):
"""Retrieve and order attempts for a given puzzle based on the sort method."""
time_diff = func.strftime("%s", Attempt.time_completed) - func.strftime(
"%s", Attempt.time_started
)
total_actions = Attempt.move_count + Attempt.rotation_count
if sort_method == "moves":
# Sort by least total actions (moves + rotations), only considering first attempts
# Sort by least total actions (moves + rotations), then time
attempts = (
s.query(Attempt)
.filter(
@ -295,16 +334,13 @@ def get_ordered_attempts(s, puzzle_id, sort_method='time'):
Attempt.attempt_number == 1,
)
.order_by(
(Attempt.move_count + Attempt.rotation_count).asc(),
(
func.julianday(Attempt.time_completed)
- func.julianday(Attempt.time_started)
).asc(),
total_actions.asc(),
time_diff.asc(),
)
.all()
)
else:
# Default sort by fastest time, only considering first attempts
# Default sort by fastest time, then total actions
attempts = (
s.query(Attempt)
.filter(
@ -314,11 +350,8 @@ def get_ordered_attempts(s, puzzle_id, sort_method='time'):
Attempt.attempt_number == 1,
)
.order_by(
(
func.julianday(Attempt.time_completed)
- func.julianday(Attempt.time_started)
).asc(),
(Attempt.move_count + Attempt.rotation_count).asc(),
time_diff.asc(),
total_actions.asc(),
)
.all()
)
@ -350,18 +383,19 @@ def login_post_view(request):
user = session.query(User).filter_by(email=email).first()
if not user:
# No valid user in session; create a new user
# Generate a unique starting username.
while True:
suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=8))
suffix = "".join(
random.choices(string.ascii_uppercase + string.digits, k=8)
)
new_tmp_username = f"User-{suffix}"
existing_user = s.query(User).filter_by(username=guest_username).first()
existing_user = (
session.query(User).filter_by(username=new_tmp_username).first()
)
if not existing_user:
# Unique username found
break
# Create new user with tmp username for now
# Create new user with temp username for now
user = User(email=email, username=new_tmp_username)
# First user becomes root admin
@ -370,7 +404,7 @@ def login_post_view(request):
user.is_admin = True
session.add(user)
session.commit()
session.flush() # Flush to assign an ID
# Generate 6-digit code
code_str = f"{random.randint(0,999999):06d}"
@ -380,7 +414,6 @@ def login_post_view(request):
user.code_hash = code_hash
user.code_expires = datetime.datetime.now() + datetime.timedelta(minutes=15)
user.is_verified = False
session.commit()
# Send code via email
email_body = f"Your verification code is: {code_str}"
@ -405,14 +438,21 @@ def verify_post_view(request):
email = request.session.get("login_email")
if not email:
return Response("No email found in session. Please start the login process again.", status=400)
return Response(
"No email found in session. Please start the login process again.",
status=400,
)
s = request.dbsession
user = s.query(User).filter(
user = (
s.query(User)
.filter(
User.email == email,
User.code_expires > datetime.datetime.now(),
User.code_hash != None,
).first()
)
.first()
)
if not user:
return Response("Code not found or expired.", status=400)
@ -423,7 +463,6 @@ def verify_post_view(request):
user.is_verified = True
user.code_hash = None
user.code_expires = None
s.commit()
# Remove the email from the session
del request.session["login_email"]
@ -494,7 +533,9 @@ def profile_post_view(request):
if not request.user:
return Response("You must be logged in to update your profile.", status=403)
if not request.user.is_verified:
return Response("This account is in guest mode, log in to update your profile.", status=403)
return Response(
"This account is in guest mode, log in to update your profile.", status=403
)
s = request.dbsession
enable_gravatar = request.POST.get("enable_gravatar") == "on"
request.user.enable_gravatar = enable_gravatar
@ -508,7 +549,6 @@ def profile_post_view(request):
return Response("Username is already in use.", status=400)
request.user.username = new_username
s.commit()
return HTTPFound(location=request.route_url("profile"))
@ -617,17 +657,25 @@ def invite_admin_post_view(request):
user = s.query(User).filter_by(email=email).first()
if not user:
# Generate a unique starting username.
while True:
suffix = "".join(
random.choices(string.ascii_uppercase + string.digits, k=8)
)
new_tmp_username = f"User-{suffix}"
existing_user = s.query(User).filter_by(username=new_tmp_username).first()
if not existing_user:
break
# Create user record
user = User(email=email, is_admin=True)
user = User(email=email, username=new_tmp_username, is_admin=True)
s.add(user)
s.commit()
# Send email notification
email_body = "You have been invited as an admin. Please log in."
send_email(email, "Admin Invitation", email_body)
else:
if not user.is_admin:
user.is_admin = True
s.commit()
email_body = "You have been granted admin access."
send_email(email, "Admin Access Granted", email_body)
@ -646,7 +694,7 @@ def manage_admins_view(request):
}
@view_config(route_name="remove_admin")
@view_config(route_name="remove_admin", request_method="POST")
@admin_required
def remove_admin_view(request):
s = request.dbsession
@ -658,7 +706,6 @@ def remove_admin_view(request):
if not user:
return Response("User not found.", status=404)
user.is_admin = False
s.commit()
return HTTPFound(location=request.route_url("manage_admins"))
@ -708,7 +755,6 @@ def edit_puzzle_post_view(request):
puzzle.date = puzzle_date
puzzle.title = title or puzzle.title
s.commit()
return HTTPFound(location=request.route_url("admin_dashboard"))
@ -722,7 +768,6 @@ def toggle_visibility_view(request):
if not puzzle:
return Response("Puzzle not found.", status=404)
puzzle.is_visible = not puzzle.is_visible
s.commit()
return HTTPFound(location=request.route_url("admin_dashboard"))
@ -797,7 +842,6 @@ def upload_post_view(request):
s.add(puzzle)
puzzles_uploaded += 1
s.commit()
if puzzles_uploaded == 0:
return Response("No puzzles were uploaded.", status=400)
@ -817,7 +861,7 @@ def daily_puzzle_view(request):
if not puzzle:
return {"request": request, "puzzle": None}
# Get the current user using request.user
# Get the current user
user = request.user
# Fetch the first attempt (attempt_number == 1) for the user and puzzle
@ -843,12 +887,11 @@ def daily_puzzle_view(request):
is_counted=True,
)
s.add(attempt)
s.commit()
# Get user's rank if they have solved it
user_rank = None
if attempt.is_solved:
sort_method = 'time' # You can change this to 'moves' if desired
sort_method = "time" # You can change this to 'moves' if desired
attempts = get_ordered_attempts(s, puzzle.id, sort_method)
for idx, att in enumerate(attempts):
if att.user_id == user.id:
@ -904,7 +947,8 @@ def record_action_view(request):
# Get request data
try:
action_data = request.json_body
except:
except Exception as e:
logger.error(f"Failed to parse JSON: {e}")
return Response(
json.dumps({"status": "error", "message": "Invalid JSON data."}),
content_type="application/json; charset=UTF-8",
@ -926,8 +970,6 @@ def record_action_view(request):
content_type="application/json; charset=UTF-8",
)
display_name = user.username
attempt = (
s.query(Attempt)
.filter_by(user_id=user.id, puzzle_id=puzzle.id, is_solved=False)
@ -943,8 +985,8 @@ def record_action_view(request):
# Update attempt stats and state
state = json.loads(attempt.state_json)
solution_state = json.loads(puzzle.solution_state_json)
try:
if action == "rotate":
attempt.rotation_count += 1
piece_uuid = data.get("pieceUUID")
@ -992,26 +1034,14 @@ def record_action_view(request):
# Check for win condition
if all(
piece["currentIndex"] == piece["correctIndex"] and piece["rotation"] % 360 == 0
piece["currentIndex"] == piece["correctIndex"]
and piece["rotation"] % 360 == 0
for piece in state
):
attempt.mark_as_solved()
attempt.time_completed = datetime.datetime.utcnow()
s.commit()
# Get user's rank
attempts = (
s.query(Attempt)
.filter_by(puzzle_id=puzzle.id, is_solved=True, is_counted=True)
.order_by(
(
func.julianday(Attempt.time_completed)
- func.julianday(Attempt.time_started)
).asc(),
(Attempt.move_count + Attempt.rotation_count).asc(),
)
.all()
)
attempts = get_ordered_attempts(s, puzzle.id, "time")
for idx, att in enumerate(attempts):
if att.user_id == user.id and att.id == attempt.id:
user_rank = idx + 1
@ -1024,9 +1054,15 @@ def record_action_view(request):
content_type="application/json; charset=UTF-8",
)
else:
s.commit()
return Response(
json.dumps({"status": "ok"}), content_type="application/json; charset=UTF-8"
json.dumps({"status": "ok"}),
content_type="application/json; charset=UTF-8",
)
except Exception as e:
logger.error(f"Error processing action '{action}': {e}")
return Response(
json.dumps({"status": "error", "message": "Server error."}),
content_type="application/json; charset=UTF-8",
)
@ -1055,34 +1091,7 @@ def daily_leaderboard_view(request):
if not puzzle:
return Response("No puzzle found for this ID.", status=404)
if sort_method == "moves":
# Sort by least total actions (moves + rotations)
attempts = (
s.query(Attempt)
.filter_by(puzzle_id=puzzle.id, is_solved=True, is_counted=True)
.order_by(
(Attempt.move_count + Attempt.rotation_count).asc(),
(
func.julianday(Attempt.time_completed)
- func.julianday(Attempt.time_started)
).asc(),
)
.all()
)
else:
# Default sort by fastest time
attempts = (
s.query(Attempt)
.filter_by(puzzle_id=puzzle.id, is_solved=True, is_counted=True)
.order_by(
(
func.julianday(Attempt.time_completed)
- func.julianday(Attempt.time_started)
).asc(),
(Attempt.move_count + Attempt.rotation_count).asc(),
)
.all()
)
attempts = get_ordered_attempts(s, puzzle.id, sort_method)
return {
"request": request,
@ -1116,10 +1125,9 @@ def king_of_the_mountain_view(request):
s.query(
func.sum(
(
func.julianday(Attempt.time_completed)
- func.julianday(Attempt.time_started)
func.strftime("%s", Attempt.time_completed)
- func.strftime("%s", Attempt.time_started)
)
* 86400
)
)
.filter(
@ -1179,13 +1187,13 @@ def main(global_config=None, **settings):
# Set up the session factory with the specified settings
session_factory = SignedCookieSessionFactory(
secret="it-is-a-secret-you-must-change",
hashalg='sha512',
hashalg="sha512",
timeout=31104000, # Approx. one year in seconds
max_age=31104000, # Set Max-Age attribute on cookie
reissue_time=15552000, # Approx. six months
samesite=None, # Allows cross-site requests if needed
httponly=True, # Helps mitigate XSS attacks
secure=False # Set to True if using HTTPS
secure=False, # Set to True if using HTTPS
)
if not settings:
@ -1199,11 +1207,14 @@ def main(global_config=None, **settings):
config.add_jinja2_renderer(".j2")
config.add_jinja2_search_path("templates", name=".j2")
# Include pyramid_tm for transaction management
config.include("pyramid_tm")
# Add user to all requests.
config.add_request_method(callable=get_current_user, name="user", reify=True)
def dbsession(request):
return Session()
return DBSession
config.add_request_method(dbsession, "dbsession", reify=True)

3
requirements.txt
View file

@ -1,10 +1,13 @@
plaster_pastedeploy
pyramid
pyramid_jinja2
pyramid_tm
pyramid_debugtoolbar
waitress
pyramid_retry
zope.sqlalchemy
bcrypt
sqlalchemy
werkzeug

4
templates/daily_leaderboard.html.j2
View file

@ -3,8 +3,8 @@
<h1>Leaderboard for {{ puzzle.title }} ({{ puzzle.date }})</h1>
<p>
<strong>Sort by:</strong>
<a href="{{ request.route_url('daily_leaderboard', puzzle_id=puzzle.id, sort='time') }}">Top Speed</a> |
<a href="{{ request.route_url('daily_leaderboard', puzzle_id=puzzle.id, sort='moves') }}">Least Moves</a>
<a href="{{ request.route_url('daily_leaderboard', puzzle_id=puzzle.id, _query={'sort': 'time'}) }}">Top Speed</a> |
<a href="{{ request.route_url('daily_leaderboard', puzzle_id=puzzle.id, _query={'sort': 'moves'}) }}">Least Moves</a>
</p>
<p>Players are ranked based on their total time and moves. Faster times and fewer moves result in better rankings.</p>
<table>