fix: proto_decode & proto_field (#31)

* fix: typing

* feat: proto_field(default_factory)

* fix: remove unused import

* Update version.py

---------

Co-authored-by: nullcat <a1025952421@outlook.com>

Author: RF-Tar-Railt

lagrange/__init__.py

@@ -2,8 +2,8 @@
 import asyncio
 
 from .client.client import Client as Client
-from .client.server_push.msg import msg_push_handler
-from .client.server_push.service import server_kick_handler
+# from .client.server_push.msg import msg_push_handler
+# from .client.server_push.service import server_kick_handler
 from .utils.log import log as log
 from .utils.log import install_loguru as install_loguru
 from .utils.sign import sign_provider

lagrange/client/base.py

@@ -1,7 +1,8 @@
 import asyncio
 import hashlib
 import time
-from typing import Callable, Coroutine, Dict, Optional, Tuple, Union, overload
+from typing import Callable, Optional, Union, overload
+from collections.abc import Coroutine
 
 from typing_extensions import Literal
 
@@ -60,7 +61,7 @@ def __init__(
         self._captcha_info = ["", "", ""]  # ticket, rand_str, aid
 
         self._server_push_queue: asyncio.Queue[SSOPacket] = asyncio.Queue()
-        self._tasks: Dict[str, asyncio.Task] = {}
+        self._tasks: dict[str, asyncio.Task] = {}
         self._network = ClientNetwork(
             sig_info,
             self._server_push_queue,
@@ -70,8 +71,8 @@ def __init__(
         )
         self._sign_provider = sign_provider
 
-        self._t106 = bytes()
-        self._t16a = bytes()
+        self._t106 = b""
+        self._t16a = b""
 
         self._online = asyncio.Event()
 
@@ -214,7 +215,7 @@ async def send_uni_packet(self, cmd, buf, send_only: bool = False, timeout=10):
             return await self._network.send(packet, wait_seq=-1, timeout=timeout)
         return await self._network.send(packet, wait_seq=seq, timeout=timeout)
 
-    async def fetch_qrcode(self) -> Union[int, Tuple[bytes, str]]:
+    async def fetch_qrcode(self) -> Union[int, tuple[bytes, str]]:
         tlv = QrCodeTlvBuilder()
         body = (
             PacketBuilder()

lagrange/client/client.py

@@ -5,13 +5,12 @@
 from typing import (
     BinaryIO,
     Callable,
-    Coroutine,
-    List,
     Optional,
     Union,
     overload,
     Literal,
 )
+from collections.abc import Coroutine
 
 from lagrange.info import AppInfo, DeviceInfo, SigInfo
 from lagrange.pb.message.msg_push import MsgPushBody
@@ -176,7 +175,7 @@ async def send_oidb_svc(
         rsp = OidbResponse.decode(
             (
                 await self.send_uni_packet(
-                    "OidbSvcTrpcTcp.0x{:0>2X}_{}".format(cmd, sub_cmd),
+                    f"OidbSvcTrpcTcp.0x{cmd:0>2X}_{sub_cmd}",
                     OidbRequest(
                         cmd=cmd, sub_cmd=sub_cmd, data=bytes(buf), is_uid=is_uid
                     ).encode(),
@@ -219,15 +218,15 @@ async def _send_msg_raw(self, pb: dict, *, grp_id=0, uid="") -> SendMsgRsp:
         packet = await self.send_uni_packet("MessageSvc.PbSendMsg", proto_encode(body))
         return SendMsgRsp.decode(packet.data)
 
-    async def send_grp_msg(self, msg_chain: List[Element], grp_id: int) -> int:
+    async def send_grp_msg(self, msg_chain: list[Element], grp_id: int) -> int:
         result = await self._send_msg_raw(
             {1: build_message(msg_chain).encode()}, grp_id=grp_id
         )
         if result.ret_code:
             raise AssertionError(result.ret_code, result.err_msg)
         return result.seq
 
-    async def send_friend_msg(self, msg_chain: List[Element], uid: str) -> int:
+    async def send_friend_msg(self, msg_chain: list[Element], uid: str) -> int:
         result = await self._send_msg_raw(
             {1: build_message(msg_chain).encode()}, uid=uid
         )
@@ -257,7 +256,7 @@ async def upload_grp_audio(self, voice: BinaryIO, grp_id: int) -> Audio:
     async def upload_friend_audio(self, voice: BinaryIO, uid: str) -> Audio:
         return await self._highway.upload_voice(voice, uid=uid)
 
-    async def fetch_audio_url(self, file_key: str, uid=None, gid=None):
+    async def fetch_audio_url(self, file_key: str, gid: int = 0, uid: str = ""):
         return await self._highway.get_audio_down_url(file_key, uid=uid, gid=gid)
 
     async def down_grp_audio(self, audio: Audio, grp_id: int) -> BytesIO:
@@ -267,7 +266,7 @@ async def down_friend_audio(self, audio: Audio) -> BytesIO:
         return await self._highway.download_audio(audio, uid=self.uid)
 
     async def fetch_image_url(
-        self, bus_type: Literal[10, 20], node: "IndexNode", uid=None, gid=None
+        self, bus_type: Literal[10, 20], node: "IndexNode", gid: int = 0, uid: str = ""
     ):
         if bus_type == 10:
             return await self._get_pri_img_url(uid, node)
@@ -327,7 +326,7 @@ async def get_grp_members(
 
     async def get_grp_msg(
         self, grp_id: int, start: int, end: int = 0, filter_deleted_msg=True
-    ) -> List[GroupMessage]:
+    ) -> list[GroupMessage]:
         if not end:
             end = start
         payload = GetGrpMsgRsp.decode(
@@ -354,9 +353,9 @@ async def get_grp_msg(
             return [*filter(lambda msg: msg.rand != -1, rsp)]
         return rsp
 
-    async def get_friend_list(self) -> List[BotFriend]:
-        nextuin_cache: List[GetFriendListUin] = []
-        rsp: List[BotFriend] = []
+    async def get_friend_list(self) -> list[BotFriend]:
+        nextuin_cache: list[GetFriendListUin] = []
+        rsp: list[BotFriend] = []
         frist_send = GetFriendListRsp.decode(
             (await self.send_oidb_svc(0xFD4, 1, PBGetFriendListRequest().encode())).data
         )
@@ -416,7 +415,7 @@ async def recall_grp_msg(self, grp_id: int, seq: int):
             PBGroupRecallRequest.build(grp_id, seq).encode(),
         )
         result = proto_decode(payload.data)
-        if result[2] != b"Success":
+        if result.into(2, bytes) != b"Success":
             raise AssertionError(result)
 
     async def rename_grp_name(self, grp_id: int, name: str) -> int:  # not test
@@ -550,11 +549,11 @@ async def set_grp_request(
     async def get_user_info(self, uid: str) -> UserInfo: ...
 
     @overload
-    async def get_user_info(self, uid: List[str]) -> List[UserInfo]: ...
+    async def get_user_info(self, uid: list[str]) -> list[UserInfo]: ...
 
     async def get_user_info(
-        self, uid: Union[str, List[str]]
-    ) -> Union[UserInfo, List[UserInfo]]:
+        self, uid: Union[str, list[str]]
+    ) -> Union[UserInfo, list[UserInfo]]:
         if isinstance(uid, str):
             uid = [uid]
         rsp = GetInfoFromUidRsp.decode(
@@ -615,7 +614,7 @@ def _gtk_1(self, skey_or_pskey: str):
             _hash += (_hash << 5) + ord(skey_or_pskey[i])
         return _hash & 2147483647
 
-    async def get_cookies(self, domains: list[str]) -> List[str]:
+    async def get_cookies(self, domains: list[str]) -> list[str]:
         """pskey"""
         return [
             i.value.decode()
@@ -631,8 +630,14 @@ async def get_cookies(self, domains: list[str]) -> List[str]:
         ]
 
     async def get_skey(self) -> str:
-        jump = "https%3A%2F%2Fh5.qzone.qq.com%2Fqqnt%2Fqzoneinpcqq%2Ffriend%3Frefresh%3D0%26clientuin%3D0%26darkMode%3D0&keyindex=19&random=2599"
-        url = f"https://ssl.ptlogin2.qq.com/jump?ptlang=1033&clientuin={self.uin}&clientkey={await self._get_client_key()}&u1={jump}"
+        jump = (
+            "https%3A%2F%2Fh5.qzone.qq.com%2Fqqnt%2Fqzoneinpcqq%2F"
+            "friend%3Frefresh%3D0%26clientuin%3D0%26darkMode%3D0&keyindex=19&random=2599"
+        )
+        url = (
+            f"https://ssl.ptlogin2.qq.com/jump?ptlang=1033&clientuin={self.uin}"
+            f"&clientkey={await self._get_client_key()}&u1={jump}"
+        )
         resp = await HttpCat.request("GET", url, follow_redirect=False)
         return resp.cookies["skey"]
 

lagrange/client/event.py

@@ -1,5 +1,6 @@
 import asyncio
-from typing import TYPE_CHECKING, Any, Callable, Awaitable, Dict, Set, Type, TypeVar
+from typing import TYPE_CHECKING, Any, Callable, TypeVar
+from collections.abc import Awaitable
 
 from lagrange.utils.log import log
 
@@ -13,26 +14,26 @@
 
 class Events:
     def __init__(self):
-        self._task_group: Set[asyncio.Task] = set()
-        self._handle_map: Dict[Type["BaseEvent"], EVENT_HANDLER] = {}
+        self._task_group: set[asyncio.Task] = set()
+        self._handle_map: dict[type["BaseEvent"], EVENT_HANDLER] = {}
 
-    def subscribe(self, event: Type[T], handler: EVENT_HANDLER[T]):
+    def subscribe(self, event: type[T], handler: EVENT_HANDLER[T]):
         if event not in self._handle_map:
             self._handle_map[event] = handler
         else:
             raise AssertionError(
-                "Event already subscribed to {}".format(self._handle_map[event])
+                f"Event already subscribed to {self._handle_map[event]}"
             )
 
-    def unsubscribe(self, event: Type["BaseEvent"]):
+    def unsubscribe(self, event: type["BaseEvent"]):
         return self._handle_map.pop(event)
 
     async def _task_exec(self, client: "Client", event: "BaseEvent", handler: EVENT_HANDLER):
         try:
             await handler(client, event)
         except Exception as e:
             log.root.error(
-                "Unhandled exception on task {}".format(event), exc_info=e
+                f"Unhandled exception on task {event}", exc_info=e
             )
 
     def emit(self, event: "BaseEvent", client: "Client"):

lagrange/client/events/friend.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, List
+from typing import TYPE_CHECKING
 from . import BaseEvent
 
 if TYPE_CHECKING:
@@ -22,4 +22,4 @@ class FriendMessage(FriendEvent):
     msg_id: int
     timestamp: int
     msg: str
-    msg_chain: List[Element]
+    msg_chain: list[Element]

lagrange/client/events/group.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, List, Optional, Union, Dict
+from typing import TYPE_CHECKING
 
 from . import BaseEvent
 
@@ -30,7 +30,7 @@ class GroupMessage(GroupEvent, MessageInfo):
     sub_id: int = field(repr=False)  # client ver identify
     sender_type: int = field(repr=False)
     msg: str
-    msg_chain: List[Element]
+    msg_chain: list[Element]
 
     @property
     def is_bot(self) -> bool:
@@ -48,7 +48,7 @@ class GroupNudge(GroupEvent):
     target_uin: int
     action: str
     suffix: str
-    attrs: Dict[str, Union[str, int]] = field(repr=False)
+    attrs: dict[str, str | int] = field(repr=False)
     attrs_xml: str = field(repr=False)
 
 
@@ -59,7 +59,7 @@ class GroupSign(GroupEvent):
     uin: int
     nickname: str
     timestamp: int
-    attrs: Dict[str, Union[str, int]] = field(repr=False)
+    attrs: dict[str, str | int] = field(repr=False)
     attrs_xml: str = field(repr=False)
 
 
@@ -75,8 +75,8 @@ class GroupMuteMember(GroupEvent):
 @dataclass
 class GroupMemberJoinRequest(GroupEvent):
     uid: str
-    invitor_uid: Optional[str] = None
-    answer: Optional[str] = None  # 问题：(.*)答案：(.*)
+    invitor_uid: str | None = None
+    answer: str | None = None  # 问题：(.*)答案：(.*)
 
 
 @dataclass
@@ -96,7 +96,7 @@ class GroupMemberQuit(GroupEvent):
     @property
     def is_kicked(self) -> bool:
         return self.exit_type in [3, 131]
-    
+
     @property
     def is_kicked_self(self) -> bool:
         return self.exit_type == 3

lagrange/client/highway/encoders.py

@@ -83,8 +83,8 @@ def encode_upload_img_req(
     fn = f"{md5.hex().upper()}.{info.name or 'jpg'}"
     c2c_info = None
     grp_info = None
-    c2c_pb = bytes()
-    grp_pb = bytes()
+    c2c_pb = b""
+    grp_pb = b""
     if grp_id:
         scene_type = 2
         grp_info = GroupInfo(grp_id=grp_id)

lagrange/client/highway/frame.py

@@ -1,5 +1,5 @@
 import struct
-from typing import BinaryIO, Tuple
+from typing import BinaryIO
 
 from lagrange.pb.highway.head import HighwayTransRespHead
 
@@ -16,7 +16,7 @@ def write_frame(head: bytes, body: bytes) -> bytes:
 
 def read_frame(
     reader: BinaryIO,
-) -> Tuple[HighwayTransRespHead, bytes]:
+) -> tuple[HighwayTransRespHead, bytes]:
     head = reader.read(9)
     if len(head) != 9 and head[0] != 0x28:
         raise ValueError("Invalid frame head", head)