Bennett/artemis
1
0
Fork 1
mirror of https://gitea.tendokyu.moe/Hay1tsme/artemis.git synced 2025-12-14 08:06:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into diva_configire_festa

Browse Source
This commit is contained in:
=
2024-06-30 12:26:53 +02:00
parent e0e63a9a13 d439483aec
commit eb275062a1
12 changed files with 335 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
titles/chuni/base.py
View File

@@ -720,9 +720,14 @@ class ChuniBase:
team_id = 65535
team_name = self.game_cfg.team.team_name
team_rank = 0
team_user_point = 0
# Get user profile
profile = await self.data.profile.get_profile_data(data["userId"], self.version)
if profile is None:
return {"userId": data["userId"], "teamId": 0}
if profile and profile["teamId"]:
# Get team by id
team = await self.data.profile.get_team_by_id(profile["teamId"])
@@ -731,7 +736,12 @@ class ChuniBase:
team_id = team["id"]
team_name = team["teamName"]
team_rank = await self.data.profile.get_team_rank(team["id"])
team_point = team["teamPoint"]
if team["userTeamPoint"] is not None and team["userTeamPoint"] != "":
user_team_point_data = json.loads(team["userTeamPoint"])
for user_point_data in user_team_point_data:
if user_point_data["user"] == data["userId"]:
team_user_point = int(user_point_data["userPoint"])
# Don't return anything if no team name has been defined for defaults and there is no team set for the player
if not profile["teamId"] and team_name == "":
return {"userId": data["userId"], "teamId": 0}
@@ -741,11 +751,12 @@ class ChuniBase:
"teamId": team_id,
"teamRank": team_rank,
"teamName": team_name,
"assaultTimeRate": 1, # TODO: Figure out assaultTime, which might be team point boost?
"userTeamPoint": {
"userId": data["userId"],
"teamId": team_id,
"orderId": 1,
"teamPoint": 1,
"orderId": 0,
"teamPoint": team_user_point,
"aggrDate": data["playDate"],
},
}

76
titles/chuni/index.py
View File

