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

Ready to update to v2.5

Browse Source 
Please wait for some time. I need to change readme. I almost forgot what I changed before.
WARNING: There will be many bugs!
This commit is contained in:
Lost-MSth
2021-06-10 19:01:26 +08:00
parent 8a3cec05b7
commit 3fbf473de2
27 changed files with 1690 additions and 662 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
latest version/server/arcpurchase.py
View File

@@ -1,5 +1,6 @@
from server.sql import Connect
import server.info
import server.item
import server.character
import time
import json
@@ -12,28 +13,37 @@ def int2b(x):
return True
def get_item(c, type='pack'):
def get_purchase(c, type='pack'):
# 读取packs内容，返回字典列表
c.execute('''select * from item where type = :a''', {'a': type})
c.execute(
'''select * from purchase where purchase_name in (select purchase_name from purchase_item where type = :a)''', {'a': type})
x = c.fetchall()
if not x:
return []
re = []
for i in x:
r = {"name": i[0],
"items": [{
"type": i[1],
"id": i[0],
"is_available": int2b(i[2])
}],
"price": i[3],
"orig_price": i[4]}
items = []
c.execute(
'''select a.* from item a, purchase_item b where a.item_id=b.item_id and a.type=b.type and b.purchase_name=:name''', {'name': i[0]})
y = c.fetchall()
if y:
for j in y:
items.append({
"type": j[1],
"id": j[0],
"is_available": int2b(j[2])
})
if i[5] > 0:
r['discount_from'] = i[5]
if i[6] > 0:
r['discount_to'] = i[6]
r = {"name": i[0],
"items": items,
"price": i[1],
"orig_price": i[2]}
if i[3] > 0:
r['discount_from'] = i[3]
if i[4] > 0:
r['discount_to'] = i[4]
re.append(r)
@@ -44,7 +54,7 @@ def get_single_purchase():
# main里面没开数据库，这里写一下代替
re = []
with Connect() as c:
re = get_item(c, type='single')
re = get_purchase(c, type='single')
return re
@@ -68,7 +78,7 @@ def buy_item(c, user_id, price):
return True, ticket - price
def buy_thing(user_id, item_id, item_type):
def buy_thing(user_id, purchase_id):
# 购买物品接口，返回字典
success_flag = False
ticket = 0
@@ -77,36 +87,46 @@ def buy_thing(user_id, item_id, item_type):
characters = []
with Connect() as c:
c.execute('''select is_available, price, orig_price, discount_from, discount_to from item where item_id=:a and type=:b''',
{'a': item_id, 'b': item_type})
c.execute('''select price, orig_price, discount_from, discount_to from purchase where purchase_name=:a''',
{'a': purchase_id})
x = c.fetchone()
price = 0
flag = False
if x:
is_available = x[0]
price = x[1]
orig_price = x[2]
discount_from = x[3]
discount_to = x[4]
if not is_available:
return False
price = x[0]
orig_price = x[1]
discount_from = x[2]
discount_to = x[3]
else:
return {
"success": False,
"error_code": 501
}
c.execute(
'''select item_id, type from purchase_item where purchase_name=:a''', {'a': purchase_id})
x = c.fetchall()
if x:
now = int(time.time() * 1000)
if not(discount_from <= now <= discount_to):
price = orig_price
flag, ticket = buy_item(c, user_id, price)
if flag:
c.execute('''insert into user_item values(:a,:b,:c)''',
{'a': user_id, 'b': item_id, 'c': item_type})
if flag:
for i in x:
server.item.claim_user_item(c, user_id, i[0], i[1])
success_flag = True
success_flag = True
else:
return {
"success": False,
"error_code": 501
}
packs = server.info.get_user_packs(c, user_id)
singles = server.info.get_user_singles(c, user_id)
characters = server.info.get_user_characters(c, user_id)
packs = server.item.get_user_items(c, user_id, 'pack')
singles = server.item.get_user_items(c, user_id, 'single')
characters = server.character.get_user_characters(c, user_id)
return {
"success": success_flag,

208
latest version/server/arcscore.py
View File

@@ -46,14 +46,19 @@ def get_score(c, user_id, song_id, difficulty):
is_char_uncapped = int2b(y[3])
if y[4] != -1:
character = y[4]
c.execute('''select is_uncapped, is_uncapped_override from user_char where user_id=:a and character_id=:b''', {
'a': user_id, 'b': character})
if not Config.CHARACTER_FULL_UNLOCK:
c.execute('''select is_uncapped, is_uncapped_override from user_char where user_id=:a and character_id=:b''', {
'a': user_id, 'b': character})
else:
c.execute('''select is_uncapped, is_uncapped_override from user_char_full where user_id=:a and character_id=:b''', {
'a': user_id, 'b': character})
z = c.fetchone()
if z:
if z[1] == 0:
is_char_uncapped = int2b(z[0])
else:
is_char_uncapped = False
return {
"user_id": x[0],
"song_id": x[1],
@@ -273,7 +278,6 @@ def get_user_ptt_float(c, user_id) -> float:
'a': user_id})
x = c.fetchall()
if x != []:
n = len(x)
for i in x:
sumr += float(i[0])
c.execute('''select * from recent30 where user_id = :a''', {'a': user_id})
@@ -305,6 +309,54 @@ def get_user_ptt(c, user_id) -> int:
return int(get_user_ptt_float(c, user_id)*100)
def get_user_world_rank(c, user_id) -> int:
# 用户世界排名计算，同时返回排名值，如果超过设定最大值，返回0
with Connect('./database/arcsong.db') as c2:
c2.execute(
'''select sid, rating_ftr, rating_byn from songs''')
x = c2.fetchall()
if x:
song_list_ftr = [user_id]
song_list_byn = [user_id]
for i in x:
if i[1] > 0:
song_list_ftr.append(i[0])
if i[2] > 0:
song_list_byn.append(i[0])
if len(song_list_ftr) >= 2:
c.execute('''select sum(score) from best_score where user_id=? and difficulty=2 and song_id in ({0})'''.format(
','.join(['?']*(len(song_list_ftr)-1))), tuple(song_list_ftr))
x = c.fetchone()
if x[0] is not None:
score_sum = x[0]
else:
score_sum = 0
if len(song_list_byn) >= 2:
c.execute('''select sum(score) from best_score where user_id=? and difficulty=3 and song_id in ({0})'''.format(
','.join(['?']*(len(song_list_byn)-1))), tuple(song_list_byn))
x = c.fetchone()
if x[0] is not None:
score_sum += x[0]
else:
score_sum += 0
c.execute('''update user set world_rank_score = :b where user_id = :a''', {
'a': user_id, 'b': score_sum})
c.execute(
'''select count(*) from user where world_rank_score > ?''', (score_sum,))
x = c.fetchone()
if x and x[0] + 1 <= Config.WORLD_RANK_MAX:
return x[0] + 1
else:
return 0
def update_recent30(c, user_id, song_id, rating, is_protected):
# 刷新r30，这里的判断方法存疑，这里的song_id结尾包含难度数字
def insert_r30table(c, user_id, a, b):
@@ -433,133 +485,20 @@ def arc_score_post(user_id, song_id, difficulty, score, shiny_perfect_count, per
ptt = get_user_ptt(c, user_id)
c.execute('''update user set rating_ptt = :a where user_id = :b''', {
'a': ptt, 'b': user_id})
# 世界模式判断
c.execute('''select stamina_multiply,fragment_multiply,prog_boost_multiply from world_songplay where user_id=:a and song_id=:b and difficulty=:c''', {
'a': user_id, 'b': song_id, 'c': difficulty})
x = c.fetchone()
re = None
if x:
stamina_multiply = x[0]
fragment_multiply = x[1]
prog_boost_multiply = x[2]
step_times = stamina_multiply * fragment_multiply / \
100 * (prog_boost_multiply+100)/100
exp_times = stamina_multiply * (prog_boost_multiply+100)/100
if prog_boost_multiply != 0:
c.execute('''update user set prog_boost = 0 where user_id = :a''', {
'a': user_id})
c.execute('''delete from world_songplay where user_id=:a and song_id=:b and difficulty=:c''', {
'a': user_id, 'b': song_id, 'c': difficulty})
c.execute('''select character_id,frag,prog,overdrive from user_char where user_id = :a and character_id = (select character_id from user where user_id=:a)''', {
'a': user_id})
y = c.fetchone()
if y:
character_id = y[0]
flag = float(y[1])
prog = float(y[2])
overdrive = float(y[3])
else:
character_id = 0
flag = 0
prog = 0
overdrive = 0
re = server.arcworld.world_update(
c, user_id, song_id, difficulty, rating, clear_type, beyond_gauge, x[0], x[1], x[2])
re['global_rank'] = get_user_world_rank(c, user_id) # 更新世界排名
re["user_rating"] = ptt
else:
re = {'global_rank': get_user_world_rank(
c, user_id), 'user_rating': ptt}
c.execute('''select current_map from user where user_id = :a''', {
'a': user_id})
map_id = c.fetchone()[0]
if beyond_gauge == 0: # 是否是beyond挑战
base_step = 2.5 + 2.45*rating**0.5
step = base_step * (prog/50) * step_times
else:
info = server.arcworld.get_world_info(map_id)
if clear_type == 0:
base_step = 8/9 + (rating/1.3)**0.5
else:
base_step = 8/3 + (rating/1.3)**0.5
if character_id in info['character_affinity']:
affinity_multiplier = info['affinity_multiplier'][info['character_affinity'].index(
character_id)]
else:
affinity_multiplier = 1
step = base_step * (prog/50) * step_times * affinity_multiplier
c.execute('''select * from user_world where user_id = :a and map_id =:b''',
{'a': user_id, 'b': map_id})
y = c.fetchone()
rewards, steps, curr_position, curr_capture, info = server.arcworld.climb_step(
user_id, map_id, step, y[3], y[2])
if beyond_gauge == 0:
re = {
"rewards": rewards,
"exp": 25000,
"level": 30,
"base_progress": base_step,
"progress": step,
"user_map": {
"user_id": user_id,
"curr_position": curr_position,
"curr_capture": curr_capture,
"is_locked": int2b(y[4]),
"map_id": map_id,
"prev_capture": y[3],
"prev_position": y[2],
"beyond_health": info['beyond_health'],
"steps": steps
},
"char_stats": {
"character_id": character_id,
"frag": flag,
"prog": prog,
"overdrive": overdrive
},
"current_stamina": 12,
"max_stamina_ts": 1586274871917,
"user_rating": ptt
}
else:
re = {
"rewards": rewards,
"exp": 25000,
"level": 30,
"base_progress": base_step,
"progress": step,
"user_map": {
"user_id": user_id,
"curr_position": curr_position,
"curr_capture": curr_capture,
"is_locked": int2b(y[4]),
"map_id": map_id,
"prev_capture": y[3],
"prev_position": y[2],
"beyond_health": info['beyond_health'],
"step_count": len(steps)
},
"char_stats": {
"character_id": character_id,
"frag": flag,
"prog": prog,
"overdrive": overdrive
},
"current_stamina": 12,
"max_stamina_ts": 1586274871917,
"user_rating": ptt
}
if stamina_multiply != 1:
re['stamina_multiply'] = stamina_multiply
if fragment_multiply != 100:
re['fragment_multiply'] = fragment_multiply
if prog_boost_multiply != 0:
re['prog_boost_multiply'] = prog_boost_multiply
if curr_position == info['step_count']-1 and info['is_repeatable']: # 循环图判断
curr_position = 0
c.execute('''update user_world set curr_position=:a, curr_capture=:b where user_id=:c and map_id=:d''', {
'a': curr_position, 'b': curr_capture, 'c': user_id, 'd': map_id})
return ptt, re
@@ -597,16 +536,14 @@ def arc_score_check(user_id, song_id, difficulty, score, shiny_perfect_count, pe
def refresh_all_score_rating():
# 刷新所有best成绩的rating
error = 'Unknown error.'
song_data = []
with Connect('./database/arcsong.db') as c:
with Connect() as c2:
c.execute(
'''select sid, rating_pst, rating_prs, rating_ftr, rating_byn from songs''')
x = c.fetchall()
c.execute(
'''select sid, rating_pst, rating_prs, rating_ftr, rating_byn from songs''')
x = c.fetchall()
if x:
song_list = [i[0] for i in song_data]
song_list = [i[0] for i in x]
with Connect() as c:
c.execute('''update best_score set rating=0 where song_id not in ({0})'''.format(
','.join(['?']*len(song_list))), tuple(song_list))
@@ -1532,7 +1469,7 @@ def arc_all_get(user_id):
"complete": 1
}, {
"unlock_key": "aegleseeker|1|103",
"complete": 103
"complete": 1
}, {
"unlock_key": "amazingmightyyyy|2|0",
"complete": 1
@@ -1586,7 +1523,7 @@ def arc_all_get(user_id):
"complete": 1
}, {
"unlock_key": "aegleseeker|2|103",
"complete": 103
"complete": 1
}, {
"unlock_key": "tiferet|2|0",
"complete": 1
@@ -1635,7 +1572,12 @@ def arc_all_get(user_id):
}, {
"unlock_key": "nirvluce|2|0",
"complete": 1
}]
},
{
"unlock_key": "aegleseeker_challenge|0|102",
"complete": 1
}
]
return {
"user_id": user_id,

156
latest version/server/arcworld.py
View File

@@ -1,5 +1,8 @@
import json
from server.sql import Connect
from setting import Config
import server.item
import server.character
import os
@@ -152,6 +155,7 @@ def play_world_song(user_id, args):
def climb_step(user_id, map_id, step, prev_capture, prev_position):
# 爬梯子，返回奖励列表，台阶列表，当前的位置和坐标，图信息
info = get_world_info(map_id)
step_count = int(info['step_count'])
@@ -245,3 +249,155 @@ def climb_step(user_id, map_id, step, prev_capture, prev_position):
steps.append(x)
return rewards, steps, curr_position, curr_capture, info
def world_update(c, user_id, song_id, difficulty, rating, clear_type, beyond_gauge, stamina_multiply=1, fragment_multiply=100, prog_boost_multiply=0):
# 成绩上传后世界模式更新，返回字典
step_times = stamina_multiply * fragment_multiply / \
100 * (prog_boost_multiply+100)/100
exp_times = stamina_multiply * (prog_boost_multiply+100)/100
if prog_boost_multiply != 0:
c.execute('''update user set prog_boost = 0 where user_id = :a''', {
'a': user_id})
c.execute('''delete from world_songplay where user_id=:a and song_id=:b and difficulty=:c''', {
'a': user_id, 'b': song_id, 'c': difficulty})
c.execute('''select character_id from user where user_id=?''', (user_id,))
x = c.fetchone()
character_id = x[0] if x else 0
c.execute('''select frag1,prog1,overdrive1,frag20,prog20,overdrive20,frag30,prog30,overdrive30 from character where character_id=?''', (character_id,))
x = c.fetchone()
if Config.CHARACTER_FULL_UNLOCK:
c.execute('''select level, exp from user_char_full where user_id = :a and character_id = :b''', {
'a': user_id, 'b': character_id})
else:
c.execute('''select level, exp from user_char where user_id = :a and character_id = :b''', {
'a': user_id, 'b': character_id})
y = c.fetchone()
if y:
level = y[0]
exp = y[1]
else:
level = 1
exp = 0
if x:
flag = server.character.calc_char_value(level, x[0], x[3], x[6])
prog = server.character.calc_char_value(level, x[1], x[4], x[7])
overdrive = server.character.calc_char_value(level, x[2], x[5], x[8])
else:
flag = 0
prog = 0
overdrive = 0
c.execute('''select current_map from user where user_id = :a''', {
'a': user_id})
map_id = c.fetchone()[0]
if beyond_gauge == 0: # 是否是beyond挑战
base_step = 2.5 + 2.45*rating**0.5
step = base_step * (prog/50) * step_times
else:
info = get_world_info(map_id)
if clear_type == 0:
base_step = 8/9 + (rating/1.3)**0.5
else:
base_step = 8/3 + (rating/1.3)**0.5
if character_id in info['character_affinity']:
affinity_multiplier = info['affinity_multiplier'][info['character_affinity'].index(
character_id)]
else:
affinity_multiplier = 1
step = base_step * (prog/50) * step_times * affinity_multiplier
c.execute('''select * from user_world where user_id = :a and map_id =:b''',
{'a': user_id, 'b': map_id})
y = c.fetchone()
rewards, steps, curr_position, curr_capture, info = climb_step(
user_id, map_id, step, y[3], y[2])
for i in rewards: # 物品分发
for j in i['items']:
amount = j['amount'] if 'amount' in j else 1
item_id = j['id'] if 'id' in j else ''
server.item.claim_user_item(c, user_id, item_id, j['type'], amount)
# 角色升级
if not Config.CHARACTER_FULL_UNLOCK:
exp, level = server.character.calc_level_up(
c, user_id, character_id, exp, exp_times*rating*6)
c.execute('''update user_char set level=?, exp=? where user_id=? and character_id=?''',
(level, exp, user_id, character_id))
else:
exp = server.character.LEVEL_STEPS[level]
if beyond_gauge == 0:
re = {
"rewards": rewards,
"exp": exp,
"level": level,
"base_progress": base_step,
"progress": step,
"user_map": {
"user_id": user_id,
"curr_position": curr_position,
"curr_capture": curr_capture,
"is_locked": int2b(y[4]),
"map_id": map_id,
"prev_capture": y[3],
"prev_position": y[2],
"beyond_health": info['beyond_health'],
"steps": steps
},
"char_stats": {
"character_id": character_id,
"frag": flag,
"prog": prog,
"overdrive": overdrive
},
"current_stamina": 12,
"max_stamina_ts": 1586274871917
}
else:
re = {
"rewards": rewards,
"exp": exp,
"level": level,
"base_progress": base_step,
"progress": step,
"user_map": {
"user_id": user_id,
"curr_position": curr_position,
"curr_capture": curr_capture,
"is_locked": int2b(y[4]),
"map_id": map_id,
"prev_capture": y[3],
"prev_position": y[2],
"beyond_health": info['beyond_health'],
"step_count": len(steps)
},
"char_stats": {
"character_id": character_id,
"frag": flag,
"prog": prog,
"overdrive": overdrive
},
"current_stamina": 12,
"max_stamina_ts": 1586274871917
}
if stamina_multiply != 1:
re['stamina_multiply'] = stamina_multiply
if fragment_multiply != 100:
re['fragment_multiply'] = fragment_multiply
if prog_boost_multiply != 0:
re['prog_boost_multiply'] = prog_boost_multiply
if curr_position == info['step_count']-1 and info['is_repeatable']: # 循环图判断
curr_position = 0
c.execute('''update user_world set curr_position=:a, curr_capture=:b where user_id=:c and map_id=:d''', {
'a': curr_position, 'b': curr_capture, 'c': user_id, 'd': map_id})
return re

15
latest version/server/auth.py
View File

@@ -91,13 +91,18 @@ def arc_register(name: str, password: str, device_id: str, email: str, ip: str):
return 2000001
def insert_user_char(c, user_id):
# 为用户添加所有可用角色
c.execute('''select * from character''')
# 为用户添加初始角色
c.execute('''insert into user_char values(?,?,?,?,?,?)''',
(user_id, 0, 1, 0, 0, 0))
c.execute('''insert into user_char values(?,?,?,?,?,?)''',
(user_id, 1, 1, 0, 0, 0))
c.execute('''select character_id, max_level, is_uncapped from character''')
x = c.fetchall()
if x != []:
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]})
exp = 25000 if i[1] == 30 else 10000
c.execute('''insert into user_char_full values(?,?,?,?,?,?)''',
(user_id, i[0], i[1], exp, i[2], 0))
user_id = None
token = None

