Compare commits

...

10 Commits

30 changed files with 3906 additions and 55 deletions

View File

@ -112,6 +112,8 @@ class AllnetServlet:
)
resp.uri = f"http://{self.config.title.hostname}:{self.config.title.port}/{req.game_id}/{req.ver.replace('.', '')}/"
resp.host = f"{self.config.title.hostname}:{self.config.title.port}"
self.logger.debug(f"Allnet response: {vars(resp)}")
return self.dict_to_http_form_string([vars(resp)])
resp.uri, resp.host = self.uri_registry[req.game_id]
@ -410,8 +412,8 @@ class AllnetPowerOnResponse3:
self.uri = ""
self.host = ""
self.place_id = "123"
self.name = ""
self.nickname = ""
self.name = "ARTEMiS"
self.nickname = "ARTEMiS"
self.region0 = "1"
self.region_name0 = "W"
self.region_name1 = ""
@ -434,8 +436,8 @@ class AllnetPowerOnResponse2:
self.uri = ""
self.host = ""
self.place_id = "123"
self.name = "Test"
self.nickname = "Test123"
self.name = "ARTEMiS"
self.nickname = "ARTEMiS"
self.region0 = "1"
self.region_name0 = "W"
self.region_name1 = "X"

View File

@ -0,0 +1,3 @@
ALTER TABLE mai2_item_card
CHANGE COLUMN startDate startDate TIMESTAMP DEFAULT "2018-01-01 00:00:00.0",
CHANGE COLUMN endDate endDate TIMESTAMP DEFAULT "2038-01-01 00:00:00.0";

View File

@ -0,0 +1,3 @@
ALTER TABLE mai2_item_card
CHANGE COLUMN startDate startDate TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
CHANGE COLUMN endDate endDate TIMESTAMP NOT NULL;

View File