@@ -41,7 +41,7 @@ class ChuniServlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
super().__init__(core_cfg, cfg_dir)
self.game_cfg = ChuniConfig()
self.hash_table: Dict[Dict[str, str]] = {}
self.hash_table: Dict[str, Dict[str, str]] = {}
if path.exists(f"{cfg_dir}/{ChuniConstants.CONFIG_NAME}"):
self.game_cfg.update(
yaml.safe_load(open(f"{cfg_dir}/{ChuniConstants.CONFIG_NAME}"))
@@ -92,32 +92,60 @@ class ChuniServlet(BaseServlet):
)
self.logger.inited = True
known_iter_counts = {
ChuniConstants.VER_CHUNITHM_CRYSTAL_PLUS: 67,
ChuniConstants.VER_CHUNITHM_PARADISE: 44,
f"{ChuniConstants.VER_CHUNITHM_PARADISE}_int": 25,
ChuniConstants.VER_CHUNITHM_NEW: 54,
f"{ChuniConstants.VER_CHUNITHM_NEW}_int": 49,
ChuniConstants.VER_CHUNITHM_NEW_PLUS: 25,
ChuniConstants.VER_CHUNITHM_SUN: 70,
ChuniConstants.VER_CHUNITHM_SUN_PLUS: 36,
ChuniConstants.VER_CHUNITHM_LUMINOUS: 8,
}
for version, keys in self.game_cfg.crypto.keys.items():
if len(keys) < 3:
continue
self.hash_table[version] = {}
if isinstance(version, int):
version_idx = version
else:
version_idx = int(version.split("_")[0])
salt = bytes.fromhex(keys[2])
if len(keys) >= 4:
iter_count = keys[3]
elif (iter_count := known_iter_counts.get(version)) is None:
self.logger.error(
"Number of iteration rounds for version %s is not known, but it is not specified in the config",
version,
)
continue
self.hash_table[version] = {}
method_list = [
method
for method in dir(self.versions[version])
for method in dir(self.versions[version_idx])
if not method.startswith("__")
]
for method in method_list:
method_fixed = inflection.camelize(method)[6:-7]
# number of iterations was changed to 70 in SUN and then to 36
if version == ChuniConstants.VER_CHUNITHM_LUMINOUS:
iter_count = 8
elif version == ChuniConstants.VER_CHUNITHM_SUN_PLUS:
iter_count = 36
elif version == ChuniConstants.VER_CHUNITHM_SUN:
iter_count = 70
else:
iter_count = 44
# This only applies for CHUNITHM NEW International and later for some reason.
# CHUNITHM SUPERSTAR (PLUS) did not add "Exp" to the endpoint when hashing.
if (
isinstance(version, str)
and version.endswith("_int")
and version_idx >= ChuniConstants.VER_CHUNITHM_NEW
):
method_fixed += "C3Exp"
hash = PBKDF2(
method_fixed,
bytes.fromhex(keys[2]),
salt,
128,
count=iter_count,
hmac_hash_module=SHA1,
@@ -127,7 +155,7 @@ class ChuniServlet(BaseServlet):
self.hash_table[version][hashed_name] = method_fixed
self.logger.debug(
f"Hashed v{version} method {method_fixed} with {bytes.fromhex(keys[2])} to get {hash.hex()}"
f"Hashed v{version} method {method_fixed} with {salt} to get {hashed_name}"
)
@classmethod
@@ -220,31 +248,39 @@ class ChuniServlet(BaseServlet):
if all(c in string.hexdigits for c in endpoint) and len(endpoint) == 32:
# If we get a 32 character long hex string, it's a hash and we're
# doing encrypted. The likelyhood of false positives is low but
# doing encrypted. The likelihood of false positives is low but
# technically not 0
if game_code == "SDGS":
crypto_cfg_key = f"{internal_ver}_int"
hash_table_key = f"{internal_ver}_int"
else:
crypto_cfg_key = internal_ver
hash_table_key = internal_ver
if internal_ver < ChuniConstants.VER_CHUNITHM_NEW:
endpoint = request.headers.get("User-Agent").split("#")[0]
else:
if internal_ver not in self.hash_table:
if hash_table_key not in self.hash_table:
self.logger.error(
f"v{version} does not support encryption or no keys entered"
)
return Response(zlib.compress(b'{"stat": "0"}'))
elif endpoint.lower() not in self.hash_table[internal_ver]:
elif endpoint.lower() not in self.hash_table[hash_table_key]:
self.logger.error(
f"No hash found for v{version} endpoint {endpoint}"
)
return Response(zlib.compress(b'{"stat": "0"}'))
endpoint = self.hash_table[internal_ver][endpoint.lower()]
endpoint = self.hash_table[hash_table_key][endpoint.lower()]
try:
crypt = AES.new(
bytes.fromhex(self.game_cfg.crypto.keys[internal_ver][0]),
bytes.fromhex(self.game_cfg.crypto.keys[crypto_cfg_key][0]),
AES.MODE_CBC,
bytes.fromhex(self.game_cfg.crypto.keys[internal_ver][1]),
bytes.fromhex(self.game_cfg.crypto.keys[crypto_cfg_key][1]),
)
req_raw = crypt.decrypt(req_raw)

35
titles/chuni/schema/profile.py
View File

@@ -1,3 +1,4 @@
import json
from typing import Dict, List, Optional
from sqlalchemy import Table, Column, UniqueConstraint, and_
from sqlalchemy.types import Integer, String, Boolean, JSON, BigInteger
@@ -389,6 +390,7 @@ team = Table(
Column("id", Integer, primary_key=True, nullable=False),
Column("teamName", String(255)),
Column("teamPoint", Integer),
Column("userTeamPoint", JSON),
mysql_charset="utf8mb4",
)
@@ -705,12 +707,36 @@ class ChuniProfileData(BaseData):
# Return the rank if found, or a default rank otherwise
return rank if rank is not None else 0
# RIP scaled team ranking. Gone, but forgotten
# def get_team_rank_scaled(self, team_id: int) -> int:
async def update_team(self, team_id: int, team_data: Dict) -> bool:
async def update_team(self, team_id: int, team_data: Dict, user_id: str, user_point_delta: int) -> bool:
# Update the team data
team_data["id"] = team_id
existing_team = self.get_team_by_id(team_id)
if existing_team is None or "userTeamPoint" not in existing_team:
self.logger.warn(
f"update_team: Failed to update team! team id: {team_id}. Existing team data not found."
)
return False
user_team_point_data = []
if existing_team["userTeamPoint"] is not None and existing_team["userTeamPoint"] != "":
user_team_point_data = json.loads(existing_team["userTeamPoint"])
updated = False
# Try to find the user in the existing data and update their points
for user_point_data in user_team_point_data:
if user_point_data["user"] == user_id:
user_point_data["userPoint"] = str(int(user_point_delta))
updated = True
break
# If the user was not found, add them to the data with the new points
if not updated:
user_team_point_data.append({"user": user_id, "userPoint": str(user_point_delta)})
# Update the team's userTeamPoint field in the team data
team_data["userTeamPoint"] = json.dumps(user_team_point_data)
# Update the team in the database
sql = insert(team).values(**team_data)
conflict = sql.on_duplicate_key_update(**team_data)
@@ -722,6 +748,7 @@ class ChuniProfileData(BaseData):
)
return False
return True
async def get_rival(self, rival_id: int) -> Optional[Row]:
sql = select(profile).where(profile.c.user == rival_id)
result = await self.execute(sql)

