diff --git a/dev-docs/event-based-agents/adapters/line.md b/dev-docs/event-based-agents/adapters/line.md new file mode 100644 index 000000000..1b55e65e4 --- /dev/null +++ b/dev-docs/event-based-agents/adapters/line.md @@ -0,0 +1,72 @@ +# LINE EBA Adapter + +## Structure + +LINE is migrated into `src/langbot/pkg/platform/adapters/line/` with the standard EBA adapter layout: + +- `adapter.py` owns lifecycle, unified webhook handling, signature verification, listener dispatch, and outbound send/reply. +- `event_converter.py` maps LINE webhook events to `message.received`, `message.deleted`, group membership events, and `platform.specific`. +- `message_converter.py` maps common `MessageChain` components to LINE message objects and maps inbound LINE text/image/voice payloads back to EBA components. +- `api_impl.py` implements the optional common APIs against the LINE Messaging API. +- `platform_api.py` declares the safe LINE-specific pass-through actions. +- `manifest.yaml` declares `line-omni`. + +The legacy `src/langbot/pkg/platform/sources/line.py` adapter is kept unchanged. + +Request handling reuses the proven behaviour of the legacy adapter: `WebhookParser` verifies `X-Line-Signature`, the webhook acknowledges immediately, and each event is converted and dispatched in a background task. + +## Configuration + +| Field | Required | Notes | +|-------|----------|-------| +| `webhook_url` | No | Generated by LangBot. Paste it into the LINE Developers console and enable "Use webhook". | +| `channel_access_token` | Yes | Long-lived channel access token, used for all outbound calls. | +| `channel_secret` | Yes | Channel secret, used to verify `X-Line-Signature`. | + +`bot_account_id` is resolved through `get_bot_info` on startup and is used to drop the adapter's own messages. + +## Events + +| Event | Notes | +|-------|-------| +| `message.received` | Text, image, voice, and other message types. Group and room chats are mapped to group chat identity. LINE mentions become `At` components. | +| `message.deleted` | Emitted for LINE `unsend` events. | +| `group.member_joined` | Emitted for `memberJoined`; the first listed member is the event subject. | +| `group.member_left` | Emitted for `memberLeft`. | +| `bot.invited_to_group` | Emitted when the Official Account joins a group or room (`join`). | +| `bot.removed_from_group` | Emitted when the Official Account leaves (`leave`). | +| `platform.specific` | `line.follow`, `line.unfollow`, `line.postback`, `line.beacon`, and any other LINE event type, with `chat_id` in `data`. | + +Legacy pipeline bindings still receive `FriendMessage` / `GroupMessage`; the adapter converts the unified event for them. + +## Common APIs + +Required: + +- `send_message` +- `reply_message` + +Optional: + +- `get_user_info` +- `get_group_info` +- `leave_group` +- `call_platform_api` + +`reply_message` uses the event's reply token and falls back to `push_message` when the token is rejected or already used, so multi-message answers still arrive. + +## Platform APIs + +| Action | Notes | +|--------|-------| +| `get_bot_info` | Returns the Official Account profile. | +| `get_profile` | Calls `get_profile` for a user id. | +| `get_group_summary` | Calls `get_group_summary` for a group id. | + +## Known Limits + +- LINE has no group member listing API, so `get_group_member_list` raises `NotSupportedError`. +- Outbound images require a public URL: the Messaging API fetches `originalContentUrl`, so a base64-only `Image` component raises `NotSupportedError` instead of sending nothing. +- LINE has no message edit API; `edit_message` raises `NotSupportedError`. Streaming output is not supported. +- Requests are capped at five messages per call and 5000 characters per text message; longer chains are split into multiple requests. +- Offline coverage lives in `tests/unit_tests/platform/test_line_eba_adapter.py`. A real channel (webhook delivery, reply tokens, media download) still needs verification against a live LINE Official Account. diff --git a/src/langbot/pkg/platform/adapter_names.py b/src/langbot/pkg/platform/adapter_names.py index be5c1f56b..debd12e5e 100644 --- a/src/langbot/pkg/platform/adapter_names.py +++ b/src/langbot/pkg/platform/adapter_names.py @@ -13,6 +13,7 @@ OMNI_ADAPTER_NAMES = frozenset( 'wecombot', 'wecomcs', 'officialaccount', + 'line', } ) diff --git a/src/langbot/pkg/platform/adapters/line/adapter.py b/src/langbot/pkg/platform/adapters/line/adapter.py new file mode 100644 index 000000000..9f2a6d52c --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/adapter.py @@ -0,0 +1,258 @@ +from __future__ import annotations + +import asyncio +import traceback +import typing + +import pydantic +import quart +from linebot.v3.exceptions import InvalidSignatureError +from linebot.v3.messaging import ( + ApiClient, + Configuration, + MessagingApi, + MessagingApiBlob, + PushMessageRequest, + ReplyMessageRequest, +) +from linebot.v3.webhook import WebhookParser +from linebot.v3.webhooks import Event + +import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter +import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger +from langbot_plugin.api.entities.builtin.platform import events as platform_events +from langbot_plugin.api.entities.builtin.platform import message as platform_message + +from langbot.pkg.platform.adapters.line.api_impl import LineAPIMixin +from langbot.pkg.platform.adapters.line.errors import NotSupportedError +from langbot.pkg.platform.adapters.line.event_converter import LineEventConverter +from langbot.pkg.platform.adapters.line.message_converter import LineMessageConverter +from langbot.pkg.platform.adapters.line.platform_api import PLATFORM_API_MAP +from langbot.pkg.platform.adapters.line.types import ADAPTER_NAME, MAX_MESSAGES_PER_REQUEST + + +class LineAdapter(LineAPIMixin, abstract_platform_adapter.AbstractMessagePlatformAdapter): + """LINE Messaging API adapter in unified webhook mode.""" + + api_client: typing.Any = pydantic.Field(exclude=True) + bot: typing.Any = pydantic.Field(exclude=True) + blob_api: typing.Any = pydantic.Field(exclude=True) + parser: typing.Any = pydantic.Field(exclude=True) + + message_converter: LineMessageConverter = LineMessageConverter() + event_converter: LineEventConverter = LineEventConverter() + + config: dict + config_secret: str = '' + bot_uuid: str | None = None + + listeners: dict[ + typing.Type[platform_events.Event], + typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None], + ] = {} + background_tasks: set = pydantic.Field(default_factory=set, exclude=True) + + def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger): + missing_keys = [key for key in ('channel_access_token', 'channel_secret') if not config.get(key)] + if missing_keys: + raise ValueError(f'LINE adapter missing config: {missing_keys}') + configuration = Configuration(access_token=config['channel_access_token']) + api_client = ApiClient(configuration) + bot = MessagingApi(api_client) + blob_api = MessagingApiBlob(api_client) + bot_account_id = config.get('bot_account_id', '') or '' + super().__init__( + config=config, + logger=logger, + bot_account_id=bot_account_id, + config_secret=config['channel_secret'], + api_client=api_client, + bot=bot, + blob_api=blob_api, + parser=WebhookParser(config['channel_secret']), + listeners={}, + background_tasks=set(), + message_converter=LineMessageConverter(), + event_converter=LineEventConverter(bot_account_id=bot_account_id, blob_api=blob_api), + ) + + def set_bot_uuid(self, bot_uuid: str): + self.bot_uuid = bot_uuid + + def get_supported_events(self) -> list[str]: + return [ + 'message.received', + 'message.deleted', + 'group.member_joined', + 'group.member_left', + 'bot.invited_to_group', + 'bot.removed_from_group', + 'platform.specific', + ] + + def get_supported_apis(self) -> list[str]: + return [ + 'send_message', + 'reply_message', + 'get_user_info', + 'get_group_info', + 'leave_group', + 'call_platform_api', + ] + + async def send_message( + self, + target_type: str, + target_id: str, + message: platform_message.MessageChain, + ) -> platform_events.MessageResult: + messages = await self.message_converter.yiri2target(message) + for batch in _batches(messages): + request = PushMessageRequest(to=str(target_id), messages=batch) + await asyncio.to_thread(self.bot.push_message, request) + return platform_events.MessageResult(raw={'sent': len(messages), 'target_id': str(target_id)}) + + async def reply_message( + self, + message_source: platform_events.MessageEvent, + message: platform_message.MessageChain, + quote_origin: bool = False, + ) -> platform_events.MessageResult: + source = await LineEventConverter.yiri2target(message_source) + messages = await self.message_converter.yiri2target(message) + reply_token = str(getattr(source, 'reply_token', '') or '') + chat_id = _source_chat_id(source) or str(getattr(message_source, 'chat_id', '') or '') + for batch in _batches(messages): + if reply_token: + try: + await asyncio.to_thread( + self.bot.reply_message, ReplyMessageRequest(reply_token=reply_token, messages=batch) + ) + # A reply token is single-use; later batches of the same turn fall back to push. + reply_token = '' + continue + except Exception: + await self.logger.warning( + f'LINE reply token rejected, falling back to push: {traceback.format_exc()}' + ) + reply_token = '' + if not chat_id: + raise NotSupportedError('reply_message:line_missing_target') + await asyncio.to_thread(self.bot.push_message, PushMessageRequest(to=chat_id, messages=batch)) + return platform_events.MessageResult(raw={'sent': len(messages)}) + + async def call_platform_api(self, action: str, params: dict = {}) -> dict: + handler = PLATFORM_API_MAP.get(action) + if handler is None: + raise NotSupportedError(f'call_platform_api:{action}') + return await handler(self, dict(params or {})) + + def register_listener( + self, + event_type: typing.Type[platform_events.Event], + callback: typing.Callable[ + [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None + ], + ): + self.listeners[event_type] = callback + + def unregister_listener( + self, + event_type: typing.Type[platform_events.Event], + callback: typing.Callable[ + [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None + ], + ): + registered = self.listeners.get(event_type) + if registered is callback: + self.listeners.pop(event_type, None) + + async def handle_unified_webhook(self, bot_uuid: str, path: str, request) -> typing.Any: + signature = request.headers.get('X-Line-Signature') + body = await request.get_data(as_text=True) + if not signature: + await self.logger.warning('LINE webhook rejected: missing X-Line-Signature header') + return quart.Response('Missing X-Line-Signature header', status=400) + try: + events = self.parser.parse(body, signature) + except InvalidSignatureError: + await self.logger.warning( + 'LINE webhook rejected: invalid signature, check channel_access_token/channel_secret' + ) + return quart.Response('Invalid signature', status=400) + for event in events or []: + task = asyncio.create_task(self._dispatch_native_event(event)) + self.background_tasks.add(task) + task.add_done_callback(self.background_tasks.discard) + # Acknowledge immediately; LINE retries any non-200 answer. + return quart.Response('OK', status=200) + + async def _dispatch_native_event(self, event: Event) -> None: + try: + if self.bot_account_id and _sender_id(event) == self.bot_account_id: + return + eba_event = await self.event_converter.target2yiri(event) + if eba_event is None: + return + if isinstance(eba_event, platform_events.MessageReceivedEvent): + legacy_event = eba_event.to_legacy_event() + listener = self.listeners.get(type(legacy_event)) + if listener: + await listener(legacy_event, self) + await self._dispatch_eba_event(eba_event) + except Exception: + await self.logger.error(f'LINE native event error: {traceback.format_exc()}') + + async def _dispatch_eba_event(self, event: platform_events.EBAEvent) -> None: + for event_type in (type(event), platform_events.EBAEvent, platform_events.Event): + callback = self.listeners.get(event_type) + if callback: + await callback(event, self) + return + + async def is_muted(self, group_id: int | None = None) -> bool: + return False + + async def run_async(self): + if not self.bot_account_id: + try: + info = await asyncio.to_thread(self.bot.get_bot_info) + self.bot_account_id = info.user_id + self.event_converter.bot_account_id = info.user_id + except Exception: + await self.logger.warning( + 'LINE adapter could not resolve its account id; self-message filtering stays disabled' + ) + await self.logger.info(f'{ADAPTER_NAME} adapter running in unified webhook mode') + while True: + await asyncio.sleep(1) + + async def kill(self) -> bool: + for task in list(self.background_tasks): + task.cancel() + self.background_tasks.clear() + await asyncio.to_thread(self.api_client.close) + return True + + +def _batches(messages: list[typing.Any]) -> list[list[typing.Any]]: + """Split a message array into LINE's per-request budget.""" + return [ + messages[index : index + MAX_MESSAGES_PER_REQUEST] + for index in range(0, len(messages), MAX_MESSAGES_PER_REQUEST) + ] + + +def _sender_id(event: Event) -> str: + source = getattr(event, 'source', None) + return str(getattr(source, 'user_id', '') or '') + + +def _source_chat_id(event: Event) -> str: + """Chat id of the conversation a LINE event belongs to.""" + source = getattr(event, 'source', None) + for attribute in ('group_id', 'room_id', 'user_id'): + value = getattr(source, attribute, '') or '' + if value: + return str(value) + return '' diff --git a/src/langbot/pkg/platform/adapters/line/api_impl.py b/src/langbot/pkg/platform/adapters/line/api_impl.py new file mode 100644 index 000000000..263bb06f2 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/api_impl.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import typing + +from langbot.pkg.platform.adapters.line.errors import NotSupportedError +from langbot_plugin.api.entities.builtin.platform import entities as platform_entities +from langbot_plugin.api.entities.builtin.platform import events as platform_events +from langbot_plugin.api.entities.builtin.platform import message as platform_message + + +class LineAPIMixin: + """Optional universal APIs; LINE only exposes profile and group summaries.""" + + async def get_user_info(self, user_id: typing.Union[int, str]) -> platform_entities.User: + import asyncio + + profile = await asyncio.to_thread(self.bot.get_profile, str(user_id)) + return platform_entities.User( + id=profile.user_id, + nickname=profile.display_name, + avatar_url=profile.picture_url or '', + ) + + async def get_group_info(self, group_id: typing.Union[int, str]) -> platform_entities.UserGroup: + import asyncio + + try: + summary = await asyncio.to_thread(self.bot.get_group_summary, str(group_id)) + except Exception as exc: + raise NotSupportedError('get_group_info:group_not_found') from exc + return platform_entities.UserGroup( + id=summary.group_id, + name=summary.group_name, + avatar_url=summary.picture_url or '', + ) + + async def leave_group(self, group_id: typing.Union[int, str]) -> None: + import asyncio + + await asyncio.to_thread(self.bot.leave_group, str(group_id)) + + async def get_group_member_list( + self, + group_id: typing.Union[int, str], + ) -> list[platform_entities.UserGroupMember]: + # LINE does not expose a group member listing API. + raise NotSupportedError('get_group_member_list') + + async def get_group_list(self) -> list[platform_entities.UserGroup]: + raise NotSupportedError('get_group_list') + + async def get_friend_list(self) -> list[platform_entities.User]: + raise NotSupportedError('get_friend_list') + + async def edit_message( + self, + chat_type: str, + chat_id: typing.Union[int, str], + message_id: typing.Union[int, str], + new_content: platform_message.MessageChain, + ) -> None: + raise NotSupportedError('edit_message') + + async def delete_message( + self, + chat_type: str, + chat_id: typing.Union[int, str], + message_id: typing.Union[int, str], + ) -> None: + raise NotSupportedError('delete_message') + + async def forward_message( + self, + from_chat_type: str, + from_chat_id: typing.Union[int, str], + message_id: typing.Union[int, str], + to_chat_type: str, + to_chat_id: typing.Union[int, str], + ) -> platform_events.MessageResult: + raise NotSupportedError('forward_message') + + async def upload_file(self, file_data: bytes, filename: str) -> str: + raise NotSupportedError('upload_file') + + async def get_file_url(self, file_id: str) -> str: + raise NotSupportedError('get_file_url') diff --git a/src/langbot/pkg/platform/adapters/line/errors.py b/src/langbot/pkg/platform/adapters/line/errors.py new file mode 100644 index 000000000..b56f459ae --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/errors.py @@ -0,0 +1,10 @@ +from __future__ import annotations + +try: + from langbot_plugin.api.entities.builtin.platform.errors import NotSupportedError +except ModuleNotFoundError: + + class NotSupportedError(Exception): + def __init__(self, api_name: str, *args): + super().__init__(f"API '{api_name}' is not supported by this adapter", *args) + self.api_name = api_name diff --git a/src/langbot/pkg/platform/adapters/line/event_converter.py b/src/langbot/pkg/platform/adapters/line/event_converter.py new file mode 100644 index 000000000..605c58910 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/event_converter.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import typing + +from linebot.v3.webhooks import ( + BeaconEvent, + Event, + FollowEvent, + GroupSource, + JoinEvent, + LeaveEvent, + MemberJoinedEvent, + MemberLeftEvent, + MessageEvent, + PostbackEvent, + RoomSource, + UnfollowEvent, + UnsendEvent, + UserSource, +) + +import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter +from langbot_plugin.api.entities.builtin.platform import entities as platform_entities +from langbot_plugin.api.entities.builtin.platform import events as platform_events + +from langbot.pkg.platform.adapters.line.message_converter import LineMessageConverter +from langbot.pkg.platform.adapters.line.types import ADAPTER_NAME + + +def _timestamp(event: Event) -> float: + try: + return float(event.timestamp) / 1000.0 + except (AttributeError, TypeError, ValueError): + return 0.0 + + +def _chat_identity(event: Event) -> tuple[platform_entities.ChatType, str, typing.Any, str]: + """Return chat type, chat id, group entity and sender id for a LINE event source.""" + source = event.source + if isinstance(source, GroupSource): + return ( + platform_entities.ChatType.GROUP, + source.group_id, + platform_entities.UserGroup(id=source.group_id, name=source.group_id), + source.user_id or source.group_id, + ) + if isinstance(source, RoomSource): + # LINE rooms are multi-user chats; treat them as group chats keyed by room id. + return ( + platform_entities.ChatType.GROUP, + source.room_id, + platform_entities.UserGroup(id=source.room_id, name=source.room_id), + source.user_id or source.room_id, + ) + user_id = source.user_id if isinstance(source, UserSource) else '' + return platform_entities.ChatType.PRIVATE, user_id, None, user_id + + +class LineEventConverter(abstract_platform_adapter.AbstractEventConverter): + """Convert LINE webhook events into unified EBA events and legacy message events.""" + + def __init__(self, bot_account_id: str = '', blob_api: typing.Any = None): + self.bot_account_id = bot_account_id + self.blob_api = blob_api + + @staticmethod + async def yiri2target(event: platform_events.Event) -> typing.Any: + return getattr(event, 'source_platform_object', None) + + async def target2yiri(self, event: Event) -> typing.Optional[platform_events.EBAEvent]: + """Convert one LINE webhook event into a unified event; None when unsupported.""" + common = dict( + adapter_name=ADAPTER_NAME, + timestamp=_timestamp(event), + source_platform_object=event, + ) + chat_type, chat_id, group, sender_id = _chat_identity(event) + chat = dict( + chat_type=chat_type, + chat_id=chat_id, + group=group, + ) + if isinstance(event, MessageEvent): + return platform_events.MessageReceivedEvent( + **common, + **chat, + message_id=str(event.message.id), + message_chain=await LineMessageConverter.target2yiri( + event, blob_api=self.blob_api, bot_account_id=self.bot_account_id + ), + sender=platform_entities.User(id=sender_id, nickname=sender_id), + ) + if isinstance(event, UnsendEvent): + return platform_events.MessageDeletedEvent( + **common, + **chat, + message_id=str(event.unsend.message_id), + operator=platform_entities.User(id=sender_id) if sender_id else None, + ) + if isinstance(event, FollowEvent): + return platform_events.PlatformSpecificEvent( + **common, action='line.follow', data={**_event_payload(event), 'chat_id': chat_id} + ) + if isinstance(event, UnfollowEvent): + return platform_events.PlatformSpecificEvent( + **common, action='line.unfollow', data={**_event_payload(event), 'chat_id': chat_id} + ) + if isinstance(event, JoinEvent): + return platform_events.BotInvitedToGroupEvent( + **common, + group=platform_entities.UserGroup(id=chat_id, name=chat_id), + ) + if isinstance(event, LeaveEvent): + return platform_events.BotRemovedFromGroupEvent( + **common, + group=platform_entities.UserGroup(id=chat_id, name=chat_id), + ) + if isinstance(event, MemberJoinedEvent): + member = _first_member(event.joined.members, chat_id) + return platform_events.MemberJoinedEvent(**common, group=group, member=member) + if isinstance(event, MemberLeftEvent): + member = _first_member(event.left.members, chat_id) + return platform_events.MemberLeftEvent(**common, group=group, member=member) + if isinstance(event, PostbackEvent): + return platform_events.PlatformSpecificEvent( + **common, + action='line.postback', + data={ + 'data': event.postback.data, + 'chat_type': chat_type.value, + 'chat_id': chat_id, + 'sender_id': sender_id, + **({'params': event.postback.params} if event.postback.params else {}), + }, + ) + if isinstance(event, BeaconEvent): + return platform_events.PlatformSpecificEvent( + **common, + action='line.beacon', + data={'hwid': event.beacon.hwid, 'type': event.beacon.type, 'chat_id': chat_id}, + ) + # Remaining LINE event types stay visible to event processors without a lossy mapping. + action = getattr(event, 'type', '') or type(event).__name__ + return platform_events.PlatformSpecificEvent( + **common, action=f'line.{action}', data={**_event_payload(event), 'chat_id': chat_id} + ) + + async def target2legacy(self, event: Event) -> platform_events.FriendMessage | platform_events.GroupMessage | None: + """Build the legacy message event used by pipeline-based bindings.""" + eba_event = await self.target2yiri(event) + if not isinstance(eba_event, platform_events.MessageReceivedEvent): + return None + return eba_event.to_legacy_event() + + +def _event_payload(event: Event) -> dict: + to_dict = getattr(event, 'to_dict', None) + if callable(to_dict): + return dict(to_dict()) + return {} + + +def _first_member(members: typing.Any, fallback_id: str) -> platform_entities.User: + """Member arrays carry only user ids; the first entry is the subject of the event.""" + for member in members or []: + user_id = getattr(member, 'user_id', '') or '' + if user_id: + return platform_entities.User(id=user_id, nickname=user_id) + return platform_entities.User(id=fallback_id, nickname=fallback_id) diff --git a/src/langbot/pkg/platform/adapters/line/line.png b/src/langbot/pkg/platform/adapters/line/line.png new file mode 100644 index 000000000..8c925c6c0 Binary files /dev/null and b/src/langbot/pkg/platform/adapters/line/line.png differ diff --git a/src/langbot/pkg/platform/adapters/line/manifest.yaml b/src/langbot/pkg/platform/adapters/line/manifest.yaml new file mode 100644 index 000000000..79c47b42d --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/manifest.yaml @@ -0,0 +1,92 @@ +apiVersion: v1 +kind: MessagePlatformAdapter + +metadata: + name: line-omni + label: + en_US: LINE + zh_Hans: LINE + zh_Hant: LINE + description: + en_US: LINE adapter with event-driven orchestration, receiving messages through the unified webhook. + zh_Hans: LINE 适配器，支持事件驱动编排，通过统一 Webhook 接收消息。 + zh_Hant: LINE 適配器，支援事件驅動編排，透過統一 Webhook 接收訊息。 + icon: line.png + +spec: + categories: + - popular + - global + help_links: + zh: https://langbot.app/docs/zh/usage/platforms/line + en: https://langbot.app/docs/en/usage/platforms/line + ja: https://langbot.app/docs/ja/usage/platforms/line + config: + - name: webhook_url + label: + en_US: Webhook Callback URL + zh_Hans: Webhook 回调地址 + zh_Hant: Webhook 回調地址 + description: + en_US: Copy this URL into the LINE Developers console webhook settings and enable "Use webhook". + zh_Hans: 将此地址填入 LINE Developers 控制台的 Webhook 设置，并启用「Use webhook」。 + zh_Hant: 將此位址填入 LINE Developers 主控台的 Webhook 設定，並啟用「Use webhook」。 + type: webhook-url + required: false + default: "" + - name: channel_access_token + label: + en_US: Channel Access Token + zh_Hans: 频道访问令牌 + zh_Hant: 頻道存取權杖 + description: + en_US: Long-lived channel access token used to send messages. + zh_Hans: 用于发送消息的长期频道访问令牌。 + zh_Hant: 用於傳送訊息的長期頻道存取權杖。 + type: string + required: true + default: "" + - name: channel_secret + label: + en_US: Channel Secret + zh_Hans: 频道密钥 + zh_Hant: 頻道密鑰 + description: + en_US: Channel secret used to verify the X-Line-Signature of incoming webhooks. + zh_Hans: 用于校验入站 Webhook 的 X-Line-Signature 的频道密钥。 + zh_Hant: 用於校驗入站 Webhook 的 X-Line-Signature 的頻道密鑰。 + type: string + required: true + default: "" + + supported_events: + - message.received + - message.deleted + - group.member_joined + - group.member_left + - bot.invited_to_group + - bot.removed_from_group + - platform.specific + + supported_apis: + required: + - send_message + - reply_message + optional: + - get_user_info + - get_group_info + - leave_group + - call_platform_api + + platform_specific_apis: + - action: get_bot_info + description: { en_US: "Return the LINE Official Account profile", zh_Hans: "返回 LINE 官方账号资料" } + - action: get_profile + description: { en_US: "Call the LINE get profile API for a user", zh_Hans: "调用 LINE 用户资料接口" } + - action: get_group_summary + description: { en_US: "Call the LINE group summary API", zh_Hans: "调用 LINE 群组摘要接口" } + +execution: + python: + path: ./adapter.py + attr: LineAdapter diff --git a/src/langbot/pkg/platform/adapters/line/message_converter.py b/src/langbot/pkg/platform/adapters/line/message_converter.py new file mode 100644 index 000000000..aaeecc748 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/message_converter.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import base64 +import datetime +import typing + +import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter +from langbot_plugin.api.entities.builtin.platform import message as platform_message +from linebot.v3.messaging import ImageMessage, TextMessage +from linebot.v3.webhooks import ( + AudioMessageContent, + FileMessageContent, + ImageMessageContent, + LocationMessageContent, + MessageEvent, + StickerMessageContent, + TextMessageContent, + UserMentionee, + VideoMessageContent, +) + +from langbot.pkg.platform.adapters.line.errors import NotSupportedError +from langbot.pkg.platform.adapters.line.types import MAX_LINE_MEDIA_BYTES, MAX_TEXT_LENGTH + +# LINE stores inbound images as JPEG and the Blob API does not return the media type. +IMAGE_MIME = 'image/jpeg' + + +def _validate_media(content: bytes) -> bytes: + """Reject inbound media above the LINE download cap.""" + if len(content) > MAX_LINE_MEDIA_BYTES: + raise ValueError(f'LINE media exceeds the {MAX_LINE_MEDIA_BYTES}-byte limit') + return content + + +def _split_text(text: str) -> list[str]: + """Split a text body into LINE-sized chunks so long answers still send.""" + return [text[index : index + MAX_TEXT_LENGTH] for index in range(0, len(text), MAX_TEXT_LENGTH)] + + +def _message_datetime(event: MessageEvent) -> datetime.datetime: + try: + return datetime.datetime.fromtimestamp(float(event.timestamp) / 1000.0) + except (AttributeError, TypeError, ValueError): + return datetime.datetime.now() + + +def _text_components(text: str, mention: typing.Any, bot_account_id: str) -> list[platform_message.MessageComponent]: + """Map LINE mention positions onto At components, as the legacy adapter does.""" + if not mention or not mention.mentionees: + return [platform_message.Plain(text=text)] if text else [] + components: list[platform_message.MessageComponent] = [] + segments = sorted( + (mentionee.index, mentionee.index + mentionee.length, mentionee) for mentionee in mention.mentionees + ) + cursor = 0 + for start, end, mentionee in segments: + if start < cursor: + start, end = cursor, min(end, len(text)) + if start < cursor or end <= start or end > len(text): + continue + if start > cursor: + components.append(platform_message.Plain(text=text[cursor:start])) + token = text[start:end] + if isinstance(mentionee, UserMentionee): + target = bot_account_id if mentionee.is_self else (mentionee.user_id or token) + else: + target = token + components.append(platform_message.At(target=str(target), display=token.lstrip('@'))) + cursor = end + if cursor < len(text): + components.append(platform_message.Plain(text=text[cursor:])) + return components + + +async def _download_content(blob_api: typing.Any, message_id: str) -> bytes: + """Fetch inbound media through the LINE Blob API without blocking the event loop.""" + import asyncio + + if blob_api is None: + raise NotSupportedError('line:media_download') + return await asyncio.to_thread(blob_api.get_message_content, message_id) + + +class LineMessageConverter(abstract_platform_adapter.AbstractMessageConverter): + """Convert between LangBot message chains and LINE Messaging API messages.""" + + @staticmethod + async def yiri2target(message_chain: platform_message.MessageChain) -> list[typing.Any]: + """Build the LINE message array for one push/reply request.""" + messages: list[typing.Any] = [] + for component in message_chain: + if isinstance(component, platform_message.Source): + continue + if isinstance(component, platform_message.Plain): + messages.extend(TextMessage(text=chunk) for chunk in _split_text(component.text or '')) + elif isinstance(component, platform_message.At): + messages.append(TextMessage(text=f'@{component.display or component.target}')) + elif isinstance(component, platform_message.AtAll): + messages.append(TextMessage(text='@All')) + elif isinstance(component, platform_message.Image): + url = component.url or '' + if not url: + # LINE fetches outgoing images from a public URL; inline base64 is not supported. + raise NotSupportedError('send_message:line_image_requires_public_url') + messages.append(ImageMessage(original_content_url=url, preview_image_url=url)) + elif isinstance(component, platform_message.Quote): + if component.origin: + messages.extend(await LineMessageConverter.yiri2target(component.origin)) + elif isinstance(component, platform_message.Forward): + for node in component.node_list: + if node.message_chain: + messages.extend(await LineMessageConverter.yiri2target(node.message_chain)) + elif isinstance(component, platform_message.Unknown): + if component.text: + messages.extend(TextMessage(text=chunk) for chunk in _split_text(component.text)) + else: + raise NotSupportedError(f'send_message:line_unsupported_component:{type(component).__name__}') + return messages + + @staticmethod + async def target2yiri( + event: MessageEvent, + blob_api: typing.Any = None, + bot_account_id: str = '', + ) -> platform_message.MessageChain: + """Build a LangBot message chain from a LINE message event.""" + message = event.message + components: list[platform_message.MessageComponent] = [ + platform_message.Source( + id=str(getattr(message, 'id', '') or event.webhook_event_id), time=_message_datetime(event) + ) + ] + if isinstance(message, TextMessageContent): + components.extend(_text_components(message.text, getattr(message, 'mention', None), bot_account_id)) + elif isinstance(message, ImageMessageContent): + encoded = base64.b64encode(_validate_media(await _download_content(blob_api, message.id))).decode('ascii') + components.append(platform_message.Image(base64=f'data:{IMAGE_MIME};base64,{encoded}')) + elif isinstance(message, AudioMessageContent): + encoded = base64.b64encode(_validate_media(await _download_content(blob_api, message.id))).decode('ascii') + components.append(platform_message.Voice(base64=encoded, length=int(getattr(message, 'duration', 0) or 0))) + elif isinstance(message, VideoMessageContent): + components.append(platform_message.Unknown(text='[video]')) + elif isinstance(message, StickerMessageContent): + components.append(platform_message.Unknown(text='[sticker]')) + elif isinstance(message, FileMessageContent): + components.append(platform_message.Unknown(text=f'[file] {getattr(message, "file_name", "")}'.strip())) + elif isinstance(message, LocationMessageContent): + components.append( + platform_message.Unknown( + text=f'[location] {getattr(message, "title", "")} {getattr(message, "address", "")}'.strip() + ) + ) + else: + components.append(platform_message.Unknown(text=f'[unsupported line message: {type(message).__name__}]')) + if len(components) == 1: + components.append(platform_message.Unknown(text=f'[unsupported line message: {type(message).__name__}]')) + return platform_message.MessageChain(components) diff --git a/src/langbot/pkg/platform/adapters/line/platform_api.py b/src/langbot/pkg/platform/adapters/line/platform_api.py new file mode 100644 index 000000000..a331b5758 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/platform_api.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +import typing + + +async def get_bot_info(adapter, params: dict) -> dict: + import asyncio + + info = await asyncio.to_thread(adapter.bot.get_bot_info) + return { + 'user_id': info.user_id, + 'basic_id': info.basic_id, + 'display_name': info.display_name, + 'chat_mode': info.chat_mode, + } + + +async def get_profile(adapter, params: dict) -> dict: + user_id = str(params.get('user_id', '')) + if not user_id: + raise ValueError('get_profile requires user_id') + import asyncio + + profile = await asyncio.to_thread(adapter.bot.get_profile, user_id) + return { + 'user_id': profile.user_id, + 'display_name': profile.display_name, + 'picture_url': profile.picture_url or '', + 'status_message': profile.status_message or '', + } + + +async def get_group_summary(adapter, params: dict) -> dict: + group_id = str(params.get('group_id', '')) + if not group_id: + raise ValueError('get_group_summary requires group_id') + import asyncio + + summary = await asyncio.to_thread(adapter.bot.get_group_summary, group_id) + return { + 'group_id': summary.group_id, + 'group_name': summary.group_name, + 'picture_url': summary.picture_url or '', + } + + +PLATFORM_API_MAP: dict[str, typing.Callable[[typing.Any, dict], typing.Awaitable[dict]]] = { + 'get_bot_info': get_bot_info, + 'get_profile': get_profile, + 'get_group_summary': get_group_summary, +} diff --git a/src/langbot/pkg/platform/adapters/line/types.py b/src/langbot/pkg/platform/adapters/line/types.py new file mode 100644 index 000000000..7a62600b5 --- /dev/null +++ b/src/langbot/pkg/platform/adapters/line/types.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +ADAPTER_NAME = 'line-omni' + +# LINE Messaging API limits (push/reply share the same message array budget). +MAX_MESSAGES_PER_REQUEST = 5 + +# A single text message body is limited to 5000 characters. +MAX_TEXT_LENGTH = 5000 + +# LINE serves inbound media through the Blob API; keep the same cap as the legacy adapter. +MAX_LINE_MEDIA_BYTES = 10 * 1024 * 1024 diff --git a/tests/unit_tests/platform/test_line_eba_adapter.py b/tests/unit_tests/platform/test_line_eba_adapter.py new file mode 100644 index 000000000..306a09176 --- /dev/null +++ b/tests/unit_tests/platform/test_line_eba_adapter.py @@ -0,0 +1,464 @@ +"""Unit coverage for the LINE Omni adapter (unified webhook mode).""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import pathlib + +import pytest +import yaml +from linebot.v3.messaging import PushMessageRequest, ReplyMessageRequest +from linebot.v3.webhook import WebhookParser + +from langbot.pkg.platform.adapters.line.adapter import LineAdapter +from langbot.pkg.platform.adapters.line.errors import NotSupportedError +from langbot.pkg.platform.adapters.line.event_converter import LineEventConverter +from langbot.pkg.platform.adapters.line.message_converter import LineMessageConverter +from langbot_plugin.api.definition.abstract.platform.event_logger import AbstractEventLogger +from langbot_plugin.api.entities.builtin.platform import entities as platform_entities +from langbot_plugin.api.entities.builtin.platform import events as platform_events +from langbot_plugin.api.entities.builtin.platform import message as platform_message + +SECRET = 'test-channel-secret' +TOKEN = 'test-channel-access-token' +PLATFORM_ROOT = pathlib.Path(__file__).parents[3] / 'src/langbot/pkg/platform' + + +class DummyLogger(AbstractEventLogger): + def __init__(self): + self.lines: list[tuple[str, str]] = [] + + async def info(self, text, images=None, message_session_id=None, no_throw=True): + self.lines.append(('info', str(text))) + + async def debug(self, text, images=None, message_session_id=None, no_throw=True): + pass + + async def warning(self, text, images=None, message_session_id=None, no_throw=True): + self.lines.append(('warning', str(text))) + + async def error(self, text, images=None, message_session_id=None, no_throw=True): + self.lines.append(('error', str(text))) + + +class DummyMessagingApi: + def __init__(self, fail_reply: bool = False): + self.push_requests: list[PushMessageRequest] = [] + self.reply_requests: list[ReplyMessageRequest] = [] + self.fail_reply = fail_reply + self.leaved_groups: list[str] = [] + + def push_message(self, request): + self.push_requests.append(request) + return {'sent': len(request.messages)} + + def reply_message(self, request): + if self.fail_reply: + raise RuntimeError('reply token expired') + self.reply_requests.append(request) + return {'sent': len(request.messages)} + + def get_bot_info(self): + return type( + 'BotInfo', (), {'user_id': 'Ubot', 'basic_id': '@bot', 'display_name': 'Bot', 'chat_mode': 'chat'} + )() + + def get_profile(self, user_id): + return type( + 'Profile', + (), + {'user_id': user_id, 'display_name': 'Alice', 'picture_url': 'https://img/1.png', 'status_message': None}, + )() + + def get_group_summary(self, group_id): + return type('Summary', (), {'group_id': group_id, 'group_name': 'Team', 'picture_url': None})() + + def leave_group(self, group_id): + self.leaved_groups.append(group_id) + + +class DummyBlobApi: + def __init__(self, content: bytes = b'image-bytes'): + self.content = content + self.requested: list[str] = [] + + def get_message_content(self, message_id): + self.requested.append(message_id) + return self.content + + +class DummyApiClient: + def __init__(self): + self.closed = False + + def close(self): + self.closed = True + + +class DummyHeaders(dict): + def get(self, key, default=None): + return dict.get(self, key, default) + + +class DummyRequest: + def __init__(self, body: str, signature: str | None): + self.headers = DummyHeaders({} if signature is None else {'X-Line-Signature': signature}) + self._body = body + + async def get_data(self, as_text: bool = False): + return self._body if as_text else self._body.encode() + + +def sign(body: str, secret: str = SECRET) -> str: + return base64.b64encode(hmac.new(secret.encode(), body.encode(), hashlib.sha256).digest()).decode() + + +def message_event( + *, + text: str = 'hello', + source: dict | None = None, + reply_token: str = 'reply-token-1', + message: dict | None = None, +) -> dict: + return { + 'type': 'message', + 'mode': 'active', + 'timestamp': 1730000000000, + 'webhookEventId': '01HWEBHOOKID', + 'deliveryContext': {'isRedelivery': False}, + 'source': source or {'type': 'user', 'userId': 'U-user-1'}, + 'replyToken': reply_token, + 'message': message or {'id': 'msg-1', 'type': 'text', 'text': text, 'quoteToken': 'quote-1'}, + } + + +def webhook_body(*events: dict) -> str: + return json.dumps({'destination': 'Ubot', 'events': list(events)}) + + +def build_adapter(logger: DummyLogger | None = None) -> LineAdapter: + return LineAdapter( + config={'channel_access_token': TOKEN, 'channel_secret': SECRET}, + logger=logger or DummyLogger(), + ) + + +class EventSink: + def __init__(self): + self.events: list[platform_events.Event] = [] + + async def __call__(self, event, adapter): + self.events.append(event) + + +async def dispatch(adapter: LineAdapter, body: str, signature: str | None = None): + response = await adapter.handle_unified_webhook( + bot_uuid='bot-uuid', path='', request=DummyRequest(body, sign(body) if signature is None else signature) + ) + pending = list(adapter.background_tasks) + if pending: + await asyncio_gather(pending) + return response + + +async def asyncio_gather(tasks): + import asyncio + + await asyncio.gather(*tasks, return_exceptions=True) + + +def parse_events(body: str) -> list: + return WebhookParser(SECRET).parse(body, sign(body)) + + +def test_manifest_is_discoverable_and_keeps_legacy_config(): + manifest = yaml.safe_load((PLATFORM_ROOT / 'adapters/line/manifest.yaml').read_text(encoding='utf-8')) + legacy = yaml.safe_load((PLATFORM_ROOT / 'sources/line.yaml').read_text(encoding='utf-8')) + assert manifest['metadata']['name'] == 'line-omni' + assert manifest['execution']['python'] == {'path': './adapter.py', 'attr': 'LineAdapter'} + assert {field['name'] for field in legacy['spec']['config']} <= { + field['name'] for field in manifest['spec']['config'] + } + assert manifest['spec']['help_links'] == legacy['spec']['help_links'] + assert 'message.received' in manifest['spec']['supported_events'] + assert (PLATFORM_ROOT / 'adapters/line' / manifest['metadata']['icon']).is_file() + + +def test_adapter_requires_credentials(): + with pytest.raises(ValueError, match='channel_access_token'): + LineAdapter(config={'channel_secret': SECRET}, logger=DummyLogger()) + + +@pytest.mark.asyncio +async def test_webhook_rejects_missing_and_invalid_signature(): + adapter = build_adapter() + body = webhook_body(message_event()) + assert (await adapter.handle_unified_webhook('bot', '', DummyRequest(body, None))).status_code == 400 + assert (await adapter.handle_unified_webhook('bot', '', DummyRequest(body, 'bogus'))).status_code == 400 + + +@pytest.mark.asyncio +async def test_webhook_dispatches_message_received_event(): + adapter = build_adapter() + sink = EventSink() + adapter.register_listener(platform_events.MessageReceivedEvent, sink) + + response = await dispatch(adapter, webhook_body(message_event(text='hi there'))) + + assert response.status_code == 200 + assert len(sink.events) == 1 + event = sink.events[0] + assert isinstance(event, platform_events.MessageReceivedEvent) + assert event.adapter_name == 'line-omni' + assert event.chat_type == platform_entities.ChatType.PRIVATE + assert event.chat_id == 'U-user-1' + assert event.message_id == 'msg-1' + assert [type(component) for component in event.message_chain] == [platform_message.Source, platform_message.Plain] + assert event.message_chain[1].text == 'hi there' + + +@pytest.mark.asyncio +async def test_webhook_serves_legacy_listeners_and_ignores_self_messages(): + adapter = build_adapter() + adapter.bot_account_id = 'Ubot' + legacy_sink = EventSink() + adapter.register_listener(platform_events.FriendMessage, legacy_sink) + + await dispatch(adapter, webhook_body(message_event(source={'type': 'user', 'userId': 'Ubot'}))) + assert legacy_sink.events == [] + + await dispatch(adapter, webhook_body(message_event(source={'type': 'user', 'userId': 'U-user-9'}))) + assert len(legacy_sink.events) == 1 + assert isinstance(legacy_sink.events[0], platform_events.FriendMessage) + assert legacy_sink.events[0].sender.id == 'U-user-9' + + +@pytest.mark.asyncio +async def test_group_message_uses_group_chat_identity(): + events = parse_events( + webhook_body(message_event(source={'type': 'group', 'groupId': 'C-group', 'userId': 'U-user-2'})) + ) + event = await LineEventConverter().target2yiri(events[0]) + + assert event.chat_type == platform_entities.ChatType.GROUP + assert event.chat_id == 'C-group' + assert event.group.id == 'C-group' + assert event.sender.id == 'U-user-2' + + +@pytest.mark.asyncio +async def test_unsend_becomes_message_deleted(): + body = webhook_body( + { + 'type': 'unsend', + 'mode': 'active', + 'timestamp': 1730000000000, + 'webhookEventId': '01HUNSEND', + 'deliveryContext': {'isRedelivery': False}, + 'source': {'type': 'user', 'userId': 'U-user-1'}, + 'unsend': {'messageId': 'msg-9'}, + } + ) + event = await LineEventConverter().target2yiri(parse_events(body)[0]) + + assert isinstance(event, platform_events.MessageDeletedEvent) + assert event.message_id == 'msg-9' + assert event.chat_id == 'U-user-1' + + +@pytest.mark.asyncio +async def test_follow_and_postback_stay_platform_specific(): + follow = webhook_body( + { + 'type': 'follow', + 'mode': 'active', + 'timestamp': 1730000000000, + 'webhookEventId': '01HFOLLOW', + 'deliveryContext': {'isRedelivery': False}, + 'source': {'type': 'user', 'userId': 'U-user-3'}, + 'replyToken': 'token', + 'follow': {'isUnblocked': True}, + } + ) + follow_event = await LineEventConverter().target2yiri(parse_events(follow)[0]) + assert follow_event.type == 'platform.specific' + assert follow_event.action == 'line.follow' + + postback = webhook_body( + { + 'type': 'postback', + 'mode': 'active', + 'timestamp': 1730000000000, + 'webhookEventId': '01HPOSTBACK', + 'deliveryContext': {'isRedelivery': False}, + 'source': {'type': 'group', 'groupId': 'C-group', 'userId': 'U-user-4'}, + 'replyToken': 'token', + 'postback': {'data': 'action=buy'}, + } + ) + postback_event = await LineEventConverter().target2yiri(parse_events(postback)[0]) + assert postback_event.action == 'line.postback' + assert postback_event.data['data'] == 'action=buy' + assert postback_event.data['chat_id'] == 'C-group' + + +@pytest.mark.asyncio +async def test_member_joined_maps_group_and_member(): + body = webhook_body( + { + 'type': 'memberJoined', + 'mode': 'active', + 'timestamp': 1730000000000, + 'webhookEventId': '01HJOINED', + 'deliveryContext': {'isRedelivery': False}, + 'source': {'type': 'group', 'groupId': 'C-group'}, + 'replyToken': 'token', + 'joined': {'members': [{'type': 'user', 'userId': 'U-user-5'}]}, + } + ) + event = await LineEventConverter().target2yiri(parse_events(body)[0]) + + assert isinstance(event, platform_events.MemberJoinedEvent) + assert event.group.id == 'C-group' + assert event.member.id == 'U-user-5' + + +@pytest.mark.asyncio +async def test_text_mention_becomes_at_component_for_the_bot(): + body = webhook_body( + message_event( + text='@bot hello', + message={ + 'id': 'msg-2', + 'type': 'text', + 'quoteToken': 'quote-2', + 'text': '@bot hello', + 'mention': { + 'mentionees': [ + {'index': 0, 'length': 4, 'userId': 'Ubot', 'isSelf': True, 'type': 'user'}, + ] + }, + }, + ) + ) + event = await LineEventConverter(bot_account_id='Ubot').target2yiri(parse_events(body)[0]) + + at_components = [component for component in event.message_chain if isinstance(component, platform_message.At)] + assert len(at_components) == 1 + assert at_components[0].target == 'Ubot' + assert at_components[0].display == 'bot' + + +@pytest.mark.asyncio +async def test_image_message_is_downloaded_through_the_blob_api(): + blob = DummyBlobApi(b'jpeg-bytes') + body = webhook_body( + message_event( + message={ + 'id': 'msg-3', + 'type': 'image', + 'quoteToken': 'quote-3', + 'contentProvider': {'type': 'line'}, + } + ) + ) + event = await LineEventConverter(blob_api=blob).target2yiri(parse_events(body)[0]) + + image = event.message_chain[1] + assert isinstance(image, platform_message.Image) + assert image.base64 == f'data:image/jpeg;base64,{base64.b64encode(b"jpeg-bytes").decode()}' + assert blob.requested == ['msg-3'] + + +@pytest.mark.asyncio +async def test_outgoing_chain_builds_line_messages(): + chain = platform_message.MessageChain( + [ + platform_message.Source(id='msg-1', time=__import__('datetime').datetime.now()), + platform_message.Plain(text='hello'), + platform_message.At(target='U-user-1', display='Alice'), + ] + ) + messages = await LineMessageConverter.yiri2target(chain) + + assert [message.type for message in messages] == ['text', 'text'] + assert messages[0].text == 'hello' + assert messages[1].text == '@Alice' + + +@pytest.mark.asyncio +async def test_outgoing_image_requires_a_public_url(): + chain = platform_message.MessageChain([platform_message.Image(base64='ZGly')]) + with pytest.raises(NotSupportedError): + await LineMessageConverter.yiri2target(chain) + + +@pytest.mark.asyncio +async def test_send_message_batches_five_messages_per_request(): + adapter = build_adapter() + adapter.bot = DummyMessagingApi() + chain = platform_message.MessageChain([platform_message.Plain(text='x'), platform_message.Plain(text='y')] * 3) + + result = await adapter.send_message('person', 'U-user-1', chain) + + assert [len(request.messages) for request in adapter.bot.push_requests] == [5, 1] + assert all(request.to == 'U-user-1' for request in adapter.bot.push_requests) + assert result.raw == {'sent': 6, 'target_id': 'U-user-1'} + + +@pytest.mark.asyncio +async def test_reply_uses_reply_token_then_falls_back_to_push(): + adapter = build_adapter() + event = await LineEventConverter().target2yiri(parse_events(webhook_body(message_event()))[0]) + chain = platform_message.MessageChain([platform_message.Plain(text='pong')]) + + adapter.bot = DummyMessagingApi() + await adapter.reply_message(event, chain) + assert adapter.bot.reply_requests[0].reply_token == 'reply-token-1' + assert adapter.bot.push_requests == [] + + adapter.bot = DummyMessagingApi(fail_reply=True) + await adapter.reply_message(event, chain) + assert adapter.bot.push_requests[0].to == 'U-user-1' + + +@pytest.mark.asyncio +async def test_profile_and_group_apis_use_the_messaging_api(): + adapter = build_adapter() + adapter.bot = DummyMessagingApi() + + user = await adapter.get_user_info('U-user-1') + group = await adapter.get_group_info('C-group') + await adapter.leave_group('C-group') + + assert user.id == 'U-user-1' + assert user.nickname == 'Alice' + assert group.name == 'Team' + assert adapter.bot.leaved_groups == ['C-group'] + with pytest.raises(NotSupportedError): + await adapter.get_group_member_list('C-group') + + +@pytest.mark.asyncio +async def test_call_platform_api_dispatches_and_rejects_unknown_actions(): + adapter = build_adapter() + adapter.bot = DummyMessagingApi() + + info = await adapter.call_platform_api('get_bot_info', {}) + assert info['user_id'] == 'Ubot' + with pytest.raises(NotSupportedError): + await adapter.call_platform_api('not_a_line_action', {}) + + +@pytest.mark.asyncio +async def test_kill_closes_the_api_client(): + adapter = build_adapter() + api_client = DummyApiClient() + adapter.api_client = api_client + + assert await adapter.kill() is True + assert api_client.closed is True + assert adapter.background_tasks == set()