Monkey Business

This commit is contained in:
drmext 2022-08-26 10:39:11 +00:00
commit e8540eac52
No known key found for this signature in database
GPG Key ID: F1ED48FFE79A6961
37 changed files with 3998 additions and 0 deletions

3
.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
*.pyc
db.json
*.db

12
README.md Normal file
View File

@ -0,0 +1,12 @@
# MonkeyBusiness
e-amusement server using FastAPI and TinyDB
for experimental testing
# Instructions
`pip install -U -r requirements.txt`
`python pyeamu.py`
Edit services url and enable url_slash

8
config.py Normal file
View File

@ -0,0 +1,8 @@
ip = '127.0.0.1'
port = 8000
services_prefix = "/core"
verbose_log = True
arcade = "Monkey Business"
paseli = 5730
maintenance_mode = False

145
core_common.py Normal file
View File

@ -0,0 +1,145 @@
import config
import random
import time
from lxml.builder import ElementMaker
from kbinxml import KBinXML
from utils.arc4 import EamuseARC4
from utils.lz77 import EamuseLZ77
def _add_val_as_str(elm, val):
new_val = str(val)
if elm is not None:
elm.text = new_val
else:
return new_val
def _add_bool_as_str(elm, val):
return _add_val_as_str(elm, 1 if val else 0)
def _add_list_as_str(elm, vals):
new_val = " ".join([str(val) for val in vals])
if elm is not None:
elm.text = new_val
elm.attrib['__count'] = str(len(vals))
else:
return new_val
E = ElementMaker(
typemap={
int: _add_val_as_str,
bool: _add_bool_as_str,
list: _add_list_as_str,
float: _add_val_as_str,
}
)
async def core_get_game_version_from_software_version(software_version):
_, model, dest, spec, rev, ext = software_version
ext = int(ext)
if model == 'LDJ' and ext >= 2021101300:
return 29
elif model == 'MDX' and ext >= 2019022600:
return 19
elif model == 'KFC' and ext >= 2020090402:
return 6
else:
return 0
async def core_process_request(request):
cl = request.headers.get('Content-Length')
data = await request.body()
if not cl or not data:
return {}
if 'X-Compress' in request.headers:
request.compress = request.headers.get('X-Compress')
else:
request.compress = None
if 'X-Eamuse-Info' in request.headers:
xeamuseinfo = request.headers.get('X-Eamuse-Info')
key = bytes.fromhex(xeamuseinfo[2:].replace("-", ""))
xml_dec = EamuseARC4(key).decrypt(data[:int(cl)])
request.is_encrypted = True
else:
xml_dec = data[:int(cl)]
request.is_encrypted = False
if request.compress == "lz77":
xml_dec = EamuseLZ77.decode(xml_dec)
xml = KBinXML(xml_dec)
root = xml.xml_doc
xml_text = xml.to_text()
request.is_binxml = KBinXML.is_binary_xml(xml_dec)
if config.verbose_log:
print("Request:")
print(xml_text)
model_parts = (root.attrib['model'], *root.attrib['model'].split(':'))
module = root[0].tag
method = root[0].attrib['method'] if 'method' in root[0].attrib else None
command = root[0].attrib['command'] if 'command' in root[0].attrib else None
game_version = await core_get_game_version_from_software_version(model_parts)
return {
'root': root,
'text': xml_text,
'module': module,
'method': method,
'command': command,
'model': model_parts[1],
'dest': model_parts[2],
'spec': model_parts[3],
'rev': model_parts[4],
'ext': model_parts[5],
'game_version': game_version,
}
async def core_prepare_response(request, xml):
binxml = KBinXML(xml)
if request.is_binxml:
xml_binary = binxml.to_binary()
else:
xml_binary = binxml.to_text().encode("utf-8") # TODO: Proper encoding
if config.verbose_log:
print("Response:")
print(binxml.to_text())
response_headers = {"User-Agent": "EAMUSE.Httpac/1.0"}
if request.is_encrypted:
xeamuseinfo = "1-%08x-%04x" % (int(time.time()), random.randint(0x0000, 0xffff))
response_headers["X-Eamuse-Info"] = xeamuseinfo
key = bytes.fromhex(xeamuseinfo[2:].replace("-", ""))
response = EamuseARC4(key).encrypt(xml_binary)
else:
response = bytes(xml_binary)
request.compress = None
# if request.compress == "lz77":
# response_headers["X-Compress"] = request.compress
# response = EamuseLZ77.encode(response)
return response, response_headers

7
core_database.py Normal file
View File

@ -0,0 +1,7 @@
from tinydb import TinyDB
db = TinyDB('db.json', indent=4)
def get_db():
return db

17
modules/__init__.py Normal file
View File

@ -0,0 +1,17 @@
from importlib import util
from os import path
from glob import glob
routers = []
for module_path in [f for f in glob(path.join(path.dirname(__file__), '**/*.py'), recursive=True)
if path.basename(f) != "__init__.py"]:
spec = util.spec_from_file_location('', module_path)
module = util.module_from_spec(spec)
spec.loader.exec_module(module)
router = getattr(module, 'router', None)
if router is not None:
routers.append(router)

0
modules/core/__init__.py Normal file
View File

135
modules/core/cardmng.py Normal file
View File

