[Enhance] [Refactor] Add API about characters

- Add API endpoints about characters
- Some small changes in refactoring

latest version/api/__init__.py

@@ -1,7 +1,7 @@
 from flask import Blueprint
 
 from . import (users, songs, token, system, items,
-               purchases, presents, redeems)
+               purchases, presents, redeems, characters)
 
 bp = Blueprint('api', __name__, url_prefix='/api/v1')
 bp.register_blueprint(users.bp)
@@ -12,3 +12,4 @@ bp.register_blueprint(items.bp)
 bp.register_blueprint(purchases.bp)
 bp.register_blueprint(presents.bp)
 bp.register_blueprint(redeems.bp)
+bp.register_blueprint(characters.bp)

latest version/api/api_auth.py

@@ -52,11 +52,12 @@ def role_required(request, powers=[]):
 
 def request_json_handle(request, required_keys: list = [], optional_keys: list = [], must_change: bool = False, is_batch: bool = False):
     '''
-        提取post参数，返回dict，写成了修饰器\ 
-        parameters: \ 
-        `request`: `Request` - 当前请求\ 
-        `required_keys`: `list` - 必须的参数\ 
-        `optional_keys`: `list` - 可选的参数\ 
+        提取post参数，返回dict，写成了修饰器
+
+        parameters:
+        `request`: `Request` - 当前请求
+        `required_keys`: `list` - 必须的参数
+        `optional_keys`: `list` - 可选的参数
         `must_change`: `bool` - 当全都是可选参数时，是否必须有至少一项修改
     '''
 
@@ -113,8 +114,7 @@ def api_try(view):
             data = view(*args, **kwargs)
             if data is None:
                 return error_return()
-            else:
-                return data
+            return data
         except ArcError as e:
             if Config.ALLOW_WARNING_LOG:
                 current_app.logger.warning(format_exc())

latest version/api/api_code.py

