Bennett/Arcaea-server
1
0
Fork 1
mirror of https://github.com/Lost-MSth/Arcaea-server.git synced 2026-02-06 23:57:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update to v1.4

Browse Source
This commit is contained in:
Lost-MSth
2020-10-30 20:17:21 +08:00
parent bb7ba249e9
commit 123554e24d
46 changed files with 9758 additions and 9638 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
latest version/server/arcdownload.py Normal file
View File

@@ -0,0 +1,48 @@
import os
import hashlib
from flask import url_for
def get_file_md5(file_path):
# 计算文件MD5
if not os.path.isfile(file_path):
return None
myhash = hashlib.md5()
f = open(file_path, 'rb')
while True:
b = f.read(8096)
if not b:
break
myhash.update(b)
f.close()
return myhash.hexdigest()
def get_one_song(song_id, file_dir='./database/songs'):
# 获取一首歌的下载链接，返回字典
dir_list = os.listdir(os.path.join(file_dir, song_id))
re = {}
for i in dir_list:
if os.path.isfile(os.path.join(file_dir, song_id, i)) and i in ['0.aff', '1.aff', '2.aff', '3.aff', 'base.ogg']:
if i == 'base.ogg':
re['audio'] = {"checksum": get_file_md5(os.path.join(file_dir, song_id, 'base.ogg')),
"url": url_for('download', file_path=song_id+'/base.ogg', _external=True)}
else:
if 'chart' not in re:
re['chart'] = {}
re['chart'][i[0]] = {"checksum": get_file_md5(os.path.join(file_dir, song_id, i)),
"url": url_for('download', file_path=song_id+'/'+i, _external=True)}
return {song_id: re}
def get_all_songs(file_dir='./database/songs'):
# 获取所有歌的下载链接，返回字典
dir_list = os.listdir(file_dir)
re = {}
for i in dir_list:
if os.path.isdir(os.path.join(file_dir, i)):
re.update(get_one_song(i))
return re

1372
latest version/server/arcscore.py Normal file
View File

File diff suppressed because it is too large Load Diff

259
latest version/server/arcworld.py Normal file
View File