@ -0,0 +1,135 @@
from fastapi import APIRouter, Request, Response
from tinydb import Query, where
from core_common import core_process_request, core_prepare_response, E
from core_database import get_db
router = APIRouter(prefix="/core", tags=["cardmng"])
def get_target_table(game_id):
target_table = {
"LDJ": "iidx_profile",
"MDX": "ddr_profile",
"KFC": "sdvx_profile",
}
return target_table[game_id]
def get_profile(game_id, cid):
target_table = get_target_table(game_id)
profile = get_db().table(target_table).get(where('card') == cid)
if profile is None:
profile = {
'card': cid,
'version': {},
}
return profile
def get_game_profile(game_id, game_version, cid):
profile = get_profile(game_id, cid)
if str(game_version) not in profile['version']:
profile['version'][str(game_version)] = {}
return profile['version'][str(game_version)]
def create_profile(game_id, game_version, cid, pin):
target_table = get_target_table(game_id)
profile = get_profile(game_id, cid)
profile['pin'] = pin
get_db().table(target_table).upsert(profile, where('card') == cid)
@router.post('/{gameinfo}/cardmng/authpass')
async def cardmng_authpass(request: Request):
request_info = await core_process_request(request)
cid = request_info['root'][0].attrib['refid']
passwd = request_info['root'][0].attrib['pass']
profile = get_profile(request_info['model'], cid)
if profile is None or passwd != profile.get('pin', None):
status = 116
else:
status = 0
response = E.response(
E.authpass(status=status)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/cardmng/bindmodel')
async def cardmng_bindmodel(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.bindmodel(
dataid=1,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/cardmng/getrefid')
async def cardmng_getrefid(request: Request):
request_info = await core_process_request(request)
cid = request_info['root'][0].attrib['cardid']
passwd = request_info['root'][0].attrib['passwd']
create_profile(request_info['model'], request_info['game_version'], cid, passwd)
response = E.response(
E.getrefid(
dataid=cid,
refid=cid,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/cardmng/inquire')
async def cardmng_inquire(request: Request):
request_info = await core_process_request(request)
cid = request_info['root'][0].attrib['cardid']
profile = get_game_profile(request_info['model'], request_info['game_version'], cid)
if profile:
binded = 1
newflag = 0
status = 0
else:
binded = 0
newflag = 1
status = 112
response = E.response(
E.inquire(
dataid=cid,
ecflag=1,
expired=0,
binded=binded,
newflag=newflag,
refid=cid,
status=status,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

58
modules/core/eacoin.py Normal file
View File

@ -0,0 +1,58 @@
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/core", tags=["eacoin"])
@router.post('/{gameinfo}/eacoin/checkin')
async def eacoin_checkin(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.eacoin(
E.sequence(1, __type="s16"),
E.acstatus(1, __type="u8"),
E.acid(1, __type="str"),
E.acname(config.arcade, __type="str"),
E.balance(config.paseli, __type="s32"),
E.sessid(1, __type="str"),
E.inshopcharge(1, __type="u8"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/eacoin/consume')
async def eacoin_consume(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.eacoin(
E.acstatus(0, __type="u8"),
E.autocharge(0, __type="u8"),
E.balance(config.paseli, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/eacoin/getbalance')
async def eacoin_getbalance(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.eacoin(
E.acstatus(0, __type="u8"),
E.balance(config.paseli, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

71
modules/core/facility.py Normal file
View File

@ -0,0 +1,71 @@
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/core", tags=["facility"])
@router.post('/{gameinfo}/facility/get')
async def facility_get(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.facility(
E.location(
E('id', 'EA000001', __type="str"),
E.country('JP', __type="str"),
E.region('JP-13', __type="str"),
E.customercode('X000000001', __type="str"),
E.companycode('X000000001', __type="str"),
E.latitude(0, __type="s32"),
E.longitude(0, __type="s32"),
E.accuracy(0, __type="u8"),
E.countryname('Japan', __type="str"),
E.regionname('Tokyo', __type="str"),
E.countryjname('日本国', __type="str"),
E.regionjname('東京都', __type="str"),
E.name(config.arcade, __type="str"),
E('type', 255, __type="u8"),
),
E.line(
E('class', 8, __type="u8"),
E.rtt(500, __type="u16"),
E.upclass(8, __type="u8"),
E('id', 3, __type="str"),
),
E.portfw(
E.globalip(config.ip, __type="ip4"),
E.globalport(5704, __type="u16"),
E.privateport(5705, __type="u16"),
),
E.public(
E.flag(0, __type="u8"),
E.name(config.arcade, __type="str"),
E.latitude(0, __type="str"),
E.longitude(0, __type="str"),
),
E.share(
E.eacoin(
E.notchamount(3000, __type="s32"),
E.notchcount(3, __type="s32"),
E.supplylimit(9999, __type="s32"),
),
E.eapass(
E.valid(365, __type="u16"),
),
E.url(
E.eapass('www.ea-pass.konami.net', __type="str"),
E.arcadefan('www.konami.jp/am', __type="str"),
E.konaminetdx('http://am.573.jp', __type="str"),
E.konamiid('https://id.konami.net', __type="str"),
E.eagate('http://eagate.573.jp', __type="str"),
),
),
expire=10800,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

26
modules/core/message.py Normal file
View File

@ -0,0 +1,26 @@
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/core", tags=["message"])
@router.post('/{gameinfo}/message/get')
async def message_get(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.message(
expire=300,
*[E.item(
name=s,
start=0,
end=604800,
) for s in ('sys.mainte', 'sys.eacoin.mainte') if config.maintenance_mode]
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

20
modules/core/package.py Normal file
View File

@ -0,0 +1,20 @@
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/core", tags=["package"])
@router.post('/{gameinfo}/package/list')
async def package_list(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.package(
expire=600,
status=0,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

19
modules/core/pcbevent.py Normal file
View File

@ -0,0 +1,19 @@
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/core", tags=["pcbevent"])
@router.post('/{gameinfo}/pcbevent/put')
async def pcbevent_put(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.pcbevent(
expire=600,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

View File

@ -0,0 +1,24 @@
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/core", tags=["pcbtracker"])
@router.post('/{gameinfo}/pcbtracker/alive')
async def pcbtracker_alive(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.pcbtracker(
expire=1200,
ecenable=not config.maintenance_mode,
eclimit=0,
limit=0,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

0
modules/ddr/__init__.py Normal file
View File

25
modules/ddr/eventlog.py Normal file
View File

@ -0,0 +1,25 @@
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 = ["MDX"]
@router.post('/{gameinfo}/eventlog/write')
async def eventlog_write(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.eventlog(
E.gamesession(9999999, __type="s64"),
E.logsendflg(1 if config.maintenance_mode else 0, __type="s32"),
E.logerrlevel(0, __type="s32"),
E.evtidnosendflg(0, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

25
modules/ddr/eventlog_2.py Normal file
View File

@ -0,0 +1,25 @@
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 = ["MDX"]
@router.post('/{gameinfo}/eventlog_2/write')
async def eventlog_2_write(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.eventlog_2(
E.gamesession(9999999, __type="s64"),
E.logsendflg(1 if config.maintenance_mode else 0, __type="s32"),
E.logerrlevel(0, __type="s32"),
E.evtidnosendflg(0, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

428
modules/ddr/playerdata.py Normal file
View File

@ -0,0 +1,428 @@
import random
import time
from tinydb import Query, where
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
from core_database import get_db
from base64 import b64decode, b64encode
router = APIRouter(prefix="/local2", tags=["local2"])
router.model_whitelist = ["MDX"]
def get_profile(cid):
return get_db().table('ddr_profile').get(
where('card') == cid
)
def get_game_profile(cid, game_version):
profile = get_profile(cid)
return profile['version'].get(str(game_version), None)
@router.post('/{gameinfo}/playerdata/usergamedata_advanced')
async def usergamedata_advanced(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
response = None
mode = request_info['root'][0].find('data/mode').text
refid = request_info['root'][0].find('data/refid').text
db = get_db()
all_profiles_for_card = db.table('ddr_profile').get(Query().card == refid)
if mode == 'usernew':
shoparea = request_info['root'][0].find('data/shoparea').text
if 'ddr_id' not in all_profiles_for_card:
ddr_id = random.randint(10000000, 99999999)
all_profiles_for_card['ddr_id'] = ddr_id
all_profiles_for_card['version'][str(game_version)] = {
'game_version': game_version,
'calories_disp': "Off",
'character': "All Character Random",
'arrow_skin': "Normal",
'filter': "Darkest",
'guideline': "Center",
'priority': "Judgment",
'timing_disp': "On",
}
db.table('ddr_profile').upsert(all_profiles_for_card, where('card') == refid)
response = E.response(
E.playerdata(
E.result(0, __type="s32"),
E.seq('-'.join([str(ddr_id)[:4], str(ddr_id)[4:]]), __type="str"),
E.code(ddr_id, __type="s32"),
E.shoparea(shoparea, __type="str")
)
)
if mode == 'userload':
all_scores = {}
for record in db.table('ddr_scores_best').search(where('game_version') == game_version):
mcode = str(record['mcode'])
if mcode not in all_scores.keys():
scores = []
for difficulty in range(10):
s = db.table('ddr_scores_best').get(
(where('mcode') == int(mcode))
& (where('difficulty') == difficulty)
)
if s == None:
scores.append([0, 0, 0, 0, 0])
else:
scores.append([1, s['rank'], s['lamp'], s['score'], s['ghostid']])
all_scores[mcode] = scores
response = E.response(
E.playerdata(
E.result(0, __type="s32"),
E.is_new(1 if all_profiles_for_card is None else 0, __type="bool"),
E.is_refid_locked(0, __type="bool"),
E.eventdata_count_all(1, __type="s16"),
*[E.music(
E.mcode(int(mcode), __type="u32"),
*[E.note(
E.count(s[0], __type="u16"),
E.rank(s[1], __type="u8"),
E.clearkind(s[2], __type="u8"),
E.score(s[3], __type="s32"),
E.ghostid(s[4], __type="s32"),
) for s in [score for score in all_scores.get(mcode)]],
) for mcode in all_scores.keys()],
*[E.eventdata(
E.eventid(event, __type="u32"),
E.eventtype(9999, __type="s32"),
E.eventno(0, __type="u32"),
E.condition(0, __type="s64"),
E.reward(0, __type="u32"),
E.comptime(1, __type="s32"),
E.savedata(0, __type="s64"),
) for event in [e for e in range(1, 100) if e not in [2, 4, 6, 7, 8, 14]]],
E.grade(
E.single_grade(0, __type="u32"),
E.double_grade(0, __type="u32"),
),
E.golden_league(
E.league_class(0, __type="s32"),
E.current(
E.id(0, __type="s32"),
E.league_name_base64("", __type="str"),
E.start_time(0, __type="u64"),
E.end_time(0, __type="u64"),
E.summary_time(0, __type="u64"),
E.league_status(0, __type="s32"),
E.league_class(0, __type="s32"),
E.league_class_result(0, __type="s32"),
E.ranking_number(0, __type="s32"),
E.total_exscore(0, __type="s32"),
E.total_play_count(0, __type="s32"),
E.join_number(0, __type="s32"),
E.promotion_ranking_number(0, __type="s32"),
E.demotion_ranking_number(0, __type="s32"),
E.promotion_exscore(0, __type="s32"),
E.demotion_exscore(0, __type="s32"),
),
),
E.championship(
E.championship_id(0, __type="s32"),
E.name_base64("", __type="str"),
E.lang(
E.destinationcodes("", __type="str"),
E.name_base64("", __type="str"),
),
E.music(
E.mcode(0, __type="u32"),
E.notetype(0, __type="s8"),
E.playstyle(0, __type="s32"),
)
),
E.preplayable(),
)
)
if mode == 'usersave':
timestamp = time.time()
data = request_info['root'][0].find('data')
if not int(data.find('isgameover').text) == 1:
ddr_id = int(data.find('ddrcode').text)
playstyle = int(data.find('playstyle').text)
note = data.findall('note')
for n in note:
if int(n.find('stagenum').text) != 0:
mcode = int(n.find('mcode').text)
difficulty = int(n.find('notetype').text)
rank = int(n.find('rank').text)
lamp = int(n.find('clearkind').text)
score = int(n.find('score').text)
exscore = int(n.find('exscore').text)
maxcombo = int(n.find('maxcombo').text)
life = int(n.find('life').text)
fastcount = int(n.find('fastcount').text)
slowcount = int(n.find('slowcount').text)
judge_marvelous = int(n.find('judge_marvelous').text)
judge_perfect = int(n.find('judge_perfect').text)
judge_great = int(n.find('judge_great').text)
judge_good = int(n.find('judge_good').text)
judge_boo = int(n.find('judge_boo').text)
judge_miss = int(n.find('judge_miss').text)
judge_ok = int(n.find('judge_ok').text)
judge_ng = int(n.find('judge_ng').text)
calorie = int(n.find('calorie').text)
ghostsize = int(n.find('ghostsize').text)
ghost = n.find('ghost').text
opt_speed = int(n.find('opt_speed').text)
opt_boost = int(n.find('opt_boost').text)
opt_appearance = int(n.find('opt_appearance').text)
opt_turn = int(n.find('opt_turn').text)
opt_dark = int(n.find('opt_dark').text)
opt_scroll = int(n.find('opt_scroll').text)
opt_arrowcolor = int(n.find('opt_arrowcolor').text)
opt_cut = int(n.find('opt_cut').text)
opt_freeze = int(n.find('opt_freeze').text)
opt_jump = int(n.find('opt_jump').text)
opt_arrowshape = int(n.find('opt_arrowshape').text)
opt_filter = int(n.find('opt_filter').text)
opt_guideline = int(n.find('opt_guideline').text)
opt_gauge = int(n.find('opt_gauge').text)
opt_judgepriority = int(n.find('opt_judgepriority').text)
opt_timing = int(n.find('opt_timing').text)
db.table('ddr_scores').insert(
{
'timestamp': timestamp,
'game_version': game_version,
'ddr_id': ddr_id,
'playstyle': playstyle,
'mcode': mcode,
'difficulty': difficulty,
'rank': rank,
'lamp': lamp,
'score': score,
'exscore': exscore,
'maxcombo': maxcombo,
'life': life,
'fastcount': fastcount,
'slowcount': slowcount,
'judge_marvelous': judge_marvelous,
'judge_perfect': judge_perfect,
'judge_great': judge_great,
'judge_good': judge_good,
'judge_boo': judge_boo,
'judge_miss': judge_miss,
'judge_ok': judge_ok,
'judge_ng': judge_ng,
'calorie': calorie,
'ghostsize': ghostsize,
'ghost': ghost,
'opt_speed': opt_speed,
'opt_boost': opt_boost,
'opt_appearance': opt_appearance,
'opt_turn': opt_turn,
'opt_dark': opt_dark,
'opt_scroll': opt_scroll,
'opt_arrowcolor': opt_arrowcolor,
'opt_cut': opt_cut,
'opt_freeze': opt_freeze,
'opt_jump': opt_jump,
'opt_arrowshape': opt_arrowshape,
'opt_filter': opt_filter,
'opt_guideline': opt_guideline,
'opt_gauge': opt_gauge,
'opt_judgepriority': opt_judgepriority,
'opt_timing': opt_timing,
},
)
best = db.table('ddr_scores_best').get(
(where('ddr_id') == ddr_id)
& (where('game_version') == game_version)
& (where('mcode') == mcode)
& (where('difficulty') == difficulty)
)
best = {} if best is None else best
best_score_data = {
'game_version': game_version,
'ddr_id': ddr_id,
'playstyle': playstyle,
'mcode': mcode,
'difficulty': difficulty,
'rank': min(rank, best.get('rank', rank)),
'lamp': max(lamp, best.get('lamp', lamp)),
'score': max(score, best.get('score', score)),
'exscore': max(exscore, best.get('exscore', exscore)),
}
ghostid = db.table('ddr_scores').get(
(where('ddr_id') == ddr_id)
& (where('game_version') == game_version)
& (where('mcode') == mcode)
& (where('difficulty') == difficulty)
& (where('score') == max(score, best.get('score', score)))
)
best_score_data['ghostid'] = ghostid.doc_id
db.table('ddr_scores_best').upsert(
best_score_data,
(where('ddr_id') == ddr_id)
& (where('game_version') == game_version)
& (where('mcode') == mcode)
& (where('difficulty') == difficulty)
)
response = E.response(
E.playerdata(
E.result(0, __type="s32"),
)
)
if mode == 'inheritance':
response = E.response(
E.playerdata(
E.result(0, __type="s32"),
E.InheritanceStatus(1, __type="s32"),
)
)
if mode == 'rivalload':
response = E.response(
E.playerdata(
E.result(0, __type="s32"),
)
)
if mode == 'ghostload':
ghostid = int(request_info['root'][0].find('data/ghostid').text)
record = db.table('ddr_scores').get(doc_id=ghostid)
response = E.response(
E.playerdata(
E.result(0, __type="s32"),
E.ghostdata(
E.code(record['ddr_id'], __type="s32"),
E.mcode(record['mcode'], __type="u32"),
E.notetype(record['difficulty'], __type="u8"),
E.ghostsize(record['ghostsize'], __type="s32"),
E.ghost(record['ghost'], __type="string"),
)
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/playerdata/usergamedata_recv')
async def usergamedata_recv(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
data = request_info['root'][0].find('data')
cid = data.find('refid').text
profile = get_game_profile(cid, game_version)
db = get_db().table('ddr_profile')
all_profiles_for_card = db.get(Query().card == cid)
if all_profiles_for_card is None:
load = [
b64encode(str.encode('1,d,1111111,1,0,0,0,0,0,ffffffffffffffff,0,0,0,0,0,0,0,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,,1010-1010,,,,,,').decode()),
b64encode(str.encode('0,3,0,0,0,0,0,3,0,0,0,0,1,2,0,0,0,10.000000,10.000000,10.000000,10.000000,0.000000,0.000000,0.000000,0.000000,,,,,,,,').decode()),
b64encode(str.encode('1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,,,,,,,,').decode()),
b64encode(str.encode('0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,,,,,,,,').decode()),
]
else:
calories_disp = ["Off", "On"]
character = ["All Character Random", "Man Random", "Female Random", "Yuni", "Rage", "Afro", "Jenny", "Emi", "Baby-Lon", "Gus", "Ruby", "Alice", "Julio", "Bonnie", "Zero", "Rinon"]
arrow_skin = ["Normal", "X", "Classic", "Cyber", "Medium", "Small", "Dot"]
screen_filter = ["Off", "Dark", "Darker", "Darkest"]
guideline = ["Off", "Border", "Center"]
priority = ["Judgment", "Arrow"]
timing_disp = ["Off", "On"]
common = profile['common'].split(',')
common[5] = calories_disp.index(profile['calories_disp'])
common[6] = character.index(profile['character'])
common_load = ",".join([str(i) for i in common])
option = profile['option'].split(',')
option[13] = arrow_skin.index(profile['arrow_skin'])
option[14] = screen_filter.index(profile['filter'])
option[15] = guideline.index(profile['guideline'])
option[17] = priority.index(profile['priority'])
option[18] = timing_disp.index(profile['timing_disp'])
option_load = ",".join([str(i) for i in option])
load = [
b64encode(str.encode(common_load.split('ffffffff,COMMON,')[1])).decode(),
b64encode(str.encode(option_load.split('ffffffff,OPTION,')[1])).decode(),
b64encode(str.encode(profile['last'].split('ffffffff,LAST,')[1])).decode(),
b64encode(str.encode(profile['rival'].split('ffffffff,RIVAL,')[1])).decode()
]
response = E.response(
E.playerdata(
E.result(0, __type="s32"),
E.player(
E.record(
*[E.d(p, __type="str")for p in load],
),
E.record_num(4, __type="u32"),
),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/playerdata/usergamedata_send')
async def usergamedata_send(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
data = request_info['root'][0].find('data')
cid = data.find('refid').text
num = int(data.find('datanum').text)
profile = get_profile(cid)
game_profile = profile['version'].get(str(game_version), {})
if num == 1:
game_profile['common'] = b64decode(data.find('record')[0].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
elif num == 4:
game_profile['common'] = b64decode(data.find('record')[0].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
game_profile['option'] = b64decode(data.find('record')[1].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
game_profile['last'] = b64decode(data.find('record')[2].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
game_profile['rival'] = b64decode(data.find('record')[3].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
profile['version'][str(game_version)] = game_profile
get_db().table('ddr_profile').upsert(profile, where('card') == cid)
response = E.response(
E.playerdata(
E.result(0, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

428
modules/ddr/playerdata_2.py Normal file
View File

@ -0,0 +1,428 @@
import random
import time
from tinydb import Query, where
import config
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
from core_database import get_db
from base64 import b64decode, b64encode
router = APIRouter(prefix="/local2", tags=["local2"])
router.model_whitelist = ["MDX"]
def get_profile(cid):
return get_db().table('ddr_profile').get(
where('card') == cid
)
def get_game_profile(cid, game_version):
profile = get_profile(cid)
return profile['version'].get(str(game_version), None)
@router.post('/{gameinfo}/playerdata_2/usergamedata_advanced')
async def usergamedata_advanced(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
response = None
mode = request_info['root'][0].find('data/mode').text
refid = request_info['root'][0].find('data/refid').text
db = get_db()
all_profiles_for_card = db.table('ddr_profile').get(Query().card == refid)
if mode == 'usernew':
shoparea = request_info['root'][0].find('data/shoparea').text
if 'ddr_id' not in all_profiles_for_card:
ddr_id = random.randint(10000000, 99999999)
all_profiles_for_card['ddr_id'] = ddr_id
all_profiles_for_card['version'][str(game_version)] = {
'game_version': game_version,
'calories_disp': "Off",
'character': "All Character Random",
'arrow_skin': "Normal",
'filter': "Darkest",
'guideline': "Center",
'priority': "Judgment",
'timing_disp': "On",
}
db.table('ddr_profile').upsert(all_profiles_for_card, where('card') == refid)
response = E.response(
E.playerdata_2(
E.result(0, __type="s32"),
E.seq('-'.join([str(ddr_id)[:4], str(ddr_id)[4:]]), __type="str"),
E.code(ddr_id, __type="s32"),
E.shoparea(shoparea, __type="str")
)
)
if mode == 'userload':
all_scores = {}
for record in db.table('ddr_scores_best').search(where('game_version') == game_version):
mcode = str(record['mcode'])
if mcode not in all_scores.keys():
scores = []
for difficulty in range(10):
s = db.table('ddr_scores_best').get(
(where('mcode') == int(mcode))
& (where('difficulty') == difficulty)
)
if s == None:
scores.append([0, 0, 0, 0, 0])
else:
scores.append([1, s['rank'], s['lamp'], s['score'], s['ghostid']])
all_scores[mcode] = scores
response = E.response(
E.playerdata_2(
E.result(0, __type="s32"),
E.is_new(1 if all_profiles_for_card is None else 0, __type="bool"),
E.is_refid_locked(0, __type="bool"),
E.eventdata_count_all(1, __type="s16"),
*[E.music(
E.mcode(int(mcode), __type="u32"),
*[E.note(
E.count(s[0], __type="u16"),
E.rank(s[1], __type="u8"),
E.clearkind(s[2], __type="u8"),
E.score(s[3], __type="s32"),
E.ghostid(s[4], __type="s32"),
) for s in [score for score in all_scores.get(mcode)]],
) for mcode in all_scores.keys()],
*[E.eventdata(
E.eventid(event, __type="u32"),
E.eventtype(9999, __type="s32"),
E.eventno(0, __type="u32"),
E.condition(0, __type="s64"),
E.reward(0, __type="u32"),
E.comptime(1, __type="s32"),
E.savedata(0, __type="s64"),
) for event in [e for e in range(1, 100) if e not in [2, 4, 6, 7, 8, 14]]],
E.grade(
E.single_grade(0, __type="u32"),
E.double_grade(0, __type="u32"),
),
E.golden_league(
E.league_class(0, __type="s32"),
E.current(
E.id(0, __type="s32"),
E.league_name_base64("", __type="str"),
E.start_time(0, __type="u64"),
E.end_time(0, __type="u64"),
E.summary_time(0, __type="u64"),
E.league_status(0, __type="s32"),
E.league_class(0, __type="s32"),
E.league_class_result(0, __type="s32"),
E.ranking_number(0, __type="s32"),
E.total_exscore(0, __type="s32"),
E.total_play_count(0, __type="s32"),
E.join_number(0, __type="s32"),
E.promotion_ranking_number(0, __type="s32"),
E.demotion_ranking_number(0, __type="s32"),
E.promotion_exscore(0, __type="s32"),
E.demotion_exscore(0, __type="s32"),
),
),
E.championship(
E.championship_id(0, __type="s32"),
E.name_base64("", __type="str"),
E.lang(
E.destinationcodes("", __type="str"),
E.name_base64("", __type="str"),
),
E.music(
E.mcode(0, __type="u32"),
E.notetype(0, __type="s8"),
E.playstyle(0, __type="s32"),
)
),
E.preplayable(),
)
)
if mode == 'usersave':
timestamp = time.time()
data = request_info['root'][0].find('data')
if not int(data.find('isgameover').text) == 1:
ddr_id = int(data.find('ddrcode').text)
playstyle = int(data.find('playstyle').text)
note = data.findall('note')
for n in note:
if int(n.find('stagenum').text) != 0:
mcode = int(n.find('mcode').text)
difficulty = int(n.find('notetype').text)
rank = int(n.find('rank').text)
lamp = int(n.find('clearkind').text)
score = int(n.find('score').text)
exscore = int(n.find('exscore').text)
maxcombo = int(n.find('maxcombo').text)
life = int(n.find('life').text)
fastcount = int(n.find('fastcount').text)
slowcount = int(n.find('slowcount').text)
judge_marvelous = int(n.find('judge_marvelous').text)
judge_perfect = int(n.find('judge_perfect').text)
judge_great = int(n.find('judge_great').text)
judge_good = int(n.find('judge_good').text)
judge_boo = int(n.find('judge_boo').text)
judge_miss = int(n.find('judge_miss').text)
judge_ok = int(n.find('judge_ok').text)
judge_ng = int(n.find('judge_ng').text)
calorie = int(n.find('calorie').text)
ghostsize = int(n.find('ghostsize').text)
ghost = n.find('ghost').text
opt_speed = int(n.find('opt_speed').text)
opt_boost = int(n.find('opt_boost').text)
opt_appearance = int(n.find('opt_appearance').text)
opt_turn = int(n.find('opt_turn').text)
opt_dark = int(n.find('opt_dark').text)
opt_scroll = int(n.find('opt_scroll').text)
opt_arrowcolor = int(n.find('opt_arrowcolor').text)
opt_cut = int(n.find('opt_cut').text)
opt_freeze = int(n.find('opt_freeze').text)
opt_jump = int(n.find('opt_jump').text)
opt_arrowshape = int(n.find('opt_arrowshape').text)
opt_filter = int(n.find('opt_filter').text)
opt_guideline = int(n.find('opt_guideline').text)
opt_gauge = int(n.find('opt_gauge').text)
opt_judgepriority = int(n.find('opt_judgepriority').text)
opt_timing = int(n.find('opt_timing').text)
db.table('ddr_scores').insert(
{
'timestamp': timestamp,
'game_version': game_version,
'ddr_id': ddr_id,
'playstyle': playstyle,
'mcode': mcode,
'difficulty': difficulty,
'rank': rank,
'lamp': lamp,
'score': score,
'exscore': exscore,
'maxcombo': maxcombo,
'life': life,
'fastcount': fastcount,
'slowcount': slowcount,
'judge_marvelous': judge_marvelous,
'judge_perfect': judge_perfect,
'judge_great': judge_great,
'judge_good': judge_good,
'judge_boo': judge_boo,
'judge_miss': judge_miss,
'judge_ok': judge_ok,
'judge_ng': judge_ng,
'calorie': calorie,
'ghostsize': ghostsize,
'ghost': ghost,
'opt_speed': opt_speed,
'opt_boost': opt_boost,
'opt_appearance': opt_appearance,
'opt_turn': opt_turn,
'opt_dark': opt_dark,
'opt_scroll': opt_scroll,
'opt_arrowcolor': opt_arrowcolor,
'opt_cut': opt_cut,
'opt_freeze': opt_freeze,
'opt_jump': opt_jump,
'opt_arrowshape': opt_arrowshape,
'opt_filter': opt_filter,
'opt_guideline': opt_guideline,
'opt_gauge': opt_gauge,
'opt_judgepriority': opt_judgepriority,
'opt_timing': opt_timing,
},
)
best = db.table('ddr_scores_best').get(
(where('ddr_id') == ddr_id)
& (where('game_version') == game_version)
& (where('mcode') == mcode)
& (where('difficulty') == difficulty)
)
best = {} if best is None else best
best_score_data = {
'game_version': game_version,
'ddr_id': ddr_id,
'playstyle': playstyle,
'mcode': mcode,
'difficulty': difficulty,
'rank': min(rank, best.get('rank', rank)),
'lamp': max(lamp, best.get('lamp', lamp)),
'score': max(score, best.get('score', score)),
'exscore': max(exscore, best.get('exscore', exscore)),
}
ghostid = db.table('ddr_scores').get(
(where('ddr_id') == ddr_id)
& (where('game_version') == game_version)
& (where('mcode') == mcode)
& (where('difficulty') == difficulty)
& (where('score') == max(score, best.get('score', score)))
)
best_score_data['ghostid'] = ghostid.doc_id
db.table('ddr_scores_best').upsert(
best_score_data,
(where('ddr_id') == ddr_id)
& (where('game_version') == game_version)
& (where('mcode') == mcode)
& (where('difficulty') == difficulty)
)
response = E.response(
E.playerdata_2(
E.result(0, __type="s32"),
)
)
if mode == 'inheritance':
response = E.response(
E.playerdata_2(
E.result(0, __type="s32"),
E.InheritanceStatus(1, __type="s32"),
)
)
if mode == 'rivalload':
response = E.response(
E.playerdata_2(
E.result(0, __type="s32"),
)
)
if mode == 'ghostload':
ghostid = int(request_info['root'][0].find('data/ghostid').text)
record = db.table('ddr_scores').get(doc_id=ghostid)
response = E.response(
E.playerdata_2(
E.result(0, __type="s32"),
E.ghostdata(
E.code(record['ddr_id'], __type="s32"),
E.mcode(record['mcode'], __type="u32"),
E.notetype(record['difficulty'], __type="u8"),
E.ghostsize(record['ghostsize'], __type="s32"),
E.ghost(record['ghost'], __type="string"),
)
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/playerdata_2/usergamedata_recv')
async def usergamedata_recv(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
data = request_info['root'][0].find('data')
cid = data.find('refid').text
profile = get_game_profile(cid, game_version)
db = get_db().table('ddr_profile')
all_profiles_for_card = db.get(Query().card == cid)
if all_profiles_for_card is None:
load = [
b64encode(str.encode('1,d,1111111,1,0,0,0,0,0,ffffffffffffffff,0,0,0,0,0,0,0,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,,1010-1010,,,,,,').decode()),
b64encode(str.encode('0,3,0,0,0,0,0,3,0,0,0,0,1,2,0,0,0,10.000000,10.000000,10.000000,10.000000,0.000000,0.000000,0.000000,0.000000,,,,,,,,').decode()),
b64encode(str.encode('1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,,,,,,,,').decode()),
b64encode(str.encode('0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,0.000000,,,,,,,,').decode()),
]
else:
calories_disp = ["Off", "On"]
character = ["All Character Random", "Man Random", "Female Random", "Yuni", "Rage", "Afro", "Jenny", "Emi", "Baby-Lon", "Gus", "Ruby", "Alice", "Julio", "Bonnie", "Zero", "Rinon"]
arrow_skin = ["Normal", "X", "Classic", "Cyber", "Medium", "Small", "Dot"]
screen_filter = ["Off", "Dark", "Darker", "Darkest"]
guideline = ["Off", "Border", "Center"]
priority = ["Judgment", "Arrow"]
timing_disp = ["Off", "On"]
common = profile['common'].split(',')
common[5] = calories_disp.index(profile['calories_disp'])
common[6] = character.index(profile['character'])
common_load = ",".join([str(i) for i in common])
option = profile['option'].split(',')
option[13] = arrow_skin.index(profile['arrow_skin'])
option[14] = screen_filter.index(profile['filter'])
option[15] = guideline.index(profile['guideline'])
option[17] = priority.index(profile['priority'])
option[18] = timing_disp.index(profile['timing_disp'])
option_load = ",".join([str(i) for i in option])
load = [
b64encode(str.encode(common_load.split('ffffffff,COMMON,')[1])).decode(),
b64encode(str.encode(option_load.split('ffffffff,OPTION,')[1])).decode(),
b64encode(str.encode(profile['last'].split('ffffffff,LAST,')[1])).decode(),
b64encode(str.encode(profile['rival'].split('ffffffff,RIVAL,')[1])).decode()
]
response = E.response(
E.playerdata_2(
E.result(0, __type="s32"),
E.player(
E.record(
*[E.d(p, __type="str")for p in load],
),
E.record_num(4, __type="u32"),
),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/playerdata_2/usergamedata_send')
async def usergamedata_send(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
data = request_info['root'][0].find('data')
cid = data.find('refid').text
num = int(data.find('datanum').text)
profile = get_profile(cid)
game_profile = profile['version'].get(str(game_version), {})
if num == 1:
game_profile['common'] = b64decode(data.find('record')[0].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
elif num == 4:
game_profile['common'] = b64decode(data.find('record')[0].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
game_profile['option'] = b64decode(data.find('record')[1].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
game_profile['last'] = b64decode(data.find('record')[2].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
game_profile['rival'] = b64decode(data.find('record')[3].text.split('<bin1')[0]).decode(encoding='utf-8', errors='ignore')
profile['version'][str(game_version)] = game_profile
get_db().table('ddr_profile').upsert(profile, where('card') == cid)
response = E.response(
E.playerdata_2(
E.result(0, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

21
modules/ddr/system.py Normal file
View File

@ -0,0 +1,21 @@
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 = ["MDX"]
@router.post('/{gameinfo}/system/convcardnumber')
async def system_convcardnumber(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.system(
E.data(E.card_number('FFFFFFFFFFFFFFFF', __type="str")),
E.result(0, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

21
modules/ddr/system_2.py Normal file
View File

@ -0,0 +1,21 @@
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 = ["MDX"]
@router.post('/{gameinfo}/system_2/convcardnumber')
async def system_2_convcardnumber(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.system_2(
E.data(E.card_number('FFFFFFFFFFFFFFFF', __type="str")),
E.result(0, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

20
modules/ddr/tax.py Normal file
View File

@ -0,0 +1,20 @@
from fastapi import APIRouter, Request, Response
from core_common import core_process_request, core_prepare_response, E
router = APIRouter(prefix="/local", tags=["local"])
router.model_whitelist = ["MDX"]
@router.post('/{gameinfo}/tax/get_phase')
async def tax_get_phase(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.tax(
E.phase(0, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

0
modules/iidx/__init__.py Normal file
View File

View File

@ -0,0 +1,76 @@
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}/IIDX29gameSystem/systemInfo')
async def iidx29gamesystem_systeminfo(request: Request):
request_info = await core_process_request(request)
unlock = () #(28008, 28065, 28073, 28088, 28089, 29027, 29094, 29095)
sp_dp = (0, 1)
response = E.response(
E.IIDX29gameSystem(
E.arena_schedule(
E.phase(2, __type="u8"),
E.start(1605784800, __type="u32"),
E.end(1605871200, __type="u32"),
),
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.music_open(
E.music_id(s, __type="s32"),
E.kind(0, __type="s32"),
) for s in unlock],
*[E.arena_reward(
E.index(unlock.index(s), __type="s32"),
E.cube_num((unlock.index(s) + 1) * 50, __type="s32"),
E.kind(0, __type="s32"),
E.value(s, __type="str"),
) for s in unlock],
*[E.arena_music_difficult(
E.play_style(s, __type="s32"),
E.arena_class(-1, __type="s32"),
E.low_difficult(1, __type="s32"),
E.high_difficult(12, __type="s32"),
E.is_leggendaria(1, __type="bool"),
E.force_music_list_id(0, __type="s32"),
) for s in sp_dp],
*[E.arena_cpu_define(
E.play_style(s, __type="s32"),
E.arena_class(-1, __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 s in sp_dp],
*[E.maching_class_range(
E.play_style(s[0], __type="s32"),
E.matching_class(s[1], __type="s32"),
E.low_arena_class(0, __type="s32"),
E.high_arena_class(19, __type="s32"),
) for s in ((0, 2), (0, 1), (1, 2), (1, 1))],
*[E.arena_force_music(
E.play_style(s, __type="s32"),
E.force_music_list_id(0, __type="s32"),
E.index(0, __type="s32"),
E.music_id(1000, __type="s32"),
E.note_grade(0, __type="s32"),
E.is_active(s, __type="bool"),
) for s in sp_dp],
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

122
modules/iidx/iidx29grade.py Normal file
View File

@ -0,0 +1,122 @@
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}/IIDX29grade/raised')
async def iidx29grade_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.IIDX29grade(
pnum=1
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

View File

@ -0,0 +1,75 @@
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}/IIDX29lobby/entry')
async def iidx29lobby_entry(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29lobby()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29lobby/update')
async def iidx29lobby_update(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29lobby()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29lobby/delete')
async def iidx29lobby_delete(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29lobby()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29lobby/bplbattle_entry')
async def iidx29lobby_bplbattle_entry(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29lobby()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29lobby/bplbattle_update')
async def iidx29lobby_bplbattle_update(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29lobby()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29lobby/bplbattle_delete')
async def iidx29lobby_bplbattle_delete(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29lobby()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

343
modules/iidx/iidx29music.py Normal file
View File

@ -0,0 +1,343 @@
import time
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}/IIDX29music/getrank')
async def iidx29music_getrank(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
iidxid = int(request_info['root'][0].attrib['iidxid'])
play_style = int(request_info['root'][0].attrib['cltype'])
all_scores = {}
db = get_db()
for record in db.table('iidx_scores_best').search(
(where('game_version') == game_version)
& (where('iidx_id') == iidxid)
& (where('play_style') == play_style)
):
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 music_id not in all_scores:
all_scores[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[music_id][chart_id]['clear_flg'] = clear_flg
all_scores[music_id][chart_id]['ex_score'] = ex_score
all_scores[music_id][chart_id]['miss_count'] = miss_count
response = E.response(
E.IIDX29music(
E.style(type=play_style),
*[E.m([
-1,
k,
*[all_scores[k][d]['clear_flg'] for d in range(5)],
*[all_scores[k][d]['ex_score'] for d in range(5)],
*[all_scores[k][d]['miss_count'] for d in range(5)],
], __type='s16') for k in all_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}/IIDX29music/crate')
async def iidx29music_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('game_version') == game_version)
)
crate = {}
fcrate = {}
for stat in all_score_stats:
if stat['music_id'] not in crate:
crate[stat['music_id']] = [[-1] * 5] * 2
if stat['music_id'] not in fcrate:
fcrate[stat['music_id']] = [[-1] * 5] * 2
crate[stat['music_id']][stat['play_style']][stat['chart_id']] = int(stat['clear_rate'])
fcrate[stat['music_id']][stat['play_style']][stat['chart_id']] = int(stat['fc_rate'])
response = E.response(
E.IIDX29music(
*[E.c(crate[k][0] + crate[k][1] + fcrate[k][0] + fcrate[k][1], 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}/IIDX29music/reg')
async def iidx29music_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'])
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('game_version') == game_version)
& (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
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': min(miss_num, best_score.get('miss_count', miss_num)),
'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('game_version') == game_version)
& (where('play_style') == play_style)
& (where('music_id') == music_id)
& (where('chart_id') == note_id)
)
score_stats = db.table('iidx_score_stats').get(
(where('game_version') == game_version)
& (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('game_version') == game_version)
& (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('game_version') == game_version)
& (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.IIDX29music(
E.ranklist(
*ranklist_data,
total_user_num=len(ranklist_data)
),
E.shopdata(
rank=myRank
),
clid=note_id,
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}/IIDX29music/appoint')
async def iidx29music_appoint(request: Request):
request_info = await core_process_request(request)
iidxid = int(request_info['root'][0].attrib['iidxid'])
music_id = int(request_info['root'][0].attrib['mid'])
chart_id = int(request_info['root'][0].attrib['clid'])
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,
))
response = E.response(
E.IIDX29music(
*vals
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

953
modules/iidx/iidx29pc.py Normal file
View File

@ -0,0 +1,953 @@
from tinydb import Query, where
import config
import random
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(cid):
return get_db().table('iidx_profile').get(
where('card') == cid
)
def get_game_profile(cid, game_version):
profile = get_profile(cid)
return profile['version'].get(str(game_version), None)
def get_id_from_profile(cid):
profile = get_db().table('iidx_profile').get(
where('card') == cid
)
djid = "%08d" % profile['iidx_id']
djid_split = '-'.join([djid[:4], djid[4:]])
return profile['iidx_id'], djid_split
def calculate_folder_mask(profile):
return profile.get('_show_category_grade', 0) << 0 \
| (profile.get('_show_category_status', 0) << 1) \
| (profile.get('_show_category_difficulty', 0) << 2) \
| (profile.get('_show_category_alphabet', 0) << 3) \
| (profile.get('_show_category_rival_play', 0) << 4) \
| (profile.get('_show_category_rival_winlose', 0) << 6) \
| (profile.get('_show_rival_shop_info', 0) << 7) \
| (profile.get('_hide_play_count', 0) << 8) \
| (profile.get('_show_score_graph_cutin', 0) << 9) \
| (profile.get('_classic_hispeed', 0) << 10) \
| (profile.get('_hide_iidx_id', 0) << 12)
@router.post('/{gameinfo}/IIDX29pc/get')
async def iidx29pc_get(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
cid = request_info['root'][0].attrib['cid']
profile = get_game_profile(cid, game_version)
djid, djid_split = get_id_from_profile(cid)
response = E.response(
E.IIDX29pc(
E.pcdata(
d_auto_adjust=profile['d_auto_adjust'],
d_auto_scrach=profile['d_auto_scrach'],
d_camera_layout=profile['d_camera_layout'],
d_disp_judge=profile['d_disp_judge'],
d_exscore=profile['d_exscore'],
d_gauge_disp=profile['d_gauge_disp'],
d_ghost_score=profile['d_ghost_score'],
d_gno=profile['d_gno'],
d_graph_score=profile['d_graph_score'],
d_gtype=profile['d_gtype'],
d_hispeed=profile['d_hispeed'],
d_judge=profile['d_judge'],
d_judgeAdj=profile['d_judgeAdj'],
d_lane_brignt=profile['d_lane_brignt'],
d_liflen=profile['d_liflen'],
d_notes=profile['d_notes'],
d_opstyle=profile['d_opstyle'],
d_pace=profile['d_pace'],
d_sdlen=profile['d_sdlen'],
d_sdtype=profile['d_sdtype'],
d_sorttype=profile['d_sorttype'],
d_sub_gno=profile['d_sub_gno'],
d_timing=profile['d_timing'],
d_tsujigiri_disp=profile['d_tsujigiri_disp'],
d_tune=profile['d_tune'],
dach=profile['dach'],
dp_opt=profile['dp_opt'],
dp_opt2=profile['dp_opt2'],
dpnum=profile["dpnum"],
gpos=profile['gpos'],
id=djid,
idstr=djid_split,
mode=profile['mode'],
name=profile['djname'],
ngrade=profile['ngrade'],
pid=profile['region'],
pmode=profile['pmode'],
rtype=profile['rtype'],
s_auto_adjust=profile['s_auto_adjust'],
s_auto_scrach=profile['s_auto_scrach'],
s_camera_layout=profile['s_camera_layout'],
s_disp_judge=profile['s_disp_judge'],
s_exscore=profile['s_exscore'],
s_gauge_disp=profile['s_gauge_disp'],
s_ghost_score=profile['s_ghost_score'],
s_gno=profile['s_gno'],
s_graph_score=profile['s_graph_score'],
s_gtype=profile['s_gtype'],
s_hispeed=profile['s_hispeed'],
s_judge=profile['s_judge'],
s_judgeAdj=profile['s_judgeAdj'],
s_lane_brignt=profile['s_lane_brignt'],
s_liflen=profile['s_liflen'],
s_notes=profile['s_notes'],
s_opstyle=profile['s_opstyle'],
s_pace=profile['s_pace'],
s_sdlen=profile['s_sdlen'],
s_sdtype=profile['s_sdtype'],
s_sorttype=profile['s_sorttype'],
s_sub_gno=profile['s_sub_gno'],
s_timing=profile['s_timing'],
s_tsujigiri_disp=profile['s_tsujigiri_disp'],
s_tune=profile['s_tune'],
sach=profile['sach'],
sp_opt=profile['sp_opt'],
spnum=profile['spnum'],
),
E.qprodata([profile["head"], profile["hair"], profile["face"], profile["hand"], profile["body"]],
__type="u32", __size=5 * 4),
E.skin(
[
profile["frame"],
profile["turntable"],
profile["explosion"],
profile["bgm"],
calculate_folder_mask(profile),
profile["sudden"],
profile["judge_pos"],
profile["categoryvoice"],
profile["note"],
profile["fullcombo"],
profile["keybeam"],
profile["judgestring"],
-1,
profile["soundpreview"],
profile["grapharea"],
profile["effector_lock"],
profile["effector_type"],
profile["explosion_size"],
profile["alternate_hcn"],
profile["kokokara_start"],
],
__type="s16"),
E.rlist(),
E.ir_data(),
E.secret_course_data(),
E.deller(deller=profile['deller'], rate=0),
E.secret(
E.flg1(profile.get('secret_flg1', [-1, -1, -1]), __type="s64"),
E.flg2(profile.get('secret_flg2', [-1, -1, -1]), __type="s64"),
E.flg3(profile.get('secret_flg3', [-1, -1, -1]), __type="s64"),
E.flg4(profile.get('secret_flg4', [-1, -1, -1]), __type="s64"),
),
E.join_shop(join_cflg=1, join_id=10, join_name=config.arcade, joinflg=1),
E.leggendaria(E.flg1(profile.get('leggendaria_flg1', [-1, -1, -1]), __type="s64")),
E.grade(
*[E.g(x, __type="u8") for x in profile['grade_values']],
dgid=profile['grade_double'],
sgid=profile['grade_single'],
),
E.world_tourism_secret_flg(
E.flg1(profile.get('wt_flg1', [-1, -1, -1]), __type="s64"),
E.flg2(profile.get('wt_flg2', [-1, -1, -1]), __type="s64"),
),
E.lightning_setting(
E.slider(profile.get('lightning_setting_slider', [0] * 7), __type="s32"),
E.light(profile.get('lightning_setting_light', [1] * 10), __type="bool"),
E.concentration(profile.get('lightning_setting_concentration', 0), __type="bool"),
headphone_vol=profile.get('lightning_setting_headphone_vol', 0),
resistance_sp_left=profile.get('lightning_setting_resistance_sp_left', 0),
resistance_sp_right=profile.get('lightning_setting_resistance_sp_right', 0),
resistance_dp_left=profile.get('lightning_setting_resistance_dp_left', 0),
resistance_dp_right=profile.get('lightning_setting_resistance_dp_right', 0),
skin_0=profile.get('lightning_setting_skin_0', 0),
flg_skin_0=profile.get('lightning_setting_flg_skin_0', 0),
),
E.arena_data(
E.achieve_data(
arena_class=-1,
counterattack_num=0,
best_top_class_continuing=0,
now_top_class_continuing=0,
play_style=0,
rating_value=90,
),
E.achieve_data(
arena_class=-1,
counterattack_num=0,
best_top_class_continuing=0,
now_top_class_continuing=0,
play_style=1,
rating_value=90,
),
E.cube_data(
cube=200,
season_id=0,
),
play_num=6,
play_num_dp=3,
play_num_sp=3,
prev_best_class_sp=18,
prev_best_class_dp=18,
),
E.follow_data(),
E.classic_course_data(),
E.bind_eaappli(),
E.ea_premium_course(),
E.enable_qr_reward(),
E.nostalgia_open(),
E.event_1(
story_prog=profile.get('event_1_story_prog', 0),
last_select_area=profile.get('event_1_last_select_area', 0),
failed_num=profile.get('event_1_failed_num', 0),
event_play_num=profile.get('event_1_event_play_num', 0),
last_select_area_id=profile.get('event_1_last_select_area_id', 0),
last_select_platform_type=profile.get('event_1_last_select_platform_type', 0),
last_select_platform_id=profile.get('event_1_last_select_platform_id', 0),
),
E.language_setting(language=profile['language_setting']),
E.movie_agreement(agreement_version=profile['movie_agreement']),
E.bpl_virtual(),
E.lightning_play_data(spnum=profile['lightning_play_data_spnum'],
dpnum=profile['lightning_play_data_dpnum']),
E.weekly(
mid=-1,
wid=1,
),
E.packinfo(
music_0=-1,
music_1=-1,
music_2=-1,
pack_id=1,
),
E.kac_entry_info(
E.enable_kac_deller(),
E.disp_kac_mark(),
E.open_kac_common_music(),
E.open_kac_new_a12_music(),
E.is_kac_entry(),
E.is_kac_evnet_entry(),
),
E.orb_data(rest_orb=100, present_orb=100),
E.visitor(anum=1, pnum=2, snum=1, vs_flg=1),
E.tonjyutsu(black_pass=-1, platinum_pass=-1),
E.pay_per_use(item_num=99),
E.old_linkage_secret_flg(
floor_infection4=-1,
bemani_janken=-1,
ichika_rush=-1,
nono_rush=-1,
song_battle=-1,
),
E.floor_infection4(music_list=-1),
E.bemani_vote(music_list=-1),
E.bemani_janken_meeting(music_list=-1),
E.bemani_rush(music_list_ichika=-1, music_list_nono=-1),
E.ultimate_mobile_link(music_list=-1),
E.bemani_musiq_fes(music_list=-1),
E.busou_linkage(music_list=-1),
E.busou_linkage_2(music_list=-1),
E.valkyrie_linkage_data(progress=-1),
E.valkyrie_linkage_2_data(progress=-1),
E.achievements(
E.trophy(
profile.get('achievements_trophy', [])[:20],
__type="s64"
),
pack=profile.get('achievements_pack_id', 0),
pack_comp=profile.get('achievements_pack_comp', 0),
last_weekly=profile.get('achievements_last_weekly', 0),
weekly_num=profile.get('achievements_weekly_num', 0),
visit_flg=profile.get('achievements_visit_flg', 0),
rival_crush=0,
),
E.notes_radar(
E.radar_score(
profile['notes_radar_single'],
__type="s32",
),
style=0,
),
E.notes_radar(
E.radar_score(
profile['notes_radar_double'],
__type="s32",
),
style=1,
),
E.dj_rank(
E.rank(
profile['dj_rank_single_rank'],
__type="s32",
),
E.point(
profile['dj_rank_single_point'],
__type="s32",
),
style=0,
),
E.dj_rank(
E.rank(
profile['dj_rank_double_rank'],
__type="s32",
),
E.point(
profile['dj_rank_double_point'],
__type="s32",
),
style=1,
),
E.step(
E.is_track_ticket(
profile['stepup_is_track_ticket'],
__type="bool",
),
dp_level=profile['stepup_dp_level'],
dp_mplay=profile['stepup_dp_mplay'],
enemy_damage=profile['stepup_enemy_damage'],
enemy_defeat_flg=profile['stepup_enemy_defeat_flg'],
mission_clear_num=profile['stepup_mission_clear_num'],
progress=profile['stepup_progress'],
sp_level=profile['stepup_sp_level'],
sp_mplay=profile['stepup_sp_mplay'],
tips_read_list=profile['stepup_tips_read_list'],
total_point=profile['stepup_total_point'],
),
E.skin_customize_flg(
skin_frame_flg=profile['skin_customize_flag_frame'],
skin_bgm_flg=profile['skin_customize_flag_bgm'],
skin_lane_flg3=profile['skin_customize_flag_lane'],
)
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29pc/common')
async def iidx29pc_common(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29pc(
E.monthly_mranking(
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
__type="u16"),
E.total_mranking(
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
__type="u16"),
# E.internet_ranking(),
# E.secret_ex_course(),
E.kac_mid([-1, -1, -1, -1, -1], __type="s32"),
E.kac_clid([2, 2, 2, 2, 2], __type="s32"),
E.ir(beat=3),
E.cm(compo='cm_ultimate', folder='cm_ultimate', id=0),
E.tdj_cm(
E.cm(filename='cm_bn_001', id=0),
E.cm(filename='cm_bn_002', id=1),
E.cm(filename='event_bn_001', id=2),
E.cm(filename='event_bn_004', id=3),
E.cm(filename='event_bn_006', id=4),
E.cm(filename='fipb_001', id=5),
E.cm(filename='year_bn_004', id=6),
E.cm(filename='year_bn_005', id=7),
E.cm(filename='year_bn_006_2', id=8),
E.cm(filename='year_bn_007', id=9),
),
# E.playvideo_disable_music(E.music(musicid=-1)),
# E.music_movie_suspend(E.music(music_id=-1, kind=0, name='')),
# E.bpl_virtual(),
E.movie_agreement(version=1),
E.license('None', __type="str"),
E.file_recovery(url=str(config.ip)),
E.movie_upload(url=str(config.ip)),
# E.button_release_frame(frame=''),
# E.trigger_logic_type(type=''),
# E.cm_movie_info(type=''),
E.escape_package_info(),
# E.expert(phase=1),
# E.expert_random_secret(phase=1),
E.boss(phase=0), # disable event
E.vip_pass_black(),
E.eisei(open=1),
E.deller_bonus(open=1),
E.newsong_another(open=1),
# E.pcb_check(flg=0)
E.expert_secret_full_open(),
E.eaorder_phase(phase=-1),
E.common_evnet(flg=-1),
E.system_voice_phase(phase=random.randint(1, 10)), # TODO: Figure out range
E.extra_boss_event(phase=6),
E.event1_phase(phase=4),
E.premium_area_news(open=1),
E.premium_area_qpro(open=1),
# E.disable_same_triger(frame=-1),
E.play_video(),
E.world_tourism(open_list=1),
E.bpl_battle(phase=1),
E.display_asio_logo(),
# E.force_rom_check(),
E.lane_gacha(),
# E.fps_fix(),
# E.save_unsync_log(),
expire=600
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29pc/save')
async def iidx29pc_save(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
xid = int(request_info['root'][0].attrib['iidxid'])
cid = request_info['root'][0].attrib['cid']
clt = int(request_info['root'][0].attrib['cltype'])
profile = get_profile(cid)
game_profile = profile['version'].get(str(game_version), {})
for k in [
'd_auto_adjust',
'd_auto_scrach',
'd_camera_layout',
'd_disp_judge',
'd_gauge_disp',
'd_ghost_score',
'd_gno',
'd_graph_score',
'd_gtype',
'd_hispeed',
'd_judge',
'd_judgeAdj',
'd_lane_brignt',
'd_notes',
'd_opstyle',
'd_pace',
'd_sdlen',
'd_sdtype',
'd_sorttype',
'd_sub_gno',
'd_timing',
'd_tsujigiri_disp',
'dp_opt',
'dp_opt2',
'gpos',
'mode',
'ngrade',
'pmode',
'rtype',
's_auto_adjust',
's_auto_scrach',
's_camera_layout',
's_disp_judge',
's_gauge_disp',
's_ghost_score',
's_gno',
's_graph_score',
's_gtype',
's_hispeed',
's_judge',
's_judgeAdj',
's_lane_brignt',
's_notes',
's_opstyle',
's_pace',
's_sdlen',
's_sdtype',
's_sorttype',
's_sub_gno',
's_timing',
's_tsujigiri_disp',
'sp_opt',
]:
if k in request_info['root'][0].attrib:
game_profile[k] = request_info['root'][0].attrib[k]
for k in [
('d_liflen', 'd_lift'),
('dach', 'd_achi'),
('s_liflen', 's_lift'),
('sach', 's_achi'),
]:
if k[1] in request_info['root'][0].attrib:
game_profile[k[0]] = request_info['root'][0].attrib[k[1]]
lightning_setting = request_info['root'][0].find('lightning_setting')
if lightning_setting is not None:
for k in [
'headphone_vol',
'resistance_dp_left',
'resistance_dp_right',
'resistance_sp_left',
'resistance_sp_right',
]:
game_profile['lightning_setting_' + k] = int(lightning_setting.attrib[k])
slider = lightning_setting.find('slider')
if slider is not None:
game_profile['lightning_setting_slider'] = [int(x) for x in slider.text.split(' ')]
light = lightning_setting.find('light')
if light is not None:
game_profile['lightning_setting_light'] = [int(x) for x in light.text.split(' ')]
concentration = lightning_setting.find('concentration')
if concentration is not None:
game_profile['lightning_setting_concentration'] = int(concentration.text)
lightning_customize_flg = request_info['root'][0].find('lightning_customize_flg')
if lightning_customize_flg is not None:
for k in [
'flg_skin_0',
]:
game_profile['lightning_setting_' + k] = int(lightning_customize_flg.attrib[k])
secret = request_info['root'][0].find('secret')
if secret is not None:
for k in ['flg1', 'flg2', 'flg3', 'flg4']:
flg = secret.find(k)
if flg is not None:
game_profile['secret_' + k] = [int(x) for x in flg.text.split(' ')]
leggendaria = request_info['root'][0].find('leggendaria')
if leggendaria is not None:
for k in ['flg1']:
flg = leggendaria.find(k)
if flg is not None:
game_profile['leggendaria_' + k] = [int(x) for x in flg.text.split(' ')]
step = request_info['root'][0].find('step')
if step is not None:
for k in [
'dp_level',
'dp_mplay',
'enemy_damage',
'enemy_defeat_flg',
'mission_clear_num',
'progress',
'sp_level',
'sp_mplay',
'tips_read_list',
'total_point',
]:
game_profile['stepup_' + k] = int(step.attrib[k])
is_track_ticket = step.find('is_track_ticket')
if is_track_ticket is not None:
game_profile['stepup_is_track_ticket'] = int(is_track_ticket.text)
dj_ranks = request_info['root'][0].findall('dj_rank')
dj_ranks = [] if dj_ranks is None else dj_ranks
for dj_rank in dj_ranks:
style = int(dj_rank.attrib['style'])
rank = dj_rank.find('rank')
game_profile['dj_rank_' + ['single', 'double'][style] + '_rank'] = [int(x) for x in rank.text.split(' ')]
point = dj_rank.find('point')
game_profile['dj_rank_' + ['single', 'double'][style] + '_point'] = [int(x) for x in point.text.split(' ')]
notes_radars = request_info['root'][0].findall('notes_radar')
notes_radars = [] if notes_radars is None else notes_radars
for notes_radar in notes_radars:
style = int(notes_radar.attrib['style'])
score = notes_radar.find('radar_score')
game_profile['notes_radar_' + ['single', 'double'][style]] = [int(x) for x in score.text.split(' ')]
achievements = request_info['root'][0].find('achievements')
if achievements is not None:
for k in [
'last_weekly',
'pack_comp',
'pack_flg',
'pack_id',
'play_pack',
'visit_flg',
'weekly_num',
]:
game_profile['achievements_' + k] = int(achievements.attrib[k])
trophy = achievements.find('trophy')
if trophy is not None:
game_profile['achievements_trophy'] = [int(x) for x in trophy.text.split(' ')]
grade = request_info['root'][0].find('grade')
if grade is not None:
grade_values = []
for g in grade.findall('g'):
grade_values.append([int(x) for x in g.text.split(' ')])
profile['grade_single'] = int(grade.attrib['sgid'])
profile['grade_double'] = int(grade.attrib['dgid'])
profile['grade_values'] = grade_values
deller_amount = game_profile.get('deller', 0)
deller = request_info['root'][0].find('deller')
if deller is not None:
deller_amount = int(deller.attrib['deller'])
game_profile['deller'] = deller_amount
language = request_info['root'][0].find('language_setting')
if language is not None:
language_value = int(language.attrib['language'])
game_profile['language_setting'] = language_value
game_profile['spnum'] = game_profile.get('spnum', 0) + (1 if clt == 0 else 0)
game_profile['dpnum'] = game_profile.get('dpnum', 0) + (1 if clt == 1 else 0)
if request_info['model'] == "TDJ":
game_profile['lightning_play_data_spnum'] = game_profile.get('lightning_play_data_spnum', 0) + (1 if clt == 0 else 0)
game_profile['lightning_play_data_dpnum'] = game_profile.get('lightning_play_data_dpnum', 0) + (1 if clt == 1 else 0)
profile['version'][str(game_version)] = game_profile
get_db().table('iidx_profile').upsert(profile, where('card') == cid)
response = E.response(
E.IIDX29pc(
iidxid=xid,
cltype=clt
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29pc/visit')
async def iidx29pc_visit(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29pc(
aflg=1,
anum=1,
pflg=1,
pnum=1,
sflg=1,
snum=1,
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29pc/reg')
async def iidx29pc_reg(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
cid = request_info['root'][0].attrib['cid']
name = request_info['root'][0].attrib['name']
pid = request_info['root'][0].attrib['pid']
db = get_db().table('iidx_profile')
all_profiles_for_card = db.get(Query().card == cid)
if all_profiles_for_card is None:
all_profiles_for_card = {
'card': cid,
'version': {}
}
if 'iidx_id' not in all_profiles_for_card:
iidx_id = random.randint(10000000, 99999999)
all_profiles_for_card['iidx_id'] = iidx_id
all_profiles_for_card['version'][str(game_version)] = {
'game_version': game_version,
'djname': name,
'region': int(pid),
'head': 0,
'hair': 0,
'face': 0,
'hand': 0,
'body': 0,
'frame': 0,
'turntable': 0,
'explosion': 0,
'bgm': 0,
'folder_mask': 0,
'sudden': 0,
'judge_pos': 0,
'categoryvoice': 0,
'note': 0,
'fullcombo': 0,
'keybeam': 0,
'judgestring': 0,
'soundpreview': 0,
'grapharea': 0,
'effector_lock': 0,
'effector_type': 0,
'explosion_size': 0,
'alternate_hcn': 0,
'kokokara_start': 0,
'd_auto_adjust': 0,
'd_auto_scrach': 0,
'd_camera_layout': 0,
'd_disp_judge': 0,
'd_exscore': 0,
'd_gauge_disp': 0,
'd_ghost_score': 0,
'd_gno': 0,
'd_graph_score': 0,
'd_gtype': 0,
'd_hispeed': 0.000000,
'd_judge': 0,
'd_judgeAdj': 0,
'd_lane_brignt': 0,
'd_liflen': 0,
'd_notes': 0.000000,
'd_opstyle': 0,
'd_pace': 0,
'd_sdlen': 0,
'd_sdtype': 0,
'd_sorttype': 0,
'd_sub_gno': 0,
'd_timing': 0,
'd_tsujigiri_disp': 0,
'd_tune': 0,
'dach': 0,
'dp_opt': 0,
'dp_opt2': 0,
'dpnum': 0,
'gpos': 0,
'mode': 0,
'ngrade': 0,
'pmode': 0,
'rtype': 0,
's_auto_adjust': 0,
's_auto_scrach': 0,
's_camera_layout': 0,
's_disp_judge': 0,
's_exscore': 0,
's_gauge_disp': 0,
's_ghost_score': 0,
's_gno': 0,
's_graph_score': 0,
's_gtype': 0,
's_hispeed': 0.000000,
's_judge': 0,
's_judgeAdj': 0,
's_lane_brignt': 0,
's_liflen': 0,
's_notes': 0.000000,
's_opstyle': 0,
's_pace': 0,
's_sdlen': 0,
's_sdtype': 0,
's_sorttype': 0,
's_sub_gno': 0,
's_timing': 0,
's_tsujigiri_disp': 0,
's_tune': 0,
'sach': 0,
'sp_opt': 0,
'spnum': 0,
'deller': 0,
# Step up mode
'stepup_dp_level': 0,
'stepup_dp_mplay': 0,
'stepup_enemy_damage': 0,
'stepup_enemy_defeat_flg': 0,
'stepup_mission_clear_num': 0,
'stepup_progress': 0,
'stepup_sp_level': 0,
'stepup_sp_mplay': 0,
'stepup_tips_read_list': 0,
'stepup_total_point': 0,
'stepup_is_track_ticket': 0,
# DJ Rank
'dj_rank_single_rank': [0] * 15,
'dj_rank_double_rank': [0] * 15,
'dj_rank_single_point': [0] * 15,
'dj_rank_double_point': [0] * 15,
# Notes Radar
'notes_radar_single': [0] * 6,
'notes_radar_double': [0] * 6,
# Grades
'grade_single': -1,
'grade_double': -1,
'grade_values': [],
# Achievements
'achievements_trophy': [0] * 160,
'achievements_last_weekly': 0,
'achievements_pack_comp': 0,
'achievements_pack_flg': 0,
'achievements_pack_id': 0,
'achievements_play_pack': 0,
'achievements_visit_flg': 0,
'achievements_weekly_num': 0,
# Other
'language_setting': 0,
'movie_agreement': 0,
'lightning_play_data_spnum': 0,
'lightning_play_data_dpnum': 0,
# Lightning model settings
'lightning_setting_slider': [0] * 7,
'lightning_setting_light': [1] * 10,
'lightning_setting_concentration': 0,
'lightning_setting_headphone_vol': 0,
'lightning_setting_resistance_sp_left': 0,
'lightning_setting_resistance_sp_right': 0,
'lightning_setting_resistance_dp_left': 0,
'lightning_setting_resistance_dp_right': 0,
'lightning_setting_skin_0': 0,
'lightning_setting_flg_skin_0': 0,
# Event_1 settings
'event_1_story_prog': 0,
'event_1_last_select_area': 0,
'event_1_failed_num': 0,
'event_1_event_play_num': 0,
'event_1_last_select_area_id': 0,
'event_1_last_select_platform_type': 0,
'event_1_last_select_platform_id': 0,
# Web UI/Other options
'_show_category_grade': 0,
'_show_category_status': 1,
'_show_category_difficulty': 1,
'_show_category_alphabet': 1,
'_show_category_rival_play': 0,
'_show_category_rival_winlose': 1,
'_show_category_all_rival_play': 0,
'_show_category_arena_winlose': 1,
'_show_rival_shop_info': 0,
'_hide_play_count': 0,
'_show_score_graph_cutin': 1,
'_hide_iidx_id': 0,
'_classic_hispeed': 0,
'_beginner_option_swap': 1,
'_show_lamps_as_no_play_in_arena': 0,
'skin_customize_flag_frame': 0,
'skin_customize_flag_bgm': 0,
'skin_customize_flag_lane': 0
}
db.upsert(all_profiles_for_card, where('card') == cid)
card, card_split = get_id_from_profile(cid)
response = E.response(
E.IIDX29pc(
id=card,
id_str=card_split
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29pc/getLaneGachaTicket')
async def iidx29pc_getlanegachaticket(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29pc(
E.ticket(
ticket_id=0,
arrange_id=0,
expire_date=0,
),
E.setting(
sp=0,
dp_left=0,
dp_right=0,
),
E.info(
last_page=0,
),
E.free(
num=10,
),
E.favorite(
arrange=0,
),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29pc/drawLaneGacha')
async def iidx29pc_drawlanegacha(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29pc(
E.ticket(
ticket_id=0,
arrange_id=0,
expire_date=0,
),
E.session(
session_id=0
),
status=0
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29pc/eaappliresult')
async def iidx29pc_eaappliresult(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29pc()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29pc/logout')
async def iidx29pc_logout(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29pc()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

View File

@ -0,0 +1,20 @@
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}/IIDX29ranking/getranker')
async def iidx29ranking_getranker(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29ranking()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

View File

@ -0,0 +1,69 @@
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}/IIDX29shop/getname')
async def iidx29shop_getname(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29shop(
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}/IIDX29shop/getconvention')
async def iidx29shop_getconvention(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29shop(
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}/IIDX29shop/sentinfo')
async def iidx29shop_sentinfo(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29shop()
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/IIDX29shop/sendescapepackageinfo')
async def iidx29shop_sendescapepackageinfo(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.IIDX29shop(
expire=1200
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

0
modules/sdvx/__init__.py Normal file
View File

25
modules/sdvx/eventlog.py Normal file
View File

@ -0,0 +1,25 @@
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 = ["KFC"]
@router.post('/{gameinfo}/eventlog/write')
async def eventlog_write(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.eventlog(
E.gamesession(9999999, __type="s64"),
E.logsendflg(1 if config.maintenance_mode else 0, __type="s32"),
E.logerrlevel(0, __type="s32"),
E.evtidnosendflg(0, __type="s32"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

648
modules/sdvx/game.py Normal file
View File

@ -0,0 +1,648 @@
from tinydb import Query, where
import config
import random
import time
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 = ["KFC"]
def get_profile(cid):
return get_db().table('sdvx_profile').get(
where('card') == cid
)
def get_game_profile(cid, game_version):
profile = get_profile(cid)
return profile['version'].get(str(game_version), None)
def get_id_from_profile(cid):
profile = get_db().table('sdvx_profile').get(
where('card') == cid
)
djid = "%08d" % profile['sdvx_id']
djid_split = '-'.join([djid[:4], djid[4:]])
return profile['sdvx_id'], djid_split
@router.post('/{gameinfo}/game/sv6_common')
async def game_sv6_common(request: Request):
request_info = await core_process_request(request)
event = [
'DEMOGAME_PLAY',
'MATCHING_MODE',
'MATCHING_MODE_FREE_IP',
'LEVEL_LIMIT_EASING',
'ACHIEVEMENT_ENABLE',
'APICAGACHADRAW\t30',
'VOLFORCE_ENABLE',
'AKANAME_ENABLE',
'PAUSE_ONLINEUPDATE',
'CONTINUATION',
'TENKAICHI_MODE',
'QC_MODE',
'KAC_MODE',
'APPEAL_CARD_GEN_PRICE\t100',
'APPEAL_CARD_GEN_NEW_PRICE\t200',
'APPEAL_CARD_UNLOCK\t0,20170914,0,20171014,0,20171116,0,20180201,0,20180607,0,20181206,0,20200326,0,20200611,4,10140732,6,10150431',
'FAVORITE_APPEALCARD_MAX\t200',
'FAVORITE_MUSIC_MAX\t200',
'EVENTDATE_APRILFOOL',
'KONAMI_50TH_LOGO',
'OMEGA_ARS_ENABLE',
'DISABLE_MONITOR_ID_CHECK',
'SKILL_ANALYZER_ABLE',
'BLASTER_ABLE',
'STANDARD_UNLOCK_ENABLE',
'PLAYERJUDGEADJ_ENABLE',
'MIXID_INPUT_ENABLE',
'EVENTDATE_ONIGO',
'EVENTDATE_GOTT',
'GENERATOR_ABLE',
'CREW_SELECT_ABLE',
'PREMIUM_TIME_ENABLE',
'OMEGA_ENABLE\t1,2,3,4,5,6,7,8,9',
'HEXA_ENABLE\t1,2,3,4,5',
'MEGAMIX_ENABLE',
'VALGENE_ENABLE',
'ARENA_ENABLE',
'DISP_PASELI_BANNER',
]
unlock = []
for i in range(2000):
for j in range(0, 5):
unlock.append([i, j])
response = E.response(
E.game(
E.event(
*[E.info(
E.event_id(s, __type="str"),
)for s in event],
),
E.music_limited(
*[E.info(
E.music_id(s[0], __type="s32"),
E.music_type(s[1], __type="u8"),
E.limited(3, __type="u8"),
)for s in unlock],
),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/game/sv6_new')
async def game_sv6_new(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
root = request_info['root'][0]
dataid = root.find("dataid").text
cardno = root.find("cardno").text
name = root.find("name").text
db = get_db().table('sdvx_profile')
all_profiles_for_card = db.get(Query().card == dataid)
if all_profiles_for_card is None:
all_profiles_for_card = {
'card': dataid,
'version': {}
}
if 'sdvx_id' not in all_profiles_for_card:
sdvx_id = random.randint(10000000, 99999999)
all_profiles_for_card['sdvx_id'] = sdvx_id
all_profiles_for_card['version'][str(game_version)] = {
'game_version': game_version,
'name': name,
'appeal_id': 0,
'skill_level': 0,
'skill_base_id': 0,
'skill_name_id': 0,
'earned_gamecoin_packet': 0,
'earned_gamecoin_block': 0,
'earned_blaster_energy': 0,
'earned_extrack_energy': 0,
'used_packet_booster': 0,
'used_block_booster': 0,
'hispeed': 0,
'lanespeed': 0,
'gauge_option': 0,
'ars_option': 0,
'notes_option': 0,
'early_late_disp': 0,
'draw_adjust': 0,
'eff_c_left': 0,
'eff_c_right': 1,
'music_id': 0,
'music_type': 0,
'sort_type': 0,
'narrow_down': 0,
'headphone': 1,
'print_count': 0,
'start_option': 0,
'bgm': 0,
'submonitor': 0,
'nemsys': 0,
'stampA': 0,
'stampB': 0,
'stampC': 0,
'stampD': 0,
'items': [],
'params': [],
}
db.upsert(all_profiles_for_card, where('card') == dataid)
response = E.response(
E.game(
E.result(0, __type="u8"),
),
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/game/sv6_load')
async def game_sv6_load(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
dataid = request_info['root'][0].find("dataid").text
profile = get_game_profile(dataid, game_version)
if profile:
djid, djid_split = get_id_from_profile(dataid)
unlock = []
for i in range(301):
unlock.append([i, 11, 15])
for i in range(6001):
unlock.append([i, 1, 1])
unlock.append([599, 4, 10])
for item in profile['items']:
unlock.append(item)
customize = [[2, 2, [profile['bgm'], profile['submonitor'], profile['nemsys'], profile['stampA'], profile['stampB'], profile['stampC'], profile['stampD']]]]
for item in profile['params']:
customize.append(item)
response = E.response(
E.game(
E.result(0, __type="u8"),
E.name(profile['name'], __type="str"),
E.code(djid_split, __type="str"),
E.sdvx_id(djid_split, __type="str"),
E.appeal_id(profile['appeal_id'], __type="u16"),
E.skill_level(profile['skill_level'], __type="s16"),
E.skill_base_id(profile['skill_base_id'], __type="s16"),
E.skill_name_id(profile['skill_name_id'], __type="s16"),
E.gamecoin_packet(profile['earned_gamecoin_packet'], __type="u32"),
E.gamecoin_block(profile['earned_gamecoin_block'], __type="u32"),
E.blaster_energy(profile['earned_blaster_energy'], __type="u32"),
E.blaster_count(9999, __type="u32"),
E.extrack_energy(profile['earned_extrack_energy'], __type="u16"),
E.play_count(1001, __type="u32"),
E.day_count(301, __type="u32"),
E.today_count(21, __type="u32"),
E.play_chain(31, __type="u32"),
E.max_play_chain(31, __type="u32"),
E.week_count(9, __type="u32"),
E.week_play_count(101, __type="u32"),
E.week_chain(31, __type="u32"),
E.max_week_chain(1001, __type="u32"),
E.creator_id(1, __type="u32"),
E.eaappli(
E.relation(1, __type="s8")
),
E.ea_shop(
E.blaster_pass_enable(1, __type="bool"),
E.blaster_pass_limit_date(1605871200, __type="u64"),
),
E.kac_id(profile['name'], __type="str"),
E.block_no(0, __type="s32"),
E.volte_factory(
*[E.info(
E.goods_id(s, __type="s32"),
E.status(1, __type="s32"),
)for s in range(1, 999)],
),
*[E.campaign(
E.campaign_id(s, __type="s32"),
E.jackpot_flg(1, __type="bool"),
)for s in range(99)],
E.cloud(
E.relation(1, __type="s8")
),
E.something(
*[E.info(
E.ranking_id(s[0], __type="s32"),
E.value(s[1], __type="s64"),
)for s in [[1402, 20000]]],
),
E.festival(
E.fes_id(1, __type="s32"),
E.live_energy(1000000, __type="s32"),
*[E.bonus(
E.energy_type(s, __type="s32"),
E.live_energy(1000000, __type="s32"),
)for s in range(1, 6)],
),
E.valgene_ticket(
E.ticket_num(0, __type="s32"),
E.limit_date(1605871200, __type="u64"),
),
E.arena(
E.last_play_season(0, __type="s32"),
E.rank_point(0, __type="s32"),
E.shop_point(0, __type="s32"),
E.ultimate_rate(0, __type="s32"),
E.ultimate_rank_num(0, __type="s32"),
E.rank_play_cnt(0, __type="s32"),
E.ultimate_play_cnt(0, __type="s32"),
),
E.hispeed(profile['hispeed'], __type="s32"),
E.lanespeed(profile['lanespeed'], __type="u32"),
E.gauge_option(profile['gauge_option'], __type="u8"),
E.ars_option(profile['ars_option'], __type="u8"),
E.notes_option(profile['notes_option'], __type="u8"),
E.early_late_disp(profile['early_late_disp'], __type="u8"),
E.draw_adjust(profile['draw_adjust'], __type="s32"),
E.eff_c_left(profile['eff_c_left'], __type="u8"),
E.eff_c_right(profile['eff_c_right'], __type="u8"),
E.last_music_id(profile['music_id'], __type="s32"),
E.last_music_type(profile['music_type'], __type="u8"),
E.sort_type(profile['sort_type'], __type="u8"),
E.narrow_down(profile['narrow_down'], __type="u8"),
E.headphone(profile['headphone'], __type="u8"),
E.item(
*[E.info(
E.id(s[0], __type="u32"),
E.type(s[1], __type="u8"),
E.param(s[2], __type="u32"),
)for s in unlock],
),
E.param(
*[E.info(
E.type(s[0], __type="s32"),
E.id(s[1], __type="s32"),
E.param(s[2], __type="s32", __count=len(s[2])),
)for s in customize],
),
),
)
else:
response = E.response(
E.game(
E.result(1, __type="u8"),
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/game/sv6_load_m')
async def game_sv6_load_m(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
dataid = request_info['root'][0].find("refid").text
profile = get_game_profile(dataid, game_version)
djid, djid_split = get_id_from_profile(dataid)
best_scores = []
db = get_db()
for record in db.table('sdvx_scores_best').search(
(where('game_version') == game_version)
& (where('sdvx_id') == djid)
):
best_scores.append([
record['music_id'],
record['music_type'],
record['score'],
record['exscore'],
record['clear_type'],
record['score_grade'],
0,
0,
record['btn_rate'],
record['long_rate'],
record['vol_rate'],
0,
0,
0,
0,
0,
0,
0,
0,
0,
0,
])
response = E.response(
E.game(
E.music(
*[E.info(
E.param(x, __type="u32"),
)for x in best_scores],
),
),
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/game/sv6_save')
async def game_sv6_save(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
dataid = request_info['root'][0].find("refid").text
profile = get_profile(dataid)
game_profile = profile['version'].get(str(game_version), {})
root = request_info['root'][0]
game_profile['appeal_id'] = int(root.find('appeal_id').text)
nodes = [
'appeal_id',
'skill_level',
'skill_base_id',
'skill_name_id',
'earned_gamecoin_packet',
'earned_gamecoin_block',
'earned_blaster_energy',
'earned_extrack_energy',
'hispeed',
'lanespeed',
'gauge_option',
'ars_option',
'notes_option',
'early_late_disp',
'draw_adjust',
'eff_c_left',
'eff_c_right',
'music_id',
'music_type',
'sort_type',
'narrow_down',
'headphone',
'start_option',
]
for node in nodes:
game_profile[node] = int(root.find(node).text)
game_profile['used_packet_booster'] = int(root.find('ea_shop')[0].text)
game_profile['used_block_booster'] = int(root.find('ea_shop')[1].text)
game_profile['print_count'] = int(root.find('print')[0].text)
items = []
for info in root.find('item'):
items.append([int(info.find('id').text), int(info.find('type').text), int(info.find('param').text)])
game_profile['items'] = items
params = []
for info in root.find('param'):
p = info.find('param')
params.append([int(info.find('type').text), int(info.find('id').text), [int(x) for x in p.text.split(' ')]])
game_profile['params'] = params
profile['version'][str(game_version)] = game_profile
get_db().table('sdvx_profile').upsert(profile, where('card') == dataid)
response = E.response(
E.game(),
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/game/sv6_save_m')
async def game_sv6_save_m(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
timestamp = time.time()
root = request_info['root'][0]
dataid = root.find("dataid").text
profile = get_game_profile(dataid, game_version)
djid, djid_split = get_id_from_profile(dataid)
track = root.find("track")
play_id = int(track.find("play_id").text)
music_id = int(track.find("music_id").text)
music_type = int(track.find("music_type").text)
score = int(track.find("score").text)
exscore = int(track.find("exscore").text)
clear_type = int(track.find("clear_type").text)
score_grade = int(track.find("score_grade").text)
max_chain = int(track.find("max_chain").text)
just = int(track.find("just").text)
critical = int(track.find("critical").text)
near = int(track.find("near").text)
error = int(track.find("error").text)
effective_rate = int(track.find("effective_rate").text)
btn_rate = int(track.find("btn_rate").text)
long_rate = int(track.find("long_rate").text)
vol_rate = int(track.find("vol_rate").text)
mode = int(track.find("mode").text)
gauge_type = int(track.find("gauge_type").text)
notes_option = int(track.find("notes_option").text)
online_num = int(track.find("online_num").text)
local_num = int(track.find("local_num").text)
challenge_type = int(track.find("challenge_type").text)
retry_cnt = int(track.find("retry_cnt").text)
judge = [int(x) for x in track.find("judge").text.split(' ')]
db = get_db()
db.table('sdvx_scores').insert(
{
'timestamp': timestamp,
'game_version': game_version,
'sdvx_id': djid,
'play_id': play_id,
'music_id': music_id,
'music_type': music_type,
'score': score,
'exscore': exscore,
'clear_type': clear_type,
'score_grade': score_grade,
'max_chain': max_chain,
'just': just,
'critical': critical,
'near': near,
'error': error,
'effective_rate': effective_rate,
'btn_rate': btn_rate,
'long_rate': long_rate,
'vol_rate': vol_rate,
'mode': mode,
'gauge_type': gauge_type,
'notes_option': notes_option,
'online_num': online_num,
'local_num': local_num,
'challenge_type': challenge_type,
'retry_cnt': retry_cnt,
'judge': judge,
},
)
best = db.table('sdvx_scores_best').get(
(where('sdvx_id') == djid)
& (where('game_version') == game_version)
& (where('music_id') == music_id)
& (where('music_type') == music_type)
)
best = {} if best is None else best
best_score_data = {
'game_version': game_version,
'sdvx_id': djid,
'name': profile['name'],
'music_id': music_id,
'music_type': music_type,
'score': max(score, best.get('score', score)),
'exscore': max(exscore, best.get('exscore', exscore)),
'clear_type': max(clear_type, best.get('clear_type', clear_type)),
'score_grade': max(score_grade, best.get('score_grade', score_grade)),
'btn_rate': max(btn_rate, best.get('btn_rate', btn_rate)),
'long_rate': max(long_rate, best.get('long_rate', long_rate)),
'vol_rate': max(vol_rate, best.get('vol_rate', vol_rate)),
}
db.table('sdvx_scores_best').upsert(
best_score_data,
(where('sdvx_id') == djid)
& (where('game_version') == game_version)
& (where('music_id') == music_id)
& (where('music_type') == music_type)
)
response = E.response(
E.game(),
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/game/sv6_hiscore')
async def game_sv6_hiscore(request: Request):
request_info = await core_process_request(request)
game_version = request_info['game_version']
best_scores = []
db = get_db()
for record in db.table('sdvx_scores_best').search(
(where('game_version') == game_version)
):
best_scores.append([
record['music_id'],
record['music_type'],
record['sdvx_id'],
record['name'],
record['score'],
])
response = E.response(
E.game(
E.sc(
*[E.d(
E.id(s[0], __type="u32"),
E.ty(s[1], __type="u32"),
E.a_sq(s[2], __type="str"),
E.a_nm(s[3], __type="str"),
E.a_sc(s[4], __type="u32"),
E.l_sq(s[2], __type="str"),
E.l_nm(s[3], __type="str"),
E.l_sc(s[4], __type="u32"),
)for s in best_scores],
),
),
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/game/sv6_lounge')
async def game_sv6_lounge(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.game(
E.interval(30, __type="u32")
),
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
@router.post('/{gameinfo}/game/sv6_shop')
async def game_sv6_shop(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.game(
E.nxt_time(1000 * 5 * 60, __type="u32")
),
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)
for stub in [
'load_r',
'frozen',
'save_e',
'save_mega',
'play_e',
'play_s',
'entry_s',
'entry_e',
'log'
]:
@router.post(f'/{{gameinfo}}/game/sv6_{stub}')
async def game_sv6_stub(request: Request):
request_info = await core_process_request(request)
response = E.response(
E.game(),
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

90
pyeamu.py Normal file
View File

@ -0,0 +1,90 @@
from urllib.parse import urlunparse, urlencode
import uvicorn
from fastapi import FastAPI, Request, Response
import config
import modules
from core_common import core_process_request, core_prepare_response, E
def urlpathjoin(parts, sep='/'):
return sep + sep.join([x.lstrip(sep) for x in parts])
server_address = f"{config.ip}:{config.port}"
server_services_url = urlunparse(('http', server_address, config.services_prefix, None, None, None))
app = FastAPI()
for router in modules.routers:
app.include_router(router)
if __name__ == "__main__":
print("https://github.com/drmext/MonkeyBusiness")
print(" __ __ _ ")
print("| \/ | ___ _ __ | | _____ _ _ ")
print("| |\/| |/ _ \| '_ \| |/ / _ \ | | |")
print("| | | | (_) | | | | < __/ |_| |")
print("|_| |_|\___/|_| |_|_|\_\___|\__, |")
print(" |___/ ")
print(" ____ _ ")
print("| __ ) _ _ ___(_)_ __ ___ ___ ___ ")
print("| _ \| | | / __| | '_ \ / _ \/ __/ __|")
print("| |_) | |_| \__ \ | | | | __/\__ \__ \\")
print("|____/ \__,_|___/_|_| |_|\___||___/___/")
print()
print(f"<services>{server_services_url}</services>")
print("<url_slash __type=\"bool\">1</url_slash>")
print()
uvicorn.run("pyeamu:app", host=config.ip, port=config.port, reload=True)
@app.post(urlpathjoin([config.services_prefix, "/{gameinfo}/services/get"]))
async def services_get(request: Request):
request_info = await core_process_request(request)
services = {}
for service in modules.routers:
model_blacklist = services.get('model_blacklist', [])
model_whitelist = services.get('model_whitelist', [])
if request_info['model'] in model_blacklist:
continue
if model_whitelist and request_info['model'] not in model_whitelist:
continue
k = (service.tags[0] if service.tags else service.prefix).strip('/')
if k not in services:
services[k] = urlunparse(('http', server_address, service.prefix, None, None, None))
keepalive_params = {
"pa": config.ip,
"ia": config.ip,
"ga": config.ip,
"ma": config.ip,
"t1": 2,
"t2": 10,
}
services["keepalive"] = urlunparse(('http', config.ip, "/keepalive", None, urlencode(keepalive_params), None))
services["ntp"] = urlunparse(('ntp', "pool.ntp.org", "/", None, None, None))
services["services"] = urlunparse(('http', server_address, "/core", None, None, None))
response = E.response(
E.services(
expire=10800,
mode='operation',
product_domain=1,
*[E.item(
name=k,
url=services[k]
) for k in services]
)
)
response_body, response_headers = await core_prepare_response(request, response)
return Response(content=response_body, headers=response_headers)

16
requirements.txt Normal file
View File

@ -0,0 +1,16 @@
anyio
asgiref
click
colorama
fastapi
h11
idna
kbinxml
lxml
pycryptodomex
pydantic
sniffio
starlette
tinydb
typing_extensions
uvicorn

15
utils/arc4.py Normal file
View File

@ -0,0 +1,15 @@
from Cryptodome.Cipher import ARC4
from Cryptodome.Hash import MD5
class EamuseARC4:
def __init__(self, eamuseKey):
self.internal_key = bytearray.fromhex("69D74627D985EE2187161570D08D93B12455035B6DF0D8205DF5")
self.key = MD5.new(eamuseKey + self.internal_key).digest()
def decrypt(self, data):
return ARC4.new(self.key).decrypt(bytes(data))
def encrypt(self, data):
return ARC4.new(self.key).encrypt(bytes(data))

33
utils/lz77.py Normal file
View File

@ -0,0 +1,33 @@
class EamuseLZ77:
@staticmethod
def decode(data):
data_length = len(data)
offset = 0
output = []
while offset < data_length:
flag = data[offset]
offset += 1
for bit in range(8):
if flag & (1 << bit):
output.append(data[offset])
offset += 1
else:
if offset >= data_length:
break
lookback_flag = int.from_bytes(data[offset:offset+2], 'big')
lookback_length = (lookback_flag & 0x000f) + 3
lookback_offset = lookback_flag >> 4
offset += 2
if lookback_flag == 0:
break
for _ in range(lookback_length):
loffset = len(output) - lookback_offset
if loffset <= 0 or loffset >= len(output):
output.append(0)
else:
output.append(output[loffset])
return bytes(output)
# @staticmethod
# def encode(data):
# return bytes(output)