from starlette.responses import HTMLResponse
from starlette.requests import Request
from starlette.routing import Route
import os
from sqlalchemy import select, update
from config import AUTHORIZATION_NEEDED
from api.database import database, user, result, daily_reward, check_blacklist, check_whitelist
from api.crypt import decrypt_fields, encryptAES
from api.templates import EXP_UNLOCKED_SONGS, TITLE_LISTS, SONG_LIST
from api.misc import inform_page
async def ranking(request: Request):
decrypted_fields, _ = await decrypt_fields(request)
if not decrypted_fields:
return HTMLResponse("""
"""
# Loop leaderboard
for rank, (username, data) in enumerate(sorted_players, start=1):
html += f"""
#{rank}
{data['score']}
"""
else:
query = select(result).where(
(result.c.id == song_id) & (result.c.mode == mode)
).order_by(result.c.score.desc())
play_results = await database.fetch_all(query)
query = select(user).where(user.c.device_id == device_id)
user_result = await database.fetch_one(query)
query = select(daily_reward).where(daily_reward.c.device_id == device_id)
device_result = await database.fetch_one(query)
user_id = user_result[0] if user_result else None
username = user_result[1] if user_result else f"Guest({device_id[-6:]})"
play_record = None
if user_id:
play_record = next((record for record in play_results if int(record[3]) == user_id), None)
if not play_record:
play_record = next((record for record in play_results if record[1] == device_id and record[3] is None), None)
player_rank = None
avatar_index = str(play_record[7]) if play_record else "1"
user_score = play_record[8] if play_record else 0
for rank, result_obj in enumerate(play_results, start=1):
if user_result and int(result_obj[3]) == user_id:
player_rank = rank
break
elif result_obj[1] == device_id and result_obj[3] is None:
player_rank = rank
break
html += f"""
You
{"#" + str(player_rank) if player_rank else "N/A"}
{username}
{user_score}
"""
html += """
"""
for rank, record in enumerate(play_results, start=1):
username = f"Guest({record[1][-6:]})"
device_info = None
if record[3]:
query = select(user.c.username).where(user.c.id == record[3])
user_data = await database.fetch_one(query)
if user_data:
username = user_data["username"]
query = select(daily_reward.c.title).where(daily_reward.c.device_id == record[1])
device_title = await database.fetch_one(query)
if device_title:
device_info = device_title["title"]
else:
device_info = "1"
avatar_id = record[7] if record[7] else 1
avatar_url = f"/files/image/icon/avatar/{avatar_id}.png"
score = record[8]
html += f"""
#{rank}
{score}
"""
html += "
"
encrypted_mass = encryptAES(("vid=" + device_id + "&dummy=").encode("utf-8"))
html += f"""
Go Back
"""
file_path = os.path.join("files", "ranking.html")
try:
with open(file_path, "r", encoding="utf-8") as file:
html_content = file.read().format(text=html)
except FileNotFoundError:
return HTMLResponse("""
Ranking file not found
""", status_code=500)
return HTMLResponse(html_content)
else:
return HTMLResponse("""
Access denied
""", status_code=403)
async def status(request: Request):
decrypted_fields, _ = await decrypt_fields(request)
if not decrypted_fields:
return HTMLResponse("""
Invalid request data
""", status_code=400)
should_serve = True
if AUTHORIZATION_NEEDED:
should_serve = await check_whitelist(decrypted_fields) and not await check_blacklist(decrypted_fields)
if should_serve:
device_id = decrypted_fields[b'vid'][0].decode()
set_title = int(decrypted_fields[b'set_title'][0].decode()) if b'set_title' in decrypted_fields else None
page_id = int(decrypted_fields[b'page_id'][0].decode()) if b'page_id' in decrypted_fields else 0
if set_title:
update_query = (
update(daily_reward)
.where(daily_reward.c.device_id == device_id)
.values(title=set_title)
)
await database.execute(update_query)
query = select(daily_reward).where(daily_reward.c.device_id == device_id)
user_data = await database.fetch_one(query)
user_name = f"Guest({device_id[-6:]})"
if user_data:
query = select(user.c.username).where(user.c.device_id == device_id)
user_result = await database.fetch_one(query)
if user_result:
user_name = user_result["username"]
html = ""
if user_data:
player_element = f"""
{user_name}
Level {user_data['lvl']}
"""
html += player_element
page_name = ["Special", "Normal", "Master", "God"]
buttons_html = '
'
html += f"
{buttons_html}
"
selected_titles = TITLE_LISTS.get(page_id, [])
titles_html = '
'
for index, num in enumerate(selected_titles):
if index % 2 == 0:
if index != 0:
titles_html += '
'
titles_html += '
'
if num == user_data["title"]:
titles_html += f"""

"""
else:
encrypted_mass = encryptAES(f"vid={device_id}&title_id={num}&page_id={page_id}&dummy=".encode("utf-8"))
titles_html += f"""
"""
titles_html += '
'
html += titles_html
file_path = os.path.join("files", "status.html")
try:
with open(file_path, "r", encoding="utf-8") as file:
html_content = file.read().format(text=html)
except FileNotFoundError:
return HTMLResponse("""