✨ 将处理过程中的 dataclass 换成 pydantic

nonebot_plugin_tetris_stats/db/models.py

@@ -5,7 +5,7 @@ from nonebot_plugin_orm import Model
 from sqlalchemy import JSON, DateTime, PickleType, String
 from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
 
-from ..game_data_processor import ProcessedData, User
+from ..game_data_processor.schemas import BaseProcessedData, BaseUser
 from ..utils.typing import CommandType, GameType
 
 
@@ -28,6 +28,6 @@ class HistoricalData(MappedAsDataclass, Model):
     game_platform: Mapped[GameType] = mapped_column(String(32), index=True, init=False)
     command_type: Mapped[CommandType] = mapped_column(String(16), index=True, init=False)
     command_args: Mapped[list[str]] = mapped_column(JSON, init=False)
-    game_user: Mapped[User] = mapped_column(PickleType, init=False)
+    game_user: Mapped[BaseUser] = mapped_column(PickleType, init=False)
-    processed_data: Mapped[ProcessedData] = mapped_column(PickleType, init=False)
+    processed_data: Mapped[BaseProcessedData] = mapped_column(PickleType, init=False)
     finish_time: Mapped[datetime] = mapped_column(DateTime, init=False)

nonebot_plugin_tetris_stats/game_data_processor/__init__.py

@@ -1,5 +1,4 @@
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
 from datetime import UTC, datetime
 from typing import Any
 
@@ -7,25 +6,11 @@ from nonebot.matcher import Matcher
 from nonebot_plugin_alconna import AlcMatches, AlconnaMatcher
 
 from ..utils.exception import MessageFormatError
+from ..utils.recorder import Recorder
 from ..utils.typing import CommandType, GameType
-
+from .schemas import BaseProcessedData as ProcessedData
-
+from .schemas import BaseRawResponse as RawResponse
-@dataclass
+from .schemas import BaseUser as User
-class User:
-    """游戏用户"""
-
-
-@dataclass
-class RawResponse:
-    """原始请求数据"""
-
-
-@dataclass
-class ProcessedData:
-    """处理/验证后的数据"""
-
-
-from ..utils.recorder import Recorder  # noqa: E402 避免循环导入
 
 
 class Processor(ABC):

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/processor.py

@@ -1,6 +1,5 @@
 from collections import defaultdict
 from collections.abc import Callable
-from dataclasses import asdict, dataclass
 from datetime import UTC, datetime, timedelta
 from math import floor
 from re import match
@@ -16,15 +15,14 @@ from ...db import create_or_update_bind
 from ...utils.exception import MessageFormatError, RequestError, WhatTheFuckError
 from ...utils.request import Request, splice_url
 from ...utils.typing import GameType
-from .. import ProcessedData as ProcessedDataMeta
 from .. import Processor as ProcessorMeta
-from .. import RawResponse as RawResponseMeta
+from ..schemas import BaseUser
-from .. import User as UserMeta
 from .constant import BASE_URL, GAME_TYPE, RANK_PERCENTILE
 from .model import IORank
 from .schemas.league_all import FailedModel as LeagueAllFailed
 from .schemas.league_all import LeagueAll
 from .schemas.league_all import ValidUser as LeagueAllUser
+from .schemas.response import ProcessedData, RawResponse
 from .schemas.user_info import FailedModel as InfoFailed
 from .schemas.user_info import (
     NeverPlayedLeague,
@@ -40,22 +38,15 @@ from .typing import Rank
 driver = get_driver()
 
 
-@dataclass
+class User(BaseUser):
-class User(UserMeta):
     ID: str | None = None
     name: str | None = None
 
-
+    @property
-@dataclass
+    def unique_identifier(self) -> str:
-class RawResponse(RawResponseMeta):
+        if self.ID is None:
-    user_info: bytes | None = None
+            raise ValueError('不完整的User!')
-    user_records: bytes | None = None
+        return self.ID
-
-
-@dataclass
-class ProcessedData(ProcessedDataMeta):
-    user_info: InfoSuccess | None = None
-    user_records: RecordsSuccess | None = None
 
 
 def identify_user_info(info: str) -> User | MessageFormatError:
@@ -214,7 +205,7 @@ async def get_io_rank_data() -> None:
         sort: Callable[[list[LeagueAllUser], Callable[[LeagueAllUser], float]], LeagueAllUser],
     ) -> tuple[dict[str, str], float]:
         user = sort(users, field)
-        return asdict(User(ID=user.id, name=user.username)), field(user)
+        return User(ID=user.id, name=user.username).dict(), field(user)
 
     users = [i for i in league_all.data.users if isinstance(i, LeagueAllUser)]
     rank_to_users: defaultdict[Rank, list[LeagueAllUser]] = defaultdict(list)

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/schemas/response.py

@@ -0,0 +1,14 @@
+from ... import ProcessedData as ProcessedDataMeta
+from ... import RawResponse as RawResponseMeta
+from .user_info import SuccessModel as InfoSuccess
+from .user_records import SuccessModel as RecordsSuccess
+
+
+class RawResponse(RawResponseMeta):
+    user_info: bytes | None = None
+    user_records: bytes | None = None
+
+
+class ProcessedData(ProcessedDataMeta):
+    user_info: InfoSuccess | None = None
+    user_records: RecordsSuccess | None = None

