This commit is contained in:
drmext 2023-10-18 19:49:22 +00:00
parent 2c934ca89d
commit 95763bd60c
No known key found for this signature in database
GPG Key ID: F1ED48FFE79A6961
9 changed files with 2425 additions and 2 deletions

View File

@ -51,7 +51,9 @@ async def core_get_game_version_from_software_version(software_version):
ext = int(ext)
if model == "LDJ":
if ext >= 2022101700:
if ext >= 2023101800:
return 31
elif ext in range(2022101700, 2023101700):
return 30
elif ext in range(2021101300, 2022101500):
return 29

View File

@ -109,7 +109,7 @@ async def iidx_profile_id_patch(iidx_id: str, item: IIDX_Profile_Main_Items):
async def iidx_profile_id_version_patch(
iidx_id: str, version: int, item: IIDX_Profile_Version_Items
):
if version != 30:
if version < 30:
# TODO: differentiate 18, 19, 20, 29, 30
return Response(status_code=406)
iidx_id = int("".join([i for i in iidx_id if i.isnumeric()]))

View File

@ -0,0 +1,94 @@
from time import time
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/local2", tags=["local2"])
router.model_whitelist = ["LDJ"]
@router.post("/{gameinfo}/IIDX31gameSystem/systemInfo")
async def iidx31gamesystem_systeminfo(request: Request):
request_info = await core_process_request(request)
unlock = ()
# force unlock LM exclusives to complete unlock all songs server side
# this makes LM exclusive folder disappear, so just use hex edits
# unlock = (28073, 28008, 29095, 29094, 29027, 30077, 30076, 30098, 30106, 30107, 30028, 30064, 30027)
current_time = round(time())
response = E.response(
E.IIDX31gameSystem(
*[
E.music_open(
E.music_id(mid, __type="s32"),
E.kind(0, __type="s32"),
)
for mid in unlock
],
# E.grade_course(),
# E.arena_schedule(
# E.phase(3, __type="u8"),
# E.start(current_time - 600, __type="u32"),
# E.end(current_time + 600, __type="u32"),
# ),
# *[
# E.arena_reward(
# E.index(unlock.index(mid), __type="s32"),
# E.cube_num((unlock.index(mid) + 1) * 50, __type="s32"),
# E.kind(0, __type="s32"),
# E.value(mid, __type="str"),
# )
# for mid in unlock
# ],
# *[
# E.arena_music_difficult(
# E.play_style(sp_dp, __type="s32"),
# E.arena_class(arena_class, __type="s32"),
# E.low_difficult(8, __type="s32"),
# E.high_difficult(12, __type="s32"),
# E.is_leggendaria(1, __type="bool"),
# E.force_music_list_id(0, __type="s32"),
# )
# for sp_dp in (0, 1)
# for arena_class in range(20)
# ],
# *[
# E.arena_cpu_define(
# E.play_style(sp_dp, __type="s32"),
# E.arena_class(arena_class, __type="s32"),
# E.grade_id(18, __type="s32"),
# E.low_music_difficult(8, __type="s32"),
# E.high_music_difficult(12, __type="s32"),
# E.is_leggendaria(0, __type="bool"),
# )
# for sp_dp in (0, 1)
# for arena_class in range(20)
# ],
# *[
# E.maching_class_range(
# E.play_style(sp_dp, __type="s32"),
# E.matching_class(arena_class, __type="s32"),
# E.low_arena_class(arena_class, __type="s32"),
# E.high_arena_class(arena_class, __type="s32"),
# )
# for sp_dp in (0, 1)
# for arena_class in range(20)
# ],
E.CommonBossPhase(val=0),
# E.Event1InternalPhase(val=0),
# E.ExtraBossEventPhase(val=0),
E.isNewSongAnother12OpenFlg(val=1),
# E.gradeOpenPhase(val=2),
# E.isEiseiOpenFlg(val=1),
# E.WorldTourismOpenList(val=-1),
E.OldBPLBattleOpenPhase(val=3),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

110
modules/iidx/iidx31grade.py Normal file
View File

@ -0,0 +1,110 @@
import time
from tinydb import Query, where
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
from core_database import get_db
router = APIRouter(prefix="/local2", tags=["local2"])
router.model_whitelist = ["LDJ"]
def get_profile(iidx_id):
return get_db().table("iidx_profile").get(where("iidx_id") == iidx_id)
@router.post("/{gameinfo}/IIDX31grade/raised")
async def iidx31grade_raised(request: Request):
request_info = await core_process_request(request)
game_version = request_info["game_version"]
timestamp = time.time()
iidx_id = int(request_info["root"][0].attrib["iidxid"])
achi = int(request_info["root"][0].attrib["achi"])
cstage = int(request_info["root"][0].attrib["cstage"])
gid = int(request_info["root"][0].attrib["gid"])
gtype = int(request_info["root"][0].attrib["gtype"])
is_ex = int(request_info["root"][0].attrib["is_ex"])
is_mirror = int(request_info["root"][0].attrib["is_mirror"])
db = get_db()
db.table("iidx_class").insert(
{
"timestamp": timestamp,
"game_version": game_version,
"iidx_id": iidx_id,
"achi": achi,
"cstage": cstage,
"gid": gid,
"gtype": gtype,
"is_ex": is_ex,
"is_mirror": is_mirror,
},
)
profile = get_profile(iidx_id)
game_profile = profile["version"].get(str(game_version), {})
best_class = db.table("iidx_class_best").get(
(where("iidx_id") == iidx_id)
& (where("game_version") == game_version)
& (where("gid") == gid)
& (where("gtype") == gtype)
)
best_class = {} if best_class is None else best_class
best_class_data = {
"game_version": game_version,
"iidx_id": iidx_id,
"achi": max(achi, best_class.get("achi", achi)),
"cstage": max(cstage, best_class.get("cstage", cstage)),
"gid": gid,
"gtype": gtype,
"is_ex": is_ex,
"is_mirror": is_mirror,
}
db.table("iidx_class_best").upsert(
best_class_data,
(where("iidx_id") == iidx_id)
& (where("game_version") == game_version)
& (where("gid") == gid)
& (where("gtype") == gtype),
)
best_class_plays = db.table("iidx_class_best").search(
(where("game_version") == game_version) & (where("iidx_id") == iidx_id)
)
grades = []
for record in best_class_plays:
grades.append(
[record["gtype"], record["gid"], record["cstage"], record["achi"]]
)
game_profile["grade_values"] = grades
grade_sp = db.table("iidx_class_best").search(
(where("iidx_id") == iidx_id) & (where("gtype") == 0) & (where("cstage") == 4)
)
game_profile["grade_single"] = max([x["gid"] for x in grade_sp], default=-1)
grade_dp = db.table("iidx_class_best").search(
(where("iidx_id") == iidx_id) & (where("gtype") == 1) & (where("cstage") == 4)
)
game_profile["grade_double"] = max([x["gid"] for x in grade_dp], default=-1)
profile["version"][str(game_version)] = game_profile
db.table("iidx_profile").upsert(profile, where("game_version") == game_version)
response = E.response(E.IIDX31grade(pnum=1))
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

169
modules/iidx/iidx31lobby.py Normal file
View File

@ -0,0 +1,169 @@
from time import time
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/lobby", tags=["lobby"])
router.model_whitelist = ["LDJ"]
arena_host = {}
bpl_host = {}
@router.post("/{gameinfo}/IIDX31lobby/entry")
async def iidx31lobby_entry(request: Request):
request_info = await core_process_request(request)
root = request_info["root"][0]
sp_dp = root.find("play_style").text
arena_class = root.find("arena_class").text
ga = root.find("address/ga").text.split()
gp = root.find("address/gp").text
la = root.find("address/la").text.split()
if arena_host and time() < arena_host["time"]:
# test menu reset
if arena_host["ga"] == ga:
is_arena_host = 1
arena_host["time"] = time() + 30
else:
is_arena_host = 0
response = E.response(
E.IIDX31lobby(
E.host(is_arena_host, __type="bool"),
E.matching_class(arena_class, __type="s32"),
E.address(
E.ga(arena_host["ga"], __type="u8"),
E.gp(arena_host["gp"], __type="u16"),
E.la(arena_host["la"], __type="u8"),
),
)
)
else:
arena_host["ga"] = ga
arena_host["gp"] = gp
arena_host["la"] = la
arena_host["time"] = time() + 30
response = E.response(
E.IIDX31lobby(
E.host(1, __type="bool"),
E.matching_class(arena_class, __type="s32"),
E.address(
E.ga(ga, __type="u8"),
E.gp(gp, __type="u16"),
E.la(la, __type="u8"),
),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31lobby/update")
async def iidx31lobby_update(request: Request):
request_info = await core_process_request(request)
response = E.response(E.IIDX31lobby())
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31lobby/delete")
async def iidx31lobby_delete(request: Request):
request_info = await core_process_request(request)
# normal reset
del arena_host["ga"]
del arena_host["gp"]
del arena_host["la"]
del arena_host["time"]
response = E.response(E.IIDX31lobby())
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31lobby/bplbattle_entry")
async def iidx31lobby_bplbattle_entry(request: Request):
request_info = await core_process_request(request)
root = request_info["root"][0]
sp_dp = root.find("play_style").text
arena_class = root.find("arena_class").text
password = root.find("passward").text # passward
ga = root.find("address/ga").text.split()
gp = root.find("address/gp").text
la = root.find("address/la").text.split()
if bpl_host and password in bpl_host and time() < bpl_host[password]["time"]:
# test menu reset
if bpl_host[password]["ga"] == ga:
is_bpl_host = 1
bpl_host[password]["time"] = time() + 30
else:
is_bpl_host = 0
response = E.response(
E.IIDX31lobby(
E.host(is_bpl_host, __type="bool"),
E.matching_class(arena_class, __type="s32"),
E.address(
E.ga(bpl_host[password]["ga"], __type="u8"),
E.gp(bpl_host[password]["gp"], __type="u16"),
E.la(bpl_host[password]["la"], __type="u8"),
),
)
)
else:
bpl_host[password] = {}
bpl_host[password]["ga"] = ga
bpl_host[password]["gp"] = gp
bpl_host[password]["la"] = la
bpl_host[password]["time"] = time() + 30
response = E.response(
E.IIDX31lobby(
E.host(1, __type="bool"),
E.matching_class(arena_class, __type="s32"),
E.address(
E.ga(ga, __type="u8"),
E.gp(gp, __type="u16"),
E.la(la, __type="u8"),
),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31lobby/bplbattle_update")
async def iidx31lobby_bplbattle_update(request: Request):
request_info = await core_process_request(request)
response = E.response(E.IIDX31lobby())
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31lobby/bplbattle_delete")
async def iidx31lobby_bplbattle_delete(request: Request):
request_info = await core_process_request(request)
root = request_info["root"][0]
ga = root.find("address/ga").text.split()
# normal reset
for host in bpl_host:
if bpl_host[host]["ga"] == ga:
del bpl_host[host]
break
response = E.response(E.IIDX31lobby())
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

543
modules/iidx/iidx31music.py Normal file
View File

@ -0,0 +1,543 @@
import time
import random
from enum import IntEnum
from fastapi import APIRouter, Request, Response
from tinydb import where
from core_common import core_process_request, core_prepare_response, E
from core_database import get_db
import config
router = APIRouter(prefix="/local2", tags=["local2"])
router.model_whitelist = ["LDJ"]
class ClearFlags(IntEnum):
NO_PLAY = 0
FAILED = 1
ASSIST_CLEAR = 2
EASY_CLEAR = 3
CLEAR = 4
HARD_CLEAR = 5
EX_HARD_CLEAR = 6
FULL_COMBO = 7
@router.post("/{gameinfo}/IIDX31music/getrank")
async def iidx31music_getrank(request: Request):
request_info = await core_process_request(request)
game_version = request_info["game_version"]
root = request_info["root"][0]
play_style = int(root.attrib["cltype"])
requested_ids = [
int(root.get("iidxid", 0)),
int(root.get("iidxid0", 0)),
int(root.get("iidxid1", 0)),
int(root.get("iidxid2", 0)),
int(root.get("iidxid3", 0)),
int(root.get("iidxid4", 0)),
int(root.get("iidxid5", 0)),
]
all_scores = {}
db = get_db()
for rival_idx, iidxid in enumerate(requested_ids, -1):
if iidxid == 0:
continue
profile = db.table("iidx_profile").get(where("iidx_id") == iidxid)["version"][
str(game_version)
]
for record in db.table("iidx_scores_best").search(
(where("music_id") < (game_version + 1) * 1000)
& (where("play_style") == play_style)
& (where("iidx_id") == iidxid)
):
music_id = record["music_id"]
clear_flg = record["clear_flg"]
ex_score = record["ex_score"]
miss_count = record["miss_count"]
chart_id = record["chart_id"]
if (rival_idx, music_id) not in all_scores:
all_scores[rival_idx, music_id] = {
0: {"clear_flg": -1, "ex_score": -1, "miss_count": -1},
1: {"clear_flg": -1, "ex_score": -1, "miss_count": -1},
2: {"clear_flg": -1, "ex_score": -1, "miss_count": -1},
3: {"clear_flg": -1, "ex_score": -1, "miss_count": -1},
4: {"clear_flg": -1, "ex_score": -1, "miss_count": -1},
}
all_scores[rival_idx, music_id][chart_id]["clear_flg"] = clear_flg
all_scores[rival_idx, music_id][chart_id]["ex_score"] = ex_score
all_scores[rival_idx, music_id][chart_id]["miss_count"] = miss_count
names = {}
profiles = get_db().table("iidx_profile")
for p in profiles:
names[p["iidx_id"]] = {}
try:
names[p["iidx_id"]]["name"] = p["version"][str(game_version)]["djname"]
except KeyError:
names[p["iidx_id"]]["name"] = "UNK"
top_scores = {}
for record in db.table("iidx_scores_best").search(
(where("music_id") < (game_version + 1) * 1000)
& (where("play_style") == play_style)
):
music_id = record["music_id"]
ex_score = record["ex_score"]
chart_id = record["chart_id"]
iidx_id = record["iidx_id"]
if music_id not in top_scores:
top_scores[music_id] = {
0: {"djname": "", "clear_flg": -1, "ex_score": -1},
1: {"djname": "", "clear_flg": -1, "ex_score": -1},
2: {"djname": "", "clear_flg": -1, "ex_score": -1},
3: {"djname": "", "clear_flg": -1, "ex_score": -1},
4: {"djname": "", "clear_flg": -1, "ex_score": -1},
}
if ex_score > top_scores[music_id][chart_id]["ex_score"]:
top_scores[music_id][chart_id]["djname"] = names[iidx_id]["name"]
top_scores[music_id][chart_id]["clear_flg"] = 1
top_scores[music_id][chart_id]["ex_score"] = ex_score
response = E.response(
E.IIDX31music(
E.style(type=play_style),
*[
E.m(
[
i,
k,
*[all_scores[i, k][d]["clear_flg"] for d in range(5)],
*[all_scores[i, k][d]["ex_score"] for d in range(5)],
*[all_scores[i, k][d]["miss_count"] for d in range(5)],
],
__type="s16",
)
for i, k in all_scores
],
*[
E.top(
E.detail(
[
k,
*[top_scores[k][d]["clear_flg"] for d in range(5)],
*[top_scores[k][d]["ex_score"] for d in range(5)],
],
__type="s16",
),
name0=top_scores[k][0]["djname"],
name1=top_scores[k][1]["djname"],
name2=top_scores[k][2]["djname"],
name3=top_scores[k][3]["djname"],
name4=top_scores[k][4]["djname"],
)
for k in top_scores
],
)
)
assert response is not None
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31music/crate")
async def iidx31music_crate(request: Request):
request_info = await core_process_request(request)
game_version = request_info["game_version"]
db = get_db()
all_score_stats = db.table("iidx_score_stats").search(
(where("music_id") < (game_version + 1) * 1000)
)
crate = {}
fcrate = {}
for stat in all_score_stats:
if stat["music_id"] not in crate:
crate[stat["music_id"]] = [1001] * 10
if stat["music_id"] not in fcrate:
fcrate[stat["music_id"]] = [1001] * 10
if stat["play_style"] == 1:
dp_idx = 5
else:
dp_idx = 0
crate[stat["music_id"]][stat["chart_id"] + dp_idx] = int(stat["clear_rate"])
fcrate[stat["music_id"]][stat["chart_id"] + dp_idx] = int(stat["fc_rate"])
response = E.response(
E.IIDX31music(*[E.c(crate[k] + fcrate[k], mid=k, __type="s32") for k in crate])
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31music/reg")
async def iidx31music_reg(request: Request):
request_info = await core_process_request(request)
game_version = request_info["game_version"]
timestamp = time.time()
log = request_info["root"][0].find("music_play_log")
clear_flg = int(request_info["root"][0].attrib["cflg"])
clid = int(request_info["root"][0].attrib["clid"])
is_death = int(request_info["root"][0].attrib["is_death"])
pid = int(request_info["root"][0].attrib["pid"])
play_style = int(log.attrib["play_style"])
ex_score = int(log.attrib["ex_score"])
folder_type = int(log.attrib["folder_type"])
gauge_type = int(log.attrib["gauge_type"])
graph_type = int(log.attrib["graph_type"])
great_num = int(log.attrib["great_num"])
iidx_id = int(log.attrib["iidx_id"])
miss_num = int(log.attrib["miss_num"])
mode_type = int(log.attrib["mode_type"])
music_id = int(log.attrib["music_id"])
note_id = int(log.attrib["note_id"])
option1 = int(log.attrib["option1"])
option2 = int(log.attrib["option2"])
pgreat_num = int(log.attrib["pgreat_num"])
ghost = log.find("ghost").text
ghost_gauge = log.find("ghost_gauge").text
db = get_db()
db.table("iidx_scores").insert(
{
"timestamp": timestamp,
"game_version": game_version,
"iidx_id": iidx_id,
"pid": pid,
"clear_flg": clear_flg,
"is_death": is_death,
"music_id": music_id,
"play_style": play_style,
"chart_id": note_id,
"pgreat_num": pgreat_num,
"great_num": great_num,
"ex_score": ex_score,
"miss_count": miss_num,
"folder_type": folder_type,
"gauge_type": gauge_type,
"graph_type": graph_type,
"mode_type": mode_type,
"option1": option1,
"option2": option2,
"ghost": ghost,
"ghost_gauge": ghost_gauge,
},
)
best_score = db.table("iidx_scores_best").get(
(where("iidx_id") == iidx_id)
& (where("play_style") == play_style)
& (where("music_id") == music_id)
& (where("chart_id") == note_id)
)
best_score = {} if best_score is None else best_score
if clear_flg < ClearFlags.EASY_CLEAR:
miss_num = -1
best_miss_count = best_score.get("miss_count", miss_num)
if best_miss_count == -1:
miss_count = max(miss_num, best_miss_count)
elif clear_flg > ClearFlags.ASSIST_CLEAR:
miss_count = min(miss_num, best_miss_count)
else:
miss_count = best_miss_count
best_ex_score = best_score.get("ex_score", ex_score)
best_score_data = {
"game_version": game_version,
"iidx_id": iidx_id,
"pid": pid,
"play_style": play_style,
"music_id": music_id,
"chart_id": note_id,
"miss_count": miss_count,
"ex_score": max(ex_score, best_ex_score),
"ghost": ghost if ex_score >= best_ex_score else best_score.get("ghost", ghost),
"ghost_gauge": ghost_gauge
if ex_score >= best_ex_score
else best_score.get("ghost_gauge", ghost_gauge),
"clear_flg": max(clear_flg, best_score.get("clear_flg", clear_flg)),
"gauge_type": gauge_type
if ex_score >= best_ex_score
else best_score.get("gauge_type", gauge_type),
}
db.table("iidx_scores_best").upsert(
best_score_data,
(where("iidx_id") == iidx_id)
& (where("play_style") == play_style)
& (where("music_id") == music_id)
& (where("chart_id") == note_id),
)
score_stats = db.table("iidx_score_stats").get(
(where("music_id") == music_id)
& (where("play_style") == play_style)
& (where("chart_id") == note_id)
)
score_stats = {} if score_stats is None else score_stats
score_stats["game_version"] = game_version
score_stats["play_style"] = play_style
score_stats["music_id"] = music_id
score_stats["chart_id"] = note_id
score_stats["play_count"] = score_stats.get("play_count", 0) + 1
score_stats["fc_count"] = score_stats.get("fc_count", 0) + (
1 if clear_flg == ClearFlags.FULL_COMBO else 0
)
score_stats["clear_count"] = score_stats.get("clear_count", 0) + (
1 if clear_flg >= ClearFlags.EASY_CLEAR else 0
)
score_stats["fc_rate"] = int(
(score_stats["fc_count"] / score_stats["play_count"]) * 1000
)
score_stats["clear_rate"] = int(
(score_stats["clear_count"] / score_stats["play_count"]) * 1000
)
db.table("iidx_score_stats").upsert(
score_stats,
(where("music_id") == music_id)
& (where("play_style") == play_style)
& (where("chart_id") == note_id),
)
ranklist_data = []
ranklist_scores = db.table("iidx_scores_best").search(
(where("play_style") == play_style)
& (where("music_id") == music_id)
& (where("chart_id") == note_id)
)
ranklist_scores = [] if ranklist_scores is None else ranklist_scores
ranklist_scores_ranked = []
for score in ranklist_scores:
profile = db.table("iidx_profile").get(where("iidx_id") == score["iidx_id"])
if profile is None or str(game_version) not in profile["version"]:
continue
game_profile = profile["version"][str(game_version)]
ranklist_scores_ranked.append(
{
"opname": config.arcade,
"name": game_profile["djname"],
"pid": game_profile["region"],
"body": game_profile["body"],
"face": game_profile["face"],
"hair": game_profile["hair"],
"hand": game_profile["hand"],
"head": game_profile["head"],
"dgrade": game_profile["grade_double"],
"sgrade": game_profile["grade_single"],
"score": score["ex_score"],
"iidx_id": score["iidx_id"],
"clflg": score["clear_flg"],
"myFlg": score["iidx_id"] == iidx_id,
}
)
ranklist_scores_ranked = sorted(
ranklist_scores_ranked, key=lambda x: (x["clflg"], x["score"]), reverse=True
)
myRank = 0
for rnum, score in enumerate(ranklist_scores_ranked):
r = E.data(
rnum=rnum + 1,
opname=score["opname"],
name=score["name"],
pid=score["pid"],
body=score["body"],
face=score["face"],
hair=score["hair"],
hand=score["hand"],
head=score["head"],
dgrade=score["dgrade"],
sgrade=score["sgrade"],
score=score["score"],
iidx_id=score["iidx_id"],
clflg=score["clflg"],
myFlg=score["myFlg"],
achieve=0,
)
ranklist_data.append(r)
if score["myFlg"]:
myRank = rnum + 1
response = E.response(
E.IIDX31music(
E.ranklist(*ranklist_data, total_user_num=len(ranklist_data)),
E.shopdata(rank=myRank),
clid=clid,
crate=score_stats["clear_rate"],
frate=score_stats["fc_rate"],
mid=music_id,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31music/appoint")
async def iidx31music_appoint(request: Request):
request_info = await core_process_request(request)
root = request_info["root"][0]
iidxid = int(root.attrib["iidxid"])
music_id = int(root.attrib["mid"])
chart_id = int(root.attrib["clid"])
ctype = int(root.attrib["ctype"])
subtype = root.attrib["subtype"]
db = get_db()
record = db.table("iidx_scores_best").get(
(where("iidx_id") == iidxid)
& (where("music_id") == music_id)
& (where("chart_id") == chart_id)
)
vals = []
if record is not None:
vals.append(
E.mydata(
record["ghost"],
score=record["ex_score"],
__type="bin",
__size=len(record["ghost"]) // 2,
)
)
if ctype == 1:
sdata = db.table("iidx_scores_best").get(
(where("iidx_id") == int(subtype))
& (where("music_id") == music_id)
& (where("chart_id") == chart_id)
)
elif ctype in (2, 4, 10):
sdata = {
"game_version": 29,
"ghost": "",
"ex_score": 0,
"iidx_id": 0,
"name": "",
"pid": 13,
}
for record in db.table("iidx_scores_best").search(
(where("music_id") == music_id) & (where("chart_id") == chart_id)
):
if record["ex_score"] > sdata["ex_score"]:
sdata["game_version"] = record["game_version"]
sdata["ghost"] = record["ghost"]
sdata["ex_score"] = record["ex_score"]
sdata["iidx_id"] = record["iidx_id"]
sdata["pid"] = record["pid"]
if ctype in (1, 2, 4, 10) and sdata["ex_score"] != 0:
vals.append(
E.sdata(
sdata["ghost"],
score=sdata["ex_score"],
name=db.table("iidx_profile").get(where("iidx_id") == sdata["iidx_id"])[
"version"
][str(sdata["game_version"])]["djname"],
pid=sdata["pid"],
__type="bin",
__size=len(sdata["ghost"]) // 2,
)
)
response = E.response(E.IIDX31music(*vals))
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31music/arenaCPU")
async def iidx31music_arenaCPU(request: Request):
request_info = await core_process_request(request)
root = request_info["root"][0]
music_list = root.findall("music_list")
music_count = len(music_list)
cpu_list = root.findall("cpu_list")
cpu_count = len(cpu_list)
cpu_scores = {}
cpu_ghosts = {}
for music in music_list:
i = int(music.find("index").text)
exscore_max = int(music.find("total_notes").text) * 2
cpu_scores[i] = {}
cpu_ghosts[i] = {}
for j in range(cpu_count):
cpu_scores[i][j] = {}
cpu_ghosts[i][j] = {}
exscore = round(exscore_max * random.uniform(0.77, 0.93))
cpu_scores[i][j]["exscore"] = exscore
ghost_len = 64
ghost_data = [0] * ghost_len
for x in range(ghost_len):
ghost_data[x] = exscore // ghost_len
if (exscore % ghost_len) > x:
ghost_data[x] += 1
cpu_ghosts[i][j]["ghost_data"] = ghost_data
response = E.response(
E.IIDX31music(
*[
E.cpu_score_list(
E.index(i, __type="s32"),
*[
E.score_list(
E.index(j, __type="s32"),
E.score(cpu_scores[i][j]["exscore"], __type="s32"),
E.ghost(cpu_ghosts[i][j]["ghost_data"], __type="u8"),
E.enable_score(1, __type="bool"),
E.enable_ghost(1, __type="bool"),
E.location_id("X000000001", __type="str"),
)
for j in range(cpu_count)
],
)
for i in range(music_count)
],
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

1423
modules/iidx/iidx31pc.py Normal file

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,18 @@
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/local2", tags=["local2"])
router.model_whitelist = ["LDJ"]
@router.post("/{gameinfo}/IIDX31ranking/getranker")
async def iidx31ranking_getranker(request: Request):
request_info = await core_process_request(request)
response = E.response(E.IIDX31ranking())
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

View File

@ -0,0 +1,64 @@
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/local2", tags=["local2"])
router.model_whitelist = ["LDJ"]
@router.post("/{gameinfo}/IIDX31shop/getname")
async def iidx31shop_getname(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX31shop(
cls_opt=0,
opname=config.arcade,
pid=13,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31shop/getconvention")
async def iidx31shop_getconvention(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX31shop(
E.valid(1, __type="bool"),
music_0=-1,
music_1=-1,
music_2=-1,
music_3=-1,
start_time=0,
end_time=0,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31shop/sentinfo")
async def iidx31shop_sentinfo(request: Request):
request_info = await core_process_request(request)
response = E.response(E.IIDX31shop())
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post("/{gameinfo}/IIDX31shop/sendescapepackageinfo")
async def iidx31shop_sendescapepackageinfo(request: Request):
request_info = await core_process_request(request)
response = E.response(E.IIDX31shop(expire=1200))
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)