@ -15,6 +15,7 @@ using the megaime database. Clean installations always create the latest databas
- [O.N.G.E.K.I.](#o-n-g-e-k-i)
- [Card Maker](#card-maker)
- [WACCA](#wacca)
- [Sword Art Online Arcade](#sao)
# Supported Games
@ -252,13 +253,13 @@ python dbutils.py --game SDDT upgrade
| Version ID | Version Name |
|------------|-----------------|
| 0 | Card Maker 1.34 |
| 0 | Card Maker 1.30 |
| 1 | Card Maker 1.35 |
### Support status
* Card Maker 1.34:
* Card Maker 1.30:
* CHUNITHM NEW!!: Yes
* maimai DX UNiVERSE: Yes
* O.N.G.E.K.I. Bright: Yes
@ -284,19 +285,46 @@ python read.py --series SDED --version <version ID> --binfolder titles/cm/cm_dat
python read.py --series SDDT --version <version ID> --binfolder /path/to/game/folder --optfolder /path/to/game/option/folder
```
Also make sure to import all maimai and Chunithm data as well:
Also make sure to import all maimai DX and CHUNITHM data as well:
```shell
python read.py --series SDED --version <version ID> --binfolder /path/to/cardmaker/CardMaker_Data
```
The importer for Card Maker will import all required Gachas (Banners) and cards (for maimai/Chunithm) and the hardcoded
The importer for Card Maker will import all required Gachas (Banners) and cards (for maimai DX/CHUNITHM) and the hardcoded
Cards for each Gacha (O.N.G.E.K.I. only).
**NOTE: Without executing the importer Card Maker WILL NOT work!**
### O.N.G.E.K.I. Gachas
### Config setup
Make sure to update your `config/cardmaker.yaml` with the correct version for each game. To get the current version required to run a specific game, open every opt (Axxx) folder descending until you find all three folders:
- `MU3`: O.N.G.E.K.I.
- `MAI`: maimai DX
- `CHU`: CHUNITHM
Inside each folder is a `DataConfig.xml` file, for example:
`MU3/DataConfig.xml`:
```xml
<cardMakerVersion>
<major>1</major>
<minor>35</minor>
<release>3</release>
</cardMakerVersion>
```
Now update your `config/cardmaker.yaml` with the correct version number, for example:
```yaml
version:
1: # Card Maker 1.35
ongeki: 1.35.03
```
### O.N.G.E.K.I.
Gacha "無料ガチャ" can only pull from the free cards with the following probabilities: 94%: R, 5% SR and 1% chance of
getting an SSR card
@ -309,20 +337,24 @@ and 3% chance of getting an SSR card
All other (limited) gachas can pull from every card added to ongeki_static_cards but with the promoted cards
(click on the green button under the banner) having a 10 times higher chance to get pulled
### Chunithm Gachas
### CHUNITHM
All cards in Chunithm (basically just the characters) have the same rarity to it just pulls randomly from all cards
All cards in CHUNITHM (basically just the characters) have the same rarity to it just pulls randomly from all cards
from a given gacha but made sure you cannot pull the same card twice in the same 5 times gacha roll.
### maimai DX
Printed maimai DX cards: Freedom (`cardTypeId=6`) or Gold Pass (`cardTypeId=4`) can now be selected during the login process. You can only have ONE Freedom and ONE Gold Pass active at a given time. The cards will expire after 15 days.
Thanks GetzeAvenue for the `selectedCardList` rarity hint!
### Notes
Card Maker 1.34 will only load an O.N.G.E.K.I. Bright profile (1.30). Card Maker 1.35 will only load an O.N.G.E.K.I.
Card Maker 1.30-1.34 will only load an O.N.G.E.K.I. Bright profile (1.30). Card Maker 1.35+ will only load an O.N.G.E.K.I.
Bright Memory profile (1.35).
The gachas inside the `ongeki.yaml` will make sure only the right gacha ids for the right CM version will be loaded.
The gachas inside the `config/ongeki.yaml` will make sure only the right gacha ids for the right CM version will be loaded.
Gacha IDs up to 1140 will be loaded for CM 1.34 and all gachas will be loaded for CM 1.35.
**NOTE: There is currently no way to load/use the (printed) maimai DX cards!**
## WACCA
### SDFE
@ -365,3 +397,47 @@ Always make sure your database (tables) are up-to-date, to do so go to the `core
```shell
python dbutils.py --game SDFE upgrade
```
## SAO
### SDEW
| Version ID | Version Name |
|------------|---------------|
| 0 | SAO |
### Importer
In order to use the importer locate your game installation folder and execute:
```shell
python read.py --series SDEW --version <version ID> --binfolder /path/to/game/extractedassets
```
The importer for SAO will import all items, heroes, support skills and titles data.
### Config
Config file is located in `config/sao.yaml`.
| Option | Info |
|--------------------|-----------------------------------------------------------------------------|
| `hostname` | Changes the server listening address for Mucha |
| `port` | Changes the listing port |
| `auto_register` | Allows the game to handle the automatic registration of new cards |
### Database upgrade
Always make sure your database (tables) are up-to-date, to do so go to the `core/data/schema/versions` folder and see which version is the latest, f.e. `SDEW_1_upgrade.sql`. In order to upgrade to version 3 in this case you need to perform all previous updates as well:
```shell
python dbutils.py --game SDEW upgrade
```
### Credits for SAO support:
- Midorica - Limited Network Support
- Dniel97 - Helping with network base
- tungnotpunk - Source

View File

@ -1,3 +1,13 @@
server:
enable: True
loglevel: "info"
version:
0:
ongeki: 1.30.01
chuni: 2.00.00
maimai: 1.20.00
1:
ongeki: 1.35.03
chuni: 2.10.00
maimai: 1.30.00

6
example_config/sao.yaml Normal file
View File

@ -0,0 +1,6 @@
server:
hostname: "localhost"
enable: True
loglevel: "info"
port: 9000
auto_register: True

View File

@ -17,7 +17,7 @@ Games listed below have been tested and confirmed working. Only game versions ol
+ All versions
+ Card Maker
+ 1.34
+ 1.30
+ 1.35
+ O.N.G.E.K.I.

View File

@ -23,19 +23,40 @@ class CardMakerBase:
self.game = CardMakerConstants.GAME_CODE
self.version = CardMakerConstants.VER_CARD_MAKER
@staticmethod
def _parse_int_ver(version: str) -> str:
return version.replace(".", "")[:3]
def handle_get_game_connect_api_request(self, data: Dict) -> Dict:
if self.core_cfg.server.is_develop:
uri = f"http://{self.core_cfg.title.hostname}:{self.core_cfg.title.port}"
else:
uri = f"http://{self.core_cfg.title.hostname}"
# CHUNITHM = 0, maimai = 1, ONGEKI = 2
# grab the dict with all games version numbers from user config
games_ver = self.game_cfg.version.version(self.version)
return {
"length": 3,
"gameConnectList": [
{"modelKind": 0, "type": 1, "titleUri": f"{uri}/SDHD/200/"},
{"modelKind": 1, "type": 1, "titleUri": f"{uri}/SDEZ/120/"},
{"modelKind": 2, "type": 1, "titleUri": f"{uri}/SDDT/130/"},
# CHUNITHM
{
"modelKind": 0,
"type": 1,
"titleUri": f"{uri}/SDHD/{self._parse_int_ver(games_ver['chuni'])}/",
},
# maimai DX
{
"modelKind": 1,
"type": 1,
"titleUri": f"{uri}/SDEZ/{self._parse_int_ver(games_ver['maimai'])}/",
},
# ONGEKI
{
"modelKind": 2,
"type": 1,
"titleUri": f"{uri}/SDDT/{self._parse_int_ver(games_ver['ongeki'])}/",
},
],
}
@ -47,12 +68,15 @@ class CardMakerBase:
datetime.now() + timedelta(hours=4), self.date_time_format
)
# grab the dict with all games version numbers from user config
games_ver = self.game_cfg.version.version(self.version)
return {
"gameSetting": {
"dataVersion": "1.30.00",
"ongekiCmVersion": "1.30.01",
"chuniCmVersion": "2.00.00",
"maimaiCmVersion": "1.20.00",
"ongekiCmVersion": games_ver["ongeki"],
"chuniCmVersion": games_ver["chuni"],
"maimaiCmVersion": games_ver["maimai"],
"requestInterval": 10,
"rebootStartTime": reboot_start,
"rebootEndTime": reboot_end,

View File

@ -1,8 +1,4 @@
from datetime import date, datetime, timedelta
from typing import Any, Dict, List
import json
import logging
from enum import Enum
from typing import Dict
from core.config import CoreConfig
from core.data.cache import cached
@ -16,23 +12,7 @@ class CardMaker135(CardMakerBase):
super().__init__(core_cfg, game_cfg)
self.version = CardMakerConstants.VER_CARD_MAKER_135
def handle_get_game_connect_api_request(self, data: Dict) -> Dict:
ret = super().handle_get_game_connect_api_request(data)
if self.core_cfg.server.is_develop:
uri = f"http://{self.core_cfg.title.hostname}:{self.core_cfg.title.port}"
else:
uri = f"http://{self.core_cfg.title.hostname}"
ret["gameConnectList"][0]["titleUri"] = f"{uri}/SDHD/205/"
ret["gameConnectList"][1]["titleUri"] = f"{uri}/SDEZ/125/"
ret["gameConnectList"][2]["titleUri"] = f"{uri}/SDDT/135/"
return ret
def handle_get_game_setting_api_request(self, data: Dict) -> Dict:
ret = super().handle_get_game_setting_api_request(data)
ret["gameSetting"]["dataVersion"] = "1.35.00"
ret["gameSetting"]["ongekiCmVersion"] = "1.35.03"
ret["gameSetting"]["chuniCmVersion"] = "2.05.00"
ret["gameSetting"]["maimaiCmVersion"] = "1.25.00"
return ret

View File

@ -1,3 +1,4 @@
from typing import Dict
from core.config import CoreConfig
@ -20,6 +21,21 @@ class CardMakerServerConfig:
)
class CardMakerVersionConfig:
def __init__(self, parent_config: "CardMakerConfig") -> None:
self.__config = parent_config
def version(self, version: int) -> Dict:
"""
in the form of:
1: {"ongeki": 1.30.01, "chuni": 2.00.00, "maimai": 1.20.00}
"""
return CoreConfig.get_config_field(
self.__config, "cardmaker", "version", default={}
)[version]
class CardMakerConfig(dict):
def __init__(self) -> None:
self.server = CardMakerServerConfig(self)
self.version = CardMakerVersionConfig(self)

View File

@ -6,7 +6,7 @@ class CardMakerConstants:
VER_CARD_MAKER = 0
VER_CARD_MAKER_135 = 1
VERSION_NAMES = ("Card Maker 1.34", "Card Maker 1.35")
VERSION_NAMES = ("Card Maker 1.30", "Card Maker 1.35")
@classmethod
def game_ver_to_string(cls, ver: int):

View File

@ -30,7 +30,7 @@ class CardMakerServlet:
self.versions = [
CardMakerBase(core_cfg, self.game_cfg),
CardMaker135(core_cfg, self.game_cfg),
CardMaker135(core_cfg, self.game_cfg)
]
self.logger = logging.getLogger("cardmaker")
@ -89,7 +89,7 @@ class CardMakerServlet:
if version >= 130 and version < 135: # Card Maker
internal_ver = CardMakerConstants.VER_CARD_MAKER
elif version >= 135 and version < 136: # Card Maker 1.35
elif version >= 135 and version < 140: # Card Maker 1.35
internal_ver = CardMakerConstants.VER_CARD_MAKER_135
if all(c in string.hexdigits for c in endpoint) and len(endpoint) == 32:

View File

@ -7,4 +7,4 @@ index = Mai2Servlet
database = Mai2Data
reader = Mai2Reader
game_codes = [Mai2Constants.GAME_CODE]
current_schema_version = 4
current_schema_version = 5

View File

@ -39,8 +39,8 @@ card = Table(
Column("cardTypeId", Integer, nullable=False),
Column("charaId", Integer, nullable=False),
Column("mapId", Integer, nullable=False),
Column("startDate", TIMESTAMP, server_default="2018-01-01 00:00:00.0"),
Column("endDate", TIMESTAMP, server_default="2038-01-01 00:00:00.0"),
Column("startDate", TIMESTAMP, nullable=False, server_default=func.now()),
Column("endDate", TIMESTAMP, nullable=False),
UniqueConstraint("user", "cardId", "cardTypeId", name="mai2_item_card_uk"),
mysql_charset="utf8mb4",
)
@ -444,6 +444,8 @@ class Mai2ItemData(BaseData):
card_kind: int,
chara_id: int,
map_id: int,
start_date: datetime,
end_date: datetime,
) -> Optional[Row]:
sql = insert(card).values(
user=user_id,
@ -451,9 +453,13 @@ class Mai2ItemData(BaseData):
cardTypeId=card_kind,
charaId=chara_id,
mapId=map_id,
startDate=start_date,
endDate=end_date,
)
conflict = sql.on_duplicate_key_update(charaId=chara_id, mapId=map_id)
conflict = sql.on_duplicate_key_update(
charaId=chara_id, mapId=map_id, startDate=start_date, endDate=end_date
)
result = self.execute(conflict)
if result is None:

View File

@ -104,8 +104,12 @@ class Mai2Universe(Mai2Base):
tmp.pop("id")
tmp.pop("user")
tmp["startDate"] = datetime.strftime(tmp["startDate"], "%Y-%m-%d %H:%M:%S")
tmp["endDate"] = datetime.strftime(tmp["endDate"], "%Y-%m-%d %H:%M:%S")
tmp["startDate"] = datetime.strftime(
tmp["startDate"], Mai2Constants.DATE_TIME_FORMAT
)
tmp["endDate"] = datetime.strftime(
tmp["endDate"], Mai2Constants.DATE_TIME_FORMAT
)
card_list.append(tmp)
return {
@ -154,6 +158,10 @@ class Mai2Universe(Mai2Base):
# set a random card serial number
serial_id = "".join([str(randint(0, 9)) for _ in range(20)])
# calculate start and end date of the card
start_date = datetime.utcnow()
end_date = datetime.utcnow() + timedelta(days=15)
user_card = upsert["userCard"]
self.data.item.put_card(
user_id,
@ -161,8 +169,26 @@ class Mai2Universe(Mai2Base):
user_card["cardTypeId"],
user_card["charaId"],
user_card["mapId"],
# add the correct start date and also the end date in 15 days
start_date,
end_date,
)
# get the profile extend to save the new bought card
extend = self.data.profile.get_profile_extend(user_id, self.version)
if extend:
extend = extend._asdict()
# parse the selectedCardList
# 6 = Freedom Pass, 4 = Gold Pass (cardTypeId)
selected_cards: list = extend["selectedCardList"]
# if no pass is already added, add the corresponding pass
if not user_card["cardTypeId"] in selected_cards:
selected_cards.insert(0, user_card["cardTypeId"])
extend["selectedCardList"] = selected_cards
self.data.profile.put_profile_extend(user_id, self.version, extend)
# properly format userPrintDetail for the database
upsert.pop("userCard")
upsert.pop("serialId")
@ -174,8 +200,8 @@ class Mai2Universe(Mai2Base):
"returnCode": 1,
"orderId": 0,
"serialId": serial_id,
"startDate": "2018-01-01 00:00:00",
"endDate": "2038-01-01 00:00:00",
"startDate": datetime.strftime(start_date, Mai2Constants.DATE_TIME_FORMAT),
"endDate": datetime.strftime(end_date, Mai2Constants.DATE_TIME_FORMAT),
}
def handle_cm_upsert_user_printlog_api_request(self, data: Dict) -> Dict:

10
titles/sao/__init__.py Normal file
View File

@ -0,0 +1,10 @@
from .index import SaoServlet
from .const import SaoConstants
from .database import SaoData
from .read import SaoReader
index = SaoServlet
database = SaoData
reader = SaoReader
game_codes = [SaoConstants.GAME_CODE]
current_schema_version = 1

514
titles/sao/base.py Normal file
View File

@ -0,0 +1,514 @@
from datetime import datetime, timedelta
import json, logging
from typing import Any, Dict
import random
import struct
import csv
from core.data import Data
from core import CoreConfig
from .config import SaoConfig
from .database import SaoData
from titles.sao.handlers.base import *
class SaoBase:
def __init__(self, core_cfg: CoreConfig, game_cfg: SaoConfig) -> None:
self.core_cfg = core_cfg
self.game_cfg = game_cfg
self.core_data = Data(core_cfg)
self.game_data = SaoData(core_cfg)
self.version = 0
self.logger = logging.getLogger("sao")
def handle_noop(self, request: Any) -> bytes:
sao_request = request
sao_id = int(sao_request[:4],16) + 1
ret = struct.pack("!HHIIIIIIb", sao_id, 0, 0, 5, 1, 1, 5, 0x01000000, 0).hex()
return bytes.fromhex(ret)
def handle_c122(self, request: Any) -> bytes:
#common/get_maintenance_info
resp = SaoGetMaintResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c12e(self, request: Any) -> bytes:
#common/ac_cabinet_boot_notification
resp = SaoCommonAcCabinetBootNotificationResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c100(self, request: Any) -> bytes:
#common/get_app_versions
resp = SaoCommonGetAppVersionsRequest(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c102(self, request: Any) -> bytes:
#common/master_data_version_check
resp = SaoMasterDataVersionCheckResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c10a(self, request: Any) -> bytes:
#common/paying_play_start
resp = SaoCommonPayingPlayStartRequest(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_ca02(self, request: Any) -> bytes:
#quest_multi_play_room/get_quest_scene_multi_play_photon_server
resp = SaoGetQuestSceneMultiPlayPhotonServerResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c11e(self, request: Any) -> bytes:
#common/get_auth_card_data
#Check authentication
access_code = bytes.fromhex(request[188:268]).decode("utf-16le")
user_id = self.core_data.card.get_user_id_from_card( access_code )
if not user_id:
user_id = self.core_data.user.create_user() #works
card_id = self.core_data.card.create_card(user_id, access_code)
if card_id is None:
user_id = -1
self.logger.error("Failed to register card!")
# Create profile with 3 basic heroes
profile_id = self.game_data.profile.create_profile(user_id)
self.game_data.item.put_hero_log(user_id, 101000010, 1, 0, 101000016, 0, 30086, 1001, 1002, 1003, 1005)
self.game_data.item.put_hero_log(user_id, 102000010, 1, 0, 103000006, 0, 30086, 1001, 1002, 1003, 1005)
self.game_data.item.put_hero_log(user_id, 103000010, 1, 0, 112000009, 0, 30086, 1001, 1002, 1003, 1005)
self.game_data.item.put_hero_party(user_id, 0, 101000010, 102000010, 103000010)
self.logger.info(f"User Authenticated: { access_code } | { user_id }")
#Grab values from profile
profile_data = self.game_data.profile.get_profile(user_id)
if user_id and not profile_data:
profile_id = self.game_data.profile.create_profile(user_id)
profile_data = self.game_data.profile.get_profile(user_id)
resp = SaoGetAuthCardDataResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, profile_data)
return resp.make()
def handle_c40c(self, request: Any) -> bytes:
#home/check_ac_login_bonus
resp = SaoHomeCheckAcLoginBonusResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c104(self, request: Any) -> bytes:
#common/login
access_code = bytes.fromhex(request[228:308]).decode("utf-16le")
user_id = self.core_data.card.get_user_id_from_card( access_code )
profile_data = self.game_data.profile.get_profile(user_id)
resp = SaoCommonLoginResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, profile_data)
return resp.make()
def handle_c404(self, request: Any) -> bytes:
#home/check_comeback_event
resp = SaoCheckComebackEventRequest(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c000(self, request: Any) -> bytes:
#ticket/ticket
resp = SaoTicketResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c500(self, request: Any) -> bytes:
#user_info/get_user_basic_data
user_id = bytes.fromhex(request[88:112]).decode("utf-16le")
profile_data = self.game_data.profile.get_profile(user_id)
resp = SaoGetUserBasicDataResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, profile_data)
return resp.make()
def handle_c600(self, request: Any) -> bytes:
#have_object/get_hero_log_user_data_list
req = bytes.fromhex(request)[24:]
req_struct = Struct(
Padding(16),
"user_id_size" / Rebuild(Int32ub, len_(this.user_id) * 2), # calculates the length of the user_id
"user_id" / PaddedString(this.user_id_size, "utf_16_le"), # user_id is a (zero) padded string
)
req_data = req_struct.parse(req)
user_id = req_data.user_id
hero_data = self.game_data.item.get_hero_logs(user_id)
resp = SaoGetHeroLogUserDataListResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, hero_data)
return resp.make()
def handle_c602(self, request: Any) -> bytes:
#have_object/get_equipment_user_data_list
equipmentIdsData = self.game_data.static.get_equipment_ids(0, True)
resp = SaoGetEquipmentUserDataListResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, equipmentIdsData)
return resp.make()
def handle_c604(self, request: Any) -> bytes:
#have_object/get_item_user_data_list
itemIdsData = self.game_data.static.get_item_ids(0, True)
resp = SaoGetItemUserDataListResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, itemIdsData)
return resp.make()
def handle_c606(self, request: Any) -> bytes:
#have_object/get_support_log_user_data_list
supportIdsData = self.game_data.static.get_support_log_ids(0, True)
resp = SaoGetSupportLogUserDataListResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, supportIdsData)
return resp.make()
def handle_c800(self, request: Any) -> bytes:
#custom/get_title_user_data_list
titleIdsData = self.game_data.static.get_title_ids(0, True)
resp = SaoGetTitleUserDataListResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, titleIdsData)
return resp.make()
def handle_c608(self, request: Any) -> bytes:
#have_object/get_episode_append_data_list
user_id = bytes.fromhex(request[88:112]).decode("utf-16le")
profile_data = self.game_data.profile.get_profile(user_id)
resp = SaoGetEpisodeAppendDataListResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, profile_data)
return resp.make()
def handle_c804(self, request: Any) -> bytes:
#custom/get_party_data_list
req = bytes.fromhex(request)[24:]
req_struct = Struct(
Padding(16),
"user_id_size" / Rebuild(Int32ub, len_(this.user_id) * 2), # calculates the length of the user_id
"user_id" / PaddedString(this.user_id_size, "utf_16_le"), # user_id is a (zero) padded string
)
req_data = req_struct.parse(req)
user_id = req_data.user_id
hero_party = self.game_data.item.get_hero_party(user_id, 0)
hero1_data = self.game_data.item.get_hero_log(user_id, hero_party[3])
hero2_data = self.game_data.item.get_hero_log(user_id, hero_party[4])
hero3_data = self.game_data.item.get_hero_log(user_id, hero_party[5])
resp = SaoGetPartyDataListResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, hero1_data, hero2_data, hero3_data)
return resp.make()
def handle_c902(self, request: Any) -> bytes: # for whatever reason, having all entries empty or filled changes nothing
#quest/get_quest_scene_prev_scan_profile_card
resp = SaoGetQuestScenePrevScanProfileCardResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c124(self, request: Any) -> bytes:
#common/get_resource_path_info
resp = SaoGetResourcePathInfoResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c900(self, request: Any) -> bytes:
#quest/get_quest_scene_user_data_list // QuestScene.csv
questIdsData = self.game_data.static.get_quests_ids(0, True)
resp = SaoGetQuestSceneUserDataListResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, questIdsData)
return resp.make()
def handle_c400(self, request: Any) -> bytes:
#home/check_yui_medal_get_condition
resp = SaoCheckYuiMedalGetConditionResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c402(self, request: Any) -> bytes:
#home/get_yui_medal_bonus_user_data
resp = SaoGetYuiMedalBonusUserDataResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c40a(self, request: Any) -> bytes:
#home/check_profile_card_used_reward
resp = SaoCheckProfileCardUsedRewardResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c806(self, request: Any) -> bytes:
#custom/change_party
req = bytes.fromhex(request)[24:]
req_struct = Struct(
Padding(20),
"ticket_id" / Bytes(1), # needs to be parsed as an int
Padding(1),
"user_id_size" / Rebuild(Int32ub, len_(this.user_id) * 2), # calculates the length of the user_id
"user_id" / PaddedString(this.user_id_size, "utf_16_le"), # user_id is a (zero) padded string
"act_type" / Int8ub, # play_mode is a byte
Padding(3),
"party_data_list_length" / Rebuild(Int8ub, len_(this.party_data_list)), # party_data_list is a byte,
"party_data_list" / Array(this.party_data_list_length, Struct(
"user_party_id_size" / Rebuild(Int32ub, len_(this.user_party_id) * 2), # calculates the length of the user_party_id
"user_party_id" / PaddedString(this.user_party_id_size, "utf_16_le"), # user_party_id is a (zero) padded string
"team_no" / Int8ub, # team_no is a byte
Padding(3),
"party_team_data_list_length" / Rebuild(Int8ub, len_(this.party_team_data_list)), # party_team_data_list is a byte
"party_team_data_list" / Array(this.party_team_data_list_length, Struct(
"user_party_team_id_size" / Rebuild(Int32ub, len_(this.user_party_team_id) * 2), # calculates the length of the user_party_team_id
"user_party_team_id" / PaddedString(this.user_party_team_id_size, "utf_16_le"), # user_party_team_id is a (zero) padded string
"arrangement_num" / Int8ub, # arrangement_num is a byte
"user_hero_log_id_size" / Rebuild(Int32ub, len_(this.user_hero_log_id) * 2), # calculates the length of the user_hero_log_id
"user_hero_log_id" / PaddedString(this.user_hero_log_id_size, "utf_16_le"), # user_hero_log_id is a (zero) padded string
"main_weapon_user_equipment_id_size" / Rebuild(Int32ub, len_(this.main_weapon_user_equipment_id) * 2), # calculates the length of the main_weapon_user_equipment_id
"main_weapon_user_equipment_id" / PaddedString(this.main_weapon_user_equipment_id_size, "utf_16_le"), # main_weapon_user_equipment_id is a (zero) padded string
"sub_equipment_user_equipment_id_size" / Rebuild(Int32ub, len_(this.sub_equipment_user_equipment_id) * 2), # calculates the length of the sub_equipment_user_equipment_id
"sub_equipment_user_equipment_id" / PaddedString(this.sub_equipment_user_equipment_id_size, "utf_16_le"), # sub_equipment_user_equipment_id is a (zero) padded string
"skill_slot1_skill_id" / Int32ub, # skill_slot1_skill_id is a int,
"skill_slot2_skill_id" / Int32ub, # skill_slot1_skill_id is a int,
"skill_slot3_skill_id" / Int32ub, # skill_slot1_skill_id is a int,
"skill_slot4_skill_id" / Int32ub, # skill_slot1_skill_id is a int,
"skill_slot5_skill_id" / Int32ub, # skill_slot1_skill_id is a int,
)),
)),
)
req_data = req_struct.parse(req)
user_id = req_data.user_id
for party_team in req_data.party_data_list[0].party_team_data_list:
hero_data = self.game_data.item.get_hero_log(user_id, party_team["user_hero_log_id"])
hero_level = 1
hero_exp = 0
if hero_data:
hero_level = hero_data["log_level"]
hero_exp = hero_data["log_exp"]
self.game_data.item.put_hero_log(
user_id,
party_team["user_hero_log_id"],
hero_level,
hero_exp,
party_team["main_weapon_user_equipment_id"],
party_team["sub_equipment_user_equipment_id"],
party_team["skill_slot1_skill_id"],
party_team["skill_slot2_skill_id"],
party_team["skill_slot3_skill_id"],
party_team["skill_slot4_skill_id"],
party_team["skill_slot5_skill_id"]
)
resp = SaoNoopResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c904(self, request: Any) -> bytes:
#quest/episode_play_start
req = bytes.fromhex(request)[24:]
req_struct = Struct(
Padding(20),
"ticket_id" / Bytes(1), # needs to be parsed as an int
Padding(1),
"user_id_size" / Rebuild(Int32ub, len_(this.user_id) * 2), # calculates the length of the user_id
"user_id" / PaddedString(this.user_id_size, "utf_16_le"), # user_id is a (zero) padded string
"episode_id" / Int32ub, # episode_id is a int,
"play_mode" / Int8ub, # play_mode is a byte
Padding(3),
"play_start_request_data_length" / Rebuild(Int8ub, len_(this.play_start_request_data)), # play_start_request_data_length is a byte,
"play_start_request_data" / Array(this.play_start_request_data_length, Struct(
"user_party_id_size" / Rebuild(Int32ub, len_(this.user_party_id) * 2), # calculates the length of the user_party_id
"user_party_id" / PaddedString(this.user_party_id_size, "utf_16_le"), # user_party_id is a (zero) padded string
"appoint_leader_resource_card_code_size" / Rebuild(Int32ub, len_(this.appoint_leader_resource_card_code) * 2), # calculates the length of the total_damage
"appoint_leader_resource_card_code" / PaddedString(this.appoint_leader_resource_card_code_size, "utf_16_le"), # total_damage is a (zero) padded string
"use_profile_card_code_size" / Rebuild(Int32ub, len_(this.use_profile_card_code) * 2), # calculates the length of the total_damage
"use_profile_card_code" / PaddedString(this.use_profile_card_code_size, "utf_16_le"), # use_profile_card_code is a (zero) padded string
"quest_drop_boost_apply_flag" / Int8ub, # quest_drop_boost_apply_flag is a byte
)),
)
req_data = req_struct.parse(req)
user_id = req_data.user_id
profile_data = self.game_data.profile.get_profile(user_id)
self.game_data.item.create_session(
user_id,
int(req_data.play_start_request_data[0].user_party_id),
req_data.episode_id,
req_data.play_mode,
req_data.play_start_request_data[0].quest_drop_boost_apply_flag
)
resp = SaoEpisodePlayStartResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, profile_data)
return resp.make()
def handle_c908(self, request: Any) -> bytes: # Level calculation missing for the profile and heroes
#quest/episode_play_end
req = bytes.fromhex(request)[24:]
req_struct = Struct(
Padding(20),
"ticket_id" / Bytes(1), # needs to be parsed as an int
Padding(1),
"user_id_size" / Rebuild(Int32ub, len_(this.user_id) * 2), # calculates the length of the user_id
"user_id" / PaddedString(this.user_id_size, "utf_16_le"), # user_id is a (zero) padded string
Padding(2),
"episode_id" / Int16ub, # episode_id is a short,
Padding(3),
"play_end_request_data" / Int8ub, # play_end_request_data is a byte
Padding(1),
"play_result_flag" / Int8ub, # play_result_flag is a byte
Padding(2),
"base_get_data_length" / Rebuild(Int8ub, len_(this.base_get_data)), # base_get_data_length is a byte,
"base_get_data" / Array(this.base_get_data_length, Struct(
"get_hero_log_exp" / Int32ub, # get_hero_log_exp is an int
"get_col" / Int32ub, # get_num is a short
)),
Padding(3),
"get_player_trace_data_list_length" / Rebuild(Int8ub, len_(this.get_player_trace_data_list)), # get_player_trace_data_list_length is a byte
"get_player_trace_data_list" / Array(this.get_player_trace_data_list_length, Struct(
"user_quest_scene_player_trace_id" / Int32ub, # user_quest_scene_player_trace_id is an int
)),
Padding(3),
"get_rare_drop_data_list_length" / Rebuild(Int8ub, len_(this.get_rare_drop_data_list)), # get_rare_drop_data_list_length is a byte
"get_rare_drop_data_list" / Array(this.get_rare_drop_data_list_length, Struct(
"quest_rare_drop_id" / Int32ub, # quest_rare_drop_id is an int
)),
Padding(3),
"get_special_rare_drop_data_list_length" / Rebuild(Int8ub, len_(this.get_special_rare_drop_data_list)), # get_special_rare_drop_data_list_length is a byte
"get_special_rare_drop_data_list" / Array(this.get_special_rare_drop_data_list_length, Struct(
"quest_special_rare_drop_id" / Int32ub, # quest_special_rare_drop_id is an int
)),
Padding(3),
"get_unanalyzed_log_tmp_reward_data_list_length" / Rebuild(Int8ub, len_(this.get_unanalyzed_log_tmp_reward_data_list)), # get_unanalyzed_log_tmp_reward_data_list_length is a byte
"get_unanalyzed_log_tmp_reward_data_list" / Array(this.get_unanalyzed_log_tmp_reward_data_list_length, Struct(
"unanalyzed_log_grade_id" / Int32ub, # unanalyzed_log_grade_id is an int,
)),
Padding(3),
"get_event_item_data_list_length" / Rebuild(Int8ub, len_(this.get_event_item_data_list)), # get_event_item_data_list_length is a byte,
"get_event_item_data_list" / Array(this.get_event_item_data_list_length, Struct(
"event_item_id" / Int32ub, # event_item_id is an int
"get_num" / Int16ub, # get_num is a short
)),
Padding(3),
"discovery_enemy_data_list_length" / Rebuild(Int8ub, len_(this.discovery_enemy_data_list)), # discovery_enemy_data_list_length is a byte
"discovery_enemy_data_list" / Array(this.discovery_enemy_data_list_length, Struct(
"enemy_kind_id" / Int32ub, # enemy_kind_id is an int
"destroy_num" / Int16ub, # destroy_num is a short
)),
Padding(3),
"destroy_boss_data_list_length" / Rebuild(Int8ub, len_(this.destroy_boss_data_list)), # destroy_boss_data_list_length is a byte
"destroy_boss_data_list" / Array(this.destroy_boss_data_list_length, Struct(
"boss_type" / Int8ub, # boss_type is a byte
"enemy_kind_id" / Int32ub, # enemy_kind_id is an int
"destroy_num" / Int16ub, # destroy_num is a short
)),
Padding(3),
"mission_data_list_length" / Rebuild(Int8ub, len_(this.mission_data_list)), # mission_data_list_length is a byte
"mission_data_list" / Array(this.mission_data_list_length, Struct(
"mission_id" / Int32ub, # enemy_kind_id is an int
"clear_flag" / Int8ub, # boss_type is a byte
"mission_difficulty_id" / Int16ub, # destroy_num is a short
)),
Padding(3),
"score_data_length" / Rebuild(Int8ub, len_(this.score_data)), # score_data_length is a byte
"score_data" / Array(this.score_data_length, Struct(
"clear_time" / Int32ub, # clear_time is an int
"combo_num" / Int32ub, # boss_type is a int
"total_damage_size" / Rebuild(Int32ub, len_(this.total_damage) * 2), # calculates the length of the total_damage
"total_damage" / PaddedString(this.total_damage_size, "utf_16_le"), # total_damage is a (zero) padded string
"concurrent_destroying_num" / Int16ub, # concurrent_destroying_num is a short
"reaching_skill_level" / Int16ub, # reaching_skill_level is a short
"ko_chara_num" / Int8ub, # ko_chara_num is a byte
"acceleration_invocation_num" / Int16ub, # acceleration_invocation_num is a short
"boss_destroying_num" / Int16ub, # boss_destroying_num is a short
"synchro_skill_used_flag" / Int8ub, # synchro_skill_used_flag is a byte
"used_friend_skill_id" / Int32ub, # used_friend_skill_id is an int
"friend_skill_used_flag" / Int8ub, # friend_skill_used_flag is a byte
"continue_cnt" / Int16ub, # continue_cnt is a short
"total_loss_num" / Int16ub, # total_loss_num is a short
)),
)
req_data = req_struct.parse(req)
# Update the profile
profile = self.game_data.profile.get_profile(req_data.user_id)
exp = int(profile["rank_exp"]) + 100 #always 100 extra exp for some reason
col = int(profile["own_col"]) + int(req_data.base_get_data[0].get_col)
# Calculate level based off experience and the CSV list
with open(r'titles/sao/data/PlayerRank.csv') as csv_file:
csv_reader = csv.reader(csv_file, delimiter=',')
line_count = 0
data = []
rowf = False
for row in csv_reader:
if rowf==False:
rowf=True
else:
data.append(row)
for i in range(0,len(data)):
if exp>=int(data[i][1]) and exp<int(data[i+1][1]):
player_level = int(data[i][0])
break
# Update profile
updated_profile = self.game_data.profile.put_profile(
req_data.user_id,
profile["user_type"],
profile["nick_name"],
player_level,
exp,
col,
profile["own_vp"],
profile["own_yui_medal"],
profile["setting_title_id"]
)
# Update heroes from the used party
play_session = self.game_data.item.get_session(req_data.user_id)
session_party = self.game_data.item.get_hero_party(req_data.user_id, play_session["user_party_team_id"])
hero_list = []
hero_list.append(session_party["user_hero_log_id_1"])
hero_list.append(session_party["user_hero_log_id_2"])
hero_list.append(session_party["user_hero_log_id_3"])
for i in range(0,len(hero_list)):
hero_data = self.game_data.item.get_hero_log(req_data.user_id, hero_list[i])
log_exp = int(hero_data["log_exp"]) + int(req_data.base_get_data[0].get_hero_log_exp)
self.game_data.item.put_hero_log(
req_data.user_id,
hero_data["user_hero_log_id"],
hero_data["log_level"],
log_exp,
hero_data["main_weapon"],
hero_data["sub_equipment"],
hero_data["skill_slot1_skill_id"],
hero_data["skill_slot2_skill_id"],
hero_data["skill_slot3_skill_id"],
hero_data["skill_slot4_skill_id"],
hero_data["skill_slot5_skill_id"]
)
resp = SaoEpisodePlayEndResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()
def handle_c914(self, request: Any) -> bytes:
#quest/trial_tower_play_start
user_id = bytes.fromhex(request[100:124]).decode("utf-16le")
floor_id = int(request[130:132], 16) # not required but nice to know
profile_data = self.game_data.profile.get_profile(user_id)
resp = SaoEpisodePlayStartResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1, profile_data)
return resp.make()
def handle_c90a(self, request: Any) -> bytes: #should be tweaked for proper item unlock
#quest/episode_play_end_unanalyzed_log_fixed
resp = SaoEpisodePlayEndUnanalyzedLogFixedResponse(int.from_bytes(bytes.fromhex(request[:4]), "big")+1)
return resp.make()

