feat: add LINE platform support with adapter and configuration (#5085)

astrbot/core/config/default.py

@@ -15,6 +15,7 @@ WEBHOOK_SUPPORTED_PLATFORMS = [
     "wecom_ai_bot",
     "slack",
     "lark",
+    "line",
 ]
 
 # 默认配置
@@ -415,6 +416,7 @@ CONFIG_METADATA_2 = {
                         "slack_webhook_port": 6197,
                         "slack_webhook_path": "/astrbot-slack-webhook/callback",
                     },
+                    # LINE's config is located in line_adapter.py
                     "Satori": {
                         "id": "satori",
                         "type": "satori",

astrbot/core/platform/manager.py

@@ -176,6 +176,10 @@ class PlatformManager:
                     from .sources.satori.satori_adapter import (
                         SatoriPlatformAdapter,  # noqa: F401
                     )
+                case "line":
+                    from .sources.line.line_adapter import (
+                        LinePlatformAdapter,  # noqa: F401
+                    )
         except (ImportError, ModuleNotFoundError) as e:
             logger.error(
                 f"加载平台适配器 {platform_config['type']} 失败，原因：{e}。请检查依赖库是否安装。提示：可以在 管理面板->平台日志->安装Pip库 中安装依赖库。",

astrbot/core/platform/sources/line/line_adapter.py

@@ -0,0 +1,474 @@
+import asyncio
+import mimetypes
+import time
+import uuid
+from pathlib import Path
+from typing import Any, cast
+
+from astrbot.api import logger
+from astrbot.api.event import MessageChain
+from astrbot.api.message_components import At, File, Image, Plain, Record, Video
+from astrbot.api.platform import (
+    AstrBotMessage,
+    Group,
+    MessageMember,
+    MessageType,
+    Platform,
+    PlatformMetadata,
+)
+from astrbot.core.platform.astr_message_event import MessageSesion
+from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
+from astrbot.core.utils.webhook_utils import log_webhook_info
+
+from ...register import register_platform_adapter
+from .line_api import LineAPIClient
+from .line_event import LineMessageEvent
+
+LINE_CONFIG_METADATA = {
+    "channel_access_token": {
+        "description": "LINE Channel Access Token",
+        "type": "string",
+        "hint": "LINE Messaging API 的 channel access token。",
+    },
+    "channel_secret": {
+        "description": "LINE Channel Secret",
+        "type": "string",
+        "hint": "用于校验 LINE Webhook 签名。",
+    },
+}
+
+LINE_I18N_RESOURCES = {
+    "zh-CN": {
+        "channel_access_token": {
+            "description": "LINE Channel Access Token",
+            "hint": "LINE Messaging API 的 channel access token。",
+        },
+        "channel_secret": {
+            "description": "LINE Channel Secret",
+            "hint": "用于校验 LINE Webhook 签名。",
+        },
+    },
+    "en-US": {
+        "channel_access_token": {
+            "description": "LINE Channel Access Token",
+            "hint": "Channel access token for LINE Messaging API.",
+        },
+        "channel_secret": {
+            "description": "LINE Channel Secret",
+            "hint": "Used to verify LINE webhook signatures.",
+        },
+    },
+}
+
+
+@register_platform_adapter(
+    "line",
+    "LINE Messaging API 适配器",
+    support_streaming_message=False,
+    default_config_tmpl={
+        "id": "line",
+        "type": "line",
+        "enable": False,
+        "channel_access_token": "",
+        "channel_secret": "",
+        "unified_webhook_mode": True,
+        "webhook_uuid": "",
+    },
+    config_metadata=LINE_CONFIG_METADATA,
+    i18n_resources=LINE_I18N_RESOURCES,
+)
+class LinePlatformAdapter(Platform):
+    def __init__(
+        self,
+        platform_config: dict,
+        platform_settings: dict,
+        event_queue: asyncio.Queue,
+    ) -> None:
+        super().__init__(platform_config, event_queue)
+        self.config["unified_webhook_mode"] = True
+        self.destination = "unknown"
+        self.settings = platform_settings
+        self._event_id_timestamps: dict[str, float] = {}
+        self.shutdown_event = asyncio.Event()
+
+        channel_access_token = str(platform_config.get("channel_access_token", ""))
+        channel_secret = str(platform_config.get("channel_secret", ""))
+        if not channel_access_token or not channel_secret:
+            raise ValueError(
+                "LINE 适配器需要 channel_access_token 和 channel_secret。",
+            )
+
+        self.line_api = LineAPIClient(
+            channel_access_token=channel_access_token,
+            channel_secret=channel_secret,
+        )
+
+    async def send_by_session(
+        self,
+        session: MessageSesion,
+        message_chain: MessageChain,
+    ) -> None:
+        messages = await LineMessageEvent.build_line_messages(message_chain)
+        if messages:
+            await self.line_api.push_message(session.session_id, messages)
+        await super().send_by_session(session, message_chain)
+
+    def meta(self) -> PlatformMetadata:
+        return PlatformMetadata(
+            name="line",
+            description="LINE Messaging API 适配器",
+            id=cast(str, self.config.get("id", "line")),
+            support_streaming_message=False,
+        )
+
+    async def run(self) -> None:
+        webhook_uuid = self.config.get("webhook_uuid")
+        if webhook_uuid:
+            log_webhook_info(f"{self.meta().id}(LINE)", webhook_uuid)
+        else:
+            logger.warning("[LINE] webhook_uuid 为空，统一 Webhook 可能无法接收消息。")
+        await self.shutdown_event.wait()
+
+    async def terminate(self) -> None:
+        self.shutdown_event.set()
+        await self.line_api.close()
+
+    async def webhook_callback(self, request: Any) -> Any:
+        raw_body = await request.get_data()
+        signature = request.headers.get("x-line-signature")
+        if not self.line_api.verify_signature(raw_body, signature):
+            logger.warning("[LINE] invalid webhook signature")
+            return "invalid signature", 400
+
+        try:
+            payload = await request.get_json(force=True, silent=False)
+        except Exception as e:
+            logger.warning("[LINE] invalid webhook body: %s", e)
+            return "bad request", 400
+
+        if not isinstance(payload, dict):
+            return "bad request", 400
+
+        await self.handle_webhook_event(payload)
+        return "ok", 200
+
+    async def handle_webhook_event(self, payload: dict[str, Any]) -> None:
+        destination = str(payload.get("destination", "")).strip()
+        if destination:
+            self.destination = destination
+
+        events = payload.get("events")
+        if not isinstance(events, list):
+            return
+
+        for event in events:
+            if not isinstance(event, dict):
+                continue
+
+            event_id = str(event.get("webhookEventId", ""))
+            if event_id and self._is_duplicate_event(event_id):
+                logger.debug("[LINE] duplicate event skipped: %s", event_id)
+                continue
+
+            abm = await self.convert_message(event)
+            if abm is None:
+                continue
+            await self.handle_msg(abm)
+
+    async def convert_message(self, event: dict[str, Any]) -> AstrBotMessage | None:
+        if str(event.get("type", "")) != "message":
+            return None
+        if str(event.get("mode", "active")) == "standby":
+            return None
+
+        source = event.get("source", {})
+        if not isinstance(source, dict):
+            return None
+
+        message = event.get("message", {})
+        if not isinstance(message, dict):
+            return None
+
+        source_type = str(source.get("type", ""))
+        user_id = str(source.get("userId", "")).strip()
+        group_id = str(source.get("groupId", "")).strip()
+        room_id = str(source.get("roomId", "")).strip()
+
+        abm = AstrBotMessage()
+        abm.self_id = self.destination or self.meta().id
+        abm.message = []
+        abm.raw_message = event
+        abm.message_id = str(
+            message.get("id")
+            or event.get("webhookEventId")
+            or event.get("deliveryContext", {}).get("deliveryId", "")
+            or uuid.uuid4().hex
+        )
+
+        event_timestamp = event.get("timestamp")
+        if isinstance(event_timestamp, int):
+            abm.timestamp = (
+                event_timestamp // 1000
+                if event_timestamp > 1_000_000_000_000
+                else event_timestamp
+            )
+        else:
+            abm.timestamp = int(time.time())
+
+        if source_type in {"group", "room"}:
+            abm.type = MessageType.GROUP_MESSAGE
+            container_id = group_id or room_id
+            abm.group = Group(group_id=container_id, group_name=container_id)
+            abm.session_id = container_id
+            sender_id = user_id or container_id
+        elif source_type == "user":
+            abm.type = MessageType.FRIEND_MESSAGE
+            abm.session_id = user_id
+            sender_id = user_id
+        else:
+            abm.type = MessageType.OTHER_MESSAGE
+            abm.session_id = user_id or group_id or room_id or "unknown"
+            sender_id = abm.session_id
+
+        abm.sender = MessageMember(user_id=sender_id, nickname=sender_id[:8])
+
+        components = await self._parse_line_message_components(message)
+        if not components:
+            return None
+        abm.message = components
+        abm.message_str = self._build_message_str(components)
+        return abm
+
+    async def _parse_line_message_components(
+        self,
+        message: dict[str, Any],
+    ) -> list:
+        msg_type = str(message.get("type", ""))
+        message_id = str(message.get("id", "")).strip()
+
+        if msg_type == "text":
+            text = str(message.get("text", ""))
+            mention = message.get("mention")
+            if isinstance(mention, dict):
+                return self._parse_text_with_mentions(text, mention)
+            return [Plain(text=text)] if text else []
+
+        if msg_type == "image":
+            image_component = await self._build_image_component(message_id, message)
+            return [image_component] if image_component else [Plain(text="[image]")]
+
+        if msg_type == "video":
+            video_component = await self._build_video_component(message_id, message)
+            return [video_component] if video_component else [Plain(text="[video]")]
+
+        if msg_type == "audio":
+            audio_component = await self._build_audio_component(message_id, message)
+            return [audio_component] if audio_component else [Plain(text="[audio]")]
+
+        if msg_type == "file":
+            file_component = await self._build_file_component(message_id, message)
+            return [file_component] if file_component else [Plain(text="[file]")]
+
+        if msg_type == "sticker":
+            return [Plain(text="[sticker]")]
+
+        return [Plain(text=f"[{msg_type}]")]
+
+    def _parse_text_with_mentions(self, text: str, mention_obj: dict[str, Any]) -> list:
+        mentions = mention_obj.get("mentionees", [])
+        if not isinstance(mentions, list) or not mentions:
+            return [Plain(text=text)] if text else []
+
+        normalized = []
+        for item in mentions:
+            if not isinstance(item, dict):
+                continue
+            start = item.get("index")
+            length = item.get("length")
+            if not isinstance(start, int) or not isinstance(length, int):
+                continue
+            normalized.append((start, length, item))
+        normalized.sort(key=lambda x: x[0])
+
+        ret = []
+        cursor = 0
+        for start, length, item in normalized:
+            if start > cursor:
+                part = text[cursor:start]
+                if part:
+                    ret.append(Plain(text=part))
+
+            label = text[start : start + length] or "@user"
+            mention_type = str(item.get("type", ""))
+            if mention_type == "user":
+                target_id = str(item.get("userId", "")).strip()
+                ret.append(At(qq=target_id, name=label.lstrip("@")))
+            else:
+                ret.append(Plain(text=label))
+            cursor = max(cursor, start + length)
+
+        if cursor < len(text):
+            tail = text[cursor:]
+            if tail:
+                ret.append(Plain(text=tail))
+        return ret
+
+    async def _build_image_component(
+        self,
+        message_id: str,
+        message: dict[str, Any],
+    ) -> Image | None:
+        external_url = self._get_external_content_url(message)
+        if external_url:
+            return Image.fromURL(external_url)
+
+        content = await self.line_api.get_message_content(message_id)
+        if not content:
+            return None
+        content_bytes, _, _ = content
+        return Image.fromBytes(content_bytes)
+
+    async def _build_video_component(
+        self,
+        message_id: str,
+        message: dict[str, Any],
+    ) -> Video | None:
+        external_url = self._get_external_content_url(message)
+        if external_url:
+            return Video.fromURL(external_url)
+
+        content = await self.line_api.get_message_content(message_id)
+        if not content:
+            return None
+        content_bytes, content_type, _ = content
+        suffix = self._guess_suffix(content_type, ".mp4")
+        file_path = self._store_temp_content("video", message_id, content_bytes, suffix)
+        return Video(file=file_path, path=file_path)
+
+    async def _build_audio_component(
+        self,
+        message_id: str,
+        message: dict[str, Any],
+    ) -> Record | None:
+        external_url = self._get_external_content_url(message)
+        if external_url:
+            return Record.fromURL(external_url)
+
+        content = await self.line_api.get_message_content(message_id)
+        if not content:
+            return None
+        content_bytes, content_type, _ = content
+        suffix = self._guess_suffix(content_type, ".m4a")
+        file_path = self._store_temp_content("audio", message_id, content_bytes, suffix)
+        return Record(file=file_path, url=file_path)
+
+    async def _build_file_component(
+        self,
+        message_id: str,
+        message: dict[str, Any],
+    ) -> File | None:
+        content = await self.line_api.get_message_content(message_id)
+        if not content:
+            return None
+        content_bytes, content_type, filename = content
+        default_name = str(message.get("fileName", "")).strip() or f"{message_id}.bin"
+        suffix = Path(default_name).suffix or self._guess_suffix(content_type, ".bin")
+        final_name = filename or default_name
+        file_path = self._store_temp_content(
+            "file",
+            message_id,
+            content_bytes,
+            suffix,
+            original_name=final_name,
+        )
+        return File(name=final_name, file=file_path, url=file_path)
+
+    @staticmethod
+    def _get_external_content_url(message: dict[str, Any]) -> str:
+        provider = message.get("contentProvider")
+        if not isinstance(provider, dict):
+            return ""
+        if str(provider.get("type", "")) != "external":
+            return ""
+        return str(provider.get("originalContentUrl", "")).strip()
+
+    @staticmethod
+    def _guess_suffix(content_type: str | None, fallback: str) -> str:
+        if not content_type:
+            return fallback
+        base_type = content_type.split(";", 1)[0].strip().lower()
+        guessed = mimetypes.guess_extension(base_type)
+        if guessed:
+            return guessed
+        return fallback
+
+    @staticmethod
+    def _store_temp_content(
+        content_type: str,
+        message_id: str,
+        content: bytes,
+        suffix: str,
+        original_name: str = "",
+    ) -> str:
+        temp_dir = Path(get_astrbot_temp_path())
+        temp_dir.mkdir(parents=True, exist_ok=True)
+        name_prefix = f"line_{content_type}"
+        if original_name:
+            safe_stem = Path(original_name).stem.strip()
+            safe_stem = "".join(
+                ch if ch.isalnum() or ch in ("-", "_", ".") else "_" for ch in safe_stem
+            )
+            safe_stem = safe_stem.strip("._")
+            if safe_stem:
+                name_prefix = safe_stem[:64]
+        file_path = temp_dir / f"{name_prefix}_{message_id}_{uuid.uuid4().hex[:6]}"
+        file_path = file_path.with_suffix(suffix)
+        file_path.write_bytes(content)
+        return str(file_path.resolve())
+
+    @staticmethod
+    def _build_message_str(components: list) -> str:
+        parts: list[str] = []
+        for comp in components:
+            if isinstance(comp, Plain):
+                parts.append(comp.text)
+            elif isinstance(comp, At):
+                parts.append(f"@{comp.name or comp.qq}")
+            elif isinstance(comp, Image):
+                parts.append("[image]")
+            elif isinstance(comp, Video):
+                parts.append("[video]")
+            elif isinstance(comp, Record):
+                parts.append("[audio]")
+            elif isinstance(comp, File):
+                parts.append(str(comp.name or "[file]"))
+            else:
+                parts.append(f"[{comp.type}]")
+        return " ".join(i for i in parts if i).strip()
+
+    def _clean_expired_events(self) -> None:
+        current = time.time()
+        expired = [
+            event_id
+            for event_id, ts in self._event_id_timestamps.items()
+            if current - ts > 1800
+        ]
+        for event_id in expired:
+            del self._event_id_timestamps[event_id]
+
+    def _is_duplicate_event(self, event_id: str) -> bool:
+        self._clean_expired_events()
+        if event_id in self._event_id_timestamps:
+            return True
+        self._event_id_timestamps[event_id] = time.time()
+        return False
+
+    async def handle_msg(self, abm: AstrBotMessage) -> None:
+        event = LineMessageEvent(
+            message_str=abm.message_str,
+            message_obj=abm,
+            platform_meta=self.meta(),
+            session_id=abm.session_id,
+            line_api=self.line_api,
+        )
+        self._event_queue.put_nowait(event)

astrbot/core/platform/sources/line/line_api.py

@@ -0,0 +1,203 @@
+import asyncio
+import base64
+import hmac
+import json
+from hashlib import sha256
+from typing import Any
+from urllib.parse import unquote
+
+import aiohttp
+
+from astrbot.api import logger
+
+
+class LineAPIClient:
+    def __init__(
+        self,
+        *,
+        channel_access_token: str,
+        channel_secret: str,
+        timeout_seconds: int = 30,
+    ) -> None:
+        self.channel_access_token = channel_access_token.strip()
+        self.channel_secret = channel_secret.strip()
+        self.timeout = aiohttp.ClientTimeout(total=timeout_seconds)
+        self._session: aiohttp.ClientSession | None = None
+
+    async def _get_session(self) -> aiohttp.ClientSession:
+        if self._session is None or self._session.closed:
+            self._session = aiohttp.ClientSession(timeout=self.timeout)
+        return self._session
+
+    async def close(self) -> None:
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+    def verify_signature(self, raw_body: bytes, signature: str | None) -> bool:
+        if not signature:
+            return False
+        digest = hmac.new(
+            self.channel_secret.encode("utf-8"),
+            raw_body,
+            sha256,
+        ).digest()
+        expected = base64.b64encode(digest).decode("utf-8")
+        return hmac.compare_digest(expected, signature.strip())
+
+    @property
+    def _auth_headers(self) -> dict[str, str]:
+        return {"Authorization": f"Bearer {self.channel_access_token}"}
+
+    async def reply_message(
+        self,
+        reply_token: str,
+        messages: list[dict[str, Any]],
+        *,
+        notification_disabled: bool = False,
+    ) -> bool:
+        payload = {
+            "replyToken": reply_token,
+            "messages": messages[:5],
+            "notificationDisabled": notification_disabled,
+        }
+        return await self._post_json(
+            "https://api.line.me/v2/bot/message/reply",
+            payload=payload,
+            op_name="reply",
+        )
+
+    async def push_message(
+        self,
+        to: str,
+        messages: list[dict[str, Any]],
+        *,
+        notification_disabled: bool = False,
+    ) -> bool:
+        payload = {
+            "to": to,
+            "messages": messages[:5],
+            "notificationDisabled": notification_disabled,
+        }
+        return await self._post_json(
+            "https://api.line.me/v2/bot/message/push",
+            payload=payload,
+            op_name="push",
+        )
+
+    async def _post_json(
+        self,
+        url: str,
+        *,
+        payload: dict[str, Any],
+        op_name: str,
+    ) -> bool:
+        session = await self._get_session()
+        headers = {
+            **self._auth_headers,
+            "Content-Type": "application/json",
+        }
+        try:
+            async with session.post(url, json=payload, headers=headers) as resp:
+                if resp.status < 400:
+                    return True
+                body = await resp.text()
+                logger.error(
+                    "[LINE] %s message failed: status=%s body=%s",
+                    op_name,
+                    resp.status,
+                    body,
+                )
+                return False
+        except Exception as e:
+            logger.error("[LINE] %s message request failed: %s", op_name, e)
+            return False
+
+    async def get_message_content(
+        self,
+        message_id: str,
+    ) -> tuple[bytes, str | None, str | None] | None:
+        session = await self._get_session()
+        url = f"https://api-data.line.me/v2/bot/message/{message_id}/content"
+        headers = self._auth_headers
+
+        async with session.get(url, headers=headers) as resp:
+            if resp.status == 202:
+                if not await self._wait_for_transcoding(message_id):
+                    return None
+                async with session.get(url, headers=headers) as retry_resp:
+                    if retry_resp.status != 200:
+                        body = await retry_resp.text()
+                        logger.warning(
+                            "[LINE] get content retry failed: message_id=%s status=%s body=%s",
+                            message_id,
+                            retry_resp.status,
+                            body,
+                        )
+                        return None
+                    return await self._read_content_response(retry_resp)
+
+            if resp.status != 200:
+                body = await resp.text()
+                logger.warning(
+                    "[LINE] get content failed: message_id=%s status=%s body=%s",
+                    message_id,
+                    resp.status,
+                    body,
+                )
+                return None
+            return await self._read_content_response(resp)
+
+    async def _read_content_response(
+        self,
+        resp: aiohttp.ClientResponse,
+    ) -> tuple[bytes, str | None, str | None]:
+        content = await resp.read()
+        content_type = resp.headers.get("Content-Type")
+        disposition = resp.headers.get("Content-Disposition")
+        filename = self._extract_filename_from_disposition(disposition)
+        return content, content_type, filename
+
+    def _extract_filename_from_disposition(self, disposition: str | None) -> str | None:
+        if not disposition:
+            return None
+        for part in disposition.split(";"):
+            token = part.strip()
+            if token.startswith("filename*="):
+                val = token.split("=", 1)[1].strip().strip('"')
+                if val.lower().startswith("utf-8''"):
+                    val = val[7:]
+                return unquote(val)
+            if token.startswith("filename="):
+                return token.split("=", 1)[1].strip().strip('"')
+        return None
+
+    async def _wait_for_transcoding(
+        self,
+        message_id: str,
+        *,
+        max_attempts: int = 10,
+        interval_seconds: float = 1.0,
+    ) -> bool:
+        session = await self._get_session()
+        url = (
+            f"https://api-data.line.me/v2/bot/message/{message_id}/content/transcoding"
+        )
+        headers = self._auth_headers
+
+        for _ in range(max_attempts):
+            try:
+                async with session.get(url, headers=headers) as resp:
+                    if resp.status != 200:
+                        await asyncio.sleep(interval_seconds)
+                        continue
+                    body = await resp.text()
+                    data = json.loads(body)
+                    status = str(data.get("status", "")).lower()
+                    if status == "succeeded":
+                        return True
+                    if status == "failed":
+                        return False
+            except Exception:
+                pass
+            await asyncio.sleep(interval_seconds)
+        return False

astrbot/core/platform/sources/line/line_event.py

@@ -0,0 +1,285 @@
+import asyncio
+import os
+import re
+import uuid
+from collections.abc import AsyncGenerator
+from pathlib import Path
+
+from astrbot.api import logger
+from astrbot.api.event import AstrMessageEvent, MessageChain
+from astrbot.api.message_components import (
+    At,
+    BaseMessageComponent,
+    File,
+    Image,
+    Plain,
+    Record,
+    Video,
+)
+from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
+from astrbot.core.utils.media_utils import get_media_duration
+
+from .line_api import LineAPIClient
+
+
+class LineMessageEvent(AstrMessageEvent):
+    def __init__(
+        self,
+        message_str,
+        message_obj,
+        platform_meta,
+        session_id,
+        line_api: LineAPIClient,
+    ) -> None:
+        super().__init__(message_str, message_obj, platform_meta, session_id)
+        self.line_api = line_api
+
+    @staticmethod
+    async def _component_to_message_object(
+        segment: BaseMessageComponent,
+    ) -> dict | None:
+        if isinstance(segment, Plain):
+            text = segment.text.strip()
+            if not text:
+                return None
+            return {"type": "text", "text": text[:5000]}
+
+        if isinstance(segment, At):
+            name = str(segment.name or segment.qq or "").strip()
+            if not name:
+                return None
+            return {"type": "text", "text": f"@{name}"[:5000]}
+
+        if isinstance(segment, Image):
+            image_url = await LineMessageEvent._resolve_image_url(segment)
+            if not image_url:
+                return None
+            return {
+                "type": "image",
+                "originalContentUrl": image_url,
+                "previewImageUrl": image_url,
+            }
+
+        if isinstance(segment, Record):
+            audio_url = await LineMessageEvent._resolve_record_url(segment)
+            if not audio_url:
+                return None
+            duration = await LineMessageEvent._resolve_record_duration(segment)
+            return {
+                "type": "audio",
+                "originalContentUrl": audio_url,
+                "duration": duration,
+            }
+
+        if isinstance(segment, Video):
+            video_url = await LineMessageEvent._resolve_video_url(segment)
+            if not video_url:
+                return None
+            preview_url = await LineMessageEvent._resolve_video_preview_url(segment)
+            if not preview_url:
+                return None
+            return {
+                "type": "video",
+                "originalContentUrl": video_url,
+                "previewImageUrl": preview_url,
+            }
+
+        if isinstance(segment, File):
+            file_url = await LineMessageEvent._resolve_file_url(segment)
+            if not file_url:
+                return None
+            file_name = str(segment.name or "").strip() or "file.bin"
+            file_size = await LineMessageEvent._resolve_file_size(segment)
+            if file_size <= 0:
+                return None
+            return {
+                "type": "file",
+                "fileName": file_name,
+                "fileSize": file_size,
+                "originalContentUrl": file_url,
+            }
+
+        return None
+
+    @staticmethod
+    async def _resolve_image_url(segment: Image) -> str:
+        candidate = (segment.url or segment.file or "").strip()
+        if candidate.startswith("http://") or candidate.startswith("https://"):
+            return candidate
+        try:
+            return await segment.register_to_file_service()
+        except Exception as e:
+            logger.debug("[LINE] resolve image url failed: %s", e)
+            return ""
+
+    @staticmethod
+    async def _resolve_record_url(segment: Record) -> str:
+        candidate = (segment.url or segment.file or "").strip()
+        if candidate.startswith("http://") or candidate.startswith("https://"):
+            return candidate
+        try:
+            return await segment.register_to_file_service()
+        except Exception as e:
+            logger.debug("[LINE] resolve record url failed: %s", e)
+            return ""
+
+    @staticmethod
+    async def _resolve_record_duration(segment: Record) -> int:
+        try:
+            file_path = await segment.convert_to_file_path()
+            duration_ms = await get_media_duration(file_path)
+            if isinstance(duration_ms, int) and duration_ms > 0:
+                return duration_ms
+        except Exception as e:
+            logger.debug("[LINE] resolve record duration failed: %s", e)
+        return 1000
+
+    @staticmethod
+    async def _resolve_video_url(segment: Video) -> str:
+        candidate = (segment.file or "").strip()
+        if candidate.startswith("http://") or candidate.startswith("https://"):
+            return candidate
+        try:
+            return await segment.register_to_file_service()
+        except Exception as e:
+            logger.debug("[LINE] resolve video url failed: %s", e)
+            return ""
+
+    @staticmethod
+    async def _resolve_video_preview_url(segment: Video) -> str:
+        cover_candidate = (segment.cover or "").strip()
+        if cover_candidate.startswith("http://") or cover_candidate.startswith(
+            "https://"
+        ):
+            return cover_candidate
+
+        if cover_candidate:
+            try:
+                cover_seg = Image(file=cover_candidate)
+                return await cover_seg.register_to_file_service()
+            except Exception as e:
+                logger.debug("[LINE] resolve video cover failed: %s", e)
+
+        try:
+            video_path = await segment.convert_to_file_path()
+            temp_dir = Path(get_astrbot_temp_path())
+            temp_dir.mkdir(parents=True, exist_ok=True)
+            thumb_path = temp_dir / f"line_video_preview_{uuid.uuid4().hex}.jpg"
+
+            process = await asyncio.create_subprocess_exec(
+                "ffmpeg",
+                "-y",
+                "-ss",
+                "00:00:01",
+                "-i",
+                video_path,
+                "-frames:v",
+                "1",
+                str(thumb_path),
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            await process.communicate()
+            if process.returncode != 0 or not thumb_path.exists():
+                return ""
+
+            cover_seg = Image.fromFileSystem(str(thumb_path))
+            return await cover_seg.register_to_file_service()
+        except Exception as e:
+            logger.debug("[LINE] generate video preview failed: %s", e)
+            return ""
+
+    @staticmethod
+    async def _resolve_file_url(segment: File) -> str:
+        if segment.url and segment.url.startswith(("http://", "https://")):
+            return segment.url
+        try:
+            return await segment.register_to_file_service()
+        except Exception as e:
+            logger.debug("[LINE] resolve file url failed: %s", e)
+            return ""
+
+    @staticmethod
+    async def _resolve_file_size(segment: File) -> int:
+        try:
+            file_path = await segment.get_file(allow_return_url=False)
+            if file_path and os.path.exists(file_path):
+                return int(os.path.getsize(file_path))
+        except Exception as e:
+            logger.debug("[LINE] resolve file size failed: %s", e)
+        return 0
+
+    @classmethod
+    async def build_line_messages(cls, message_chain: MessageChain) -> list[dict]:
+        messages: list[dict] = []
+        for segment in message_chain.chain:
+            obj = await cls._component_to_message_object(segment)
+            if obj:
+                messages.append(obj)
+
+        if not messages:
+            return []
+
+        if len(messages) > 5:
+            logger.warning(
+                "[LINE] message count exceeds 5, extra segments will be dropped."
+            )
+            messages = messages[:5]
+        return messages
+
+    async def send(self, message: MessageChain) -> None:
+        messages = await self.build_line_messages(message)
+        if not messages:
+            return
+
+        raw = self.message_obj.raw_message
+        reply_token = ""
+        if isinstance(raw, dict):
+            reply_token = str(raw.get("replyToken") or "")
+
+        sent = False
+        if reply_token:
+            sent = await self.line_api.reply_message(reply_token, messages)
+
+        if not sent:
+            target_id = self.get_group_id() or self.get_sender_id()
+            if target_id:
+                await self.line_api.push_message(target_id, messages)
+
+        await super().send(message)
+
+    async def send_streaming(
+        self,
+        generator: AsyncGenerator,
+        use_fallback: bool = False,
+    ):
+        if not use_fallback:
+            buffer = None
+            async for chain in generator:
+                if not buffer:
+                    buffer = chain
+                else:
+                    buffer.chain.extend(chain.chain)
+            if not buffer:
+                return None
+            buffer.squash_plain()
+            await self.send(buffer)
+            return await super().send_streaming(generator, use_fallback)
+
+        buffer = ""
+        pattern = re.compile(r"[^。？！~…]+[。？！~…]+")
+
+        async for chain in generator:
+            if isinstance(chain, MessageChain):
+                for comp in chain.chain:
+                    if isinstance(comp, Plain):
+                        buffer += comp.text
+                        if any(p in buffer for p in "。？！~…"):
+                            buffer = await self.process_buffer(buffer, pattern)
+                    else:
+                        await self.send(MessageChain(chain=[comp]))
+                        await asyncio.sleep(1.5)
+
+        if buffer.strip():
+            await self.send(MessageChain([Plain(buffer)]))
+        return await super().send_streaming(generator, use_fallback)

astrbot/core/star/filter/platform_adapter_type.py

@@ -20,6 +20,7 @@ class PlatformAdapterType(enum.Flag):
     WEIXIN_OFFICIAL_ACCOUNT = enum.auto()
     SATORI = enum.auto()
     MISSKEY = enum.auto()
+    LINE = enum.auto()
     ALL = (
         AIOCQHTTP
         | QQOFFICIAL
@@ -34,6 +35,7 @@ class PlatformAdapterType(enum.Flag):
         | WEIXIN_OFFICIAL_ACCOUNT
         | SATORI
         | MISSKEY
+        | LINE
     )
 
 
@@ -51,6 +53,7 @@ ADAPTER_NAME_2_TYPE = {
     "weixin_official_account": PlatformAdapterType.WEIXIN_OFFICIAL_ACCOUNT,
     "satori": PlatformAdapterType.SATORI,
     "misskey": PlatformAdapterType.MISSKEY,
+    "line": PlatformAdapterType.LINE,
 }
 
 

dashboard/src/utils/platformUtils.js

@@ -34,6 +34,8 @@ export function getPlatformIcon(name) {
     return new URL('@/assets/images/platform_logos/satori.png', import.meta.url).href
   } else if (name === 'misskey') {
     return new URL('@/assets/images/platform_logos/misskey.png', import.meta.url).href
+  } else if (name === 'line') {
+    return new URL('@/assets/images/platform_logos/line.png', import.meta.url).href
   }
 }