diff --git a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/__init__.py b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/__init__.py index 320942f..d1a1fd0 100644 --- a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/__init__.py +++ b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/__init__.py @@ -7,8 +7,7 @@ import inspect import json import uuid -from functools import partial -from typing import Any, Optional, Union +from typing import Any, Optional, TypeVar, Union from typing_extensions import TypeGuard, override import aiohttp @@ -37,11 +36,6 @@ Response, Update, ) -from .utils import ( - camel_to_snake_case, - lower_camel_to_snake_case, - snake_to_lower_camel_case, -) __all__ = ["TelegramAdapter"] @@ -52,7 +46,9 @@ EVENT_MODELS: EventModels = {} for _, model in inspect.getmembers(event, inspect.isclass): if issubclass(model, TelegramEvent): - EVENT_MODELS[camel_to_snake_case(model.__name__)] = model + EVENT_MODELS[model.__event_type__] = model + +_T = TypeVar("_T") class TelegramAdapter(Adapter[TelegramEvent, Config], TelegramAPI): @@ -69,12 +65,6 @@ class TelegramAdapter(Adapter[TelegramEvent, Config], TelegramAPI): _secret_token: Optional[str] _update_offset: Optional[int] = None - @override - def __getattribute__(self, name: str) -> Any: - if not name.startswith("_") and hasattr(TelegramAPI, name): - return partial(self.call_api, name) - return super().__getattribute__(name) - @override async def startup(self) -> None: if self.config.adapter_type == "webhook": @@ -155,7 +145,7 @@ async def handle_telegram_event(self, update: Update) -> None: event_class_name = "" for k, v in update: if v is not None: - event_class_name = f"{k}_event" + event_class_name = k if event_class_name not in EVENT_MODELS: logger.warning( "Unknown event type", @@ -203,11 +193,19 @@ def is_file(v: Any) -> TypeGuard[InputFile]: if v is not None } - async def call_api(self, api: str, **params: Any) -> Any: + @override + async def call_api( + self, + api: str, + *, + response_type: Optional[type[_T]] = None, + **params: Any, + ) -> Optional[_T]: """调用 Telegram Bot API，协程会等待直到获得 API 响应。 Args: api: API 名称。 + response_type: API 响应类型。 **params: API 参数。 Returns: @@ -217,12 +215,10 @@ async def call_api(self, api: str, **params: Any) -> Any: NetworkError: 网络错误。 ActionFailed: API 请求响应 failed， API 操作失败。 """ - return_type_adapter: TypeAdapter[Response[Any]] = TypeAdapter(Response[Any]) - if hasattr(TelegramAPI, api): - sign = inspect.signature(getattr(TelegramAPI, api)) - return_type_adapter = TypeAdapter(Response[sign.return_annotation]) # type: ignore - - api = snake_to_lower_camel_case(api) + if response_type is None: + return_type_adapter = TypeAdapter(Response[Any]) + else: + return_type_adapter = TypeAdapter(Response[response_type]) data = self._format_telegram_api_params(**params) if isinstance(data, aiohttp.FormData): @@ -319,7 +315,7 @@ async def send( else: fields[k] = v return await self.call_api( - "send_" + lower_camel_to_snake_case(message.__class__.__name__), + "send" + message.__class__.__name__, chat_id=chat_id, **fields, **kwargs, diff --git a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/api.py b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/api.py index c95574b..bd7d861 100644 --- a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/api.py +++ b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/api.py @@ -2,9 +2,10 @@ # autogenerated by codegen.py, do not edit manually. # ruff: noqa: D101, D102, A002 # mypy: ignore-errors -# pylint: disable=redefined-builtin, unused-argument, missing-class-docstring, too-many-locals +# pylint: disable=redefined-builtin, missing-class-docstring, too-many-locals -from typing import Optional, Union +from abc import ABC, abstractmethod +from typing import Any, Optional, TypeVar, Union from .model import ( BotCommand, @@ -59,8 +60,21 @@ WebhookInfo, ) +_T = TypeVar("_T") -class TelegramAPI: + +class TelegramAPIBase(ABC): + @abstractmethod + async def call_api( + self, + api: str, + *, + response_type: Optional[type[_T]] = None, + **params: Any, + ) -> Any: ... + + +class TelegramAPI(TelegramAPIBase): async def get_updates( self, *, @@ -68,7 +82,15 @@ async def get_updates( limit: Optional[int] = None, timeout: Optional[int] = None, allowed_updates: Optional[list[str]] = None, - ) -> list[Update]: ... + ) -> list[Update]: + return await self.call_api( + "getUpdates", + response_type=list[Update], + offset=offset, + limit=limit, + timeout=timeout, + allowed_updates=allowed_updates, + ) async def set_webhook( self, @@ -80,29 +102,39 @@ async def set_webhook( allowed_updates: Optional[list[str]] = None, drop_pending_updates: Optional[bool] = None, secret_token: Optional[str] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "setWebhook", + response_type=bool, + url=url, + certificate=certificate, + ip_address=ip_address, + max_connections=max_connections, + allowed_updates=allowed_updates, + drop_pending_updates=drop_pending_updates, + secret_token=secret_token, + ) async def delete_webhook( - self, - *, - drop_pending_updates: Optional[bool] = None, - ) -> bool: ... + self, *, drop_pending_updates: Optional[bool] = None + ) -> bool: + return await self.call_api( + "deleteWebhook", + response_type=bool, + drop_pending_updates=drop_pending_updates, + ) - async def get_webhook_info( - self, - ) -> WebhookInfo: ... + async def get_webhook_info(self) -> WebhookInfo: + return await self.call_api("getWebhookInfo", response_type=WebhookInfo) - async def get_me( - self, - ) -> User: ... + async def get_me(self) -> User: + return await self.call_api("getMe", response_type=User) - async def log_out( - self, - ) -> bool: ... + async def log_out(self) -> bool: + return await self.call_api("logOut", response_type=bool) - async def close( - self, - ) -> bool: ... + async def close(self) -> bool: + return await self.call_api("close", response_type=bool) async def send_message( self, @@ -126,7 +158,23 @@ async def send_message( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendMessage", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + text=text, + parse_mode=parse_mode, + entities=entities, + link_preview_options=link_preview_options, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def forward_message( self, @@ -137,7 +185,17 @@ async def forward_message( message_thread_id: Optional[int] = None, disable_notification: Optional[bool] = None, protect_content: Optional[bool] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "forwardMessage", + response_type=Message, + chat_id=chat_id, + message_thread_id=message_thread_id, + from_chat_id=from_chat_id, + disable_notification=disable_notification, + protect_content=protect_content, + message_id=message_id, + ) async def forward_messages( self, @@ -148,7 +206,17 @@ async def forward_messages( message_thread_id: Optional[int] = None, disable_notification: Optional[bool] = None, protect_content: Optional[bool] = None, - ) -> list[MessageId]: ... + ) -> list[MessageId]: + return await self.call_api( + "forwardMessages", + response_type=list[MessageId], + chat_id=chat_id, + message_thread_id=message_thread_id, + from_chat_id=from_chat_id, + message_ids=message_ids, + disable_notification=disable_notification, + protect_content=protect_content, + ) async def copy_message( self, @@ -172,7 +240,23 @@ async def copy_message( ForceReply, ] ] = None, - ) -> MessageId: ... + ) -> MessageId: + return await self.call_api( + "copyMessage", + response_type=MessageId, + chat_id=chat_id, + message_thread_id=message_thread_id, + from_chat_id=from_chat_id, + message_id=message_id, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + show_caption_above_media=show_caption_above_media, + disable_notification=disable_notification, + protect_content=protect_content, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def copy_messages( self, @@ -184,7 +268,18 @@ async def copy_messages( disable_notification: Optional[bool] = None, protect_content: Optional[bool] = None, remove_caption: Optional[bool] = None, - ) -> list[MessageId]: ... + ) -> list[MessageId]: + return await self.call_api( + "copyMessages", + response_type=list[MessageId], + chat_id=chat_id, + message_thread_id=message_thread_id, + from_chat_id=from_chat_id, + message_ids=message_ids, + disable_notification=disable_notification, + protect_content=protect_content, + remove_caption=remove_caption, + ) async def send_photo( self, @@ -210,7 +305,25 @@ async def send_photo( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendPhoto", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + photo=photo, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + show_caption_above_media=show_caption_above_media, + has_spoiler=has_spoiler, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_audio( self, @@ -238,7 +351,27 @@ async def send_audio( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendAudio", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + audio=audio, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + duration=duration, + performer=performer, + title=title, + thumbnail=thumbnail, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_document( self, @@ -264,7 +397,25 @@ async def send_document( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendDocument", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + document=document, + thumbnail=thumbnail, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + disable_content_type_detection=disable_content_type_detection, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_video( self, @@ -295,7 +446,30 @@ async def send_video( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendVideo", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + video=video, + duration=duration, + width=width, + height=height, + thumbnail=thumbnail, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + show_caption_above_media=show_caption_above_media, + has_spoiler=has_spoiler, + supports_streaming=supports_streaming, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_animation( self, @@ -325,7 +499,29 @@ async def send_animation( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendAnimation", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + animation=animation, + duration=duration, + width=width, + height=height, + thumbnail=thumbnail, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + show_caption_above_media=show_caption_above_media, + has_spoiler=has_spoiler, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_voice( self, @@ -350,7 +546,24 @@ async def send_voice( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendVoice", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + voice=voice, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + duration=duration, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_video_note( self, @@ -374,7 +587,23 @@ async def send_video_note( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendVideoNote", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + video_note=video_note, + duration=duration, + length=length, + thumbnail=thumbnail, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_paid_media( self, @@ -399,7 +628,24 @@ async def send_paid_media( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendPaidMedia", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + star_count=star_count, + media=media, + payload=payload, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + show_caption_above_media=show_caption_above_media, + disable_notification=disable_notification, + protect_content=protect_content, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_media_group( self, @@ -417,7 +663,19 @@ async def send_media_group( protect_content: Optional[bool] = None, message_effect_id: Optional[str] = None, reply_parameters: Optional[ReplyParameters] = None, - ) -> list[Message]: ... + ) -> list[Message]: + return await self.call_api( + "sendMediaGroup", + response_type=list[Message], + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + media=media, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + ) async def send_location( self, @@ -443,7 +701,25 @@ async def send_location( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendLocation", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + latitude=latitude, + longitude=longitude, + horizontal_accuracy=horizontal_accuracy, + live_period=live_period, + heading=heading, + proximity_alert_radius=proximity_alert_radius, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_venue( self, @@ -471,7 +747,27 @@ async def send_venue( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendVenue", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + latitude=latitude, + longitude=longitude, + title=title, + address=address, + foursquare_id=foursquare_id, + foursquare_type=foursquare_type, + google_place_id=google_place_id, + google_place_type=google_place_type, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_contact( self, @@ -495,7 +791,23 @@ async def send_contact( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendContact", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + phone_number=phone_number, + first_name=first_name, + last_name=last_name, + vcard=vcard, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_poll( self, @@ -529,7 +841,33 @@ async def send_poll( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendPoll", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + question=question, + question_parse_mode=question_parse_mode, + question_entities=question_entities, + options=options, + is_anonymous=is_anonymous, + type=type, + allows_multiple_answers=allows_multiple_answers, + correct_option_id=correct_option_id, + explanation=explanation, + explanation_parse_mode=explanation_parse_mode, + explanation_entities=explanation_entities, + open_period=open_period, + close_date=close_date, + is_closed=is_closed, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_dice( self, @@ -550,7 +888,20 @@ async def send_dice( ForceReply, ] ] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendDice", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + emoji=emoji, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def send_chat_action( self, @@ -559,7 +910,15 @@ async def send_chat_action( action: str, business_connection_id: Optional[str] = None, message_thread_id: Optional[int] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "sendChatAction", + response_type=bool, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + action=action, + ) async def set_message_reaction( self, @@ -568,21 +927,29 @@ async def set_message_reaction( message_id: int, reaction: Optional[list[ReactionType]] = None, is_big: Optional[bool] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "setMessageReaction", + response_type=bool, + chat_id=chat_id, + message_id=message_id, + reaction=reaction, + is_big=is_big, + ) async def get_user_profile_photos( - self, - *, - user_id: int, - offset: Optional[int] = None, - limit: Optional[int] = None, - ) -> UserProfilePhotos: ... - - async def get_file( - self, - *, - file_id: str, - ) -> File: ... + self, *, user_id: int, offset: Optional[int] = None, limit: Optional[int] = None + ) -> UserProfilePhotos: + return await self.call_api( + "getUserProfilePhotos", + response_type=UserProfilePhotos, + user_id=user_id, + offset=offset, + limit=limit, + ) + + async def get_file(self, *, file_id: str) -> File: + return await self.call_api("getFile", response_type=File, file_id=file_id) async def ban_chat_member( self, @@ -591,7 +958,15 @@ async def ban_chat_member( user_id: int, until_date: Optional[int] = None, revoke_messages: Optional[bool] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "banChatMember", + response_type=bool, + chat_id=chat_id, + user_id=user_id, + until_date=until_date, + revoke_messages=revoke_messages, + ) async def unban_chat_member( self, @@ -599,7 +974,14 @@ async def unban_chat_member( chat_id: Union[int, str], user_id: int, only_if_banned: Optional[bool] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "unbanChatMember", + response_type=bool, + chat_id=chat_id, + user_id=user_id, + only_if_banned=only_if_banned, + ) async def restrict_chat_member( self, @@ -609,7 +991,16 @@ async def restrict_chat_member( permissions: ChatPermissions, use_independent_chat_permissions: Optional[bool] = None, until_date: Optional[int] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "restrictChatMember", + response_type=bool, + chat_id=chat_id, + user_id=user_id, + permissions=permissions, + use_independent_chat_permissions=use_independent_chat_permissions, + until_date=until_date, + ) async def promote_chat_member( self, @@ -631,29 +1022,59 @@ async def promote_chat_member( can_edit_messages: Optional[bool] = None, can_pin_messages: Optional[bool] = None, can_manage_topics: Optional[bool] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "promoteChatMember", + response_type=bool, + chat_id=chat_id, + user_id=user_id, + is_anonymous=is_anonymous, + can_manage_chat=can_manage_chat, + can_delete_messages=can_delete_messages, + can_manage_video_chats=can_manage_video_chats, + can_restrict_members=can_restrict_members, + can_promote_members=can_promote_members, + can_change_info=can_change_info, + can_invite_users=can_invite_users, + can_post_stories=can_post_stories, + can_edit_stories=can_edit_stories, + can_delete_stories=can_delete_stories, + can_post_messages=can_post_messages, + can_edit_messages=can_edit_messages, + can_pin_messages=can_pin_messages, + can_manage_topics=can_manage_topics, + ) async def set_chat_administrator_custom_title( - self, - *, - chat_id: Union[int, str], - user_id: int, - custom_title: str, - ) -> bool: ... + self, *, chat_id: Union[int, str], user_id: int, custom_title: str + ) -> bool: + return await self.call_api( + "setChatAdministratorCustomTitle", + response_type=bool, + chat_id=chat_id, + user_id=user_id, + custom_title=custom_title, + ) async def ban_chat_sender_chat( - self, - *, - chat_id: Union[int, str], - sender_chat_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], sender_chat_id: int + ) -> bool: + return await self.call_api( + "banChatSenderChat", + response_type=bool, + chat_id=chat_id, + sender_chat_id=sender_chat_id, + ) async def unban_chat_sender_chat( - self, - *, - chat_id: Union[int, str], - sender_chat_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], sender_chat_id: int + ) -> bool: + return await self.call_api( + "unbanChatSenderChat", + response_type=bool, + chat_id=chat_id, + sender_chat_id=sender_chat_id, + ) async def set_chat_permissions( self, @@ -661,13 +1082,19 @@ async def set_chat_permissions( chat_id: Union[int, str], permissions: ChatPermissions, use_independent_chat_permissions: Optional[bool] = None, - ) -> bool: ... - - async def export_chat_invite_link( - self, - *, - chat_id: Union[int, str], - ) -> str: ... + ) -> bool: + return await self.call_api( + "setChatPermissions", + response_type=bool, + chat_id=chat_id, + permissions=permissions, + use_independent_chat_permissions=use_independent_chat_permissions, + ) + + async def export_chat_invite_link(self, *, chat_id: Union[int, str]) -> str: + return await self.call_api( + "exportChatInviteLink", response_type=str, chat_id=chat_id + ) async def create_chat_invite_link( self, @@ -677,7 +1104,16 @@ async def create_chat_invite_link( expire_date: Optional[int] = None, member_limit: Optional[int] = None, creates_join_request: Optional[bool] = None, - ) -> ChatInviteLink: ... + ) -> ChatInviteLink: + return await self.call_api( + "createChatInviteLink", + response_type=ChatInviteLink, + chat_id=chat_id, + name=name, + expire_date=expire_date, + member_limit=member_limit, + creates_join_request=creates_join_request, + ) async def edit_chat_invite_link( self, @@ -688,7 +1124,17 @@ async def edit_chat_invite_link( expire_date: Optional[int] = None, member_limit: Optional[int] = None, creates_join_request: Optional[bool] = None, - ) -> ChatInviteLink: ... + ) -> ChatInviteLink: + return await self.call_api( + "editChatInviteLink", + response_type=ChatInviteLink, + chat_id=chat_id, + invite_link=invite_link, + name=name, + expire_date=expire_date, + member_limit=member_limit, + creates_join_request=creates_join_request, + ) async def create_chat_subscription_invite_link( self, @@ -697,63 +1143,83 @@ async def create_chat_subscription_invite_link( subscription_period: int, subscription_price: int, name: Optional[str] = None, - ) -> ChatInviteLink: ... + ) -> ChatInviteLink: + return await self.call_api( + "createChatSubscriptionInviteLink", + response_type=ChatInviteLink, + chat_id=chat_id, + name=name, + subscription_period=subscription_period, + subscription_price=subscription_price, + ) async def edit_chat_subscription_invite_link( - self, - *, - chat_id: Union[int, str], - invite_link: str, - name: Optional[str] = None, - ) -> ChatInviteLink: ... + self, *, chat_id: Union[int, str], invite_link: str, name: Optional[str] = None + ) -> ChatInviteLink: + return await self.call_api( + "editChatSubscriptionInviteLink", + response_type=ChatInviteLink, + chat_id=chat_id, + invite_link=invite_link, + name=name, + ) async def revoke_chat_invite_link( - self, - *, - chat_id: Union[int, str], - invite_link: str, - ) -> ChatInviteLink: ... + self, *, chat_id: Union[int, str], invite_link: str + ) -> ChatInviteLink: + return await self.call_api( + "revokeChatInviteLink", + response_type=ChatInviteLink, + chat_id=chat_id, + invite_link=invite_link, + ) async def approve_chat_join_request( - self, - *, - chat_id: Union[int, str], - user_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], user_id: int + ) -> bool: + return await self.call_api( + "approveChatJoinRequest", + response_type=bool, + chat_id=chat_id, + user_id=user_id, + ) async def decline_chat_join_request( - self, - *, - chat_id: Union[int, str], - user_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], user_id: int + ) -> bool: + return await self.call_api( + "declineChatJoinRequest", + response_type=bool, + chat_id=chat_id, + user_id=user_id, + ) async def set_chat_photo( - self, - *, - chat_id: Union[int, str], - photo: InputFile, - ) -> bool: ... - - async def delete_chat_photo( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... - - async def set_chat_title( - self, - *, - chat_id: Union[int, str], - title: str, - ) -> bool: ... + self, *, chat_id: Union[int, str], photo: InputFile + ) -> bool: + return await self.call_api( + "setChatPhoto", response_type=bool, chat_id=chat_id, photo=photo + ) + + async def delete_chat_photo(self, *, chat_id: Union[int, str]) -> bool: + return await self.call_api( + "deleteChatPhoto", response_type=bool, chat_id=chat_id + ) + + async def set_chat_title(self, *, chat_id: Union[int, str], title: str) -> bool: + return await self.call_api( + "setChatTitle", response_type=bool, chat_id=chat_id, title=title + ) async def set_chat_description( - self, - *, - chat_id: Union[int, str], - description: Optional[str] = None, - ) -> bool: ... + self, *, chat_id: Union[int, str], description: Optional[str] = None + ) -> bool: + return await self.call_api( + "setChatDescription", + response_type=bool, + chat_id=chat_id, + description=description, + ) async def pin_chat_message( self, @@ -762,7 +1228,15 @@ async def pin_chat_message( message_id: int, business_connection_id: Optional[str] = None, disable_notification: Optional[bool] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "pinChatMessage", + response_type=bool, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + disable_notification=disable_notification, + ) async def unpin_chat_message( self, @@ -770,61 +1244,66 @@ async def unpin_chat_message( chat_id: Union[int, str], business_connection_id: Optional[str] = None, message_id: Optional[int] = None, - ) -> bool: ... - - async def unpin_all_chat_messages( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... - - async def leave_chat( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... - - async def get_chat( - self, - *, - chat_id: Union[int, str], - ) -> ChatFullInfo: ... + ) -> bool: + return await self.call_api( + "unpinChatMessage", + response_type=bool, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + ) + + async def unpin_all_chat_messages(self, *, chat_id: Union[int, str]) -> bool: + return await self.call_api( + "unpinAllChatMessages", response_type=bool, chat_id=chat_id + ) + + async def leave_chat(self, *, chat_id: Union[int, str]) -> bool: + return await self.call_api("leaveChat", response_type=bool, chat_id=chat_id) + + async def get_chat(self, *, chat_id: Union[int, str]) -> ChatFullInfo: + return await self.call_api( + "getChat", response_type=ChatFullInfo, chat_id=chat_id + ) async def get_chat_administrators( - self, - *, - chat_id: Union[int, str], - ) -> list[ChatMember]: ... + self, *, chat_id: Union[int, str] + ) -> list[ChatMember]: + return await self.call_api( + "getChatAdministrators", response_type=list[ChatMember], chat_id=chat_id + ) - async def get_chat_member_count( - self, - *, - chat_id: Union[int, str], - ) -> int: ... + async def get_chat_member_count(self, *, chat_id: Union[int, str]) -> int: + return await self.call_api( + "getChatMemberCount", response_type=int, chat_id=chat_id + ) async def get_chat_member( - self, - *, - chat_id: Union[int, str], - user_id: int, - ) -> ChatMember: ... + self, *, chat_id: Union[int, str], user_id: int + ) -> ChatMember: + return await self.call_api( + "getChatMember", response_type=ChatMember, chat_id=chat_id, user_id=user_id + ) async def set_chat_sticker_set( - self, - *, - chat_id: Union[int, str], - sticker_set_name: str, - ) -> bool: ... - - async def delete_chat_sticker_set( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... - - async def get_forum_topic_icon_stickers( - self, - ) -> list[Sticker]: ... + self, *, chat_id: Union[int, str], sticker_set_name: str + ) -> bool: + return await self.call_api( + "setChatStickerSet", + response_type=bool, + chat_id=chat_id, + sticker_set_name=sticker_set_name, + ) + + async def delete_chat_sticker_set(self, *, chat_id: Union[int, str]) -> bool: + return await self.call_api( + "deleteChatStickerSet", response_type=bool, chat_id=chat_id + ) + + async def get_forum_topic_icon_stickers(self) -> list[Sticker]: + return await self.call_api( + "getForumTopicIconStickers", response_type=list[Sticker] + ) async def create_forum_topic( self, @@ -833,7 +1312,15 @@ async def create_forum_topic( name: str, icon_color: Optional[int] = None, icon_custom_emoji_id: Optional[str] = None, - ) -> ForumTopic: ... + ) -> ForumTopic: + return await self.call_api( + "createForumTopic", + response_type=ForumTopic, + chat_id=chat_id, + name=name, + icon_color=icon_color, + icon_custom_emoji_id=icon_custom_emoji_id, + ) async def edit_forum_topic( self, @@ -842,72 +1329,89 @@ async def edit_forum_topic( message_thread_id: int, name: Optional[str] = None, icon_custom_emoji_id: Optional[str] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "editForumTopic", + response_type=bool, + chat_id=chat_id, + message_thread_id=message_thread_id, + name=name, + icon_custom_emoji_id=icon_custom_emoji_id, + ) async def close_forum_topic( - self, - *, - chat_id: Union[int, str], - message_thread_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], message_thread_id: int + ) -> bool: + return await self.call_api( + "closeForumTopic", + response_type=bool, + chat_id=chat_id, + message_thread_id=message_thread_id, + ) async def reopen_forum_topic( - self, - *, - chat_id: Union[int, str], - message_thread_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], message_thread_id: int + ) -> bool: + return await self.call_api( + "reopenForumTopic", + response_type=bool, + chat_id=chat_id, + message_thread_id=message_thread_id, + ) async def delete_forum_topic( - self, - *, - chat_id: Union[int, str], - message_thread_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], message_thread_id: int + ) -> bool: + return await self.call_api( + "deleteForumTopic", + response_type=bool, + chat_id=chat_id, + message_thread_id=message_thread_id, + ) async def unpin_all_forum_topic_messages( - self, - *, - chat_id: Union[int, str], - message_thread_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], message_thread_id: int + ) -> bool: + return await self.call_api( + "unpinAllForumTopicMessages", + response_type=bool, + chat_id=chat_id, + message_thread_id=message_thread_id, + ) async def edit_general_forum_topic( - self, - *, - chat_id: Union[int, str], - name: str, - ) -> bool: ... - - async def close_general_forum_topic( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... - - async def reopen_general_forum_topic( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... - - async def hide_general_forum_topic( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... - - async def unhide_general_forum_topic( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... + self, *, chat_id: Union[int, str], name: str + ) -> bool: + return await self.call_api( + "editGeneralForumTopic", response_type=bool, chat_id=chat_id, name=name + ) + + async def close_general_forum_topic(self, *, chat_id: Union[int, str]) -> bool: + return await self.call_api( + "closeGeneralForumTopic", response_type=bool, chat_id=chat_id + ) + + async def reopen_general_forum_topic(self, *, chat_id: Union[int, str]) -> bool: + return await self.call_api( + "reopenGeneralForumTopic", response_type=bool, chat_id=chat_id + ) + + async def hide_general_forum_topic(self, *, chat_id: Union[int, str]) -> bool: + return await self.call_api( + "hideGeneralForumTopic", response_type=bool, chat_id=chat_id + ) + + async def unhide_general_forum_topic(self, *, chat_id: Union[int, str]) -> bool: + return await self.call_api( + "unhideGeneralForumTopic", response_type=bool, chat_id=chat_id + ) async def unpin_all_general_forum_topic_messages( - self, - *, - chat_id: Union[int, str], - ) -> bool: ... + self, *, chat_id: Union[int, str] + ) -> bool: + return await self.call_api( + "unpinAllGeneralForumTopicMessages", response_type=bool, chat_id=chat_id + ) async def answer_callback_query( self, @@ -917,20 +1421,35 @@ async def answer_callback_query( show_alert: Optional[bool] = None, url: Optional[str] = None, cache_time: Optional[int] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "answerCallbackQuery", + response_type=bool, + callback_query_id=callback_query_id, + text=text, + show_alert=show_alert, + url=url, + cache_time=cache_time, + ) async def get_user_chat_boosts( - self, - *, - chat_id: Union[int, str], - user_id: int, - ) -> UserChatBoosts: ... + self, *, chat_id: Union[int, str], user_id: int + ) -> UserChatBoosts: + return await self.call_api( + "getUserChatBoosts", + response_type=UserChatBoosts, + chat_id=chat_id, + user_id=user_id, + ) async def get_business_connection( - self, - *, - business_connection_id: str, - ) -> BusinessConnection: ... + self, *, business_connection_id: str + ) -> BusinessConnection: + return await self.call_api( + "getBusinessConnection", + response_type=BusinessConnection, + business_connection_id=business_connection_id, + ) async def set_my_commands( self, @@ -938,86 +1457,132 @@ async def set_my_commands( commands: list[BotCommand], scope: Optional[BotCommandScope] = None, language_code: Optional[str] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "setMyCommands", + response_type=bool, + commands=commands, + scope=scope, + language_code=language_code, + ) async def delete_my_commands( self, *, scope: Optional[BotCommandScope] = None, language_code: Optional[str] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "deleteMyCommands", + response_type=bool, + scope=scope, + language_code=language_code, + ) async def get_my_commands( self, *, scope: Optional[BotCommandScope] = None, language_code: Optional[str] = None, - ) -> list[BotCommand]: ... + ) -> list[BotCommand]: + return await self.call_api( + "getMyCommands", + response_type=list[BotCommand], + scope=scope, + language_code=language_code, + ) async def set_my_name( - self, - *, - name: Optional[str] = None, - language_code: Optional[str] = None, - ) -> bool: ... + self, *, name: Optional[str] = None, language_code: Optional[str] = None + ) -> bool: + return await self.call_api( + "setMyName", response_type=bool, name=name, language_code=language_code + ) - async def get_my_name( - self, - *, - language_code: Optional[str] = None, - ) -> BotName: ... + async def get_my_name(self, *, language_code: Optional[str] = None) -> BotName: + return await self.call_api( + "getMyName", response_type=BotName, language_code=language_code + ) async def set_my_description( - self, - *, - description: Optional[str] = None, - language_code: Optional[str] = None, - ) -> bool: ... + self, *, description: Optional[str] = None, language_code: Optional[str] = None + ) -> bool: + return await self.call_api( + "setMyDescription", + response_type=bool, + description=description, + language_code=language_code, + ) async def get_my_description( - self, - *, - language_code: Optional[str] = None, - ) -> BotDescription: ... + self, *, language_code: Optional[str] = None + ) -> BotDescription: + return await self.call_api( + "getMyDescription", + response_type=BotDescription, + language_code=language_code, + ) async def set_my_short_description( self, *, short_description: Optional[str] = None, language_code: Optional[str] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "setMyShortDescription", + response_type=bool, + short_description=short_description, + language_code=language_code, + ) async def get_my_short_description( - self, - *, - language_code: Optional[str] = None, - ) -> BotShortDescription: ... + self, *, language_code: Optional[str] = None + ) -> BotShortDescription: + return await self.call_api( + "getMyShortDescription", + response_type=BotShortDescription, + language_code=language_code, + ) async def set_chat_menu_button( - self, - *, - chat_id: Optional[int] = None, - menu_button: Optional[MenuButton] = None, - ) -> bool: ... + self, *, chat_id: Optional[int] = None, menu_button: Optional[MenuButton] = None + ) -> bool: + return await self.call_api( + "setChatMenuButton", + response_type=bool, + chat_id=chat_id, + menu_button=menu_button, + ) async def get_chat_menu_button( - self, - *, - chat_id: Optional[int] = None, - ) -> MenuButton: ... + self, *, chat_id: Optional[int] = None + ) -> MenuButton: + return await self.call_api( + "getChatMenuButton", response_type=MenuButton, chat_id=chat_id + ) async def set_my_default_administrator_rights( self, *, rights: Optional[ChatAdministratorRights] = None, for_channels: Optional[bool] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "setMyDefaultAdministratorRights", + response_type=bool, + rights=rights, + for_channels=for_channels, + ) async def get_my_default_administrator_rights( - self, - *, - for_channels: Optional[bool] = None, - ) -> ChatAdministratorRights: ... + self, *, for_channels: Optional[bool] = None + ) -> ChatAdministratorRights: + return await self.call_api( + "getMyDefaultAdministratorRights", + response_type=ChatAdministratorRights, + for_channels=for_channels, + ) async def edit_message_text( self, @@ -1031,7 +1596,20 @@ async def edit_message_text( entities: Optional[list[MessageEntity]] = None, link_preview_options: Optional[LinkPreviewOptions] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Union[Message, bool]: ... + ) -> Union[Message, bool]: + return await self.call_api( + "editMessageText", + response_type=Union[Message, bool], + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + inline_message_id=inline_message_id, + text=text, + parse_mode=parse_mode, + entities=entities, + link_preview_options=link_preview_options, + reply_markup=reply_markup, + ) async def edit_message_caption( self, @@ -1045,7 +1623,20 @@ async def edit_message_caption( caption_entities: Optional[list[MessageEntity]] = None, show_caption_above_media: Optional[bool] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Union[Message, bool]: ... + ) -> Union[Message, bool]: + return await self.call_api( + "editMessageCaption", + response_type=Union[Message, bool], + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + inline_message_id=inline_message_id, + caption=caption, + parse_mode=parse_mode, + caption_entities=caption_entities, + show_caption_above_media=show_caption_above_media, + reply_markup=reply_markup, + ) async def edit_message_media( self, @@ -1056,7 +1647,17 @@ async def edit_message_media( message_id: Optional[int] = None, inline_message_id: Optional[str] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Union[Message, bool]: ... + ) -> Union[Message, bool]: + return await self.call_api( + "editMessageMedia", + response_type=Union[Message, bool], + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + inline_message_id=inline_message_id, + media=media, + reply_markup=reply_markup, + ) async def edit_message_live_location( self, @@ -1072,7 +1673,22 @@ async def edit_message_live_location( heading: Optional[int] = None, proximity_alert_radius: Optional[int] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Union[Message, bool]: ... + ) -> Union[Message, bool]: + return await self.call_api( + "editMessageLiveLocation", + response_type=Union[Message, bool], + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + inline_message_id=inline_message_id, + latitude=latitude, + longitude=longitude, + live_period=live_period, + horizontal_accuracy=horizontal_accuracy, + heading=heading, + proximity_alert_radius=proximity_alert_radius, + reply_markup=reply_markup, + ) async def stop_message_live_location( self, @@ -1082,7 +1698,16 @@ async def stop_message_live_location( message_id: Optional[int] = None, inline_message_id: Optional[str] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Union[Message, bool]: ... + ) -> Union[Message, bool]: + return await self.call_api( + "stopMessageLiveLocation", + response_type=Union[Message, bool], + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + inline_message_id=inline_message_id, + reply_markup=reply_markup, + ) async def edit_message_reply_markup( self, @@ -1092,7 +1717,16 @@ async def edit_message_reply_markup( message_id: Optional[int] = None, inline_message_id: Optional[str] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Union[Message, bool]: ... + ) -> Union[Message, bool]: + return await self.call_api( + "editMessageReplyMarkup", + response_type=Union[Message, bool], + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + inline_message_id=inline_message_id, + reply_markup=reply_markup, + ) async def stop_poll( self, @@ -1101,21 +1735,32 @@ async def stop_poll( message_id: int, business_connection_id: Optional[str] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Poll: ... + ) -> Poll: + return await self.call_api( + "stopPoll", + response_type=Poll, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_id=message_id, + reply_markup=reply_markup, + ) async def delete_message( - self, - *, - chat_id: Union[int, str], - message_id: int, - ) -> bool: ... + self, *, chat_id: Union[int, str], message_id: int + ) -> bool: + return await self.call_api( + "deleteMessage", response_type=bool, chat_id=chat_id, message_id=message_id + ) async def delete_messages( - self, - *, - chat_id: Union[int, str], - message_ids: list[int], - ) -> bool: ... + self, *, chat_id: Union[int, str], message_ids: list[int] + ) -> bool: + return await self.call_api( + "deleteMessages", + response_type=bool, + chat_id=chat_id, + message_ids=message_ids, + ) async def send_sticker( self, @@ -1137,27 +1782,44 @@ async def send_sticker( ForceReply, ] ] = None, - ) -> Message: ... - - async def get_sticker_set( - self, - *, - name: str, - ) -> StickerSet: ... + ) -> Message: + return await self.call_api( + "sendSticker", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + sticker=sticker, + emoji=emoji, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) + + async def get_sticker_set(self, *, name: str) -> StickerSet: + return await self.call_api("getStickerSet", response_type=StickerSet, name=name) async def get_custom_emoji_stickers( - self, - *, - custom_emoji_ids: list[str], - ) -> list[Sticker]: ... + self, *, custom_emoji_ids: list[str] + ) -> list[Sticker]: + return await self.call_api( + "getCustomEmojiStickers", + response_type=list[Sticker], + custom_emoji_ids=custom_emoji_ids, + ) async def upload_sticker_file( - self, - *, - user_id: int, - sticker: InputFile, - sticker_format: str, - ) -> File: ... + self, *, user_id: int, sticker: InputFile, sticker_format: str + ) -> File: + return await self.call_api( + "uploadStickerFile", + response_type=File, + user_id=user_id, + sticker=sticker, + sticker_format=sticker_format, + ) async def create_new_sticker_set( self, @@ -1168,65 +1830,85 @@ async def create_new_sticker_set( stickers: list[InputSticker], sticker_type: Optional[str] = None, needs_repainting: Optional[bool] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "createNewStickerSet", + response_type=bool, + user_id=user_id, + name=name, + title=title, + stickers=stickers, + sticker_type=sticker_type, + needs_repainting=needs_repainting, + ) async def add_sticker_to_set( - self, - *, - user_id: int, - name: str, - sticker: InputSticker, - ) -> bool: ... - - async def set_sticker_position_in_set( - self, - *, - sticker: str, - position: int, - ) -> bool: ... - - async def delete_sticker_from_set( - self, - *, - sticker: str, - ) -> bool: ... + self, *, user_id: int, name: str, sticker: InputSticker + ) -> bool: + return await self.call_api( + "addStickerToSet", + response_type=bool, + user_id=user_id, + name=name, + sticker=sticker, + ) + + async def set_sticker_position_in_set(self, *, sticker: str, position: int) -> bool: + return await self.call_api( + "setStickerPositionInSet", + response_type=bool, + sticker=sticker, + position=position, + ) + + async def delete_sticker_from_set(self, *, sticker: str) -> bool: + return await self.call_api( + "deleteStickerFromSet", response_type=bool, sticker=sticker + ) async def replace_sticker_in_set( - self, - *, - user_id: int, - name: str, - old_sticker: str, - sticker: InputSticker, - ) -> bool: ... + self, *, user_id: int, name: str, old_sticker: str, sticker: InputSticker + ) -> bool: + return await self.call_api( + "replaceStickerInSet", + response_type=bool, + user_id=user_id, + name=name, + old_sticker=old_sticker, + sticker=sticker, + ) async def set_sticker_emoji_list( - self, - *, - sticker: str, - emoji_list: list[str], - ) -> bool: ... + self, *, sticker: str, emoji_list: list[str] + ) -> bool: + return await self.call_api( + "setStickerEmojiList", + response_type=bool, + sticker=sticker, + emoji_list=emoji_list, + ) async def set_sticker_keywords( - self, - *, - sticker: str, - keywords: Optional[list[str]] = None, - ) -> bool: ... + self, *, sticker: str, keywords: Optional[list[str]] = None + ) -> bool: + return await self.call_api( + "setStickerKeywords", response_type=bool, sticker=sticker, keywords=keywords + ) async def set_sticker_mask_position( - self, - *, - sticker: str, - mask_position: Optional[MaskPosition] = None, - ) -> bool: ... - - async def set_sticker_set_title( - self, - *, - name: str, - title: str, - ) -> bool: ... + self, *, sticker: str, mask_position: Optional[MaskPosition] = None + ) -> bool: + return await self.call_api( + "setStickerMaskPosition", + response_type=bool, + sticker=sticker, + mask_position=mask_position, + ) + + async def set_sticker_set_title(self, *, name: str, title: str) -> bool: + return await self.call_api( + "setStickerSetTitle", response_type=bool, name=name, title=title + ) async def set_sticker_set_thumbnail( self, @@ -1235,20 +1917,28 @@ async def set_sticker_set_thumbnail( user_id: int, format: str, thumbnail: Optional[Union[InputFile, str]] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "setStickerSetThumbnail", + response_type=bool, + name=name, + user_id=user_id, + thumbnail=thumbnail, + format=format, + ) async def set_custom_emoji_sticker_set_thumbnail( - self, - *, - name: str, - custom_emoji_id: Optional[str] = None, - ) -> bool: ... - - async def delete_sticker_set( - self, - *, - name: str, - ) -> bool: ... + self, *, name: str, custom_emoji_id: Optional[str] = None + ) -> bool: + return await self.call_api( + "setCustomEmojiStickerSetThumbnail", + response_type=bool, + name=name, + custom_emoji_id=custom_emoji_id, + ) + + async def delete_sticker_set(self, *, name: str) -> bool: + return await self.call_api("deleteStickerSet", response_type=bool, name=name) async def answer_inline_query( self, @@ -1259,14 +1949,27 @@ async def answer_inline_query( is_personal: Optional[bool] = None, next_offset: Optional[str] = None, button: Optional[InlineQueryResultsButton] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "answerInlineQuery", + response_type=bool, + inline_query_id=inline_query_id, + results=results, + cache_time=cache_time, + is_personal=is_personal, + next_offset=next_offset, + button=button, + ) async def answer_web_app_query( - self, - *, - web_app_query_id: str, - result: InlineQueryResult, - ) -> SentWebAppMessage: ... + self, *, web_app_query_id: str, result: InlineQueryResult + ) -> SentWebAppMessage: + return await self.call_api( + "answerWebAppQuery", + response_type=SentWebAppMessage, + web_app_query_id=web_app_query_id, + result=result, + ) async def send_invoice( self, @@ -1299,7 +2002,39 @@ async def send_invoice( message_effect_id: Optional[str] = None, reply_parameters: Optional[ReplyParameters] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendInvoice", + response_type=Message, + chat_id=chat_id, + message_thread_id=message_thread_id, + title=title, + description=description, + payload=payload, + provider_token=provider_token, + currency=currency, + prices=prices, + max_tip_amount=max_tip_amount, + suggested_tip_amounts=suggested_tip_amounts, + start_parameter=start_parameter, + provider_data=provider_data, + photo_url=photo_url, + photo_size=photo_size, + photo_width=photo_width, + photo_height=photo_height, + need_name=need_name, + need_phone_number=need_phone_number, + need_email=need_email, + need_shipping_address=need_shipping_address, + send_phone_number_to_provider=send_phone_number_to_provider, + send_email_to_provider=send_email_to_provider, + is_flexible=is_flexible, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def create_invoice_link( self, @@ -1324,7 +2059,31 @@ async def create_invoice_link( send_phone_number_to_provider: Optional[bool] = None, send_email_to_provider: Optional[bool] = None, is_flexible: Optional[bool] = None, - ) -> str: ... + ) -> str: + return await self.call_api( + "createInvoiceLink", + response_type=str, + title=title, + description=description, + payload=payload, + provider_token=provider_token, + currency=currency, + prices=prices, + max_tip_amount=max_tip_amount, + suggested_tip_amounts=suggested_tip_amounts, + provider_data=provider_data, + photo_url=photo_url, + photo_size=photo_size, + photo_width=photo_width, + photo_height=photo_height, + need_name=need_name, + need_phone_number=need_phone_number, + need_email=need_email, + need_shipping_address=need_shipping_address, + send_phone_number_to_provider=send_phone_number_to_provider, + send_email_to_provider=send_email_to_provider, + is_flexible=is_flexible, + ) async def answer_shipping_query( self, @@ -1333,7 +2092,15 @@ async def answer_shipping_query( ok: bool, shipping_options: Optional[list[ShippingOption]] = None, error_message: Optional[str] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "answerShippingQuery", + response_type=bool, + shipping_query_id=shipping_query_id, + ok=ok, + shipping_options=shipping_options, + error_message=error_message, + ) async def answer_pre_checkout_query( self, @@ -1341,28 +2108,41 @@ async def answer_pre_checkout_query( pre_checkout_query_id: str, ok: bool, error_message: Optional[str] = None, - ) -> bool: ... + ) -> bool: + return await self.call_api( + "answerPreCheckoutQuery", + response_type=bool, + pre_checkout_query_id=pre_checkout_query_id, + ok=ok, + error_message=error_message, + ) async def get_star_transactions( - self, - *, - offset: Optional[int] = None, - limit: Optional[int] = None, - ) -> StarTransactions: ... + self, *, offset: Optional[int] = None, limit: Optional[int] = None + ) -> StarTransactions: + return await self.call_api( + "getStarTransactions", + response_type=StarTransactions, + offset=offset, + limit=limit, + ) async def refund_star_payment( - self, - *, - user_id: int, - telegram_payment_charge_id: str, - ) -> bool: ... + self, *, user_id: int, telegram_payment_charge_id: str + ) -> bool: + return await self.call_api( + "refundStarPayment", + response_type=bool, + user_id=user_id, + telegram_payment_charge_id=telegram_payment_charge_id, + ) async def set_passport_data_errors( - self, - *, - user_id: int, - errors: list[PassportElementError], - ) -> bool: ... + self, *, user_id: int, errors: list[PassportElementError] + ) -> bool: + return await self.call_api( + "setPassportDataErrors", response_type=bool, user_id=user_id, errors=errors + ) async def send_game( self, @@ -1376,7 +2156,20 @@ async def send_game( message_effect_id: Optional[str] = None, reply_parameters: Optional[ReplyParameters] = None, reply_markup: Optional[InlineKeyboardMarkup] = None, - ) -> Message: ... + ) -> Message: + return await self.call_api( + "sendGame", + response_type=Message, + business_connection_id=business_connection_id, + chat_id=chat_id, + message_thread_id=message_thread_id, + game_short_name=game_short_name, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + ) async def set_game_score( self, @@ -1388,7 +2181,18 @@ async def set_game_score( chat_id: Optional[int] = None, message_id: Optional[int] = None, inline_message_id: Optional[str] = None, - ) -> Union[Message, bool]: ... + ) -> Union[Message, bool]: + return await self.call_api( + "setGameScore", + response_type=Union[Message, bool], + user_id=user_id, + score=score, + force=force, + disable_edit_message=disable_edit_message, + chat_id=chat_id, + message_id=message_id, + inline_message_id=inline_message_id, + ) async def get_game_high_scores( self, @@ -1397,4 +2201,12 @@ async def get_game_high_scores( chat_id: Optional[int] = None, message_id: Optional[int] = None, inline_message_id: Optional[str] = None, - ) -> list[GameHighScore]: ... + ) -> list[GameHighScore]: + return await self.call_api( + "getGameHighScores", + response_type=list[GameHighScore], + user_id=user_id, + chat_id=chat_id, + message_id=message_id, + inline_message_id=inline_message_id, + ) diff --git a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/__init__.py b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/__init__.py new file mode 100644 index 0000000..c6931ce --- /dev/null +++ b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/__init__.py @@ -0,0 +1,6 @@ +"""Telegram 适配器事件。""" + +# ruff: noqa: F403 +from .base import * +from .message import * +from .other import * diff --git a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/base.py b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/base.py new file mode 100644 index 0000000..c50ed88 --- /dev/null +++ b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/base.py @@ -0,0 +1,23 @@ +"""事件基类。""" + +from typing import TYPE_CHECKING + +from alicebot.event import Event + +from ..model import Update # noqa: TID252 + +if TYPE_CHECKING: + from .. import TelegramAdapter # noqa: TID252 + + +class TelegramEvent(Event["TelegramAdapter"]): + """Telegram Event Baseclass.""" + + __event_type__: str = "" + + update: Update + + @property + def update_id(self) -> int: + """The update's unique identifier.""" + return self.update.update_id diff --git a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/message.py b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/message.py new file mode 100644 index 0000000..0425152 --- /dev/null +++ b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/message.py @@ -0,0 +1,84 @@ +"""Telegram 适配器事件。""" + +# ruff: noqa: TID252 +from typing import TYPE_CHECKING, Any, Optional, Union +from typing_extensions import override + +from alicebot.event import MessageEvent as BaseMessageEvent + +from ..media import TelegramMedia +from ..message import TelegramMessage +from ..model import ( + ForceReply, + InlineKeyboardMarkup, + Message, + ReplyKeyboardMarkup, + ReplyKeyboardRemove, + ReplyParameters, +) +from .base import TelegramEvent + +if TYPE_CHECKING: + from .. import TelegramAdapter + + +class MessageEvent(TelegramEvent, BaseMessageEvent["TelegramAdapter"]): + """New incoming message of any kind - text, photo, sticker, etc.""" + + __event_type__ = "message" + + @property + def message(self) -> Message: + """The message object.""" + assert self.update.message is not None + return self.update.message + + def get_message(self) -> TelegramMessage: + """获取 TelegramMessage 对象。 + + Returns: + TelegramMessage 对象。 + """ + return TelegramMessage.from_entities( + text=self.message.text or "", + entities=self.message.entities or [], + ) + + @override + def get_sender_id(self) -> Union[None, int, str]: + if self.message.from_ is None: + return None + return self.message.from_.id + + @override + def get_plain_text(self) -> str: + return self.message.text or "" + + @override + async def reply( + self, + message: Union[str, TelegramMessage, TelegramMedia], + disable_notification: Optional[bool] = None, + protect_content: Optional[bool] = None, + message_effect_id: Optional[str] = None, + reply_parameters: Optional[ReplyParameters] = None, + reply_markup: Optional[ + Union[ + InlineKeyboardMarkup, + ReplyKeyboardMarkup, + ReplyKeyboardRemove, + ForceReply, + ] + ] = None, + **kwargs: Any, + ) -> Any: + return await self.adapter.send( + message, + chat_id=self.message.chat.id, + disable_notification=disable_notification, + protect_content=protect_content, + message_effect_id=message_effect_id, + reply_parameters=reply_parameters, + reply_markup=reply_markup, + **kwargs, + ) diff --git a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event.py b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/other.py similarity index 73% rename from packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event.py rename to packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/other.py index 7cbbec4..2a53b4f 100644 --- a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event.py +++ b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/event/other.py @@ -1,14 +1,8 @@ """Telegram 适配器事件。""" +# autogenerated by codegen.py, do not edit manually. +# ruff: noqa: TID252 -from typing import TYPE_CHECKING, Any, Optional, Union -from typing_extensions import override - -from alicebot.event import Event -from alicebot.event import MessageEvent as BaseMessageEvent - -from .media import TelegramMedia -from .message import TelegramMessage -from .model import ( +from ..model import ( BusinessConnection, BusinessMessagesDeleted, CallbackQuery, @@ -17,100 +11,24 @@ ChatJoinRequest, ChatMemberUpdated, ChosenInlineResult, - ForceReply, - InlineKeyboardMarkup, InlineQuery, Message, MessageReactionCountUpdated, MessageReactionUpdated, + PaidMediaPurchased, Poll, PollAnswer, PreCheckoutQuery, - ReplyKeyboardMarkup, - ReplyKeyboardRemove, - ReplyParameters, ShippingQuery, - Update, ) - -if TYPE_CHECKING: - from . import TelegramAdapter - - -class TelegramEvent(Event["TelegramAdapter"]): - """Telegram Event Baseclass.""" - - update: Update - - @property - def update_id(self) -> int: - """The update's unique identifier.""" - return self.update.update_id - - -class MessageEvent(TelegramEvent, BaseMessageEvent["TelegramAdapter"]): - """New incoming message of any kind - text, photo, sticker, etc.""" - - @property - def message(self) -> Message: - """The message object.""" - assert self.update.message is not None - return self.update.message - - def get_message(self) -> TelegramMessage: - """获取 TelegramMessage 对象。 - - Returns: - TelegramMessage 对象。 - """ - return TelegramMessage.from_entities( - text=self.message.text or "", - entities=self.message.entities or [], - ) - - @override - def get_sender_id(self) -> Union[None, int, str]: - if self.message.from_ is None: - return None - return self.message.from_.id - - @override - def get_plain_text(self) -> str: - return self.message.text or "" - - @override - async def reply( - self, - message: Union[str, TelegramMessage, TelegramMedia], - disable_notification: Optional[bool] = None, - protect_content: Optional[bool] = None, - message_effect_id: Optional[str] = None, - reply_parameters: Optional[ReplyParameters] = None, - reply_markup: Optional[ - Union[ - InlineKeyboardMarkup, - ReplyKeyboardMarkup, - ReplyKeyboardRemove, - ForceReply, - ] - ] = None, - **kwargs: Any, - ) -> Any: - return await self.adapter.send( - message, - chat_id=self.message.chat.id, - disable_notification=disable_notification, - protect_content=protect_content, - message_effect_id=message_effect_id, - reply_parameters=reply_parameters, - reply_markup=reply_markup, - **kwargs, - ) +from .base import TelegramEvent class EditedMessageEvent(TelegramEvent): """New version of a message that is known to the bot and was edited.""" + __event_type__ = "edited_message" + @property def edited_message(self) -> Message: """The edited_message object.""" @@ -119,7 +37,9 @@ def edited_message(self) -> Message: class ChannelPostEvent(TelegramEvent): - """New incoming channel post of any kind - text, photo, sticker, etc.""" + """New incoming channel post of any kind - text, photo, sticker, etc..""" + + __event_type__ = "channel_post" @property def channel_post(self) -> Message: @@ -131,6 +51,8 @@ def channel_post(self) -> Message: class EditedChannelPostEvent(TelegramEvent): """New version of a channel post that is known to the bot and was edited.""" + __event_type__ = "edited_channel_post" + @property def edited_channel_post(self) -> Message: """The edited_channel_post object.""" @@ -139,7 +61,9 @@ def edited_channel_post(self) -> Message: class BusinessConnectionEvent(TelegramEvent): - """The bot was connected to or disconnected from a business account.""" + """The bot was connected to or disconnected from a business account, or a user edited an existing connection with the bot.""" + + __event_type__ = "business_connection" @property def business_connection(self) -> BusinessConnection: @@ -151,6 +75,8 @@ def business_connection(self) -> BusinessConnection: class BusinessMessageEvent(TelegramEvent): """New message from a connected business account.""" + __event_type__ = "business_message" + @property def business_message(self) -> Message: """The business_message object.""" @@ -161,6 +87,8 @@ def business_message(self) -> Message: class EditedBusinessMessageEvent(TelegramEvent): """New version of a message from a connected business account.""" + __event_type__ = "edited_business_message" + @property def edited_business_message(self) -> Message: """The edited_business_message object.""" @@ -171,6 +99,8 @@ def edited_business_message(self) -> Message: class DeletedBusinessMessagesEvent(TelegramEvent): """Messages were deleted from a connected business account.""" + __event_type__ = "deleted_business_messages" + @property def deleted_business_messages(self) -> BusinessMessagesDeleted: """The deleted_business_messages object.""" @@ -181,6 +111,8 @@ def deleted_business_messages(self) -> BusinessMessagesDeleted: class MessageReactionEvent(TelegramEvent): """A reaction to a message was changed by a user.""" + __event_type__ = "message_reaction" + @property def message_reaction(self) -> MessageReactionUpdated: """The message_reaction object.""" @@ -191,6 +123,8 @@ def message_reaction(self) -> MessageReactionUpdated: class MessageReactionCountEvent(TelegramEvent): """Reactions to a message with anonymous reactions were changed.""" + __event_type__ = "message_reaction_count" + @property def message_reaction_count(self) -> MessageReactionCountUpdated: """The message_reaction_count object.""" @@ -201,6 +135,8 @@ def message_reaction_count(self) -> MessageReactionCountUpdated: class InlineQueryEvent(TelegramEvent): """New incoming inline query.""" + __event_type__ = "inline_query" + @property def inline_query(self) -> InlineQuery: """The inline_query object.""" @@ -211,6 +147,8 @@ def inline_query(self) -> InlineQuery: class ChosenInlineResultEvent(TelegramEvent): """The result of an inline query that was chosen by a user and sent to their chat partner.""" + __event_type__ = "chosen_inline_result" + @property def chosen_inline_result(self) -> ChosenInlineResult: """The chosen_inline_result object.""" @@ -221,6 +159,8 @@ def chosen_inline_result(self) -> ChosenInlineResult: class CallbackQueryEvent(TelegramEvent): """New incoming callback query.""" + __event_type__ = "callback_query" + @property def callback_query(self) -> CallbackQuery: """The callback_query object.""" @@ -231,6 +171,8 @@ def callback_query(self) -> CallbackQuery: class ShippingQueryEvent(TelegramEvent): """New incoming shipping query.""" + __event_type__ = "shipping_query" + @property def shipping_query(self) -> ShippingQuery: """The shipping_query object.""" @@ -241,6 +183,8 @@ def shipping_query(self) -> ShippingQuery: class PreCheckoutQueryEvent(TelegramEvent): """New incoming pre-checkout query.""" + __event_type__ = "pre_checkout_query" + @property def pre_checkout_query(self) -> PreCheckoutQuery: """The pre_checkout_query object.""" @@ -248,9 +192,23 @@ def pre_checkout_query(self) -> PreCheckoutQuery: return self.update.pre_checkout_query +class PurchasedPaidMediaEvent(TelegramEvent): + """A user purchased paid media with a non-empty payload sent by the bot in a non-channel chat.""" + + __event_type__ = "purchased_paid_media" + + @property + def purchased_paid_media(self) -> PaidMediaPurchased: + """The purchased_paid_media object.""" + assert self.update.purchased_paid_media is not None + return self.update.purchased_paid_media + + class PollEvent(TelegramEvent): """New poll state.""" + __event_type__ = "poll" + @property def poll(self) -> Poll: """The poll object.""" @@ -261,6 +219,8 @@ def poll(self) -> Poll: class PollAnswerEvent(TelegramEvent): """A user changed their answer in a non-anonymous poll.""" + __event_type__ = "poll_answer" + @property def poll_answer(self) -> PollAnswer: """The poll_answer object.""" @@ -271,6 +231,8 @@ def poll_answer(self) -> PollAnswer: class MyChatMemberEvent(TelegramEvent): """The bot's chat member status was updated in a chat.""" + __event_type__ = "my_chat_member" + @property def my_chat_member(self) -> ChatMemberUpdated: """The my_chat_member object.""" @@ -281,6 +243,8 @@ def my_chat_member(self) -> ChatMemberUpdated: class ChatMemberEvent(TelegramEvent): """A chat member's status was updated in a chat.""" + __event_type__ = "chat_member" + @property def chat_member(self) -> ChatMemberUpdated: """The chat_member object.""" @@ -291,6 +255,8 @@ def chat_member(self) -> ChatMemberUpdated: class ChatJoinRequestEvent(TelegramEvent): """A request to join the chat has been sent.""" + __event_type__ = "chat_join_request" + @property def chat_join_request(self) -> ChatJoinRequest: """The chat_join_request object.""" @@ -301,6 +267,8 @@ def chat_join_request(self) -> ChatJoinRequest: class ChatBoostEvent(TelegramEvent): """A chat boost was added or changed.""" + __event_type__ = "chat_boost" + @property def chat_boost(self) -> ChatBoostUpdated: """The chat_boost object.""" @@ -311,6 +279,8 @@ def chat_boost(self) -> ChatBoostUpdated: class RemovedChatBoostEvent(TelegramEvent): """A boost was removed from a chat.""" + __event_type__ = "removed_chat_boost" + @property def removed_chat_boost(self) -> ChatBoostRemoved: """The removed_chat_boost object.""" diff --git a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/utils.py b/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/utils.py deleted file mode 100644 index ff9910b..0000000 --- a/packages/alicebot-adapter-telegram/alicebot/adapter/telegram/utils.py +++ /dev/null @@ -1,55 +0,0 @@ -"""Telegram 适配器内部使用的实用工具。""" - - -def snake_to_camel_case(snake_str: str) -> str: - """将 snake_case 转换为 CamelCase。 - - Args: - snake_str: snake_case 字符串。 - - Returns: - CamelCase 字符串。 - """ - return "".join(x.capitalize() for x in snake_str.lower().split("_")) - - -def snake_to_lower_camel_case(snake_str: str) -> str: - """将 snake_case 转换为 lowerCamelCase。 - - Args: - snake_str: snake_case 字符串。 - - Returns: - lowerCamelCase 字符串。 - """ - camel_string = snake_to_camel_case(snake_str) - return snake_str[0].lower() + camel_string[1:] - - -def camel_to_snake_case(camel_case: str) -> str: - """将 CamelCase 转换为 snake_case。 - - Args: - camel_case: CamelCase 字符串。 - - Returns: - snake_case 字符串。 - """ - return lower_camel_to_snake_case(camel_case[0].lower() + camel_case[1:]) - - -def lower_camel_to_snake_case(lower_camel_case: str) -> str: - """将 lowerCamelCase 转换为 snake_case。 - - Args: - lower_camel_case: lowerCamelCase 字符串。 - - Returns: - snake_case 字符串。 - """ - snake_case = "" - for s in lower_camel_case: - if s.isupper(): - snake_case += "_" - snake_case += s.lower() - return snake_case diff --git a/packages/alicebot-adapter-telegram/codegen.py b/packages/alicebot-adapter-telegram/codegen.py index 4edfcf8..dfa8ade 100644 --- a/packages/alicebot-adapter-telegram/codegen.py +++ b/packages/alicebot-adapter-telegram/codegen.py @@ -20,6 +20,7 @@ MODEL_FILE = BASE_PATH / "model.py" API_FILE = BASE_PATH / "api.py" MEDIA_FILE = BASE_PATH / "media.py" +EVENT_FILE = BASE_PATH / "event" / "other.py" ENTITY_FILE = BASE_PATH / "entity.py" @@ -54,13 +55,28 @@ class Response(BaseModel, Generic[_T]): # autogenerated by codegen.py, do not edit manually. # ruff: noqa: D101, D102, A002 # mypy: ignore-errors -# pylint: disable=redefined-builtin, unused-argument, missing-class-docstring, too-many-locals +# pylint: disable=redefined-builtin, missing-class-docstring, too-many-locals -from typing import Optional, Union +from abc import ABC, abstractmethod +from typing import Any, Optional, TypeVar, Union from .model import {} +_T = TypeVar("_T") + + +class TelegramAPIBase(ABC): + @abstractmethod + async def call_api( + self, + api: str, + *, + response_type: Optional[type[_T]] = None, + **params: Any, + ) -> Any: ... + + """ MEDIA_CODE_PREFIX = """\"\"\"Telegram Media 模型。\"\"\" # autogenerated by codegen.py, do not edit manually. @@ -79,6 +95,14 @@ class TelegramMedia(BaseModel): pass +""" +EVENT_CODE_PREFIX = """\"\"\"Telegram 适配器事件。\"\"\" +# autogenerated by codegen.py, do not edit manually. +# ruff: noqa: TID252 + +from .base import TelegramEvent +from ..model import {} + """ ENTITY_CODE_PREFIX = """\"\"\"Telegram Entity 模型。\"\"\" # autogenerated by codegen.py, do not edit manually. @@ -136,6 +160,7 @@ def to_api_method(self) -> str: *(field.to_python() for field in self.fields if field.required), *(field.to_python() for field in self.fields if not field.required), ) + return_type = to_annotation(self.returns) return ( f"async def {lower_camel_to_snake_case(self.name)}(\n" + indent( @@ -146,7 +171,20 @@ def to_api_method(self) -> str: ) ) ) - + f",\n) -> {to_annotation(self.returns)}: ..." + + f") -> {return_type}:\n" + + indent( + "return await self.call_api(" + + indent( + ",\n".join( + ( + f'"{self.name}"', + f"response_type={return_type}", + *(f"{field.name}={field.name}" for field in self.fields), + ) + ) + ) + + ")" + ) ) def to_media_model(self) -> Optional[str]: @@ -256,7 +294,7 @@ def gen_model(self) -> str: def gen_api(self) -> str: """生成 API 类的代码。""" USED_MODELS.clear() - api_code = "class TelegramAPI:\n" + indent( + api_code = "class TelegramAPI(TelegramAPIBase):\n" + indent( "\n\n".join(method.to_api_method() for method in self.methods.values()) ) return API_CODE_PREFIX.format(", ".join(USED_MODELS)) + api_code @@ -271,6 +309,30 @@ def gen_media(self) -> str: ) return MEDIA_CODE_PREFIX.format(", ".join(USED_MODELS)) + media_code + def gen_event(self) -> str: + """生成 Event 模型代码。""" + USED_MODELS.clear() + exclude_fields = {"update_id", "message"} + update_type = self.types["Update"] + result = "\n\n".join( + f""" +class {snake_to_camel_case(field.name)}Event(TelegramEvent): + \"\"\"{field.description.removeprefix('Optional. ').split('. ')[0] + '.'}\"\"\" + + __event_type__ = "{field.name}" + + @property + def {field.name}(self) -> {to_annotation(field.types)}: + \"\"\"The {field.name} object.\"\"\" + assert self.update.{field.name} is not None + return self.update.{field.name} + +""" + for field in update_type.fields + if field.name not in exclude_fields + ) + return EVENT_CODE_PREFIX.format(", ".join(USED_MODELS)) + result + def gen_entity(self) -> str: """生成 Entity 模型代码。""" USED_MODELS.clear() @@ -334,6 +396,31 @@ def to_annotation(t: Union[list[str], str]) -> str: return f'"{t}"' +def snake_to_camel_case(snake_str: str) -> str: + """将 snake_case 转换为 CamelCase。 + + Args: + snake_str: snake_case 字符串。 + + Returns: + CamelCase 字符串。 + """ + return "".join(x.capitalize() for x in snake_str.lower().split("_")) + + +def snake_to_lower_camel_case(snake_str: str) -> str: + """将 snake_case 转换为 lowerCamelCase。 + + Args: + snake_str: snake_case 字符串。 + + Returns: + lowerCamelCase 字符串。 + """ + camel_string = snake_to_camel_case(snake_str) + return snake_str[0].lower() + camel_string[1:] + + def lower_camel_to_snake_case(lower_camel_case: str) -> str: """将 lowerCamelCase 转换为 snake_case。 @@ -381,10 +468,12 @@ async def main() -> None: MODEL_FILE.write_text(api.gen_model()) API_FILE.write_text(api.gen_api()) MEDIA_FILE.write_text(api.gen_media()) + EVENT_FILE.write_text(api.gen_event()) ENTITY_FILE.write_text(api.gen_entity()) ruff_format(MODEL_FILE) ruff_format(API_FILE) ruff_format(MEDIA_FILE) + ruff_format(EVENT_FILE) ruff_format(ENTITY_FILE)