47
titles/sao/config.py Normal file
View File

@ -0,0 +1,47 @@
from core.config import CoreConfig
class SaoServerConfig:
def __init__(self, parent_config: "SaoConfig"):
self.__config = parent_config
@property
def hostname(self) -> str:
return CoreConfig.get_config_field(
self.__config, "sao", "server", "hostname", default="localhost"
)
@property
def enable(self) -> bool:
return CoreConfig.get_config_field(
self.__config, "sao", "server", "enable", default=True
)
@property
def loglevel(self) -> int:
return CoreConfig.str_to_loglevel(
CoreConfig.get_config_field(
self.__config, "sao", "server", "loglevel", default="info"
)
)
@property
def port(self) -> int:
return CoreConfig.get_config_field(
self.__config, "sao", "server", "port", default=9000
)
@property
def auto_register(self) -> bool:
"""
Automatically register users in `aime_user` on first carding in with sao
if they don't exist already. Set to false to display an error instead.
"""
return CoreConfig.get_config_field(
self.__config, "sao", "server", "auto_register", default=True
)
class SaoConfig(dict):
def __init__(self) -> None:
self.server = SaoServerConfig(self)

15
titles/sao/const.py Normal file
View File

@ -0,0 +1,15 @@
from enum import Enum
class SaoConstants:
GAME_CODE = "SDEW"
CONFIG_NAME = "sao.yaml"
VER_SAO = 0
VERSION_NAMES = ("Sword Art Online Arcade")
@classmethod
def game_ver_to_string(cls, ver: int):
return cls.VERSION_NAMES[ver]