22
titles/mai2/base.py
View File

@@ -471,7 +471,27 @@ class Mai2Base:
}
async def handle_get_user_present_api_request(self, data: Dict) -> Dict:
return { "userId": data.get("userId", 0), "length": 0, "userPresentList": []}
items: List[Dict[str, Any]] = []
user_pres_list = await self.data.item.get_presents_by_version_user(self.version, data["userId"])
if user_pres_list:
self.logger.debug(f"Found {len(user_pres_list)} possible presents")
for present in user_pres_list:
if (present['startDate'] and present['startDate'].timestamp() > datetime.now().timestamp()):
self.logger.debug(f"Present {present['id']} distribution hasn't started yet (begins {present['startDate']})")
continue # present period hasn't started yet, move onto the next one
if (present['endDate'] and present['endDate'].timestamp() < datetime.now().timestamp()):
self.logger.warn(f"Present {present['id']} ended on {present['endDate']} and should be removed")
continue # present period ended, move onto the next one
test = await self.data.item.get_item(data["userId"], present['itemKind'], present['itemId'])
if not test: # Don't send presents for items the user already has
pres_id = present['itemKind'] * 1000000
pres_id += present['itemId']
items.append({"itemId": pres_id, "itemKind": 4, "stock": present['stock'], "isValid": True})
self.logger.info(f"Give user {data['userId']} {present['stock']}x item {present['itemId']} (kind {present['itemKind']}) as present")
return { "userId": data.get("userId", 0), "length": len(items), "userPresentList": items}
async def handle_get_transfer_friend_api_request(self, data: Dict) -> Dict:
return {}

54
titles/mai2/dx.py
View File

@@ -196,10 +196,17 @@ class Mai2DX(Mai2Base):
if "userItemList" in upsert and len(upsert["userItemList"]) > 0:
for item in upsert["userItemList"]:
if item["itemKind"] == 4:
item_id = item["itemId"] % 1000000
item_kind = item["itemId"] // 1000000
else:
item_id = item["itemId"]
item_kind = item["itemKind"]
await self.data.item.put_item(
user_id,
int(item["itemKind"]),
item["itemId"],
item_kind,
item_id,
item["stock"],
item["isValid"],
)
@@ -325,18 +332,39 @@ class Mai2DX(Mai2Base):
}
async def handle_get_user_item_api_request(self, data: Dict) -> Dict:
kind = int(data["nextIndex"] / 10000000000)
next_idx = int(data["nextIndex"] % 10000000000)
user_item_list = await self.data.item.get_items(data["userId"], kind)
kind = data["nextIndex"] // 10000000000
next_idx = data["nextIndex"] % 10000000000
items: List[Dict[str, Any]] = []
for i in range(next_idx, len(user_item_list)):
tmp = user_item_list[i]._asdict()
tmp.pop("user")
tmp.pop("id")
items.append(tmp)
if len(items) >= int(data["maxCount"]):
break
if kind == 4: # presents
user_pres_list = await self.data.item.get_presents_by_version_user(self.version, data["userId"])
if user_pres_list:
self.logger.debug(f"Found {len(user_pres_list)} possible presents")
for present in user_pres_list:
if (present['startDate'] and present['startDate'].timestamp() > datetime.now().timestamp()):
self.logger.debug(f"Present {present['id']} distribution hasn't started yet (begins {present['startDate']})")
continue # present period hasn't started yet, move onto the next one
if (present['endDate'] and present['endDate'].timestamp() < datetime.now().timestamp()):
self.logger.warn(f"Present {present['id']} ended on {present['endDate']} and should be removed")
continue # present period ended, move onto the next one
test = await self.data.item.get_item(data["userId"], present['itemKind'], present['itemId'])
if not test: # Don't send presents for items the user already has
pres_id = present['itemKind'] * 1000000
pres_id += present['itemId']
items.append({"itemId": pres_id, "itemKind": 4, "stock": present['stock'], "isValid": True})
self.logger.info(f"Give user {data['userId']} {present['stock']}x item {present['itemId']} (kind {present['itemKind']}) as present")
else:
user_item_list = await self.data.item.get_items(data["userId"], kind)
for i in range(next_idx, len(user_item_list)):
tmp = user_item_list[i]._asdict()
tmp.pop("user")
tmp.pop("id")
items.append(tmp)
if len(items) >= int(data["maxCount"]):
break
xout = kind * 10000000000 + next_idx + len(items)