255
latest version/server/character.py Normal file
View File

@@ -0,0 +1,255 @@
from setting import Config
from server.sql import Connect
import server.item
import server.setme
LEVEL_STEPS = {1: 0, 2: 50, 3: 100, 4: 150, 5: 200, 6: 300, 7: 450, 8: 650, 9: 900, 10: 1200, 11: 1600, 12: 2100, 13: 2700, 14: 3400, 15: 4200, 16: 5100,
17: 6100, 18: 7200, 19: 8500, 20: 10000, 21: 11500, 22: 13000, 23: 14500, 24: 16000, 25: 17500, 26: 19000, 27: 20500, 28: 22000, 29: 23500, 30: 25000}
def int2b(x):
# int与布尔值转换
if x is None or x == 0:
return False
else:
return True
def get_level_steps():
# 返回level_steps字典数组
return [{'level': i, 'level_exp': LEVEL_STEPS[i]} for i in LEVEL_STEPS]
def calc_char_value(level, value1, value20, value30):
# 计算搭档数值的核心函数，返回浮点数
def calc_char_value_20(level, stata, statb, lva=1, lvb=20):
# 计算1~20级搭档数值的核心函数，返回浮点数，来自https://redive.estertion.win/arcaea/calc/
n = [0, 0, 0.0005831753900000081, 0.004665403120000065, 0.015745735529959858, 0.03732322495992008, 0.07289692374980007, 0.12596588423968, 0.2000291587694801, 0.29858579967923987, 0.42513485930893946,
0.5748651406910605, 0.7014142003207574, 0.7999708412305152, 0.8740341157603029, 0.9271030762501818, 0.962676775040091, 0.9842542644700301, 0.9953345968799998, 0.9994168246100001, 1]
e = n[lva] - n[lvb]
a = stata - statb
r = a / e
d = stata - n[lva] * r
return d + r * n[level]
def calc_char_value_30(level, stata, statb, lva=20, lvb=30):
# 计算21~30级搭档数值，返回浮点数
return (level - lva) * (statb - stata) / (lvb - lva) + stata
if level < 1 or level > 30:
return 0
elif 1 <= level <= 20:
return calc_char_value_20(level, value1, value20)
else:
return calc_char_value_30(level, value20, value30)
def get_char_core(c, character_id):
# 得到对应角色觉醒所需的核心，返回字典列表
r = []
c.execute(
'''select item_id, amount from char_item where character_id = ? and type="core"''', (character_id,))
x = c.fetchall()
if x:
for i in x:
r.append({'core_type': i[0], 'amount': i[1]})
return r
def get_user_characters(c, user_id):
# 获取用户所拥有角色，返回列表
c.execute('''select character_id from user_char where user_id = :user_id''',
{'user_id': user_id})
x = c.fetchall()
characters = []
if x:
for i in x:
characters.append(i[0])
return characters
def get_user_character(c, user_id):
# 得到用户拥有的角色列表，返回列表
if Config.CHARACTER_FULL_UNLOCK:
c.execute('''select * from user_char_full a,character b where a.user_id = :user_id and a.character_id=b.character_id''',
{'user_id': user_id})
else:
c.execute('''select * from user_char a,character b where a.user_id = :user_id and a.character_id=b.character_id''',
{'user_id': user_id})
x = c.fetchall()
if not x and not Config.CHARACTER_FULL_UNLOCK:
# 添加初始角色
c.execute('''insert into user_char values(?,?,?,?,?,?)''',
(user_id, 0, 1, 0, 0, 0))
c.execute('''insert into user_char values(?,?,?,?,?,?)''',
(user_id, 1, 1, 0, 0, 0))
c.execute('''select * from user_char a,character b where a.user_id = :user_id and a.character_id=b.character_id''',
{'user_id': user_id})
x = c.fetchall()
if not x:
return []
r = []
for i in x:
char = {
"is_uncapped_override": int2b(i[5]),
"is_uncapped": int2b(i[4]),
"uncap_cores": get_char_core(c, i[1]),
"char_type": i[22],
"skill_id_uncap": i[21],
"skill_requires_uncap": int2b(i[20]),
"skill_unlock_level": i[19],
"skill_id": i[18],
"overdrive": calc_char_value(i[2], i[11], i[14], i[17]),
"prog": calc_char_value(i[2], i[10], i[13], i[16]),
"frag": calc_char_value(i[2], i[9], i[12], i[15]),
"level_exp": LEVEL_STEPS[i[2]],
"exp": i[3],
"level": i[2],
"name": i[7],
"character_id": i[1]
}
if i[1] == 21:
char["voice"] = [0, 1, 2, 3, 100, 1000, 1001]
r.append(char)
return r
def get_one_character(c, user_id, character_id):
# 得到用户某个拥有的角色列表，返回字典
if Config.CHARACTER_FULL_UNLOCK:
c.execute('''select * from user_char_full a,character b where a.user_id = :user_id and a.character_id=b.character_id and a.character_id=:a''',
{'user_id': user_id, 'a': character_id})
else:
c.execute('''select * from user_char a,character b where a.user_id = :user_id and a.character_id=b.character_id and a.character_id=:a''',
{'user_id': user_id, 'a': character_id})
x = c.fetchone()
if not x:
return {}
r = {
"is_uncapped_override": int2b(x[5]),
"is_uncapped": int2b(x[4]),
"uncap_cores": get_char_core(c, x[1]),
"char_type": x[22],
"skill_id_uncap": x[21],
"skill_requires_uncap": int2b(x[20]),
"skill_unlock_level": x[19],
"skill_id": x[18],
"overdrive": calc_char_value(x[2], x[11], x[14], x[17]),
"prog": calc_char_value(x[2], x[10], x[13], x[16]),
"frag": calc_char_value(x[2], x[9], x[12], x[15]),
"level_exp": LEVEL_STEPS[x[2]],
"exp": x[3],
"level": x[2],
"name": x[7],
"character_id": x[1]
}
if x[1] == 21:
r["voice"] = [0, 1, 2, 3, 100, 1000, 1001]
return r
def calc_level_up(c, user_id, character_id, exp, exp_addition):
# 计算角色升级，返回当前经验和等级
exp += exp_addition
if exp >= LEVEL_STEPS[20]: # 未觉醒溢出
c.execute('''select is_uncapped from user_char where user_id=? and character_id=?''',
(user_id, character_id))
x = c.fetchone()
if x and x[0] == 0:
return LEVEL_STEPS[20], 20
a = []
b = []
for i in LEVEL_STEPS:
a.append(i)
b.append(LEVEL_STEPS[i])
if exp >= b[-1]: # 溢出
return b[-1], a[-1]
if exp < b[0]: # 向下溢出，是异常状态
return 0, 1
i = len(a) - 1
while exp < b[i]:
i -= 1
return exp, a[i]
def char_use_core(user_id, character_id, amount):
# 以太之滴升级，返回user_id，core状态，角色状态的字典
r = None
with Connect() as c:
c.execute(
'''select amount from user_item where user_id=? and item_id="core_generic" and type="core"''', (user_id,))
x = c.fetchone()
if x:
pre_amount = x[0]
else:
pre_amount = 0
if amount <= pre_amount:
c.execute(
'''select exp from user_char where user_id=? and character_id=?''', (user_id, character_id))
x = c.fetchone()
if x:
exp, level = calc_level_up(
c, user_id, character_id, x[0], amount*250)
c.execute('''update user_char set level=?, exp=? where user_id=? and character_id=?''',
(level, exp, user_id, character_id))
server.item.claim_user_item(
c, user_id, 'core_generic', 'core', -amount)
r = {'character': [get_one_character(c, user_id, character_id)]}
r['cores'] = server.item.get_user_cores(c, user_id)
r['user_id'] = user_id
return r
def char_uncap(user_id, character_id):
# 角色觉醒，返回user_id，core状态，角色状态的字典
r = None
with Connect() as c:
c.execute('''select * from char_item where character_id=?''',
(character_id,))
x = c.fetchall()
if not x:
return None
success = True
for i in x:
c.execute(
'''select amount from user_item where user_id=? and item_id=? and type=?''', (user_id, i[1], i[2]))
y = c.fetchone()
if not y or i[3] > y[0]:
success = False
break
if success:
for i in x:
server.item.claim_user_item(c, user_id, i[1], i[2], -i[3])
c.execute('''update user_char set is_uncapped=1, is_uncapped_override=0 where user_id=? and character_id=?''',
(user_id, character_id))
r = {'character': [get_one_character(c, user_id, character_id)]}
r['cores'] = server.item.get_user_cores(c, user_id)
r['user_id'] = user_id
return r

248
latest version/server/info.py
View File

@@ -2,6 +2,8 @@ from server.sql import Connect
import server.arcworld
import server.arcpurchase
import server.arcdownload
import server.character
import server.item
import time
from setting import Config
@@ -46,47 +48,6 @@ def get_recent_score(c, user_id):
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]
char = {
"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]
}
if i[1] == 21:
char["voice"] = [0, 1, 2, 3, 100, 1000, 1001]
s.append(char)
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''', {
@@ -111,8 +72,12 @@ def get_user_friend(c, user_id):
is_char_uncapped_override = int2b(y[9])
if y[23] != -1:
character = y[23]
c.execute('''select is_uncapped, is_uncapped_override from user_char where user_id=:a and character_id=:b''', {
'a': i[0], 'b': character})
if not Config.CHARACTER_FULL_UNLOCK:
c.execute('''select is_uncapped, is_uncapped_override from user_char where user_id=:a and character_id=:b''', {
'a': i[0], 'b': character})
else:
c.execute('''select is_uncapped, is_uncapped_override from user_char_full where user_id=:a and character_id=:b''', {
'a': i[0], 'b': character})
z = c.fetchone()
if z:
is_char_uncapped = int2b(z[0])
@@ -138,62 +103,44 @@ def get_user_friend(c, user_id):
return s
def get_user_singles(c, user_id):
# 得到用户的单曲，返回列表
c.execute('''select * from user_item where user_id = :user_id and type = "single"''',
{'user_id': user_id})
x = c.fetchall()
if not x:
return []
re = []
for i in x:
re.append(i[1])
return re
def get_user_packs(c, user_id):
# 得到用户的曲包，返回列表
c.execute('''select * from user_item where user_id = :user_id and type = "pack"''',
{'user_id': user_id})
x = c.fetchall()
if not x:
return []
re = []
for i in x:
re.append(i[1])
return re
def get_user_characters(c, user_id):
# 获取用户所拥有角色，返回列表
c.execute('''select character_id from user_char where user_id = :user_id''',
{'user_id': user_id})
x = c.fetchall()
characters = []
if x:
for i in x:
characters.append(i[0])
return characters
def get_user_me(c, user_id):
# 构造user/me的数据，返回字典
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)
user_character = server.character.get_user_character(c, user_id)
# 下面没有使用get_user_characters函数是为了节省一次查询
characters = []
for i in user_character:
characters.append(i['character_id'])
character_id = x[6]
if character_id not in characters:
character_id = 0
c.execute(
'''update user set character_id=0 where user_id=?''', (user_id,))
favorite_character_id = x[23]
if favorite_character_id not in characters:
favorite_character_id = -1
c.execute(
'''update user set favorite_character=-1 where user_id=?''', (user_id,))
c.execute(
'''select world_rank_score from user where user_id=?''', (user_id,))
y = c.fetchone()
if y and y[0]:
c.execute(
'''select count(*) from user where world_rank_score > ?''', (y[0],))
y = c.fetchone()
if y and y[0] + 1 <= Config.WORLD_RANK_MAX:
world_rank = y[0] + 1
else:
world_rank = 0
else:
world_rank = 0
prog_boost = 0
if x[27] and x[27] != 0:
prog_boost = 300
@@ -203,7 +150,7 @@ def get_user_me(c, user_id):
"character_stats": user_character,
"friends": get_user_friend(c, user_id),
"settings": {
"favorite_character": x[23],
"favorite_character": favorite_character_id,
"is_hide_rating": int2b(x[10]),
"max_stamina_notification_enabled": int2b(x[24])
},
@@ -212,7 +159,7 @@ def get_user_me(c, user_id):
"user_code": x[4],
"display_name": x[1],
"ticket": x[26],
"character": x[6],
"character": character_id,
"is_locked_name_duplicate": False,
"is_skill_sealed": int2b(x[7]),
"current_map": x[25],
@@ -220,17 +167,17 @@ def get_user_me(c, user_id):
"next_fragstam_ts": -1,
"max_stamina_ts": 1586274871917,
"stamina": 12,
"world_unlocks": ["scenery_chap1", "scenery_chap2", "scenery_chap3", "scenery_chap4", "scenery_chap5"],
"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", "solitarydream", "lumia3", "purpleverse", "moonheart3", "glow", "enchantedlove", "take", "lifeispiano", "vandalism", "nexttoyou3", "lostcivilization3"],
"singles": get_user_singles(c, user_id),
"packs": get_user_packs(c, user_id),
"world_unlocks": server.item.get_user_items(c, user_id, 'world_unlock'),
"world_songs": server.item.get_user_items(c, user_id, 'world_song'),
"singles": server.item.get_user_items(c, user_id, 'single'),
"packs": server.item.get_user_items(c, user_id, 'pack'),
"characters": characters,
"cores": [],
"cores": server.item.get_user_cores(c, user_id),
"recent_score": get_recent_score(c, user_id),
"max_friend": 50,
"rating": x[5],
"join_date": int(x[3]),
"global_rank": 114514
"global_rank": world_rank
}
return r
@@ -253,16 +200,23 @@ def arc_aggregate_big(user_id):
# 返回比较全的用户数据
r = {"success": False}
with Connect() as c:
# 防止数据库锁
id_2 = server.arcdownload.get_all_songs(user_id, url_flag=False)
id_5 = {
"current_map": server.arcworld.get_current_map(user_id),
"user_id": user_id,
"maps": server.arcworld.get_world_all(user_id)
}
r = {"success": True,
"value": [{
"id": 0,
"value": get_user_me(c, user_id)
}, {
"id": 1,
"value": server.arcpurchase.get_item(c, 'pack')
"value": server.arcpurchase.get_purchase(c, 'pack')
}, {
"id": 2,
"value": server.arcdownload.get_all_songs(user_id, url_flag=False)
"value": id_2
}, {
"id": 3,
"value": {
@@ -270,97 +224,7 @@ def arc_aggregate_big(user_id):
"stamina_recover_tick": 1800000,
"core_exp": 250,
"curr_ts": int(time.time())*1000,
"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
}],
"level_steps": server.character.get_level_steps(),
"world_ranking_enabled": False,
"is_byd_chapter_unlocked": True
}
@@ -369,11 +233,7 @@ def arc_aggregate_big(user_id):
"value": server.arcpurchase.get_user_present(c, user_id)
}, {
"id": 5,
"value": {
"current_map": server.arcworld.get_current_map(user_id),
"user_id": user_id,
"maps": server.arcworld.get_world_all(user_id)
}
"value": id_5
}
]}

106
latest version/server/item.py Normal file
View File

@@ -0,0 +1,106 @@
from setting import Config
def get_user_items(c, user_id, item_type):
# 得到用户的物品，返回列表，不包含数量信息
if item_type == 'world_song' and Config.WORLD_SONG_FULL_UNLOCK or item_type == 'world_unlock' and Config.WORLD_SCENERY_FULL_UNLOCK:
c.execute('''select item_id from item where type=?''', (item_type,))
else:
c.execute('''select item_id from user_item where user_id = :user_id and type = :item_type''',
{'user_id': user_id, 'item_type': item_type})
x = c.fetchall()
if not x:
return []
re = []
for i in x:
re.append(i[0])
return re
def get_user_cores(c, user_id):
# 得到用户的core，返回字典列表
r = []
c.execute(
'''select item_id, amount from user_item where user_id = ? and type="core"''', (user_id,))
x = c.fetchall()
if x:
for i in x:
if i[1]:
amount = i[1]
else:
amount = 0
r.append({'core_type': i[0], 'amount': amount})
return r
def claim_user_item(c, user_id, item_id, item_type, amount=1):
# 处理用户物品，包括添加和删除操作，返回成功与否布尔值
# 这个接口允许memory变成负数，所以不能用来购买
try:
amount = int(amount)
except:
return False
if item_type not in ['memory', 'character']:
c.execute('''select is_available from item where item_id=? and type=?''',
(item_id, item_type))
x = c.fetchone()
if x:
if x[0] == 0:
return False
else:
return False
if item_type == 'core':
c.execute(
'''select amount from user_item where user_id=? and item_id=? and type="core"''', (user_id, item_id))
x = c.fetchone()
if x:
if x[0] + amount < 0: # 数量不足
return False
c.execute('''update user_item set amount=? where user_id=? and item_id=? and type="core"''',
(x[0]+amount, user_id, item_id))
else:
if amount < 0: # 添加数量错误
return False
c.execute('''insert into user_item values(?,?,"core",?)''',
(user_id, item_id, amount))
elif item_type == 'memory':
c.execute('''select ticket from user where user_id=?''', (user_id,))
x = c.fetchone()
if x is not None:
c.execute('''update user set ticket=? where user_id=?''',
(x[0]+amount, user_id))
else:
return False
elif item_type == 'character':
if not item_id.isdigit():
c.execute(
'''select character_id from character where name=?''', (item_id,))
x = c.fetchone()
if x:
character_id = x[0]
else:
return False
else:
character_id = int(item_id)
c.execute(
'''select exists(select * from user_char where user_id=? and character_id=?)''', (user_id, character_id))
if c.fetchone() == (0,):
c.execute('''insert into user_char values(?,?,1,0,0,0)''',
(user_id, character_id))
elif item_type in ['world_song', 'world_unlock', 'single', 'pack']:
c.execute('''select exists(select * from user_item where user_id=? and item_id=? and type=?)''',
(user_id, item_id, item_type))
if c.fetchone() == (0,):
c.execute('''insert into user_item values(:a,:b,:c,1)''',
{'a': user_id, 'b': item_id, 'c': item_type})
return True

85
latest version/server/setme.py
View File

@@ -1,5 +1,7 @@
from server.sql import Connect
from setting import Config
import server.info
import server.character
def b2int(x):
@@ -23,18 +25,27 @@ def change_char(user_id, character_id, skill_sealed):
re = False
with Connect() as c:
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})
if Config.CHARACTER_FULL_UNLOCK:
c.execute('''select is_uncapped, is_uncapped_override from user_char_full where user_id = :a and character_id = :b''',
{'a': user_id, 'b': character_id})
else:
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})
if x:
is_uncapped = x[0]
is_uncapped_override = x[1]
else:
return False
re = True
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': is_uncapped, 'd': is_uncapped_override, 'e': user_id})
re = True
return re
@@ -43,40 +54,46 @@ def change_char_uncap(user_id, character_id):
# 角色觉醒改变，返回字典
r = None
with Connect() as c:
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})
if not Config.CHARACTER_FULL_UNLOCK:
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})
else:
c.execute('''select is_uncapped, is_uncapped_override from user_char_full where user_id = :a and character_id = :b''',
{'a': user_id, 'b': character_id})
x = c.fetchone()
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})
if not Config.CHARACTER_FULL_UNLOCK:
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 a,character b where a.user_id=? and a.character_id=b.character_id and a.character_id=?''', (user_id, character_id))
else:
c.execute('''update user_char_full 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_full a,character b where a.user_id=? and a.character_id=b.character_id and a.character_id=?''', (user_id, 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],
"is_uncapped_override": int2b(y[5]),
"is_uncapped": int2b(y[4]),
"uncap_cores": server.character.get_char_core(c, y[1]),
"char_type": y[22],
"skill_id_uncap": y[21],
"skill_requires_uncap": int2b(y[20]),
"skill_unlock_level": y[19],
"skill_id": y[18],
"overdrive": server.character.calc_char_value(y[2], y[11], y[14], y[17]),
"prog": server.character.calc_char_value(y[2], y[10], y[13], y[16]),
"frag": server.character.calc_char_value(y[2], y[9], y[12], y[15]),
"level_exp": server.character.LEVEL_STEPS[y[2]],
"exp": y[3],
"level": y[2],
"name": char_name,
"name": y[7],
"character_id": y[1]
}