View File

@ -0,0 +1,121 @@
HeroLogLevelId,RequireExp
1,0,
2,1000,
3,1200,
4,1400,
5,1600,
6,1900,
7,2200,
8,2500,
9,2800,
10,3100,
11,3500,
12,3900,
13,4300,
14,4700,
15,5100,
16,5550,
17,6000,
18,6450,
19,6900,
20,7350,
21,7850,
22,8350,
23,8850,
24,9350,
25,9850,
26,10450,
27,11050,
28,11650,
29,12250,
30,12850,
31,13550,
32,14250,
33,14950,
34,15650,
35,16350,
36,17150,
37,17950,
38,18750,
39,19550,
40,20350,
41,21250,
42,22150,
43,23050,
44,23950,
45,24850,
46,25850,
47,26850,
48,27850,
49,28850,
50,29850,
51,30950,
52,32050,
53,33150,
54,34250,
55,35350,
56,36550,
57,37750,
58,38950,
59,40150,
60,41350,
61,42650,
62,43950,
63,45250,
64,46550,
65,47850,
66,49250,
67,50650,
68,52050,
69,53450,
70,54850,
71,56350,
72,57850,
73,59350,
74,60850,
75,62350,
76,63950,
77,65550,
78,67150,
79,68750,
80,70350,
81,72050,
82,73750,
83,75450,
84,77150,
85,78850,
86,80650,
87,82450,
88,84250,
89,86050,
90,87850,
91,89750,
92,91650,
93,93550,
94,95450,
95,97350,
96,99350,
97,101350,
98,103350,
99,105350,
100,107350,
101,200000,
102,300000,
103,450000,
104,600000,
105,800000,
106,1000000,
107,1250000,
108,1500000,
109,1800000,
110,2100000,
111,2100000,
112,2100000,
113,2100000,
114,2100000,
115,2100000,
116,2100000,
117,2100000,
118,2100000,
119,2100000,
120,2100000,
1 HeroLogLevelId,RequireExp
2 1,0,
3 2,1000,
4 3,1200,
5 4,1400,
6 5,1600,
7 6,1900,
8 7,2200,
9 8,2500,
10 9,2800,
11 10,3100,
12 11,3500,
13 12,3900,
14 13,4300,
15 14,4700,
16 15,5100,
17 16,5550,
18 17,6000,
19 18,6450,
20 19,6900,
21 20,7350,
22 21,7850,
23 22,8350,
24 23,8850,
25 24,9350,
26 25,9850,
27 26,10450,
28 27,11050,
29 28,11650,
30 29,12250,
31 30,12850,
32 31,13550,
33 32,14250,
34 33,14950,
35 34,15650,
36 35,16350,
37 36,17150,
38 37,17950,
39 38,18750,
40 39,19550,
41 40,20350,
42 41,21250,
43 42,22150,
44 43,23050,
45 44,23950,
46 45,24850,
47 46,25850,
48 47,26850,
49 48,27850,
50 49,28850,
51 50,29850,
52 51,30950,
53 52,32050,
54 53,33150,
55 54,34250,
56 55,35350,
57 56,36550,
58 57,37750,
59 58,38950,
60 59,40150,
61 60,41350,
62 61,42650,
63 62,43950,
64 63,45250,
65 64,46550,
66 65,47850,
67 66,49250,
68 67,50650,
69 68,52050,
70 69,53450,
71 70,54850,
72 71,56350,
73 72,57850,
74 73,59350,
75 74,60850,
76 75,62350,
77 76,63950,
78 77,65550,
79 78,67150,
80 79,68750,
81 80,70350,
82 81,72050,
83 82,73750,
84 83,75450,
85 84,77150,
86 85,78850,
87 86,80650,
88 87,82450,
89 88,84250,
90 89,86050,
91 90,87850,
92 91,89750,
93 92,91650,
94 93,93550,
95 94,95450,
96 95,97350,
97 96,99350,
98 97,101350,
99 98,103350,
100 99,105350,
101 100,107350,
102 101,200000,
103 102,300000,
104 103,450000,
105 104,600000,
106 105,800000,
107 106,1000000,
108 107,1250000,
109 108,1500000,
110 109,1800000,
111 110,2100000,
112 111,2100000,
113 112,2100000,
114 113,2100000,
115 114,2100000,
116 115,2100000,
117 116,2100000,
118 117,2100000,
119 118,2100000,
120 119,2100000,
121 120,2100000,

