feat: recv music share

partial send music share
See #42

python/ichika/client.py

@@ -5,7 +5,17 @@
 
 from .core import PlumbingClient, RawMessageReceipt
 from .message import _serialize_message as _serialize_msg
-from .message.elements import At, AtAll, Audio, Face, FlashImage, Image, Reply, Text
+from .message.elements import (
+    At,
+    AtAll,
+    Audio,
+    Face,
+    FlashImage,
+    Image,
+    MusicShare,
+    Reply,
+    Text,
+)
 
 
 class Client(PlumbingClient):
@@ -39,13 +49,16 @@ async def _validate_chain(self, chain: MessageChain) -> MessageChain | Element:
         return chain
 
     async def _send_special_element(self, uin: int, kind: str, element: Element) -> RawMessageReceipt:
-        method = getattr(self, f"send_{kind}_{element.__class__.__name__.lower()}")
         if Audio._check(element):
             if element.raw is None:
-                sealed = (await getattr(self, f"upload_{kind}_audio")(uin, await element.fetch())).raw
+                uploader = self.upload_friend_audio if kind == "friend" else self.upload_group_audio
+                sealed = (await uploader(uin, await element.fetch())).raw
             else:
                 sealed = element.raw
-            return await method(uin, sealed)
+            sender = self.send_friend_audio if kind == "friend" else self.send_group_audio
+            return await sender(uin, sealed)
+        if isinstance(element, MusicShare):
+            raise TypeError("音乐分享无法因发送后无法获得消息元数据，无法使用 send_xxx_message API 发送，请直接调用底层 API")
         raise TypeError(f"无法发送元素: {element!r}")
 
     async def send_group_message(self, uin: int, chain: MessageChain) -> RawMessageReceipt:

python/ichika/core/__init__.pyi

@@ -1,8 +1,11 @@
 import asyncio
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Literal, TypeVar
 from typing_extensions import Any, TypeAlias
 