@@ -23,6 +23,8 @@ CODE_MSG = {
     -122: 'Item already exists',
     -123: 'The collection already has this item',
     -124: 'The collection does not have this item',
+    -130: 'No such character',
+    -131: 'Invalid skill ID',
     -200: 'No permission',  # 2xx用户相关错误
     -201: 'Wrong username or password',
     -202: 'User is banned',

latest version/api/characters.py

@@ -0,0 +1,134 @@
+from flask import Blueprint, request
+
+from core.error import InputError, NoData
+from core.item import ItemFactory
+from core.character import Character
+from core.sql import Connect, Query, Sql
+
+from .api_auth import api_try, request_json_handle, role_required
+from .api_code import success_return
+from .constant import Constant
+
+
+bp = Blueprint('characters', __name__, url_prefix='/characters')
+
+
+@bp.route('', methods=['GET'])
+@role_required(request, ['select'])
+@request_json_handle(request, optional_keys=Constant.QUERY_KEYS)
+@api_try
+def characters_get(data, user):
+    '''查询全角色信息'''
+    A = ['character_id', 'name', 'skill_id',
+         'skill_id_uncap', 'char_type', 'is_uncapped']
+    B = ['name', 'skill_id', 'skill_id_uncap']
+    C = ['name', 'frag1', 'prog1', 'overdrive1', 'frag20',
+         'prog20', 'overdrive20', 'frag30', 'prog30', 'overdrive30']
+    with Connect() as c:
+        query = Query(A, B, C).from_dict(data)
+        x = Sql(c).select('character', query=query)
+        r = [Character().from_list(i) for i in x]
+
+        if not r:
+            raise NoData(api_error_code=-2)
+
+        return success_return([x.to_dict() for x in r])
+
+
+@bp.route('/<int:character_id>', methods=['GET'])
+@role_required(request, ['select'])
+@api_try
+def characters_character_get(user, character_id: int):
+    # 包含core
+    with Connect() as c:
+        c = Character(c).select(character_id)
+        c.select_character_core()
+        return success_return(c.to_dict(has_cores=True))
+
+
+@bp.route('/<int:character_id>', methods=['PUT'])
+@role_required(request, ['change'])
+@request_json_handle(request, optional_keys=['max_level', 'skill_id', 'skill_id_uncap', 'skill_unlock_level', 'skill_requires_uncap', 'char_type', 'is_uncapped', 'frag1', 'prog1', 'overdrive1', 'frag20', 'prog20', 'overdrive20', 'frag30', 'prog30', 'overdrive30'], must_change=True)
+@api_try
+def characters_character_put(data, user, character_id: int):
+    '''修改角色信息'''
+    if ('skill_id' in data and data['skill_id'] != '' and data['skill_id'] not in Constant.SKILL_IDS) or ('skill_id_uncap' in data and data['skill_id_uncap'] != '' and data['skill_id_uncap'] not in Constant.SKILL_IDS):
+        raise InputError('Invalid skill_id', api_error_code=-131)
+    with Connect() as c:
+        c = Character(c).select(character_id)
+        try:
+            if 'max_level' in data:
+                c.max_level = int(data['max_level'])
+            if 'skill_id' in data:
+                c.skill_id = data['skill_id']
+            if 'skill_id_uncap' in data:
+                c.skill_id_uncap = data['skill_id_uncap']
+            if 'skill_unlock_level' in data:
+                c.skill_unlock_level = int(data['skill_unlock_level'])
+            if 'skill_requires_uncap' in data:
+                c.skill_requires_uncap = data['skill_requires_uncap'] == 1
+            if 'char_type' in data:
+                c.char_type = int(data['char_type'])
+            if 'is_uncapped' in data:
+                c.is_uncapped = data['is_uncapped'] == 1
+            t = ['frag1', 'prog1', 'overdrive1', 'frag20', 'prog20',
+                 'overdrive20', 'frag30', 'prog30', 'overdrive30']
+            for i in t:
+                if i not in data:
+                    continue
+                if i.endswith('1'):
+                    x = getattr(c, i[:-1])
+                    x.start = float(data[i])
+                elif i.endswith('20'):
+                    x = getattr(c, i[:-2])
+                    x.mid = float(data[i])
+                else:
+                    x = getattr(c, i[:-2])
+                    x.end = float(data[i])
+        except ValueError as e:
+            raise InputError('Invalid input', api_error_code=-101) from e
+        c.update()
+        return success_return(c.to_dict())
+
+
+@bp.route('/<int:character_id>/cores', methods=['GET'])
+@role_required(request, ['select'])
+@api_try
+def characters_character_cores_get(user, character_id: int):
+    with Connect() as c:
+        c = Character(c)
+        c.character_id = character_id
+        c.select_character_core()
+        return success_return(c.uncap_cores_to_dict())
+
+
+@bp.route('/<int:character_id>/cores', methods=['PATCH'])
+@role_required(request, ['change'])
+@request_json_handle(request, is_batch=True)
+@api_try
+def characters_character_cores_patch(data, user, character_id: int):
+    '''修改角色觉醒cores'''
+    def force_type_core(x: dict) -> dict:
+        x['item_type'] = 'core'
+        x['type'] = 'core'
+        return x
+
+    with Connect() as c:
+        ch = Character(c)
+        ch.character_id = character_id
+        ch.select_character_core()
+        ch.remove_items([ItemFactory.from_dict(x, c=c)
+                         for x in map(force_type_core, data.get('remove', []))])
+        ch.add_items([ItemFactory.from_dict(x, c=c)
+                      for x in map(force_type_core, data.get('create', []))])
+        updates = list(map(force_type_core, data.get('update', [])))
+        for x in updates:
+            if 'amount' not in x:
+                raise InputError('`amount` is required in `update`')
+            if not isinstance(x['amount'], int) or x['amount'] <= 0:
+                raise InputError(
+                    '`amount` must be a positive integer', api_error_code=-101)
+
+        ch.update_items(
+            [ItemFactory.from_dict(x, c=c) for x in updates])
+        return success_return(ch.uncap_cores_to_dict())

latest version/api/constant.py

@@ -2,3 +2,6 @@ class Constant:
     QUERY_KEYS = ['limit', 'offset', 'query', 'fuzzy_query', 'sort']
 
     PATCH_KEYS = ['create', 'update', 'remove']
+
+    SKILL_IDS = ['gauge_easy', 'note_mirror', 'gauge_hard', 'frag_plus_10_pack_stellights', 'gauge_easy|frag_plus_15_pst&prs', 'gauge_hard|fail_frag_minus_100', 'frag_plus_5_side_light', 'visual_hide_hp', 'frag_plus_5_side_conflict', 'challenge_fullcombo_0gauge', 'gauge_overflow', 'gauge_easy|note_mirror', 'note_mirror', 'visual_tomato_pack_tonesphere', 'frag_rng_ayu', 'gaugestart_30|gaugegain_70', 'combo_100-frag_1', 'audio_gcemptyhit_pack_groovecoaster', 'gauge_saya', 'gauge_chuni', 'kantandeshou', 'gauge_haruna', 'frags_nono', 'gauge_pandora', 'gauge_regulus', 'omatsuri_daynight',
+                 'sometimes(note_mirror|frag_plus_5)', 'scoreclear_aa|visual_scoregauge', 'gauge_tempest', 'gauge_hard', 'gauge_ilith_summer', 'frags_kou', 'visual_ink', 'shirabe_entry_fee', 'frags_yume', 'note_mirror|visual_hide_far', 'frags_ongeki', 'gauge_areus', 'gauge_seele', 'gauge_isabelle', 'gauge_exhaustion', 'skill_lagrange', 'gauge_safe_10', 'frags_nami', 'skill_elizabeth', 'skill_lily', 'skill_kanae_midsummer', 'eto_uncap', 'luna_uncap', 'frags_preferred_song', 'visual_ghost_skynotes', 'ayu_uncap', 'skill_vita', 'skill_fatalis', 'skill_reunion', 'frags_ongeki_slash', 'frags_ongeki_hard', 'skill_amane', 'skill_kou_winter', 'gauge_hard|note_mirror', 'skill_shama', 'skill_milk', 'skill_shikoku', 'skill_mika', 'ilith_awakened_skill']

latest version/api/token.py

@@ -17,21 +17,22 @@ bp = Blueprint('token', __name__, url_prefix='/token')
 @api_try
 def token_post(data):
     '''
-        登录，获取token\ 
+        登录，获取token
+
         {'auth': base64('<user_id>:<password>')}
     '''
     try:
         auth_decode = bytes.decode(b64decode(data['auth']))
-    except:
-        raise PostError(api_error_code=-100)
-    if not ':' in auth_decode:
+    except Exception as e:
+        raise PostError(api_error_code=-100) from e
+    if ':' not in auth_decode:
         raise PostError(api_error_code=-100)
     name, password = auth_decode.split(':', 1)
 
     with Connect() as c:
         user = APIUser(c)
         user.login(name, password, request.remote_addr)
-        return success_return({'token': user.token, 'user_id': user.user_id})
+        return success_return({'token': user.api_token, 'user_id': user.user_id})
 
 
 @bp.route('', methods=['GET'])

latest version/api/users.py

@@ -128,7 +128,7 @@ def users_user_b30_get(user, user_id):
                 f'No best30 data of user `{user_id}`', api_error_code=-3)
         x.select_song_name()
         r = x.to_dict_list()
-        rating_sum = sum([i.rating for i in x.scores])
+        rating_sum = sum(i.rating for i in x.scores)
         return success_return({'user_id': user_id, 'b30_ptt': rating_sum / 30, 'data': r})