View File

@ -0,0 +1,301 @@
PlayerRankId,TotalExp,
1,0,
2,100,
3,300,
4,500,
5,800,
6,1100,
7,1400,
8,1800,
9,2200,
10,2600,,
11,3000,,
12,3500,,
13,4000,,
14,4500,,
15,5000,,
16,5500,,
17,6100,,
18,6700,,
19,7300,,
20,7900,,
21,8500,,
22,9100,,
23,9800,,
24,10500,
25,11200,
26,11900,
27,12600,
28,13300,
29,14000,
30,14800,
31,15600,
32,16400,
33,17200,
34,18000,
35,18800,
36,19600,
37,20400,
38,21300,
39,22200,
40,23100,
41,24000,
42,24900,
43,25800,
44,26700,
45,27600,
46,28500,
47,29500,
48,30600,
49,31800,
50,33100,
51,34500,
52,36000,
53,37600,
54,39300,
55,41100,
56,43000,
57,45000,
58,47100,
59,49300,
60,51600,
61,54000,
62,56500,
63,59100,
64,61800,
65,64600,
66,67500,
67,70500,
68,73600,
69,76800,
70,80100,
71,83500,
72,87000,
73,90600,
74,94300,
75,98100,
76,102000,
77,106000,
78,110100,
79,114300,
80,118600,
81,123000,
82,127500,
83,132100,
84,136800,
85,141600,
86,146500,
87,151500,
88,156600,
89,161800,
90,167100,
91,172500,
92,178000,
93,183600,
94,189300,
95,195100,
96,201000,
97,207000,
98,213100,
99,219300,
100,225600,
101,232000,
102,238500,
103,245100,
104,251800,
105,258600,
106,265500,
107,272500,
108,279600,
109,286800,
110,294100,
111,301500,
112,309000,
113,316600,
114,324300,
115,332100,
116,340000,
117,348000,
118,356100,
119,364300,
120,372600,
121,381000,
122,389500,
123,398100,
124,406800,
125,415600,
126,424500,
127,433500,
128,442600,
129,451800,
130,461100,
131,470500,
132,480000,
133,489600,
134,499300,
135,509100,
136,519000,
137,529000,
138,539100,
139,549300,
140,559600,
141,570000,
142,580500,
143,591100,
144,601800,
145,612600,
146,623500,
147,634500,
148,645600,
149,656800,
150,668100,
151,679500,
152,691000,
153,702600,
154,714300,
155,726100,
156,738000,
157,750000,
158,762100,
159,774300,
160,786600,
161,799000,
162,811500,
163,824100,
164,836800,
165,849600,
166,862500,
167,875500,
168,888600,
169,901800,
170,915100,
171,928500,
172,942000,
173,955600,
174,969300,
175,983100,
176,997000,
177,1011000,
178,1025100,
179,1039300,
180,1053600,
181,1068000,
182,1082500,
183,1097100,
184,1111800,
185,1126600,
186,1141500,
187,1156500,
188,1171600,
189,1186800,
190,1202100,
191,1217500,
192,1233000,
193,1248600,
194,1264300,
195,1280100,
196,1296000,
197,1312000,
198,1328100,
199,1344300,
200,1360600,
201,1377000,
202,1393500,
203,1410100,
204,1426800,
205,1443600,
206,1460500,
207,1477500,
208,1494600,
209,1511800,
210,1529100,
211,1546500,
212,1564000,
213,1581600,
214,1599300,
215,1617100,
216,1635000,
217,1653000,
218,1671100,
219,1689300,
220,1707600,
221,1726000,
222,1744500,
223,1763100,
224,1781800,
225,1800600,
226,1819500,
227,1838500,
228,1857600,
229,1876800,
230,1896100,
231,1915500,
232,1935000,
233,1954600,
234,1974300,
235,1994100,
236,2014000,
237,2034000,
238,2054100,
239,2074300,
240,2094600,
241,2115000,
242,2135500,
243,2156100,
244,2176800,
245,2197600,
246,2218500,
247,2239500,
248,2260600,
249,2281800,
250,2303100,
251,2324500,
252,2346000,
253,2367600,
254,2389300,
255,2411100,
256,2433000,
257,2455000,
258,2477100,
259,2499300,
260,2521600,
261,2544000,
262,2566500,
263,2589100,
264,2611800,
265,2634600,
266,2657500,
267,2680500,
268,2703600,
269,2726800,
270,2750100,
271,2773500,
272,2797000,
273,2820600,
274,2844300,
275,2868100,
276,2892000,
277,2916000,
278,2940100,
279,2964300,
280,2988600,
281,3013000,
282,3037500,
283,3062100,
284,3086800,
285,3111600,
286,3136500,
287,3161500,
288,3186600,
289,3211800,
290,3237100,
291,3262500,
292,3288000,
293,3313600,
294,3339300,
295,3365100,
296,3391000,
297,3417000,
298,3443100,
299,3469300,
300,3495600,
Can't render this file because it has a wrong number of fields in line 11.

