✨ 添加冗余 platform 字段

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/constant.py

@@ -1,7 +1,8 @@
-from ...utils.typing import GameType
+from typing import Literal
+
 from .typing import Rank
 
-GAME_TYPE: GameType = 'IO'
+GAME_TYPE: Literal['IO'] = 'IO'
 BASE_URL = 'https://ch.tetr.io/api/'
 RANK_PERCENTILE: dict[Rank, float] = {
     'x': 1,

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/processor.py

@@ -4,6 +4,7 @@ from datetime import UTC, datetime, timedelta
 from math import floor
 from re import match
 from statistics import mean
+from typing import Literal
 
 from nonebot import get_driver
 from nonebot_plugin_apscheduler import scheduler  # type: ignore[import-untyped]
@@ -15,7 +16,6 @@ from ...db import create_or_update_bind
 from ...utils.exception import MessageFormatError, RequestError, WhatTheFuckError
 from ...utils.request import splice_url
 from ...utils.retry import retry
-from ...utils.typing import GameType
 from .. import Processor as ProcessorMeta
 from .cache import Cache
 from .constant import BASE_URL, GAME_TYPE, RANK_PERCENTILE
@@ -55,7 +55,7 @@ class Processor(ProcessorMeta):
         self.processed_data = ProcessedData()
 
     @property
-    def game_platform(self) -> GameType:
+    def game_platform(self) -> Literal['IO']:
         return GAME_TYPE
 
     async def handle_bind(self, platform: str, account: str) -> str:

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/schemas/response.py

@@ -1,14 +1,21 @@
+from typing import Literal
+
 from ... import ProcessedData as ProcessedDataMeta
 from ... import RawResponse as RawResponseMeta
+from ..constant import GAME_TYPE
 from .user_info import SuccessModel as InfoSuccess
 from .user_records import SuccessModel as RecordsSuccess
 
 
 class RawResponse(RawResponseMeta):
+    platform: Literal['IO'] = GAME_TYPE
+
     user_info: bytes | None = None
     user_records: bytes | None = None
 
 
 class ProcessedData(ProcessedDataMeta):
+    platform: Literal['IO'] = GAME_TYPE
+
     user_info: InfoSuccess | None = None
     user_records: RecordsSuccess | None = None

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/schemas/user.py

@@ -1,7 +1,12 @@
+from typing import Literal
+
 from ...schemas import BaseUser
+from ..constant import GAME_TYPE
 
 
 class User(BaseUser):
+    platform: Literal['IO'] = GAME_TYPE
+
     ID: str | None = None
     name: str | None = None
 

nonebot_plugin_tetris_stats/game_data_processor/schemas.py

@@ -2,8 +2,14 @@ from abc import ABC, abstractmethod
 
 from pydantic import BaseModel
 
+from ..utils.typing import GameType
 
-class BaseUser(ABC, BaseModel):
+
+class Base(BaseModel):
+    platform: GameType
+
+
+class BaseUser(ABC, Base):
     """游戏用户"""
 
     def __eq__(self, __value: object) -> bool:
@@ -17,9 +23,9 @@ class BaseUser(ABC, BaseModel):
         raise NotImplementedError
 
 
-class BaseRawResponse(BaseModel):
+class BaseRawResponse(Base):
     """原始请求数据"""
 
 
-class BaseProcessedData(BaseModel):
+class BaseProcessedData(Base):
     """处理/验证后的数据"""

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/constant.py

@@ -1,4 +1,4 @@
-from ...utils.typing import GameType
+from typing import Literal
 
-GAME_TYPE: GameType = 'TOP'
+GAME_TYPE: Literal['TOP'] = 'TOP'
 BASE_URL = 'http://tetrisonline.pl/top/'

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/processor.py

@@ -2,7 +2,7 @@ from contextlib import suppress
 from dataclasses import dataclass
 from io import StringIO
 from re import match
-from typing import NoReturn
+from typing import Literal, NoReturn
 from urllib.parse import urlencode
 
 from lxml import etree
@@ -12,7 +12,6 @@ from pandas import read_html
 from ...db import create_or_update_bind
 from ...utils.exception import MessageFormatError, RequestError
 from ...utils.request import Request, splice_url
-from ...utils.typing import GameType
 from .. import Processor as ProcessorMeta
 from ..schemas import BaseUser
 from .constant import BASE_URL, GAME_TYPE
@@ -20,6 +19,8 @@ from .schemas.response import ProcessedData, RawResponse
 
 
 class User(BaseUser):
+    platform: Literal['TOP'] = GAME_TYPE
+
     name: str
 
     @property
@@ -56,7 +57,7 @@ class Processor(ProcessorMeta):
         self.processed_data = ProcessedData()
 
     @property
-    def game_platform(self) -> GameType:
+    def game_platform(self) -> Literal['TOP']:
         return GAME_TYPE
 
     async def handle_bind(self, platform: str, account: str) -> str:

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/schemas/response.py

@@ -1,9 +1,16 @@
+from typing import Literal
+
 from ...schemas import BaseProcessedData, BaseRawResponse
+from ..constant import GAME_TYPE
 
 
 class RawResponse(BaseRawResponse):
+    platform: Literal['TOP'] = GAME_TYPE
+
     user_profile: bytes | None = None
 
 
 class ProcessedData(BaseProcessedData):
+    platform: Literal['TOP'] = GAME_TYPE
+
     user_profile: str | None = None

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/constant.py

@@ -1,4 +1,4 @@
-from ...utils.typing import GameType
+from typing import Literal
 
-GAME_TYPE: GameType = 'TOS'
+GAME_TYPE: Literal['TOS'] = 'TOS'
 BASE_URL = 'https://teatube.cn:8888/'

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/processor.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from re import match
+from typing import Literal
 from urllib.parse import urlencode
 
 from nonebot_plugin_orm import get_session
@@ -8,7 +9,6 @@ from pydantic import parse_raw_as
 from ...db import create_or_update_bind
 from ...utils.exception import MessageFormatError, RequestError
 from ...utils.request import Request, splice_url
-from ...utils.typing import GameType
 from .. import Processor as ProcessorMeta
 from ..schemas import BaseUser
 from .constant import BASE_URL, GAME_TYPE
@@ -19,6 +19,8 @@ from .schemas.user_profile import UserProfile
 
 
 class User(BaseUser):
+    platform: Literal['TOS'] = GAME_TYPE
+
     teaid: str | None = None
     name: str | None = None
 
@@ -67,7 +69,7 @@ class Processor(ProcessorMeta):
         self.processed_data = ProcessedData(user_profile={})
 
     @property
-    def game_platform(self) -> GameType:
+    def game_platform(self) -> Literal['TOS']:
         return GAME_TYPE
 
     async def handle_bind(self, platform: str, account: str) -> str:

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/schemas/response.py

@@ -1,13 +1,20 @@
+from typing import Literal
+
 from ...schemas import BaseProcessedData, BaseRawResponse
+from ..constant import GAME_TYPE
 from .user_info import SuccessModel as InfoSuccess
 from .user_profile import UserProfile
 
 
 class RawResponse(BaseRawResponse):
+    platform: Literal['TOS'] = GAME_TYPE
+
     user_profile: dict[str, bytes]
     user_info: bytes | None = None
 
 
 class ProcessedData(BaseProcessedData):
+    platform: Literal['TOS'] = GAME_TYPE
+
     user_profile: dict[str, UserProfile]
     user_info: InfoSuccess | None = None