nonebot_plugin_tetris_stats/game_data_processor/schemas.py

@@ -0,0 +1,24 @@
+from abc import ABC, abstractmethod
+from typing import Self
+
+from pydantic import BaseModel
+
+
+class BaseUser(ABC, BaseModel):
+    """游戏用户"""
+
+    def __eq__(self, __value: Self) -> bool:
+        return self.unique_identifier == __value.unique_identifier
+
+    @property
+    @abstractmethod
+    def unique_identifier(self) -> str:
+        raise NotImplementedError
+
+
+class BaseRawResponse(BaseModel):
+    """原始请求数据"""
+
+
+class BaseProcessedData(BaseModel):
+    """处理/验证后的数据"""

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/processor.py

@@ -13,26 +13,18 @@ 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 ProcessedData as ProcessedDataMeta
 from .. import Processor as ProcessorMeta
-from .. import RawResponse as RawResponseMeta
+from ..schemas import BaseUser
-from .. import User as UserMeta
 from .constant import BASE_URL, GAME_TYPE
+from .schemas.response import ProcessedData, RawResponse
 
 
-@dataclass
+class User(BaseUser):
-class User(UserMeta):
     name: str
 
-
+    @property
-@dataclass
+    def unique_identifier(self) -> str:
-class RawResponse(RawResponseMeta):
+        return self.name
-    user_profile: bytes | None = None
-
-
-@dataclass
-class ProcessedData(ProcessedDataMeta):
-    user_profile: str | None = None
 
 
 @dataclass

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/schemas/response.py

@@ -0,0 +1,9 @@
+from ...schemas import BaseProcessedData, BaseRawResponse
+
+
+class RawResponse(BaseRawResponse):
+    user_profile: bytes | None = None
+
+
+class ProcessedData(BaseProcessedData):
+    user_profile: str | None = None

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/processor.py

@@ -1,6 +1,5 @@
 from dataclasses import dataclass
 from re import match
-from typing import Any
 from urllib.parse import urlencode
 
 from nonebot_plugin_orm import get_session
@@ -10,32 +9,24 @@ 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 ProcessedData as ProcessedDataMeta
 from .. import Processor as ProcessorMeta
-from .. import RawResponse as RawResponseMeta
+from ..schemas import BaseUser
-from .. import User as UserMeta
 from .constant import BASE_URL, GAME_TYPE
+from .schemas.response import ProcessedData, RawResponse
 from .schemas.user_info import SuccessModel as InfoSuccess
 from .schemas.user_info import UserInfo
 from .schemas.user_profile import UserProfile
 
 
-@dataclass
+class User(BaseUser):
-class User(UserMeta):
     teaid: str | None = None
     name: str | None = None
 
-
+    @property
-@dataclass
+    def unique_identifier(self) -> str:
-class RawResponse(RawResponseMeta):
+        if self.teaid is None:
-    user_profile: dict[frozenset[tuple[str, Any]], bytes]
+            raise ValueError('不完整的User!')
-    user_info: bytes | None = None
+        return self.teaid
-
-
-@dataclass
-class ProcessedData(ProcessedDataMeta):
-    user_profile: dict[frozenset[tuple[str, Any]], UserProfile]
-    user_info: InfoSuccess | None = None
 
 
 @dataclass
@@ -135,18 +126,18 @@ class Processor(ProcessorMeta):
             self.processed_data.user_info = user_info
         return self.processed_data.user_info
 
-    async def get_user_profile(self, other_parameter: dict[str, Any] | None = None) -> UserProfile:
+    async def get_user_profile(self, other_parameter: dict[str, str | bytes] | None = None) -> UserProfile:
         """获取用户数据"""
         if other_parameter is None:
             other_parameter = {}
-        fset = frozenset(other_parameter.items())
+        fset: frozenset[tuple[str, str | bytes]] = frozenset(other_parameter.items())
         if self.processed_data.user_profile.get(fset) is None:
             self.raw_response.user_profile[fset] = await Request.request(
                 splice_url(
                     [
                         BASE_URL,
                         'getProfile',
-                        f'?{urlencode({"id":self.user.teaid or self.user.name},**other_parameter)}',
+                        f'?{urlencode({"id":self.user.teaid or self.user.name,**other_parameter})}',
                     ]
                 )
             )

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/schemas/response.py

@@ -0,0 +1,13 @@
+from ...schemas import BaseProcessedData, BaseRawResponse
+from .user_info import SuccessModel as InfoSuccess
+from .user_profile import UserProfile
+
+
+class RawResponse(BaseRawResponse):
+    user_profile: dict[frozenset[tuple[str, str | bytes]], bytes]
+    user_info: bytes | None = None
+
+
+class ProcessedData(BaseProcessedData):
+    user_profile: dict[frozenset[tuple[str, str | bytes]], UserProfile]
+    user_info: InfoSuccess | None = None