12
titles/sao/database.py Normal file
View File

@ -0,0 +1,12 @@
from core.data import Data
from core.config import CoreConfig
from .schema import *
class SaoData(Data):
def __init__(self, cfg: CoreConfig) -> None:
super().__init__(cfg)
self.profile = SaoProfileData(cfg, self.session)
self.static = SaoStaticData(cfg, self.session)

View File

@ -0,0 +1 @@
from titles.sao.handlers.base import *

1739
titles/sao/handlers/base.py Normal file

File diff suppressed because it is too large Load Diff

117
titles/sao/index.py Normal file
View File

@ -0,0 +1,117 @@
from typing import Tuple
from twisted.web.http import Request
from twisted.web import resource
import json, ast
from datetime import datetime
import yaml
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
import inflection
from os import path
from core import CoreConfig, Utils
from titles.sao.config import SaoConfig
from titles.sao.const import SaoConstants
from titles.sao.base import SaoBase
from titles.sao.handlers.base import *
class SaoServlet(resource.Resource):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.isLeaf = True
self.core_cfg = core_cfg
self.config_dir = cfg_dir
self.game_cfg = SaoConfig()
if path.exists(f"{cfg_dir}/sao.yaml"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/sao.yaml")))
self.logger = logging.getLogger("sao")
if not hasattr(self.logger, "inited"):
log_fmt_str = "[%(asctime)s] SAO | %(levelname)s | %(message)s"
log_fmt = logging.Formatter(log_fmt_str)
fileHandler = TimedRotatingFileHandler(
"{0}/{1}.log".format(self.core_cfg.server.log_dir, "sao"),
encoding="utf8",
when="d",
backupCount=10,
)
fileHandler.setFormatter(log_fmt)
consoleHandler = logging.StreamHandler()
consoleHandler.setFormatter(log_fmt)
self.logger.addHandler(fileHandler)
self.logger.addHandler(consoleHandler)
self.logger.setLevel(self.game_cfg.server.loglevel)
coloredlogs.install(
level=self.game_cfg.server.loglevel, logger=self.logger, fmt=log_fmt_str
)
self.logger.inited = True
self.base = SaoBase(core_cfg, self.game_cfg)
@classmethod
def get_allnet_info(
cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str
) -> Tuple[bool, str, str]:
game_cfg = SaoConfig()
if path.exists(f"{cfg_dir}/{SaoConstants.CONFIG_NAME}"):
game_cfg.update(
yaml.safe_load(open(f"{cfg_dir}/{SaoConstants.CONFIG_NAME}"))
)
if not game_cfg.server.enable:
return (False, "", "")
return (
True,
f"http://{game_cfg.server.hostname}:{game_cfg.server.port}/{game_code}/$v/",
f"{game_cfg.server.hostname}/SDEW/$v/",
)
@classmethod
def get_mucha_info(
cls, core_cfg: CoreConfig, cfg_dir: str
) -> Tuple[bool, str, str]:
game_cfg = SaoConfig()
if path.exists(f"{cfg_dir}/{SaoConstants.CONFIG_NAME}"):
game_cfg.update(
yaml.safe_load(open(f"{cfg_dir}/{SaoConstants.CONFIG_NAME}"))
)
if not game_cfg.server.enable:
return (False, "")
return (True, "SAO1")
def setup(self) -> None:
pass
def render_POST(
self, request: Request, version: int = 0, endpoints: str = ""
) -> bytes:
req_url = request.uri.decode()
if req_url == "/matching":
self.logger.info("Matching request")
request.responseHeaders.addRawHeader(b"content-type", b"text/html; charset=utf-8")
sao_request = request.content.getvalue().hex()
#sao_request = sao_request[:32]
handler = getattr(self.base, f"handle_{sao_request[:4]}", None)
if handler is None:
self.logger.info(f"Generic Handler for {req_url} - {sao_request[:4]}")
#self.logger.debug(f"Request: {request.content.getvalue().hex()}")
resp = SaoNoopResponse(int.from_bytes(bytes.fromhex(sao_request[:4]), "big")+1)
self.logger.debug(f"Response: {resp.make().hex()}")
return resp.make()
self.logger.info(f"Handler {req_url} - {sao_request[:4]} request")
self.logger.debug(f"Request: {request.content.getvalue().hex()}")
self.logger.debug(f"Response: {handler(sao_request).hex()}")
return handler(sao_request)

230
titles/sao/read.py Normal file
View File

@ -0,0 +1,230 @@
from typing import Optional, Dict, List
from os import walk, path
import urllib
import csv
from read import BaseReader
from core.config import CoreConfig
from titles.sao.database import SaoData
from titles.sao.const import SaoConstants
class SaoReader(BaseReader):
def __init__(
self,
config: CoreConfig,
version: int,
bin_arg: Optional[str],
opt_arg: Optional[str],
extra: Optional[str],
) -> None:
super().__init__(config, version, bin_arg, opt_arg, extra)
self.data = SaoData(config)
try:
self.logger.info(
f"Start importer for {SaoConstants.game_ver_to_string(version)}"
)
except IndexError:
self.logger.error(f"Invalid project SAO version {version}")
exit(1)
def read(self) -> None:
pull_bin_ram = True
if not path.exists(f"{self.bin_dir}"):
self.logger.warn(f"Couldn't find csv file in {self.bin_dir}, skipping")
pull_bin_ram = False
if pull_bin_ram:
self.read_csv(f"{self.bin_dir}")
def read_csv(self, bin_dir: str) -> None:
self.logger.info(f"Read csv from {bin_dir}")
self.logger.info("Now reading QuestScene.csv")
try:
fullPath = bin_dir + "/QuestScene.csv"
with open(fullPath, encoding="UTF-8") as fp:
reader = csv.DictReader(fp)
for row in reader:
questSceneId = row["QuestSceneId"]
sortNo = row["SortNo"]
name = row["Name"]
enabled = True
self.logger.info(f"Added quest {questSceneId} | Name: {name}")
try:
self.data.static.put_quest(
questSceneId,
0,
sortNo,
name,
enabled
)
except Exception as err:
print(err)
except:
self.logger.warn(f"Couldn't read csv file in {self.bin_dir}, skipping")
self.logger.info("Now reading HeroLog.csv")
try:
fullPath = bin_dir + "/HeroLog.csv"
with open(fullPath, encoding="UTF-8") as fp:
reader = csv.DictReader(fp)
for row in reader:
heroLogId = row["HeroLogId"]
name = row["Name"]
nickname = row["Nickname"]
rarity = row["Rarity"]
skillTableSubId = row["SkillTableSubId"]
awakeningExp = row["AwakeningExp"]
flavorText = row["FlavorText"]
enabled = True
self.logger.info(f"Added hero {heroLogId} | Name: {name}")
try:
self.data.static.put_hero(
0,
heroLogId,
name,
nickname,
rarity,
skillTableSubId,
awakeningExp,
flavorText,
enabled
)
except Exception as err:
print(err)
except:
self.logger.warn(f"Couldn't read csv file in {self.bin_dir}, skipping")
self.logger.info("Now reading Equipment.csv")
try:
fullPath = bin_dir + "/Equipment.csv"
with open(fullPath, encoding="UTF-8") as fp:
reader = csv.DictReader(fp)
for row in reader:
equipmentId = row["EquipmentId"]
equipmentType = row["EquipmentType"]
weaponTypeId = row["WeaponTypeId"]
name = row["Name"]
rarity = row["Rarity"]
flavorText = row["FlavorText"]
enabled = True
self.logger.info(f"Added equipment {equipmentId} | Name: {name}")
try:
self.data.static.put_equipment(
0,
equipmentId,
name,
equipmentType,
weaponTypeId,
rarity,
flavorText,
enabled
)
except Exception as err:
print(err)
except:
self.logger.warn(f"Couldn't read csv file in {self.bin_dir}, skipping")
self.logger.info("Now reading Item.csv")
try:
fullPath = bin_dir + "/Item.csv"
with open(fullPath, encoding="UTF-8") as fp:
reader = csv.DictReader(fp)
for row in reader:
itemId = row["ItemId"]
itemTypeId = row["ItemTypeId"]
name = row["Name"]
rarity = row["Rarity"]
flavorText = row["FlavorText"]
enabled = True
self.logger.info(f"Added item {itemId} | Name: {name}")
try:
self.data.static.put_item(
0,
itemId,
name,
itemTypeId,
rarity,
flavorText,
enabled
)
except Exception as err:
print(err)
except:
self.logger.warn(f"Couldn't read csv file in {self.bin_dir}, skipping")
self.logger.info("Now reading SupportLog.csv")
try:
fullPath = bin_dir + "/SupportLog.csv"
with open(fullPath, encoding="UTF-8") as fp:
reader = csv.DictReader(fp)
for row in reader:
supportLogId = row["SupportLogId"]
charaId = row["CharaId"]
name = row["Name"]
rarity = row["Rarity"]
salePrice = row["SalePrice"]
skillName = row["SkillName"]
enabled = True
self.logger.info(f"Added support log {supportLogId} | Name: {name}")
try:
self.data.static.put_support_log(
0,
supportLogId,
charaId,
name,
rarity,
salePrice,
skillName,
enabled
)
except Exception as err:
print(err)
except:
self.logger.warn(f"Couldn't read csv file in {self.bin_dir}, skipping")
self.logger.info("Now reading Title.csv")
try:
fullPath = bin_dir + "/Title.csv"
with open(fullPath, encoding="UTF-8") as fp:
reader = csv.DictReader(fp)
for row in reader:
titleId = row["TitleId"]
displayName = row["DisplayName"]
requirement = row["Requirement"]
rank = row["Rank"]
imageFilePath = row["ImageFilePath"]
enabled = True
self.logger.info(f"Added title {titleId} | Name: {displayName}")
if len(titleId) > 5:
try:
self.data.static.put_title(
0,
titleId,
displayName,
requirement,
rank,
imageFilePath,
enabled
)
except Exception as err:
print(err)
elif len(titleId) < 6: # current server code cannot have multiple lengths for the id
continue
except:
self.logger.warn(f"Couldn't read csv file in {self.bin_dir}, skipping")

View File

@ -0,0 +1,3 @@
from .profile import SaoProfileData
from .static import SaoStaticData
from .item import SaoItemData

212
titles/sao/schema/item.py Normal file
View File

@ -0,0 +1,212 @@
from typing import Optional, Dict, List
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_, case
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean
from sqlalchemy.schema import ForeignKey
from sqlalchemy.sql import func, select, update, delete
from sqlalchemy.engine import Row
from sqlalchemy.dialects.mysql import insert
from core.data.schema import BaseData, metadata
hero_log_data = Table(
"sao_hero_log_data",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
Column("user_hero_log_id", Integer, nullable=False),
Column("log_level", Integer, nullable=False),
Column("log_exp", Integer, nullable=False),
Column("main_weapon", Integer, nullable=False),
Column("sub_equipment", Integer, nullable=False),
Column("skill_slot1_skill_id", Integer, nullable=False),
Column("skill_slot2_skill_id", Integer, nullable=False),
Column("skill_slot3_skill_id", Integer, nullable=False),
Column("skill_slot4_skill_id", Integer, nullable=False),
Column("skill_slot5_skill_id", Integer, nullable=False),
Column("get_date", TIMESTAMP, nullable=False, server_default=func.now()),
UniqueConstraint("user", "user_hero_log_id", name="sao_hero_log_data_uk"),
mysql_charset="utf8mb4",
)
hero_party = Table(
"sao_hero_party",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
Column("user_party_team_id", Integer, nullable=False),
Column("user_hero_log_id_1", Integer, nullable=False),
Column("user_hero_log_id_2", Integer, nullable=False),
Column("user_hero_log_id_3", Integer, nullable=False),
UniqueConstraint("user", "user_party_team_id", name="sao_hero_party_uk"),
mysql_charset="utf8mb4",
)
sessions = Table(
"sao_play_sessions",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
),
Column("user_party_team_id", Integer, nullable=False),
Column("episode_id", Integer, nullable=False),
Column("play_mode", Integer, nullable=False),
Column("quest_drop_boost_apply_flag", Integer, nullable=False),
Column("play_date", TIMESTAMP, nullable=False, server_default=func.now()),
UniqueConstraint("user", "user_party_team_id", "play_date", name="sao_play_sessions_uk"),
mysql_charset="utf8mb4",
)
class SaoItemData(BaseData):
def create_session(self, user_id: int, user_party_team_id: int, episode_id: int, play_mode: int, quest_drop_boost_apply_flag: int) -> Optional[int]:
sql = insert(sessions).values(
user=user_id,
user_party_team_id=user_party_team_id,
episode_id=episode_id,
play_mode=play_mode,
quest_drop_boost_apply_flag=quest_drop_boost_apply_flag
)
conflict = sql.on_duplicate_key_update(user=user_id)
result = self.execute(conflict)
if result is None:
self.logger.error(f"Failed to create SAO session for user {user_id}!")
return None
return result.lastrowid
def put_hero_log(self, user_id: int, user_hero_log_id: int, log_level: int, log_exp: int, main_weapon: int, sub_equipment: int, skill_slot1_skill_id: int, skill_slot2_skill_id: int, skill_slot3_skill_id: int, skill_slot4_skill_id: int, skill_slot5_skill_id: int) -> Optional[int]:
sql = insert(hero_log_data).values(
user=user_id,
user_hero_log_id=user_hero_log_id,
log_level=log_level,
log_exp=log_exp,
main_weapon=main_weapon,
sub_equipment=sub_equipment,
skill_slot1_skill_id=skill_slot1_skill_id,
skill_slot2_skill_id=skill_slot2_skill_id,
skill_slot3_skill_id=skill_slot3_skill_id,
skill_slot4_skill_id=skill_slot4_skill_id,
skill_slot5_skill_id=skill_slot5_skill_id,
)
conflict = sql.on_duplicate_key_update(
log_level=log_level,
log_exp=log_exp,
main_weapon=main_weapon,
sub_equipment=sub_equipment,
skill_slot1_skill_id=skill_slot1_skill_id,
skill_slot2_skill_id=skill_slot2_skill_id,
skill_slot3_skill_id=skill_slot3_skill_id,
skill_slot4_skill_id=skill_slot4_skill_id,
skill_slot5_skill_id=skill_slot5_skill_id,
)
result = self.execute(conflict)
if result is None:
self.logger.error(
f"{__name__} failed to insert hero! user: {user_id}, user_hero_log_id: {user_hero_log_id}"
)
return None
return result.lastrowid
def put_hero_party(self, user_id: int, user_party_team_id: int, user_hero_log_id_1: int, user_hero_log_id_2: int, user_hero_log_id_3: int) -> Optional[int]:
sql = insert(hero_party).values(
user=user_id,
user_party_team_id=user_party_team_id,
user_hero_log_id_1=user_hero_log_id_1,
user_hero_log_id_2=user_hero_log_id_2,
user_hero_log_id_3=user_hero_log_id_3,
)
conflict = sql.on_duplicate_key_update(
user_hero_log_id_1=user_hero_log_id_1,
user_hero_log_id_2=user_hero_log_id_2,
user_hero_log_id_3=user_hero_log_id_3,
)
result = self.execute(conflict)
if result is None:
self.logger.error(
f"{__name__} failed to insert hero party! user: {user_id}, user_party_team_id: {user_party_team_id}"
)
return None
return result.lastrowid
def get_hero_log(
self, user_id: int, user_hero_log_id: int = None
) -> Optional[List[Row]]:
"""
A catch-all hero lookup given a profile and user_party_team_id and ID specifiers
"""
sql = hero_log_data.select(
and_(
hero_log_data.c.user == user_id,
hero_log_data.c.user_hero_log_id == user_hero_log_id if user_hero_log_id is not None else True,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def get_hero_logs(
self, user_id: int
) -> Optional[List[Row]]:
"""
A catch-all hero lookup given a profile and user_party_team_id and ID specifiers
"""
sql = hero_log_data.select(
and_(
hero_log_data.c.user == user_id,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchall()
def get_hero_party(
self, user_id: int, user_party_team_id: int = None
) -> Optional[List[Row]]:
sql = hero_party.select(
and_(
hero_party.c.user == user_id,
hero_party.c.user_party_team_id == user_party_team_id if user_party_team_id is not None else True,
)
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()
def get_session(
self, user_id: int = None
) -> Optional[List[Row]]:
sql = sessions.select(
and_(
sessions.c.user == user_id,
)
).order_by(
sessions.c.play_date.asc()
)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()

View File

@ -0,0 +1,80 @@
from typing import Optional, Dict, List
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_, case
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON
from sqlalchemy.schema import ForeignKey
from sqlalchemy.sql import func, select, update, delete
from sqlalchemy.engine import Row
from sqlalchemy.dialects.mysql import insert
from core.data.schema import BaseData, metadata
from ..const import SaoConstants
profile = Table(
"sao_profile",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column(
"user",
ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"),
nullable=False,
unique=True,
),
Column("user_type", Integer, server_default="1"),
Column("nick_name", String(16), server_default="PLAYER"),
Column("rank_num", Integer, server_default="1"),
Column("rank_exp", Integer, server_default="0"),
Column("own_col", Integer, server_default="0"),
Column("own_vp", Integer, server_default="0"),
Column("own_yui_medal", Integer, server_default="0"),
Column("setting_title_id", Integer, server_default="20005"),
)
class SaoProfileData(BaseData):
def create_profile(self, user_id: int) -> Optional[int]:
sql = insert(profile).values(user=user_id)
conflict = sql.on_duplicate_key_update(user=user_id)
result = self.execute(conflict)
if result is None:
self.logger.error(f"Failed to create SAO profile for user {user_id}!")
return None
return result.lastrowid
def put_profile(self, user_id: int, user_type: int, nick_name: str, rank_num: int, rank_exp: int, own_col: int, own_vp: int, own_yui_medal: int, setting_title_id: int) -> Optional[int]:
sql = insert(profile).values(
user=user_id,
user_type=user_type,
nick_name=nick_name,
rank_num=rank_num,
rank_exp=rank_exp,
own_col=own_col,
own_vp=own_vp,
own_yui_medal=own_yui_medal,
setting_title_id=setting_title_id
)
conflict = sql.on_duplicate_key_update(
rank_num=rank_num,
rank_exp=rank_exp,
own_col=own_col,
own_vp=own_vp,
own_yui_medal=own_yui_medal,
setting_title_id=setting_title_id
)
result = self.execute(conflict)
if result is None:
self.logger.error(
f"{__name__} failed to insert profile! user: {user_id}"
)
return None
print(result.lastrowid)
return result.lastrowid
def get_profile(self, user_id: int) -> Optional[Row]:
sql = profile.select(profile.c.user == user_id)
result = self.execute(sql)
if result is None:
return None
return result.fetchone()

297
titles/sao/schema/static.py Normal file
View File

@ -0,0 +1,297 @@
from typing import Dict, List, Optional
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON, Float
from sqlalchemy.engine.base import Connection
from sqlalchemy.engine import Row
from sqlalchemy.schema import ForeignKey
from sqlalchemy.sql import func, select
from sqlalchemy.dialects.mysql import insert
from core.data.schema import BaseData, metadata
quest = Table(
"sao_static_quest",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer),
Column("questSceneId", Integer),
Column("sortNo", Integer),
Column("name", String(255)),
Column("enabled", Boolean),
UniqueConstraint(
"version", "questSceneId", name="sao_static_quest_uk"
),
mysql_charset="utf8mb4",
)
hero = Table(
"sao_static_hero_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer),
Column("heroLogId", Integer),
Column("name", String(255)),
Column("nickname", String(255)),
Column("rarity", Integer),
Column("skillTableSubId", Integer),
Column("awakeningExp", Integer),
Column("flavorText", String(255)),
Column("enabled", Boolean),
UniqueConstraint(
"version", "heroLogId", name="sao_static_hero_list_uk"
),
mysql_charset="utf8mb4",
)
equipment = Table(
"sao_static_equipment_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer),
Column("equipmentId", Integer),
Column("equipmentType", Integer),
Column("weaponTypeId", Integer),
Column("name", String(255)),
Column("rarity", Integer),
Column("flavorText", String(255)),
Column("enabled", Boolean),
UniqueConstraint(
"version", "equipmentId", name="sao_static_equipment_list_uk"
),
mysql_charset="utf8mb4",
)
item = Table(
"sao_static_item_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer),
Column("itemId", Integer),
Column("itemTypeId", Integer),
Column("name", String(255)),
Column("rarity", Integer),
Column("flavorText", String(255)),
Column("enabled", Boolean),
UniqueConstraint(
"version", "itemId", name="sao_static_item_list_uk"
),
mysql_charset="utf8mb4",
)
support = Table(
"sao_static_support_log_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer),
Column("supportLogId", Integer),
Column("charaId", Integer),
Column("name", String(255)),
Column("rarity", Integer),
Column("salePrice", Integer),
Column("skillName", String(255)),
Column("enabled", Boolean),
UniqueConstraint(
"version", "supportLogId", name="sao_static_support_log_list_uk"
),
mysql_charset="utf8mb4",
)
title = Table(
"sao_static_title_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer),
Column("titleId", Integer),
Column("displayName", String(255)),
Column("requirement", Integer),
Column("rank", Integer),
Column("imageFilePath", String(255)),
Column("enabled", Boolean),
UniqueConstraint(
"version", "titleId", name="sao_static_title_list_uk"
),
mysql_charset="utf8mb4",
)
class SaoStaticData(BaseData):
def put_quest( self, questSceneId: int, version: int, sortNo: int, name: str, enabled: bool ) -> Optional[int]:
sql = insert(quest).values(
questSceneId=questSceneId,
version=version,
sortNo=sortNo,
name=name,
tutorial=tutorial,
)
conflict = sql.on_duplicate_key_update(
name=name, questSceneId=questSceneId, version=version
)
result = self.execute(conflict)
if result is None:
return None
return result.lastrowid
def put_hero( self, version: int, heroLogId: int, name: str, nickname: str, rarity: int, skillTableSubId: int, awakeningExp: int, flavorText: str, enabled: bool ) -> Optional[int]:
sql = insert(hero).values(
version=version,
heroLogId=heroLogId,
name=name,
nickname=nickname,
rarity=rarity,
skillTableSubId=skillTableSubId,
awakeningExp=awakeningExp,
flavorText=flavorText,
enabled=enabled
)
conflict = sql.on_duplicate_key_update(
name=name, heroLogId=heroLogId
)
result = self.execute(conflict)
if result is None:
return None
return result.lastrowid
def put_equipment( self, version: int, equipmentId: int, name: str, equipmentType: int, weaponTypeId:int, rarity: int, flavorText: str, enabled: bool ) -> Optional[int]:
sql = insert(equipment).values(
version=version,
equipmentId=equipmentId,
name=name,
equipmentType=equipmentType,
weaponTypeId=weaponTypeId,
rarity=rarity,
flavorText=flavorText,
enabled=enabled
)
conflict = sql.on_duplicate_key_update(
name=name, equipmentId=equipmentId
)
result = self.execute(conflict)
if result is None:
return None
return result.lastrowid
def put_item( self, version: int, itemId: int, name: str, itemTypeId: int, rarity: int, flavorText: str, enabled: bool ) -> Optional[int]:
sql = insert(item).values(
version=version,
itemId=itemId,
name=name,
itemTypeId=itemTypeId,
rarity=rarity,
flavorText=flavorText,
enabled=enabled
)
conflict = sql.on_duplicate_key_update(
name=name, itemId=itemId
)
result = self.execute(conflict)
if result is None:
return None
return result.lastrowid
def put_support_log( self, version: int, supportLogId: int, charaId: int, name: str, rarity: int, salePrice: int, skillName: str, enabled: bool ) -> Optional[int]:
sql = insert(support).values(
version=version,
supportLogId=supportLogId,
charaId=charaId,
name=name,
rarity=rarity,
salePrice=salePrice,
skillName=skillName,
enabled=enabled
)
conflict = sql.on_duplicate_key_update(
name=name, supportLogId=supportLogId
)
result = self.execute(conflict)
if result is None:
return None
return result.lastrowid
def put_title( self, version: int, titleId: int, displayName: str, requirement: int, rank: int, imageFilePath: str, enabled: bool ) -> Optional[int]:
sql = insert(title).values(
version=version,
titleId=titleId,
displayName=displayName,
requirement=requirement,
rank=rank,
imageFilePath=imageFilePath,
enabled=enabled
)
conflict = sql.on_duplicate_key_update(
displayName=displayName, titleId=titleId
)
result = self.execute(conflict)
if result is None:
return None
return result.lastrowid
def get_quests_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]:
sql = quest.select(quest.c.version == version and quest.c.enabled == enabled).order_by(
quest.c.questSceneId.asc()
)
result = self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
def get_hero_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]:
sql = hero.select(hero.c.version == version and hero.c.enabled == enabled).order_by(
hero.c.heroLogId.asc()
)
result = self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
def get_equipment_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]:
sql = equipment.select(equipment.c.version == version and equipment.c.enabled == enabled).order_by(
equipment.c.equipmentId.asc()
)
result = self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
def get_item_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]:
sql = item.select(item.c.version == version and item.c.enabled == enabled).order_by(
item.c.itemId.asc()
)
result = self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
def get_support_log_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]:
sql = support.select(support.c.version == version and support.c.enabled == enabled).order_by(
support.c.supportLogId.asc()
)
result = self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]
def get_title_ids(self, version: int, enabled: bool) -> Optional[List[Dict]]:
sql = title.select(title.c.version == version and title.c.enabled == enabled).order_by(
title.c.titleId.asc()
)
result = self.execute(sql)
if result is None:
return None
return [list[2] for list in result.fetchall()]