from starlette.responses import Response, FileResponse, HTMLResponse from starlette.requests import Request from starlette.routing import Route import os import json import math from sqlalchemy import select, update import xml.etree.ElementTree as ET from config import START_COIN, AUTHORIZATION_NEEDED, STAGE_PRICE, START_COIN, AVATAR_PRICE, ITEM_PRICE, FMAX_PRICE, EX_PRICE from api.crypt import decrypt_fields from api.misc import inform_page, parse_res, FMAX_VER, FMAX_RES from api.database import database, daily_reward, check_blacklist, check_whitelist from api.templates import START_AVATARS, START_STAGES, EXCLUDE_STAGE_EXP, SONG_LIST, AVATAR_LIST, ITEM_LIST async def web_shop(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: cnt_type = decrypted_fields[b'cnt_type'][0].decode() device_id = decrypted_fields[b'vid'][0].decode() if (decrypted_fields.get(b'page')): page = int(decrypted_fields.get(b'page')[0].decode()) else: page = 0 inc = 0 fmax_inc = 0 buttons_html = "" spawn_prev_page = False spawn_next_page = False query = select(daily_reward.c.my_stage, daily_reward.c.my_avatar, daily_reward.c.coin).where(daily_reward.c.device_id == device_id) result = await database.fetch_one(query) my_stage = set(json.loads(result["my_stage"])) if result and result["my_stage"] else START_STAGES my_avatar = set(json.loads(result["my_avatar"])) if result and result["my_avatar"] else START_AVATARS coin = result["coin"] if result and result["coin"] else START_COIN if cnt_type == "1": low_range = 100 up_range = 615 if page < 0 or page > math.ceil(up_range - low_range) / 80: return HTMLResponse("""

Invalid page number

""", status_code=400) if page > 0: spawn_prev_page = True if page < math.ceil(up_range - low_range) / 80 - 1: spawn_next_page = True low_range = low_range + page * 80 up_range = min(up_range, low_range + 80) if 700 not in my_stage and os.path.isfile('./files/dlc_4max.html') and not spawn_prev_page: buttons_html += """

""" fmax_inc = 1 elif 700 in my_stage and os.path.isfile('./files/dlc_4max.html') and not spawn_prev_page: buttons_html += """

""" if 980 not in my_stage and os.path.isfile('./files/dlc_extra.html') and not spawn_prev_page: buttons_html += """

""" fmax_inc = 1 elif 980 in my_stage and os.path.isfile('./files/dlc_4max.html') and not spawn_prev_page: buttons_html += """

""" for i in range(low_range, up_range): if i not in my_stage and i not in EXCLUDE_STAGE_EXP: buttons_html += f""" """ inc += 1 if inc % 4 == 0: buttons_html += "
" if spawn_prev_page: buttons_html += """
""".format(page - 1) if spawn_next_page: buttons_html += """
""".format(page + 1) elif cnt_type == "2": low_range = 15 up_range = 173 if FMAX_VER == 0 else 267 if page < 0 or page > math.ceil(up_range - low_range) / 80: return HTMLResponse("""

Invalid page number

""", status_code=400) if page > 0: spawn_prev_page = True if page < math.ceil(up_range - low_range) / 80 - 1: spawn_next_page = True low_range = low_range + page * 80 up_range = min(up_range, low_range + 80) for i in range(low_range, up_range): if i not in my_avatar and i not in EXCLUDE_STAGE_EXP: buttons_html += f""" """ inc += 1 if inc % 4 == 0: buttons_html += "
" if spawn_prev_page: buttons_html += """
""".format(page - 1) if spawn_next_page: buttons_html += """
""".format(page + 1) elif cnt_type == "3": for i in range(1, 11): buttons_html += f""" """ if i % 4 == 0: buttons_html += "
" if inc == 0 and fmax_inc == 0 and cnt_type != "3": buttons_html += """

Everything has been purchased!

""" html_path = f"files/web_shop_{cnt_type}.html" try: with open(html_path, "r", encoding="utf-8") as file: html_content = file.read().format(text=buttons_html, coin=coin) except FileNotFoundError: return HTMLResponse("""

Shop template not found

""", status_code=500) return HTMLResponse(html_content) else: return HTMLResponse("""

Access denied

""", status_code=403) async def web_shop_detail(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: cnt_type = decrypted_fields[b'cnt_type'][0].decode() cnt_id = int(decrypted_fields[b'cnt_id'][0].decode()) device_id = decrypted_fields[b'vid'][0].decode() if b'return_page' in decrypted_fields: return_page = decrypted_fields[b'return_page'][0].decode() else: return_page = "1" query = select(daily_reward.c.coin).where(daily_reward.c.device_id == device_id) result = await database.fetch_one(query) coin = result["coin"] if result and result["coin"] else 0 html = "" if cnt_type == "1": if cnt_id > -1: song = SONG_LIST[cnt_id] difficulty_levels = "/".join(map(str, song.get("difficulty_levels", []))) song_stage_price = STAGE_PRICE * 2 if len(song["difficulty_levels"]) == 6 else STAGE_PRICE html = f"""

Would you like to purchase this song?

{song.get("name_en")} - {song.get("author_en")}

Difficulty Levels: {difficulty_levels}

{song_stage_price}

""" elif cnt_id == -3: log = parse_res(FMAX_RES) html = f"""

You have unlocked the GC4MAX expansion!

Please report bugs/missing tracks to Discord: #AnTcfgss, or QQ 3421587952.

This server has version {FMAX_VER}.

Update log:

{log}

""" elif cnt_id == -4: html = f"""

You have unlocked the EXTRA Challenge!

Please report bugs/missing tracks to Discord: #AnTcfgss, or QQ 3421587952.

""" elif cnt_id == -2: html = f"""

Are you looking for a bad time?

If so, this is the Ultimate - Extra - Challenge.

180+ Arcade Extra difficulty charts await you.

You have been warned.

""" elif cnt_id == -1: html = f"""

Experience the arcade with the GC4MAX expansion! This DLC unlocks 320+ exclusive songs for your 2OS experience.

Note that these songs don't have mobile difficulties. A short placeholder is used, and GCoin reward is not available for playing them. You must clear the Normal difficulty to unlock AC content.

Due to technical limitations, Extra level charts cannot be ported as of now. After purchasing, you will have access to support information and update logs.

""" elif cnt_type == "2": avatar = next((item for item in AVATAR_LIST if item.get("id") == cnt_id), None) if avatar: html = f"""

Would you like to purchase this avatar?

{avatar.get("name")}

Effect: {avatar.get("effect")}

{AVATAR_PRICE}

""" else: html = "

Avatar not found.

" elif cnt_type == "3": item = next((item for item in ITEM_LIST if item.get("id") == cnt_id), None) if item: html = f"""

Would you like to purchase this item?

{item.get("name")}

Effect: {item.get("effect")}

{ITEM_PRICE}

""" else: html = "

Item not found.

" if cnt_type == "1" and (cnt_id == -1 or cnt_id == -3): source_html = f"files/dlc_4max.html" elif cnt_type == "1" and (cnt_id == -2 or cnt_id == -4): source_html = f"files/dlc_extra.html" else: source_html = f"files/web_shop_detail.html" html += f"""

Buy
Go Back

""" try: with open(source_html, "r", encoding="utf-8") as file: html_content = file.read().format(text=html, coin=coin) except FileNotFoundError: return HTMLResponse("""

Shop detail template not found

""", status_code=500) return HTMLResponse(html_content) else: return HTMLResponse("""

Access denied

""", status_code=403) async def buy_by_coin(request: Request): decrypted_fields, _ = await decrypt_fields(request) if not decrypted_fields: return Response("""1coin_error.php""", media_type="application/xml") should_serve = True if AUTHORIZATION_NEEDED: should_serve = await check_whitelist(decrypted_fields) and not await check_blacklist(decrypted_fields) if should_serve: cnt_type = decrypted_fields[b'cnt_type'][0].decode() cnt_id = int(decrypted_fields[b'cnt_id'][0].decode()) num = int(decrypted_fields[b'num'][0].decode()) device_id = decrypted_fields[b'vid'][0].decode() fail_url = """1coin_error.php""" query = select(daily_reward.c.my_stage, daily_reward.c.my_avatar, daily_reward.c.coin, daily_reward.c.item).where(daily_reward.c.device_id == device_id) result = await database.fetch_one(query) if not result: return Response(fail_url, media_type="application/xml") my_stage = set(json.loads(result["my_stage"])) if result["my_stage"] else set() my_avatar = set(json.loads(result["my_avatar"])) if result["my_avatar"] else set() coin = int(result["coin"]) if result["coin"] else 0 item = json.loads(result["item"]) if result["item"] else [] if cnt_type == "1": if cnt_id == -1: song_stage_price = FMAX_PRICE if coin < song_stage_price: return Response(fail_url, media_type="application/xml") for i in range(615, 926): my_stage.add(i) coin -= song_stage_price elif cnt_id == -2: song_stage_price = EX_PRICE if coin < song_stage_price: return Response(fail_url, media_type="application/xml") for i in range(926, 985): my_stage.add(i) coin -= song_stage_price else: song_stage_price = STAGE_PRICE * 2 if len(SONG_LIST[cnt_id]["difficulty_levels"]) == 6 else STAGE_PRICE if coin < song_stage_price or cnt_id in my_stage: return Response(fail_url, media_type="application/xml") coin -= song_stage_price my_stage.add(cnt_id) elif cnt_type == "2": if coin < AVATAR_PRICE or cnt_id in my_avatar: return Response(fail_url, media_type="application/xml") coin -= AVATAR_PRICE my_avatar.add(cnt_id) elif cnt_type == "3": if coin < ITEM_PRICE: return Response(fail_url, media_type="application/xml") coin -= ITEM_PRICE item.append(cnt_id) else: return Response(fail_url, media_type="application/xml") update_query = ( update(daily_reward) .where(daily_reward.c.device_id == device_id) .values( my_stage=json.dumps(list(my_stage)), my_avatar=json.dumps(list(my_avatar)), coin=coin, item=json.dumps(item) ) ) await database.execute(update_query) response = ET.Element("response") ET.SubElement(response, "code").text = "0" ET.SubElement(response, "result_url").text = "web_shop_result.php" ET.SubElement(response, "cnt_type").text = cnt_type ET.SubElement(response, "cnt_id").text = str(cnt_id) ET.SubElement(response, "num").text = str(num) if cnt_type == "1": ET.SubElement(response, "stage_id").text = str(cnt_id) response_string = ET.tostring(response, encoding="utf-8", method="xml").decode("utf-8") return Response(response_string, media_type="application/xml") else: return Response("""1coin_error.php""", media_type="application/xml") async def web_shop_result(request: Request): decrypted_fields, _ = await decrypt_fields(request) cnt_type = decrypted_fields[b'cnt_type'][0].decode() return HTMLResponse(inform_page(f"""SUCCESS:
Purchase successful.
Please close this page and the reward will arrive shortly.
If it took too long, try restarting the game.
Go Back""", 2)) async def coin_error(request: Request): return HTMLResponse(inform_page(f"""FAILED:
Either you don't have enough coin,
or there were a duplicate order, and the reward will arrive shortly.""", 2)) routes = [ Route('/web_shop.php', web_shop, methods=['GET', 'POST']), Route('/web_shop_detail.php', web_shop_detail, methods=['GET', 'POST']), Route('/buy_by_coin.php', buy_by_coin, methods=['GET']), Route('/web_shop_result.php', web_shop_result, methods=['GET']), Route('/coin_error.php', coin_error, methods=['GET']), ]