@@ -0,0 +1,259 @@
import json
import sqlite3
import os
def int2b(x):
# int与布尔值转换
if x is None or x == 0:
return False
else:
return True
def get_world_name(file_dir='./database/map'):
# 获取所有地图名称，返回列表
L = []
for root, dirs, files in os.walk(file_dir):
for file in files:
if os.path.splitext(file)[1] == '.json':
L.append(os.path.splitext(file)[0])
return L
def get_world_info(map_id):
# 读取json文件内容，返回字典
f = open('./database/map/'+map_id+'.json', 'r')
world_info = json.load(f)
f.close()
return world_info
def get_user_world_info(user_id, map_id):
# 读取json文件内容，加上用户信息，返回字典
info = get_world_info(map_id)
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select * from user_world where map_id = :a and user_id = :b''',
{'a': map_id, 'b': user_id})
x = c.fetchone()
if x:
info['curr_position'] = x[2]
info['curr_capture'] = x[3]
info['is_locked'] = int2b(x[4])
else:
c.execute('''insert into user_world values(:a,:b,0,0,0)''', {
'a': user_id, 'b': map_id})
conn.commit()
conn.close()
return info
def get_current_map(user_id):
# 得到user的当前图，返回字符串
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select current_map from user where user_id = :a''',
{'a': user_id})
x = c.fetchone()
re = ''
if x:
re = x[0]
conn.commit()
conn.close()
return re
def get_world_all(user_id):
# 读取所有地图信息并处理，返回字典列表
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
worlds = get_world_name()
re = []
for map_id in worlds:
info = get_world_info(map_id)
steps = info['steps']
del info['steps']
rewards = []
for step in steps:
if 'items' in step:
rewards.append(
{'items': step['items'], 'position': step['position']})
info['rewards'] = rewards
c.execute('''select * from user_world where map_id = :a and user_id = :b''',
{'a': map_id, 'b': user_id})
x = c.fetchone()
if x:
info['curr_position'] = x[2]
info['curr_capture'] = x[3]
info['is_locked'] = int2b(x[4])
else:
c.execute('''insert into user_world values(:a,:b,0,0,0)''', {
'a': user_id, 'b': map_id})
re.append(info)
conn.commit()
conn.close()
return re
def get_user_world(user_id, map_id):
# 获取用户图信息，返回字典
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select * from user_world where map_id = :a and user_id = :b''',
{'a': map_id, 'b': user_id})
x = c.fetchone()
re = {
"user_id": user_id,
"curr_position": 0,
"curr_capture": 0,
"is_locked": False,
"map_id": map_id
}
if x:
re['curr_position'] = x[2]
re['curr_capture'] = x[3]
re['is_locked'] = int2b(x[4])
else:
c.execute('''insert into user_world values(:a,:b,0,0,0)''', {
'a': user_id, 'b': map_id})
conn.commit()
conn.close()
return re
def change_user_current_map(user_id, map_id):
# 改变用户当前图
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''update user set current_map = :a where user_id=:b''', {
'a': map_id, 'b': user_id})
conn.commit()
conn.close()
return None
def play_world_song(user_id, args):
# 声明是世界模式的打歌，并且记录加成信息
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
stamina_multiply = 1
fragment_multiply = 100
prog_boost_multiply = 0
if 'stamina_multiply' in args:
stamina_multiply = int(args['stamina_multiply'])
if 'fragment_multiply' in args:
fragment_multiply = int(args['fragment_multiply'])
if 'prog_boost_multiply' in args:
prog_boost_multiply = int(args['prog_boost_multiply'])
c.execute('''delete from world_songplay where user_id=:a and song_id=:b and difficulty=:c''', {
'a': user_id, 'b': args['song_id'], 'c': args['difficulty']})
c.execute('''insert into world_songplay values(:a,:b,:c,:d,:e,:f)''', {
'a': user_id, 'b': args['song_id'], 'c': args['difficulty'], 'd': stamina_multiply, 'e': fragment_multiply, 'f': prog_boost_multiply})
conn.commit()
conn.close()
return None
def climb_step(user_id, map_id, step, prev_capture, prev_position):
# 爬梯子，返回奖励列表，台阶列表，当前的位置和坐标，图信息
info = get_world_info(map_id)
step_count = int(info['step_count'])
restrict_ids = [[]] * step_count
capture = [0] * step_count
reward_bundle = [""] * step_count # 暂且不用
restrict_id = [""] * step_count
restrict_type = [""] * step_count
items = [[]] * step_count
step_type = [[]] * step_count
speed_limit_value = [0] * step_count
plus_stamina_value = [0] * step_count
for i in info['steps']:
capture[i['position']] = i['capture']
if 'items' in i:
items[i['position']] = i['items']
if 'restrict_id' in i:
restrict_id[i['position']] = i['restrict_id']
if 'restrict_ids' in i:
restrict_ids[i['position']] = i['restrict_ids']
if 'restrict_type' in i:
restrict_type[i['position']] = i['restrict_type']
if 'step_type' in i:
step_type[i['position']] = i['step_type']
if "speedlimit" in i['step_type']:
speed_limit_value[i['position']] = i['speed_limit_value']
if "plusstamina" in i['step_type']:
plus_stamina_value[i['position']] = i['plus_stamina_value']
if info['is_beyond']: # beyond判断
dt = info['beyond_health'] - prev_capture
if dt >= step:
curr_capture = prev_capture + step
else:
curr_capture = info['beyond_health']
i = 0
t = prev_capture + step
while i < step_count and t > 0:
dt = capture[i]
if dt > t:
t = 0
else:
t -= dt
i += 1
if i >= step_count:
curr_position = step_count - 1
else:
curr_position = i
else:
i = prev_position
j = prev_capture
t = step
while t > 0 and i < step_count:
dt = capture[i] - j
if dt > t:
j += t
t = 0
else:
t -= dt
j = 0
i += 1
if i >= step_count:
curr_position = step_count - 1
curr_capture = 0
else:
curr_position = i
curr_capture = j
rewards = []
steps = []
for i in range(prev_position, curr_position+1):
if items[i]:
rewards.append({'position': i, 'items': items[i]})
x = {
"map_id": map_id,
"position": i,
"restrict_ids": restrict_ids[i],
"capture": capture[i],
"reward_bundle": reward_bundle[i],
"restrict_id": restrict_id[i],
"restrict_type": restrict_type[i]
}
if step_type[i]:
x['step_type'] = step_type[i]
if speed_limit_value[i]:
x['speed_limit_value'] = speed_limit_value[i]
if plus_stamina_value[i]:
x['plus_stamina_value'] = plus_stamina_value[i]
steps.append(x)
return rewards, steps, curr_position, curr_capture, info

