ayaka.extension.user_input

 1import re
 2from typing import Optional
 3from pydantic import BaseModel, Field
 4from ..ayaka import AyakaApp
 5from ..depend import AyakaInput
 6from ..driver import MessageSegment
 7
 8
 9def find_user_by_uid(users: list, uid: int):
10    for user in users:
11        if uid == user["user_id"]:
12            return user
13
14
15def find_user_by_uname(users: list, uname: str):
16    for user in users:
17        _uname = user["card"] or user["nickname"]
18        if _uname == uname:
19            return user
20
21
22async def get_user(app: AyakaApp, at: MessageSegment):
23    str_at = str(at)
24    users = await app.bot.get_group_member_list(group_id=app.group_id)
25
26    if at.type == "at":
27        user = find_user_by_uid(users, int(at.data["qq"]))
28    elif at.type != "text":
29        return
30    elif re.search(r"^\d+$", str_at):
31        user = find_user_by_uid(users, int(str_at))
32    elif str_at.startswith("@"):
33        user = find_user_by_uname(users, str_at[1:])
34    else:
35        user = find_user_by_uname(users, str_at)
36
37    if user:
38        uid = user["user_id"]
39        uname = user["card"] or user["nickname"]
40        return User(id=uid, name=uname)
41
42
43class User(BaseModel):
44    id: int
45    name: str
46
47
48class UserInput(AyakaInput):
49    '''将用户输入的第一个参数（QQ号/群名片/@xxx）自动转换为User对象'''
50    user: Optional[User] = Field(description="查询目标的QQ号/名称/@xx")
51
52    @classmethod
53    async def create_by_app_get_params(cls, app: AyakaApp):
54        args = app.args
55        props = cls.props()
56        params = {k: v for k, v in zip(props, args)}
57        if args:
58            params["user"] = await get_user(app, args[0])
59        return params