66
titles/mai2/schema/item.py
View File

@@ -2,8 +2,8 @@ from core.data.schema import BaseData, metadata
from datetime import datetime
from typing import Optional, Dict, List
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON
from sqlalchemy import Table, Column, UniqueConstraint, PrimaryKeyConstraint, and_, or_
from sqlalchemy.types import Integer, String, TIMESTAMP, Boolean, JSON, BIGINT, INTEGER
from sqlalchemy.schema import ForeignKey
from sqlalchemy.sql import func, select
from sqlalchemy.dialects.mysql import insert
@@ -198,6 +198,20 @@ print_detail = Table(
mysql_charset="utf8mb4",
)
present = Table(
"mai2_item_present",
metadata,
Column('id', BIGINT, primary_key=True, nullable=False),
Column('version', INTEGER),
Column("user", Integer, ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
Column("itemKind", INTEGER, nullable=False),
Column("itemId", INTEGER, nullable=False),
Column("stock", INTEGER, nullable=False, server_default="1"),
Column("startDate", TIMESTAMP),
Column("endDate", TIMESTAMP),
UniqueConstraint("version", "user", "itemKind", "itemId", name="mai2_item_present_uk"),
mysql_charset="utf8mb4",
)
class Mai2ItemData(BaseData):
async def put_item(
@@ -476,7 +490,7 @@ class Mai2ItemData(BaseData):
musicId = music_id
)
conflict = sql.on_duplicate_key_do_nothing()
conflict = sql.on_duplicate_key_update(musicId = music_id)
result = await self.execute(conflict)
if result:
@@ -586,3 +600,49 @@ class Mai2ItemData(BaseData):
)
return None
return result.lastrowid
async def put_present(self, item_kind: int, item_id: int, version: int = None, user_id: int = None, start_date: datetime = None, end_date: datetime = None) -> Optional[int]:
sql = insert(present).values(
version = version,
user = user_id,
itemKind = item_kind,
itemId = item_id,
startDate = start_date,
endDate = end_date
)
conflict = sql.on_duplicate_key_update(
startDate = start_date,
endDate = end_date
)
result = await self.execute(conflict)
if result:
return result.lastrowid
self.logger.error(f"Failed to add present item {item_id}!")
async def get_presents_by_user(self, user_id: int = None) -> Optional[List[Row]]:
result = await self.execute(present.select(or_(present.c.user == user_id, present.c.user is None)))
if result:
return result.fetchall()
async def get_presents_by_version(self, ver: int = None) -> Optional[List[Row]]:
result = await self.execute(present.select(or_(present.c.version == ver, present.c.version is None)))
if result:
return result.fetchall()
async def get_presents_by_version_user(self, ver: int = None, user_id: int = None) -> Optional[List[Row]]:
result = await self.execute(present.select(
and_(
or_(present.c.user == user_id, present.c.user == None),
or_(present.c.version == ver, present.c.version == None)
)
))
if result:
return result.fetchall()
async def get_present_by_id(self, present_id: int) -> Optional[Row]:
result = await self.execute(present.select(present.c.id == present_id))
if result:
return result.fetchone()

2
titles/mai2/schema/profile.py
View File

@@ -892,7 +892,7 @@ class Mai2ProfileData(BaseData):
rival = rival_id
)
conflict = sql.on_duplicate_key_do_nothing()
conflict = sql.on_duplicate_key_update(rival = rival_id)
result = await self.execute(conflict)
if result: