[Enhance] Cloud save update scores & capture auth error

- Restore the feature that cloud save can be used to cover best scores
- Capture error that the request does not have `Authorization` in header

latest version/core/download.py

@@ -167,7 +167,7 @@ class UserDownload:
 
 class DownloadList(UserDownload):
     '''
-        下载列表类\ 
+        下载列表类
         properties: `user` - `User`类或子类的实例
     '''
 

latest version/core/operation.py

@@ -1,18 +1,20 @@
+from .save import SaveData
 from .sql import Connect, Sql
 from .score import Score
 from .download import DownloadList
+from .user import User
 
 
 class BaseOperation:
     name: str = None
 
-    def __init__(self):
+    def __init__(self, *args, **kwargs):
         pass
 
-    def __call__(self, *args, **kwargs):
+    def __call__(self, *args, **kwargs) -> None:
         return self.run(*args, **kwargs)
 
-    def run(self, *args, **kwargs):
+    def run(self, *args, **kwargs) -> None:
         raise NotImplementedError
 
 
@@ -65,3 +67,102 @@ class RefreshSongFileCache(BaseOperation):
     def run(self):
         DownloadList.clear_all_cache()
         DownloadList.initialize_cache()
+
+
+class SaveUpdateScore(BaseOperation):
+    '''
+        云存档更新成绩，是覆盖式更新\ 
+        提供user参数时，只更新该用户的成绩，否则更新所有用户的成绩
+    '''
+    name = 'save_update_score'
+
+    def __init__(self, user=None):
+        self.user = user
+
+    def run(self, user=None):
+        '''
+        parameter:
+            `user` - `User`类或子类的实例
+        '''
+        if user is not None:
+            self.user = user
+        if self.user is not None and self.user.user_id is not None:
+            self._one_user_update()
+        else:
+            self._all_update()
+
+    def _one_user_update(self):
+        with Connect() as c:
+            save = SaveData(c)
+            save.select_scores(self.user)
+
+            clear_state = {f'{i["song_id"]}{i["difficulty"]}': i['clear_type']
+                           for i in save.clearlamps_data}
+
+            song_id_1 = [i['song_id'] for i in save.scores_data]
+            song_id_2 = [i['song_id'] for i in save.clearlamps_data]
+            song_id = list(set(song_id_1 + song_id_2))
+
+            c.execute(
+                f'''select song_id, rating_pst, rating_prs, rating_ftr, rating_byn from chart where song_id in ({','.join(['?']*len(song_id))})''', song_id)
+            x = c.fetchall()
+            song_chart_const = {i[0]: [i[1], i[2], i[3], i[4]]
+                                for i in x}  # chart const * 10
+
+            new_scores = []
+            for i in save.scores_data:
+                rating = 0
+                if i['song_id'] in song_chart_const:
+                    rating = Score.calculate_rating(
+                        song_chart_const[i['song_id']][i['difficulty']] / 10, i['score'])
+                    if rating < 0:
+                        rating = 0
+
+                y = f'{i["song_id"]}{i["difficulty"]}'
+                if y in clear_state:
+                    clear_type = clear_state[y]
+                else:
+                    clear_type = 0
+
+                new_scores.append((self.user.user_id, i['song_id'], i['difficulty'], i['score'], i['shiny_perfect_count'], i['perfect_count'],
+                                   i['near_count'], i['miss_count'], i['health'], i['modifier'], i['time_played'], clear_type, clear_type, rating))
+
+            c.executemany(
+                '''insert or replace into best_score values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', new_scores)
+
+    def _all_update(self):
+        with Connect() as c:
+            c.execute(
+                f'''select song_id, rating_pst, rating_prs, rating_ftr, rating_byn from chart''')
+            song_chart_const = {i[0]: [i[1], i[2], i[3], i[4]]
+                                for i in c.fetchall()}  # chart const * 10
+            c.execute('''select user_id from user_save''')
+            for y in c.fetchall():
+                user = User()
+                user.user_id = y[0]
+                save = SaveData(c)
+                save.select_scores(user)
+
+                clear_state = {f'{i["song_id"]}{i["difficulty"]}': i['clear_type']
+                               for i in save.clearlamps_data}
+
+                new_scores = []
+                for i in save.scores_data:
+                    rating = 0
+                    if i['song_id'] in song_chart_const:
+                        rating = Score.calculate_rating(
+                            song_chart_const[i['song_id']][i['difficulty']] / 10, i['score'])
+                        if rating < 0:
+                            rating = 0
+
+                    y = f'{i["song_id"]}{i["difficulty"]}'
+                    if y in clear_state:
+                        clear_type = clear_state[y]
+                    else:
+                        clear_type = 0
+
+                    new_scores.append((user.user_id, i['song_id'], i['difficulty'], i['score'], i['shiny_perfect_count'], i['perfect_count'],
+                                       i['near_count'], i['miss_count'], i['health'], i['modifier'], i['time_played'], clear_type, clear_type, rating))
+
+                c.executemany(
+                    '''insert or replace into best_score values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', new_scores)

latest version/core/save.py

@@ -3,7 +3,7 @@ from time import time
 
 from .config_manager import Config
 from .constant import Constant
-from .error import InputError
+from .error import InputError, NoData
 from .util import md5
 
 
@@ -54,6 +54,20 @@ class SaveData:
             }
         }
 
+    def select_scores(self, user) -> None:
+        '''
+            parameter: `user` - `User`类或子类的实例
+        '''
+        self.user = user
+        self.c.execute('''select scores_data, clearlamps_data from user_save where user_id=:a''',
+                       {'a': user.user_id})
+        x = self.c.fetchone()
+        if not x:
+            raise NoData(f'User `{user.user_id}` has no cloud save data')
+
+        self.scores_data: list = json.loads(x[0])[""]
+        self.clearlamps_data: list = json.loads(x[1])[""]
+
     def select_all(self, user) -> None:
         '''
             parameter: `user` - `User`类或子类的实例

latest version/core/score.py

@@ -102,7 +102,7 @@ class Score:
         return True
 
     @staticmethod
-    def calculate_rating(defnum: int, score: int) -> float:
+    def calculate_rating(defnum: float, score: int) -> float:
         '''计算rating，谱面定数小于等于0视为Unrank，返回值会为-1，这里的defnum = Chart const'''
         if not defnum or defnum <= 0:
             # 谱面没定数或者定数小于等于0被视作Unrank

latest version/core/user.py

@@ -243,8 +243,9 @@ class UserLogin(User):
                        'name': self.name})
         x = self.c.fetchone()
         if x is None:
-            raise NoData('Username does not exist.', 104)
+            raise NoData(f'Username `{self.name}` does not exist.', 104)
 
+        self.user_id = x[0]
         self.now = int(time.time() * 1000)
         if x[2] is not None and x[2] != '':
             # 自动封号检查
@@ -255,14 +256,14 @@ class UserLogin(User):
 
         if x[1] == '':
             # 账号封禁
-            raise UserBan('The account has been banned.', 106)
+            raise UserBan(
+                f'The account `{self.user_id}` has been banned.', 106)
 
         if x[1] != self.hash_pwd:
             raise NoAccess('Wrong password.', 104)
 
-        self.user_id = str(x[0])
         self.token = base64.b64encode(hashlib.sha256(
-            (self.user_id + str(self.now)).encode("utf8") + urandom(8)).digest()).decode()
+            (str(self.user_id) + str(self.now)).encode("utf8") + urandom(8)).digest()).decode()
 
         self.c.execute(
             '''select login_device from login where user_id = :user_id''', {"user_id": self.user_id})