+from ichika.message.elements import MusicShare
+
 from ..client import Client
 from ..login import (
     BaseLoginCredentialStore,
@@ -230,6 +233,18 @@ class RawMessageReceipt:
     target: int
     """发送目标"""
 
+@_internal_repr
+class OCRText:
+    detected_text: str
+    confidence: int
+    polygon: VTuple[tuple[int, int]] | None
+    advanced_info: str
+
+@_internal_repr
+class OCRResult:
+    texts: VTuple[OCRText]
+    language: str
+
 __OnlineStatus: TypeAlias = (  # TODO: Wrapper
     tuple[int, str]  # (face_index, wording)
     | tuple[
@@ -291,6 +306,7 @@ class PlumbingClient:
     ) -> None: ...
     async def get_other_clients(self) -> VTuple[OtherClientInfo]: ...
     async def modify_online_status(self, status: __OnlineStatus) -> None: ...
+    async def image_ocr(self, url: str, md5: str, width: int, height: int) -> OCRResult: ...
     # [impl 2]
     async def get_friend_list(self) -> FriendList: ...
     async def get_friend_list_raw(self) -> FriendList: ...
@@ -323,11 +339,13 @@ class PlumbingClient:
     async def upload_friend_audio(self, uin: int, data: bytes) -> dict[str, Any]: ...
     async def upload_group_image(self, uin: int, data: bytes) -> dict[str, Any]: ...
     async def upload_group_audio(self, uin: int, data: bytes) -> dict[str, Any]: ...
+    async def send_friend_audio(self, uin: int, audio: _SealedAudio) -> RawMessageReceipt: ...
+    async def send_group_audio(self, uin: int, audio: _SealedAudio) -> RawMessageReceipt: ...
+    async def send_friend_music_share(self, uin: int, share: MusicShare) -> None: ...
+    async def send_group_music_share(self, uin: int, share: MusicShare) -> None: ...
     # [impl 6]
     async def send_friend_message(self, uin: int, chain: list[dict[str, Any]]) -> RawMessageReceipt: ...
-    async def send_friend_audio(self, uin: int, audio: _SealedAudio) -> RawMessageReceipt: ...
     async def send_group_message(self, uin: int, chain: list[dict[str, Any]]) -> RawMessageReceipt: ...
-    async def send_group_audio(self, uin: int, audio: _SealedAudio) -> RawMessageReceipt: ...
     async def recall_friend_message(self, uin: int, time: int, seq: int, rand: int) -> None: ...
     async def recall_group_message(self, uin: int, seq: int, rand: int) -> None: ...
     async def modify_group_essence(self, uin: int, seq: int, rand: int, flag: bool) -> None: ...
@@ -342,3 +360,24 @@ class PlumbingClient:
 
 def face_id_from_name(name: str) -> int | None: ...
 def face_name_from_id(id: int) -> str: ...
+@_internal_repr
+class MessageSource:
+    """消息元信息"""
+
+    seqs: tuple[int, ...]
+    """消息的 SEQ
+    建议搭配聊天类型与上下文 ID （例如 `("group", 123456, seq)`）作为索引的键
+    """
+    rands: tuple[int, ...]
+    """消息的随机信息，撤回需要"""
+    time: datetime
+    """消息发送时间"""
+
+@_internal_repr
+class FriendInfo:
+    """事件中的好友信息"""
+
+    uin: int
+    """好友账号"""
+    nickname: str
+    """好友实际昵称"""

python/ichika/events.py

@@ -4,8 +4,7 @@
 
 from graia.amnesia.message import MessageChain
 
-from ichika.core import Group, Member
-from ichika.core.events.structs import FriendInfo, MessageSource
+from ichika.core import FriendInfo, Group, Member, MessageSource
 
 
 class LoginEvent(TypedDict):

python/ichika/message/__init__.py

@@ -6,14 +6,14 @@
 from graia.amnesia.message.element import Element, Unknown
 from loguru import logger
 
-from .elements import TYPE_MAP
+from .elements import DESERIALIZE_INV
 from .serializer import SERIALIZE_INV
 
 
-def deserialize_message(elements: list[dict[str, Any]]) -> MessageChain:
+def _deserialize_message(elements: list[dict[str, Any]]) -> MessageChain:
     elem_seq: list[Element] = []
     for e_data in elements:
-        cls = TYPE_MAP.get(e_data.pop("type"), None)
+        cls = DESERIALIZE_INV.get(e_data.pop("type"), None)
         if cls is None:
             logger.warning(f"未知元素: {e_data!r}")
             elem_seq.append(Unknown("Unknown", e_data))

python/ichika/message/elements.py

@@ -7,7 +7,7 @@
 from enum import Enum
 from functools import total_ordering
 from io import BytesIO
-from typing import Generic, Literal, Optional
+from typing import Callable, Generic, Literal, Optional
 from typing_extensions import Self, TypeAlias, TypeGuard, TypeVar
 
 import aiohttp
@@ -108,9 +108,33 @@ def __str__(self) -> str:
         return f"[表情: {self.name}]"
 
 
+@dataclass
+class MusicShare(Element):
+    """音乐分享
+
+    音乐分享本质为 “小程序”
+    但是可以用不同方式发送
+    并且风控几率较小
+    """
+
+    kind: Literal["QQ", "Netease", "Migu", "Kugou", "Kuwo"]
+    title: str
+    summary: str
+    jump_url: str
+    picture_url: str
+    music_url: str
+    brief: str
+
+
 @dataclass
 class LightApp(Element):
+    """小程序
+
+    本框架不辅助音乐分享外的小程序构造与发送
+    """
+
     content: str
+    """JSON 内容"""
 
     def __str__(self) -> str:
         return "[小程序]"
@@ -275,7 +299,7 @@ def __repr__(self) -> str:
         return f"MarketFace(name={self.name})"
 
 
-TYPE_MAP = {
+DESERIALIZE_INV: dict[str, Callable[..., Element]] = {
     cls.__name__: cls
     for cls in (
         Reply,
@@ -293,3 +317,36 @@ def __repr__(self) -> str:
         Audio,
     )
 }
+
+__MUSIC_SHARE_APPID_MAP: dict[int, Literal["QQ", "Netease", "Migu", "Kugou", "Kuwo"]] = {
+    100497308: "QQ",
+    100495085: "Netease",
+    1101053067: "Migu",
+    205141: "Kugou",
+    100243533: "Kuwo",
+}
+
+
+def _light_app_deserializer(**data) -> Element:
+    import json
+    from contextlib import suppress
+
+    with suppress(ValueError, KeyError):
+        # MusicShare resolver
+        # https://github.com/mamoe/mirai/blob/893fb3e9f653623056f9c4bff73b4dac957cd2a2/mirai-core/src/commonMain/kotlin/message/data/lightApp.kt
+        app_data = json.loads(data["content"])
+        music_info = app_data["meta"]["music"]
+        return MusicShare(
+            kind=__MUSIC_SHARE_APPID_MAP[app_data["extra"]["appid"]],
+            title=music_info["title"],
+            summary=music_info["desc"],
+            jump_url=music_info["jumpUrl"],
+            picture_url=music_info["preview"],
+            music_url=music_info["musicUrl"],
+            brief=data["prompt"],
+        )
+
+    return LightApp(content=data["content"])
+
+
+DESERIALIZE_INV["LightApp"] = _light_app_deserializer

python/ichika/message/serializer.py

@@ -15,6 +15,7 @@
     FingerGuessing,
     FlashImage,
     Image,
+    LightApp,
     MarketFace,
     Reply,
 )
@@ -48,17 +49,18 @@ def wrapper(elem: Elem_T) -> dict[str, Any]:
 _serialize(FingerGuessing)(lambda t: {"choice": t.choice.name})
 _serialize(Face)(lambda t: {"index": t.index})
 _serialize(MarketFace)(lambda t: {"raw": t.raw})
+_serialize(LightApp)(lambda t: {"content": t.content})
 
 
 @_serialize(Image)
-def _(i: Image):
-    if i.raw is None:
+def _serialize_image(elem: Image):
+    if elem.raw is None:
         raise ValueError
-    return {"raw": i.raw}
+    return {"raw": elem.raw}
 
 
 @_serialize(FlashImage)
-def _(i: FlashImage):
-    if i.raw is None:
+def _serialize_flash_image(elem: FlashImage):
+    if elem.raw is None:
         raise ValueError
-    return {"raw": i.raw}
+    return {"raw": elem.raw}