150
latest version/server/auth.py Normal file
View File

@@ -0,0 +1,150 @@
import sqlite3
import hashlib
import time
import server.arcworld
def arc_login(name: str, password: str) -> str: # 登录判断
# 查询数据库中的user表，验证账号密码，返回并记录token
# token采用user_id和时间戳连接后hash生成（真的是瞎想的，没用bear）
# 密码和token的加密方式为 SHA-256
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
hash_pwd = hashlib.sha256(password.encode("utf8")).hexdigest()
c.execute('''select user_id from user where name = :name and password = :password''', {
'name': name, 'password': hash_pwd})
x = c.fetchone()
if x is not None:
user_id = str(x[0])
now = int(time.time() * 1000)
token = hashlib.sha256((user_id + str(now)).encode("utf8")).hexdigest()
c.execute(
'''select exists(select * from login where user_id = :user_id)''', {"user_id": user_id})
if c.fetchone() == (1,): # 删掉多余token
c.execute('''delete from login where user_id = :user_id''',
{'user_id': user_id})
c.execute('''insert into login(access_token, user_id) values(:access_token, :user_id)''', {
'user_id': user_id, 'access_token': token})
conn.commit()
conn.close()
return token
conn.commit()
conn.close()
return None
def arc_register(name: str, password: str): # 注册
# 账号注册，只记录hash密码和用户名，生成user_id和user_code，自动登录返回token
# token和密码的处理同登录部分
def build_user_code(c):
# 生成9位的user_code，用的自然是随机
import random
flag = True
while flag:
user_code = ''.join([str(random.randint(0, 9)) for i in range(9)])
c.execute('''select exists(select * from user where user_code = :user_code)''',
{'user_code': user_code})
if c.fetchone() == (0,):
flag = False
return user_code
def build_user_id(c):
# 生成user_id，往后加1
c.execute('''select max(user_id) from user''')
x = c.fetchone()
if x[0] is not None:
return x[0] + 1
else:
return 2000001
# def insert_user_char(c, user_id):
# # 为用户添加所有可用角色
# for i in range(0, 38):
# if i in [0, 1, 2, 4, 13, 26, 27, 28, 29, 36, 21]:
# sql = 'insert into user_char values('+str(user_id)+','+str(
# i)+''',30,25000,25000,90,90,90,'',0,0,'',0,1,1)'''
# c.execute(sql)
# else:
# if i != 5:
# sql = 'insert into user_char values('+str(user_id)+','+str(
# i)+''',30,25000,25000,90,90,90,'',0,0,'',0,0,0)'''
# c.execute(sql)
def insert_user_char(c, user_id):
# 为用户添加所有可用角色
c.execute('''select * from character''')
x = c.fetchall()
if x != []:
for i in x:
c.execute('''insert into user_char values(:a,:b,:c,:d,:e,:f,:g,:h,:i,:j,:k,:l,:m,:n,:o)''', {
'a': user_id, 'b': i[0], 'c': i[2], 'd': i[3], 'e': i[4], 'f': i[5], 'g': i[6], 'h': i[7], 'i': i[8], 'j': i[9], 'k': i[10], 'l': i[11], 'm': i[12], 'n': i[14], 'o': i[15]})
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
hash_pwd = hashlib.sha256(password.encode("utf8")).hexdigest()
c.execute(
'''select exists(select * from user where name = :name)''', {'name': name})
if c.fetchone() == (0,):
user_code = build_user_code(c)
user_id = build_user_id(c)
now = int(time.time() * 1000)
c.execute('''insert into user(user_id, name, password, join_date, user_code, rating_ptt,
character_id, is_skill_sealed, is_char_uncapped, is_char_uncapped_override, is_hide_rating, favorite_character, max_stamina_notification_enabled, current_map)
values(:user_id, :name, :password, :join_date, :user_code, 0, 0, 0, 0, 0, 0, -1, 0, '')
''', {'user_code': user_code, 'user_id': user_id, 'join_date': now, 'name': name, 'password': hash_pwd})
c.execute('''insert into recent30(user_id) values(:user_id)''', {
'user_id': user_id})
token = hashlib.sha256(
(str(user_id) + str(now)).encode("utf8")).hexdigest()
c.execute('''insert into login(access_token, user_id) values(:access_token, :user_id)''', {
'user_id': user_id, 'access_token': token})
insert_user_char(c, user_id)
conn.commit()
conn.close()
return user_id, token, 0
else:
conn.commit()
conn.close()
return None, None, 101
def token_get_id(token: str):
# 用token获取id，没有考虑不同用户token相同情况，说不定会有bug
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select user_id from login where access_token = :token''', {
'token': token})
x = c.fetchone()
if x is not None:
conn.commit()
conn.close()
return x[0]
else:
conn.commit()
conn.close()
return None
def code_get_id(user_code):
# 用user_code获取id
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select user_id from user where user_code = :a''',
{'a': user_code})
x = c.fetchone()
if x is not None:
conn.commit()
conn.close()
return x[0]
else:
conn.commit()
conn.close()
return None

