engineering/puzzle.unturf.com
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
puzzle.unturf.com/app.py
Russell Ballestrini db81a7aa8e auto create guest users 
modified:   app.py
	modified:   templates/base.html.j2
	modified:   templates/daily_puzzle.html.j2
2024-12-28 14:44:29 -05:00

1236 lines
39 KiB
Python
Raw Blame History

# app.py
import os
import base64
import datetime
import json
import random
import string
import bcrypt
import re
import uuid
import hashlib
import smtplib
from email.mime.text import MIMEText
from pyramid.config import Configurator
from pyramid.view import view_config
from pyramid.response import Response
from pyramid.httpexceptions import HTTPFound, HTTPForbidden
from pyramid.session import SignedCookieSessionFactory
from sqlalchemy import (
create_engine,
Column,
Integer,
String,
Float,
Boolean,
Date,
DateTime,
ForeignKey,
desc,
func,
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from waitress import serve
# Import renderers for templates
from pyramid.renderers import render_to_response
################################################################################
# Database Setup
################################################################################
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.
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
is_verified = Column(Boolean, default=False)
enable_gravatar = Column(Boolean, default=False) # Gravatar support
is_admin = Column(Boolean, default=False) # Admin flag
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'
else:
return 'application/octet-stream'
class Puzzle(Base):
__tablename__ = "puzzles"
id = Column(Integer, primary_key=True)
date = Column(Date, nullable=False)
size = Column(Integer, nullable=False) # 3, 4, or 5
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):
self.date = date
self.size = size
self.image_b64 = image_b64
self.image_extension = image_extension
self.title = title
self.is_visible = is_visible
# Generate initial and solution states
self.generate_states()
@property
def filename_slug(self):
return f"{slugify(self.title)}.{self.image_extension}"
@property
def image_mime_type(self):
return get_mime_type(self.image_extension)
def generate_states(self):
total_pieces = self.size * self.size
pieces = []
for idx in range(total_pieces):
piece_uuid = str(uuid.uuid4())
piece = {
"uuid": piece_uuid,
"correctIndex": idx,
"currentIndex": idx,
"rotation": random.choice([0, 90, 180, 270]), # Random initial rotation
}
pieces.append(piece)
# Shuffle the pieces to create the initial state
random.shuffle(pieces)
# Update currentIndex after shuffling
for idx, piece in enumerate(pieces):
piece['currentIndex'] = idx
# Save states to the database
self.initial_state_json = json.dumps(pieces)
# For solution, pieces are in correct order with rotation = 0
solution_pieces = [
{
"uuid": piece["uuid"],
"correctIndex": piece["correctIndex"],
"currentIndex": piece["correctIndex"],
"rotation": 0
}
for piece in pieces
]
self.solution_state_json = json.dumps(solution_pieces)
class Attempt(Base):
__tablename__ = "attempts"
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
user_display_name = Column(String, default="anonymous")
puzzle_id = Column(Integer, ForeignKey("puzzles.id"))
attempt_date = Column(Date, default=datetime.date.today())
attempt_number = Column(Integer, default=1)
move_count = Column(Integer, default=0)
rotation_count = Column(Integer, default=0)
time_started = Column(DateTime, default=datetime.datetime.utcnow)
time_completed = Column(DateTime, nullable=True)
is_solved = Column(Boolean, default=False)
state_json = Column(String, nullable=False) # current game state
is_counted = Column(
Boolean, default=True
) # whether attempt counts towards leaderboard
puzzle = relationship("Puzzle", backref="attempts")
user = relationship("User", backref="attempts")
@property
def time_started_ms(self):
"""Returns time_started as milliseconds since epoch."""
if self.time_started:
return int(self.time_started.timestamp() * 1000)
return 0
def mark_as_solved(self, time_completed=None):
"""Mark the attempt as solved and set time_completed."""
self.is_solved = True
if self.time_completed is None:
self.time_completed = time_completed or datetime.datetime.utcnow()
@property
def total_elapsed_ms(self):
"""Calculate total elapsed time in milliseconds if solved."""
if self.time_completed and self.time_started:
delta = self.time_completed - self.time_started
return int(delta.total_seconds() * 1000)
return None
@property
def current_elapsed_ms(self):
"""Calculate current elapsed time in milliseconds."""
if self.time_started:
end_time = self.time_completed or datetime.datetime.utcnow()
delta = end_time - self.time_started
return int(delta.total_seconds() * 1000)
return None
DB_URL = "sqlite:///puzzle_game.db"
engine = create_engine(DB_URL, echo=False)
Session = sessionmaker(bind=engine)
Base.metadata.create_all(engine)
################################################################################
# Session Factory
################################################################################
session_factory = SignedCookieSessionFactory(secret="my_insecure_secret")
################################################################################
# Helper Functions
################################################################################
def slugify(text):
text = text.lower()
text = re.sub(r"\s+", "-", text)
text = re.sub(r"[^\w\-]", "", text)
return text
def get_gravatar_url(email, size=100):
email = email.strip().lower()
hash_code = hashlib.md5(email.encode("utf-8")).hexdigest()
return f"https://www.gravatar.com/avatar/{hash_code}?s={size}&d=identicon"
def send_email(to_email, subject, body):
# For testing purposes, print the email content to the console
print("======= Email Sent =======")
print(f"To: {to_email}")
print(f"Subject: {subject}")
print(f"Body:\n{body}")
print("==========================")
msg = MIMEText(body)
msg["Subject"] = subject
msg["From"] = "noreply@example.com"
msg["To"] = to_email
try:
s = smtplib.SMTP("localhost", 25)
s.sendmail("noreply@example.com", [to_email], msg.as_string())
s.quit()
except Exception as e:
print(f"Error sending email: {e}")
def admin_required(view_func):
def wrapper(request):
user = get_current_user(request)
if not user or not user.is_admin:
raise HTTPForbidden("You must be an admin to access this page.")
return view_func(request)
return wrapper
def get_current_user(request):
"""Return the current user (authenticated or guest) from session."""
user_id = request.session.get("user_id")
s = request.dbsession
if user_id:
# Try to get the user from the database
user = s.query(User).filter_by(id=user_id).first()
if user:
return user
else:
# User ID in session does not exist in the database; remove it
del request.session["user_id"]
# No valid user in session; create a guest user
# Generate a unique guest username
while True:
suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=8))
guest_username = f"Guest-{suffix}"
existing_user = s.query(User).filter_by(username=guest_username).first()
if not existing_user:
break # Unique username found
# Create a new guest user
guest_user = User(
email=guest_username,
username=guest_username,
is_verified=False,
)
s.add(guest_user)
s.commit()
# Store the user ID in the session
request.session["user_id"] = guest_user.id
return guest_user
def get_ordered_attempts(s, puzzle_id, sort_method='time'):
"""Retrieve and order attempts for a given puzzle based on the sort method."""
if sort_method == "moves":
# Sort by least total actions (moves + rotations), only considering first attempts
attempts = (
s.query(Attempt)
.filter(
Attempt.puzzle_id == puzzle_id,
Attempt.is_solved == True,
Attempt.is_counted == True,
Attempt.attempt_number == 1,
)
.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, only considering first attempts
attempts = (
s.query(Attempt)
.filter(
Attempt.puzzle_id == puzzle_id,
Attempt.is_solved == True,
Attempt.is_counted == True,
Attempt.attempt_number == 1,
)
.order_by(
(
func.julianday(Attempt.time_completed)
- func.julianday(Attempt.time_started)
).asc(),
(Attempt.move_count + Attempt.rotation_count).asc(),
)
.all()
)
return attempts
################################################################################
# Routes
################################################################################
@view_config(route_name="home", renderer="home.html.j2")
def home_view(request):
return {
"request": request,
}
@view_config(route_name="login", request_method="GET", renderer="login.html.j2")
def login_get_view(request):
return {"request": request}
@view_config(route_name="login", request_method="POST")
def login_post_view(request):
email = request.POST.get("email", "").strip().lower()
if not email:
return Response("Email required.", status=400)
session = request.dbsession
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))
new_tmp_username = f"User-{suffix}"
existing_user = s.query(User).filter_by(username=guest_username).first()
if not existing_user:
# Unique username found
break
# Create new user with tmp username for now
user = User(email=email, username=new_tmp_username)
# First user becomes root admin
user_count = session.query(func.count(User.id)).scalar()
if user_count == 0:
user.is_admin = True
session.add(user)
session.commit()
# Generate 6-digit code
code_str = f"{random.randint(0,999999):06d}"
code_hash = bcrypt.hashpw(code_str.encode("utf-8"), bcrypt.gensalt()).decode(
"utf-8"
)
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}"
send_email(user.email, "Your Verification Code", email_body)
# Store the email in the session for verification
request.session["login_email"] = email
return HTTPFound(location=request.route_url("verify"))
@view_config(route_name="verify", request_method="GET", renderer="verify.html.j2")
def verify_get_view(request):
return {"request": request}
@view_config(route_name="verify", request_method="POST")
def verify_post_view(request):
code_entered = request.POST.get("code", "").strip()
if not code_entered or len(code_entered) != 6:
return Response("Invalid code.", status=400)
email = request.session.get("login_email")
if not email:
return Response("No email found in session. Please start the login process again.", status=400)
s = request.dbsession
user = s.query(User).filter(
User.email == email,
User.code_expires > datetime.datetime.now(),
User.code_hash != None,
).first()
if not user:
return Response("Code not found or expired.", status=400)
if not bcrypt.checkpw(code_entered.encode("utf-8"), user.code_hash.encode("utf-8")):
return Response("Invalid code.", status=400)
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"]
request.session["user_id"] = user.id
return HTTPFound(location=request.route_url("home"))
@view_config(route_name="logout")
def logout_view(request):
request.session.invalidate()
return HTTPFound(location=request.route_url("home"))
################################################################################
# Profile and Gravatar
################################################################################
@view_config(route_name="profile", request_method="GET", renderer="profile.html.j2")
def profile_get_view(request):
user = request.user
if not user:
return Response("You must be logged in to access your profile.", status=403)
gravatar_url = get_gravatar_url(user.email) if user.enable_gravatar else ""
s = request.dbsession
# Fetch user stats
solved_attempts = (
s.query(Attempt)
.join(Puzzle)
.filter(
Attempt.user_id == user.id,
Attempt.is_solved == True,
Attempt.is_counted == True,
)
.order_by(Attempt.time_completed.desc())
.all()
)
total_puzzles_solved = len(solved_attempts)
# Calculate streaks
solved_dates = sorted({a.puzzle.date for a in solved_attempts})
current_streak = 0
max_streak = 0
last_date = None
for date in solved_dates:
if last_date and (date - last_date).days == 1:
current_streak += 1
else:
current_streak = 1
if current_streak > max_streak:
max_streak = current_streak
last_date = date
return {
"request": request,
"user": user,
"gravatar_url": gravatar_url,
"total_puzzles_solved": total_puzzles_solved,
"current_streak": current_streak,
"max_streak": max_streak,
"solved_attempts": solved_attempts,
}
@view_config(route_name="profile", request_method="POST")
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)
s = request.dbsession
enable_gravatar = request.POST.get("enable_gravatar") == "on"
request.user.enable_gravatar = enable_gravatar
# Handle username update if provided
new_username = request.POST.get("new_username", "").strip()
if new_username:
# Check if the new username is already taken
existing = s.query(User).filter(User.username == new_username).first()
if existing and existing.id != request.user.id:
return Response("Username is already in use.", status=400)
request.user.username = new_username
s.commit()
return HTTPFound(location=request.route_url("profile"))
@view_config(route_name="view_profile", renderer="profile.html.j2")
def view_profile(request):
s = request.dbsession
user_id = request.matchdict.get("user_id")
user = s.query(User).filter_by(id=user_id).first()
if not user:
return Response("User not found.", status=404)
gravatar_url = get_gravatar_url(user.email) if user.enable_gravatar else ""
# Fetch user stats
solved_attempts = (
s.query(Attempt)
.join(Puzzle)
.filter(
Attempt.user_id == user.id,
Attempt.is_solved == True,
Attempt.is_counted == True,
)
.order_by(Attempt.time_completed.desc())
.all()
)
total_puzzles_solved = len(solved_attempts)
# Calculate streaks
solved_dates = sorted({a.puzzle.date for a in solved_attempts})
current_streak = 0
max_streak = 0
last_date = None
for date in solved_dates:
if last_date and (date - last_date).days == 1:
current_streak += 1
else:
current_streak = 1
if current_streak > max_streak:
max_streak = current_streak
last_date = date
return {
"request": request,
"user": user,
"gravatar_url": gravatar_url,
"total_puzzles_solved": total_puzzles_solved,
"current_streak": current_streak,
"max_streak": max_streak,
"solved_attempts": solved_attempts,
}
################################################################################
# Puzzle Menu
################################################################################
@view_config(route_name="puzzle_menu", renderer="puzzle_menu.html.j2")
def puzzle_menu_view(request):
s = request.dbsession
today = datetime.date.today()
puzzles = {}
puzzle_sizes = [3, 4, 5]
for size in puzzle_sizes:
puzzle = (
s.query(Puzzle).filter_by(date=today, size=size, is_visible=True).first()
)
puzzles[size] = puzzle
return {
"request": request,
"puzzles": puzzles,
"puzzle_sizes": puzzle_sizes,
}
################################################################################
# Admin Views
################################################################################
@view_config(route_name="admin_dashboard", renderer="admin_dashboard.html.j2")
@admin_required
def admin_dashboard_view(request):
s = request.dbsession
puzzles = s.query(Puzzle).order_by(Puzzle.date.desc(), Puzzle.size.asc()).all()
return {
"request": request,
"puzzles": puzzles,
}
@view_config(
route_name="invite_admin", request_method="GET", renderer="invite_admin.html.j2"
)
@admin_required
def invite_admin_get_view(request):
return {"request": request}
@view_config(route_name="invite_admin", request_method="POST")
@admin_required
def invite_admin_post_view(request):
email = request.POST.get("email", "").strip().lower()
if not email:
return Response("Email required.", status=400)
s = request.dbsession
user = s.query(User).filter_by(email=email).first()
if not user:
# Create user record
user = User(email=email, 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)
return HTTPFound(location=request.route_url("admin_dashboard"))
@view_config(route_name="manage_admins", renderer="manage_admins.html.j2")
@admin_required
def manage_admins_view(request):
s = request.dbsession
admins = s.query(User).filter_by(is_admin=True).all()
return {
"request": request,
"admins": admins,
"user": request.user,
}
@view_config(route_name="remove_admin")
@admin_required
def remove_admin_view(request):
s = request.dbsession
current_user = request.user
user_id = int(request.matchdict.get("user_id"))
if user_id == current_user.id or user_id == 1:
return Response("Cannot remove root admin or yourself.", status=400)
user = s.query(User).filter_by(id=user_id).first()
if not user:
return Response("User not found.", status=404)
user.is_admin = False
s.commit()
return HTTPFound(location=request.route_url("manage_admins"))
@view_config(
route_name="edit_puzzle", request_method="GET", renderer="edit_puzzle.html.j2"
)
@admin_required
def edit_puzzle_get_view(request):
s = request.dbsession
puzzle_id = request.matchdict.get("puzzle_id")
puzzle = s.query(Puzzle).filter_by(id=puzzle_id).first()
if not puzzle:
return Response("Puzzle not found.", status=404)
return {
"request": request,
"puzzle": puzzle,
}
@view_config(route_name="edit_puzzle", request_method="POST")
@admin_required
def edit_puzzle_post_view(request):
s = request.dbsession
puzzle_id = request.matchdict.get("puzzle_id")
puzzle = s.query(Puzzle).filter_by(id=puzzle_id).first()
if not puzzle:
return Response("Puzzle not found.", status=404)
puzzle_date_str = request.POST.get("puzzle_date")
title = request.POST.get("title")
try:
puzzle_date = datetime.datetime.strptime(puzzle_date_str, "%Y-%m-%d").date()
except ValueError:
return Response("Invalid date format. Use YYYY-MM-DD.", status=400)
image_file = request.POST.get("image_file")
if image_file and getattr(image_file, "filename", "").strip():
raw_bytes = image_file.file.read()
max_size = 5 * 1024 * 1024 # 5 MB
if len(raw_bytes) > max_size:
return Response(
f"File size exceeds the 5MB limit for the puzzle.", status=400
)
encoded_str = base64.b64encode(raw_bytes).decode("utf-8")
puzzle.image_b64 = encoded_str
puzzle.date = puzzle_date
puzzle.title = title or puzzle.title
s.commit()
return HTTPFound(location=request.route_url("admin_dashboard"))
@view_config(route_name="toggle_visibility")
@admin_required
def toggle_visibility_view(request):
s = request.dbsession
puzzle_id = request.matchdict.get("puzzle_id")
puzzle = s.query(Puzzle).filter_by(id=puzzle_id).first()
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"))
################################################################################
# Puzzle Upload
################################################################################
@view_config(route_name="upload", request_method="GET", renderer="upload.html.j2")
@admin_required
def upload_get_view(request):
return {"request": request}
@view_config(route_name="upload", request_method="POST")
@admin_required
def upload_post_view(request):
puzzle_date_str = request.POST.get("puzzle_date")
puzzle_sizes = [3, 4, 5]
puzzles_uploaded = 0
if not puzzle_date_str:
return Response("Missing date.", status=400)
try:
puzzle_date = datetime.datetime.strptime(puzzle_date_str, "%Y-%m-%d").date()
except ValueError:
return Response("Invalid date format. Use YYYY-MM-DD.", status=400)
s = request.dbsession
for size in puzzle_sizes:
title = request.POST.get(f"title_{size}") or f"Puzzle {size}x{size}"
image_file = request.POST.get(f"image_file_{size}")
if image_file is None or not getattr(image_file, "filename", "").strip():
continue # Skip if no image uploaded for this size
# Validate file type
allowed_extensions = {"png", "jpg", "jpeg", "gif"}
filename = image_file.filename.lower()
file_extension = os.path.splitext(filename)[1][1:] # Get extension without dot
if file_extension not in allowed_extensions:
return Response(
f"Unsupported file type for {size}x{size} puzzle. Allowed types: png, jpg, jpeg, gif.",
status=400,
)
raw_bytes = image_file.file.read()
max_size = 5 * 1024 * 1024 # 5 MB
if len(raw_bytes) > max_size:
return Response(
f"File size exceeds the 5MB limit for {size}x{size} puzzle.", status=400
)
encoded_str = base64.b64encode(raw_bytes).decode("utf-8")
existing = s.query(Puzzle).filter_by(date=puzzle_date, size=size).first()
if existing:
return Response(
f"Puzzle of size {size}x{size} already exists for {puzzle_date}.",
status=400,
)
# Create the puzzle instance; states are generated in the constructor
puzzle = Puzzle(
date=puzzle_date,
size=size,
image_b64=encoded_str,
image_extension=file_extension, # Pass the file extension here
title=title,
)
s.add(puzzle)
puzzles_uploaded += 1
s.commit()
if puzzles_uploaded == 0:
return Response("No puzzles were uploaded.", status=400)
return HTTPFound(location=request.route_url("admin_dashboard"))
################################################################################
# Daily Puzzle
################################################################################
@view_config(route_name="daily_puzzle", renderer="daily_puzzle.html.j2")
def daily_puzzle_view(request):
s = request.dbsession
size = int(request.matchdict.get("size", 4))
today = datetime.date.today()
puzzle = s.query(Puzzle).filter_by(date=today, size=size, is_visible=True).first()
if not puzzle:
return {"request": request, "puzzle": None}
# Get the current user using request.user
user = request.user
# Fetch the first attempt (attempt_number == 1) for the user and puzzle
attempt = (
s.query(Attempt)
.filter_by(user_id=user.id, puzzle_id=puzzle.id, attempt_number=1)
.first()
)
if not attempt:
# Create the first attempt
attempt = Attempt(
user_id=user.id,
user_display_name=user.username,
puzzle_id=puzzle.id,
attempt_date=today,
attempt_number=1,
move_count=0,
rotation_count=0,
is_solved=False,
state_json=puzzle.initial_state_json,
time_started=datetime.datetime.utcnow(),
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
attempts = get_ordered_attempts(s, puzzle.id, sort_method)
for idx, att in enumerate(attempts):
if att.user_id == user.id:
user_rank = idx + 1
break
# Calculate elapsed time
if attempt.is_solved:
# Use total elapsed time when solved
total_elapsed_ms = attempt.total_elapsed_ms or 0
else:
# Calculate current elapsed time
total_elapsed_ms = attempt.current_elapsed_ms or 0
return {
"request": request,
"puzzle": puzzle,
"attempt": attempt,
"attempt_state": json.loads(attempt.state_json),
"attempt_is_solved": attempt.is_solved,
"user_rank": user_rank,
"image_mime_type": puzzle.image_mime_type,
"total_elapsed_ms": total_elapsed_ms,
}
@view_config(route_name="puzzle_image")
def puzzle_image_view(request):
s = request.dbsession
puzzle_id = request.matchdict.get("puzzle_id")
puzzle = s.query(Puzzle).filter_by(id=puzzle_id).first()
if not puzzle:
return Response("Puzzle not found.", status=404)
# Check if user has solved this puzzle
solved_attempt = (
s.query(Attempt)
.filter_by(user_id=request.user.id, puzzle_id=puzzle.id, is_solved=True)
.first()
)
if not solved_attempt:
return Response("You must solve the puzzle to view the image.", status=403)
image_data = base64.b64decode(puzzle.image_b64)
return Response(body=image_data, content_type="image/png")
@view_config(route_name="record_action", request_method="POST")
def record_action_view(request):
s = request.dbsession
# Get request data
try:
action_data = request.json_body
except:
return Response(
json.dumps({"status": "error", "message": "Invalid JSON data."}),
content_type="application/json; charset=UTF-8",
)
action = action_data.get("action")
data = action_data.get("data")
# Retrieve user, puzzle, and attempt
user = request.user
size = int(request.matchdict.get("size", 4))
today = datetime.date.today()
puzzle = s.query(Puzzle).filter_by(date=today, size=size, is_visible=True).first()
if not puzzle:
return Response(
json.dumps({"status": "error", "message": "No puzzle found."}),
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)
.order_by(Attempt.attempt_number.desc())
.first()
)
if not attempt:
return Response(
json.dumps({"status": "error", "message": "No active attempt found."}),
content_type="application/json; charset=UTF-8",
)
# Update attempt stats and state
state = json.loads(attempt.state_json)
solution_state = json.loads(puzzle.solution_state_json)
if action == "rotate":
attempt.rotation_count += 1
piece_uuid = data.get("pieceUUID")
rotation = int(data.get("rotation"))
# Update rotation in state
for piece in state:
if piece["uuid"] == piece_uuid:
piece["rotation"] = rotation
break
else:
return Response(
json.dumps({"status": "error", "message": "Invalid piece UUID."}),
content_type="application/json; charset=UTF-8",
)
elif action == "swap":
attempt.move_count += 1
piece_uuid_a = data.get("pieceUUIDA")
piece_uuid_b = data.get("pieceUUIDB")
# Find pieces by UUID
idxA = idxB = None
for idx, piece in enumerate(state):
if piece["uuid"] == piece_uuid_a:
idxA = idx
if piece["uuid"] == piece_uuid_b:
idxB = idx
if idxA is not None and idxB is not None:
# Swap currentIndex values
state[idxA]["currentIndex"], state[idxB]["currentIndex"] = (
state[idxB]["currentIndex"],
state[idxA]["currentIndex"],
)
else:
return Response(
json.dumps({"status": "error", "message": "Invalid piece UUIDs."}),
content_type="application/json; charset=UTF-8",
)
else:
return Response(
json.dumps({"status": "error", "message": "Invalid action."}),
content_type="application/json; charset=UTF-8",
)
# Save updated state
attempt.state_json = json.dumps(state)
# Check for win condition
if all(
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()
)
for idx, att in enumerate(attempts):
if att.user_id == user.id and att.id == attempt.id:
user_rank = idx + 1
break
else:
user_rank = "N/A"
return Response(
json.dumps({"status": "win", "rank": user_rank}),
content_type="application/json; charset=UTF-8",
)
else:
s.commit()
return Response(
json.dumps({"status": "ok"}), content_type="application/json; charset=UTF-8"
)
################################################################################
# Leaderboards
################################################################################
@view_config(route_name="leaderboard_index", renderer="leaderboard_index.html.j2")
def leaderboard_index_view(request):
s = request.dbsession
puzzles = s.query(Puzzle).order_by(Puzzle.date.desc(), Puzzle.size.asc()).all()
return {"request": request, "puzzles": puzzles}
@view_config(route_name="daily_leaderboard", renderer="daily_leaderboard.html.j2")
def daily_leaderboard_view(request):
s = request.dbsession
puzzle_id = request.matchdict.get("puzzle_id")
sort_method = request.params.get("sort", "time")
puzzle = s.query(Puzzle).filter_by(id=puzzle_id).first()
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()
)
return {
"request": request,
"puzzle": puzzle,
"attempts": attempts,
}
@view_config(route_name="king_of_the_mountain", renderer="king_of_the_mountain.html.j2")
def king_of_the_mountain_view(request):
s = request.dbsession
# Aggregate scores
users = s.query(User).all()
leaderboard = []
for user in users:
total_puzzles = (
s.query(func.count(Attempt.id))
.filter(
Attempt.user_id == user.id,
Attempt.is_solved == True,
Attempt.is_counted == True,
)
.scalar()
)
if total_puzzles == 0:
continue # Skip users with no solves
total_time = (
s.query(
func.sum(
(
func.julianday(Attempt.time_completed)
- func.julianday(Attempt.time_started)
)
* 86400
)
)
.filter(
Attempt.user_id == user.id,
Attempt.is_solved == True,
Attempt.is_counted == True,
)
.scalar()
) # Time in seconds
total_moves = (
s.query(func.sum(Attempt.move_count))
.filter(
Attempt.user_id == user.id,
Attempt.is_solved == True,
Attempt.is_counted == True,
)
.scalar()
)
total_rotations = (
s.query(func.sum(Attempt.rotation_count))
.filter(
Attempt.user_id == user.id,
Attempt.is_solved == True,
Attempt.is_counted == True,
)
.scalar()
)
leaderboard.append(
{
"user_id": user.id,
"username": user.username,
"total_puzzles": total_puzzles,
"total_time": total_time,
"total_moves": total_moves,
"total_rotations": total_rotations,
}
)
# Sort leaderboard
leaderboard.sort(key=lambda x: (-x["total_puzzles"], x["total_time"]))
return {
"request": request,
"global_leaderboard": leaderboard,
}
################################################################################
# Main
################################################################################
def main(global_config=None, **settings):
from pyramid.decorator import reify
if not settings:
settings = {}
settings["sqlalchemy.url"] = DB_URL
config = Configurator(settings=settings, session_factory=session_factory)
config.include("pyramid_jinja2")
# Add .html.j2 extension for Jinja2 templates
# Set up Jinja2 template search path
config.add_jinja2_renderer(".j2")
config.add_jinja2_search_path("templates", name=".j2")
# Add user to all requests.
config.add_request_method(callable=get_current_user, name="user", reify=True)
def dbsession(request):
return Session()
config.add_request_method(dbsession, "dbsession", reify=True)
# Routes
config.add_route("home", "/")
# Auth
config.add_route("login", "/auth/login")
config.add_route("verify", "/auth/verify")
config.add_route("logout", "/auth/logout")
config.add_route("profile", "/auth/profile")
config.add_route("view_profile", "/user/{user_id}")
# Admin
config.add_route("admin_dashboard", "/admin")
config.add_route("invite_admin", "/admin/invite")
config.add_route("manage_admins", "/admin/manage")
config.add_route("remove_admin", "/admin/remove/{user_id}")
config.add_route("edit_puzzle", "/admin/puzzle/{puzzle_id}/edit")
config.add_route("toggle_visibility", "/admin/puzzle/{puzzle_id}/toggle")
# Puzzle
config.add_route("upload", "/admin/upload")
config.add_route("puzzle_menu", "/puzzle/menu")
config.add_route("daily_puzzle", "/puzzle/{size}")
config.add_route("record_action", "/puzzle/{size}/record_action")
config.add_route("puzzle_image", "/puzzle/{puzzle_id}/image")
# Leaderboards
config.add_route("leaderboard_index", "/leaderboards")
config.add_route("daily_leaderboard", "/leaderboard/{puzzle_id}")
config.add_route("king_of_the_mountain", "/king_of_the_mountain")
config.scan()
return config.make_wsgi_app()
if __name__ == "__main__":
app = main()
print("Serving on http://localhost:6543")
serve(app, host="0.0.0.0", port=6543)
Reference in a new issue View git blame Copy permalink