491
latest version/server/info.py Normal file
View File

@@ -0,0 +1,491 @@
import sqlite3
import server.arcworld
def int2b(x):
# int与布尔值转换
if x is None or x == 0:
return False
else:
return True
def get_recent_score(c, user_id):
# 得到用户最近一次的成绩，返回列表
c.execute('''select * from user where user_id = :x''', {'x': user_id})
x = c.fetchone()
if x is not None:
if x[11] is not None:
c.execute('''select best_clear_type from best_score where user_id=:u and song_id=:s and difficulty=:d''', {
'u': user_id, 's': x[11], 'd': x[12]})
y = c.fetchone()
if y is not None:
best_clear_type = y[0]
else:
best_clear_type = x[21]
return [{
"rating": x[22],
"modifier": x[19],
"time_played": x[20],
"health": x[18],
"best_clear_type": best_clear_type,
"clear_type": x[21],
"miss_count": x[17],
"near_count": x[16],
"perfect_count": x[15],
"shiny_perfect_count": x[14],
"score": x[13],
"difficulty": x[12],
"song_id": x[11]
}]
return []
def get_user_character(c, user_id):
# 得到用户拥有的角色列表，返回列表
c.execute('''select * from user_char where user_id = :user_id''',
{'user_id': user_id})
x = c.fetchall()
if x != []:
s = []
for i in x:
char_name = ''
c.execute(
'''select name from character where character_id = :x''', {'x': i[1]})
y = c.fetchone()
if y is not None:
char_name = y[0]
s.append({
"is_uncapped_override": int2b(i[14]),
"is_uncapped": int2b(i[13]),
"uncap_cores": [],
"char_type": i[12],
"skill_id_uncap": i[11],
"skill_requires_uncap": int2b(i[10]),
"skill_unlock_level": i[9],
"skill_id": i[8],
"overdrive": i[7],
"prog": i[6],
"frag": i[5],
"level_exp": i[4],
"exp": i[3],
"level": i[2],
"name": char_name,
"character_id": i[1]
})
return s
else:
return []
def get_user_friend(c, user_id):
# 得到用户的朋友列表，返回列表
c.execute('''select user_id_other from friend where user_id_me = :user_id''', {
'user_id': user_id})
x = c.fetchall()
s = []
if x != [] and x[0][0] is not None:
for i in x:
c.execute('''select exists(select * from friend where user_id_me = :x and user_id_other = :y)''',
{'x': i[0], 'y': user_id})
if c.fetchone() == (1,):
is_mutual = True
else:
is_mutual = False
c.execute('''select * from user where user_id = :x''', {'x': i[0]})
y = c.fetchone()
if y is not None:
s.append({
"is_mutual": is_mutual,
"is_char_uncapped_override": int2b(y[9]),
"is_char_uncapped": int2b(y[8]),
"is_skill_sealed": int2b(y[7]),
"rating": y[5],
"join_date": int(y[3]),
"character": y[6],
"recent_score": get_recent_score(c, i[0]),
"name": y[1],
"user_id": i[0]
})
return s
def get_value_0(c, user_id):
# 构造value id=0的数据，返回字典
c.execute('''select * from user where user_id = :x''', {'x': user_id})
x = c.fetchone()
r = {}
if x is not None:
user_character = get_user_character(c, user_id)
characters = []
for i in user_character:
characters.append(i['character_id'])
r = {"is_aprilfools": False,
"curr_available_maps": [],
"character_stats": user_character,
"friends": get_user_friend(c, user_id),
"settings": {
"favorite_character": x[23],
"is_hide_rating": int2b(x[10]),
"max_stamina_notification_enabled": int2b(x[24])
},
"user_id": user_id,
"name": x[1],
"user_code": x[4],
"display_name": x[1],
"ticket": 114514,
"character": x[6],
"is_locked_name_duplicate": False,
"is_skill_sealed": int2b(x[7]),
"current_map": x[25],
"prog_boost": 0,
"next_fragstam_ts": -1,
"max_stamina_ts": 1586274871917,
"stamina": 12,
"world_unlocks": [],
"world_songs": ["babaroque", "shadesoflight", "kanagawa", "lucifer", "anokumene", "ignotus", "rabbitintheblackroom", "qualia", "redandblue", "bookmaker", "darakunosono", "espebranch", "blacklotus", "givemeanightmare", "vividtheory", "onefr", "gekka", "vexaria3", "infinityheaven3", "fairytale3", "goodtek3", "suomi", "rugie", "faintlight", "harutopia", "goodtek", "dreaminattraction", "syro", "diode", "freefall", "grimheart", "blaster", "cyberneciacatharsis", "monochromeprincess", "revixy", "vector", "supernova", "nhelv", "purgatorium3", "dement3", "crossover", "guardina", "axiumcrisis", "worldvanquisher", "sheriruth", "pragmatism", "gloryroad", "etherstrike", "corpssansorganes", "lostdesire", "blrink", "essenceoftwilight", "lapis"],
"singles": ["dataerror", "yourvoiceso", "crosssoul", "impurebird", "auxesia", "modelista", "yozakurafubuki", "surrender", "metallicpunisher", "carminescythe", "bethere", "callmyname", "fallensquare", "dropdead", "alexandrite", "astraltale", "phantasia", "empireofwinter", "libertas", "dottodot", "dreadnought", "mirzam", "heavenlycaress", "filament", "avantraze", "battlenoone", "saikyostronger", "izana", "einherjar", "laqryma", "amygdata", "altale", "feelssoright", "scarletcage", "teriqma", "mahoroba", "badtek", "maliciousmischance", "buchigireberserker", "galaxyfriends", "buchigireberserker2", "xeraphinite"],
"packs": ["vs", "extend", "dynamix", "prelude", "core", "yugamu", "omatsuri", "zettai", "mirai", "shiawase", "chunithm", "nijuusei", "groovecoaster", "rei", "tonesphere", "lanota"],
"characters": characters,
"cores": [],
"recent_score": get_recent_score(c, user_id),
"max_friend": 50,
"rating": x[5],
"join_date": int(x[3])
}
return r
def arc_aggregate_small(user_id):
# 返回用户数据
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
r = {"success": True,
"value": [{
"id": 0,
"value": get_value_0(c, user_id)
}]}
conn.commit()
conn.close()
return r
def arc_aggregate_big(user_id):
# 返回用户数据和地图歌曲信息
# 因为没有整理地图和曲包数据（不需要世界模式），所以直接复制了
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
r = {"success": True,
"value": [{
"id": 0,
"value": get_value_0(c, user_id)
}, {
"id": 1,
"value": [{
"name": "core",
"items": [{
"id": "core",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "shiawase",
"items": [{
"id": "shiawase",
"type": "pack",
"is_available": True
}, {
"id": "kou",
"type": "character",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1552089600000,
"discount_to": 1552694399000
}, {
"name": "dynamix",
"items": [{
"id": "dynamix",
"type": "pack",
"is_available": True
}, {
"id": "sapphire",
"type": "character",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "mirai",
"items": [{
"id": "mirai",
"type": "pack",
"is_available": True
}, {
"id": "lethe",
"type": "character",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1552089600000,
"discount_to": 1552694399000
}, {
"name": "yugamu",
"items": [{
"id": "yugamu",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "lanota",
"items": [{
"id": "lanota",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "nijuusei",
"items": [{
"id": "nijuusei",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "rei",
"items": [{
"id": "rei",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "tonesphere",
"items": [{
"id": "tonesphere",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "groovecoaster",
"items": [{
"id": "groovecoaster",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "zettai",
"items": [{
"id": "zettai",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500,
"discount_from": 1583712000000,
"discount_to": 1584316799000
}, {
"name": "chunithm",
"items": [{
"id": "chunithm",
"type": "pack",
"is_available": True
}],
"price": 300,
"orig_price": 300
}, {
"name": "prelude",
"items": [{
"id": "prelude",
"type": "pack",
"is_available": True
}],
"price": 400,
"orig_price": 400
}, {
"name": "omatsuri",
"items": [{
"id": "omatsuri",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500
}, {
"name": "vs",
"items": [{
"id": "vs",
"type": "pack",
"is_available": True
}],
"price": 500,
"orig_price": 500
}, {
"name": "extend",
"items": [{
"id": "extend",
"type": "pack",
"is_available": True
}],
"price": 700,
"orig_price": 700
}]
}, {
"id": 2,
"value": {}
}, {
"id": 3,
"value": {
"max_stamina": 12,
"stamina_recover_tick": 1800000,
"core_exp": 250,
"curr_ts": 1599547606825,
"level_steps": [{
"level": 1,
"level_exp": 0
}, {
"level": 2,
"level_exp": 50
}, {
"level": 3,
"level_exp": 100
}, {
"level": 4,
"level_exp": 150
}, {
"level": 5,
"level_exp": 200
}, {
"level": 6,
"level_exp": 300
}, {
"level": 7,
"level_exp": 450
}, {
"level": 8,
"level_exp": 650
}, {
"level": 9,
"level_exp": 900
}, {
"level": 10,
"level_exp": 1200
}, {
"level": 11,
"level_exp": 1600
}, {
"level": 12,
"level_exp": 2100
}, {
"level": 13,
"level_exp": 2700
}, {
"level": 14,
"level_exp": 3400
}, {
"level": 15,
"level_exp": 4200
}, {
"level": 16,
"level_exp": 5100
}, {
"level": 17,
"level_exp": 6100
}, {
"level": 18,
"level_exp": 7200
}, {
"level": 19,
"level_exp": 8500
}, {
"level": 20,
"level_exp": 10000
}, {
"level": 21,
"level_exp": 11500
}, {
"level": 22,
"level_exp": 13000
}, {
"level": 23,
"level_exp": 14500
}, {
"level": 24,
"level_exp": 16000
}, {
"level": 25,
"level_exp": 17500
}, {
"level": 26,
"level_exp": 19000
}, {
"level": 27,
"level_exp": 20500
}, {
"level": 28,
"level_exp": 22000
}, {
"level": 29,
"level_exp": 23500
}, {
"level": 30,
"level_exp": 25000
}],
"world_ranking_enabled": False,
"is_byd_chapter_unlocked": True
}
}, {
"id": 4,
"value": []
}, {
"id": 5,
"value": {
"current_map": server.arcworld.get_current_map(user_id),
"user_id": user_id,
"maps": server.arcworld.get_world_all(user_id)
}
}
]}
conn.commit()
conn.close()
return r

155
latest version/server/setme.py Normal file
View File

@@ -0,0 +1,155 @@
import sqlite3
import server.info
def b2int(x):
# int与布尔值转换
if x:
return 1
else:
return 0
def int2b(x):
# int与布尔值转换
if x is None or x == 0:
return False
else:
return True
def change_char(user_id, character_id, skill_sealed):
# 角色改变，包括技能封印的改变，返回成功与否的布尔值
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select is_uncapped, is_uncapped_override from user_char where user_id = :a and character_id = :b''',
{'a': user_id, 'b': character_id})
x = c.fetchone()
if x is not None:
if skill_sealed == 'false':
skill_sealed = False
else:
skill_sealed = True
c.execute('''update user set is_skill_sealed = :a, character_id = :b, is_char_uncapped = :c, is_char_uncapped_override = :d where user_id = :e''', {
'a': b2int(skill_sealed), 'b': character_id, 'c': x[0], 'd': x[1], 'e': user_id})
conn.commit()
conn.close()
return True
conn.commit()
conn.close()
return False
def change_char_uncap(user_id, character_id):
# 角色觉醒改变，返回字典
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select is_uncapped, is_uncapped_override from user_char where user_id = :a and character_id = :b''',
{'a': user_id, 'b': character_id})
x = c.fetchone()
r = None
if x is not None and x[0] == 1:
c.execute('''update user set is_char_uncapped_override = :a where user_id = :b''', {
'a': b2int(x[1] == 0), 'b': user_id})
c.execute('''update user_char set is_uncapped_override = :a where user_id = :b and character_id = :c''', {
'a': b2int(x[1] == 0), 'b': user_id, 'c': character_id})
c.execute('''select * from user_char where user_id = :a and character_id = :b''',
{'a': user_id, 'b': character_id})
y = c.fetchone()
c.execute(
'''select name from character where character_id = :x''', {'x': y[1]})
z = c.fetchone()
if z is not None:
char_name = z[0]
if y is not None:
r = {
"is_uncapped_override": int2b(y[14]),
"is_uncapped": int2b(y[13]),
"uncap_cores": [],
"char_type": y[12],
"skill_id_uncap": y[11],
"skill_requires_uncap": int2b(y[10]),
"skill_unlock_level": y[9],
"skill_id": y[8],
"overdrive": y[7],
"prog": y[6],
"frag": y[5],
"level_exp": y[4],
"exp": y[3],
"level": y[2],
"name": char_name,
"character_id": y[1]
}
conn.commit()
conn.close()
return r
def arc_sys_set(user_id, value, set_arg):
# 三个设置，PTT隐藏、体力满通知、最爱角色，无返回
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
if 'favorite_character' in set_arg:
value = int(value)
c.execute('''update user set favorite_character = :a where user_id = :b''', {
'a': value, 'b': user_id})
else:
if value == 'false':
value = False
else:
value = True
if 'is_hide_rating' in set_arg:
c.execute('''update user set is_hide_rating = :a where user_id = :b''', {
'a': b2int(value), 'b': user_id})
if 'max_stamina_notification_enabled' in set_arg:
c.execute('''update user set max_stamina_notification_enabled = :a where user_id = :b''', {
'a': b2int(value), 'b': user_id})
conn.commit()
conn.close()
return None
def arc_add_friend(user_id, friend_id):
# 加好友，返回好友列表，或者是错误码602、604
if user_id == friend_id:
return 604
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select exists(select * from friend where user_id_me = :x and user_id_other = :y)''',
{'x': user_id, 'y': friend_id})
r = None
if c.fetchone() == (0,):
c.execute('''insert into friend values(:a, :b)''',
{'a': user_id, 'b': friend_id})
r = server.info.get_user_friend(c, user_id)
else:
return 602
conn.commit()
conn.close()
return r
def arc_delete_friend(user_id, friend_id):
# 删好友，返回好友列表
conn = sqlite3.connect('./database/arcaea_database.db')
c = conn.cursor()
c.execute('''select exists(select * from friend where user_id_me = :x and user_id_other = :y)''',
{'x': user_id, 'y': friend_id})
r = None
if c.fetchone() == (1,):
c.execute('''delete from friend where user_id_me = :x and user_id_other = :y''',
{'x': user_id, 'y': friend_id})
r = server.info.get_user_friend(c, user_id)
conn.commit()
conn.close()
return r