From f736a0ebb3c4ba2a4ca6163fb5f44e1f917827cd Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:11:04 +0800 Subject: [PATCH] =?UTF-8?q?refactor(line):=20=E9=87=8D=E6=9E=84Line?= =?UTF-8?q?=E9=80=82=E9=85=8D=E5=99=A8=E7=9B=B8=E5=85=B3=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=EF=BC=8C=E4=BC=98=E5=8C=96=E5=8A=9F=E8=83=BD=E4=B8=8E=E5=8F=AF?= =?UTF-8?q?=E8=AF=BB=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 抽离alt文本截断逻辑为公共工具函数 2. 重构快速回复构建函数，支持更多action类型 3. 新增 narrowcast 消息发送与进度查询能力 4. 新增用户粉丝数、好友人口统计API调用 5. 优化回复消息逻辑，支持quote token 6. 改进审批ID生成逻辑，避免重复 7. 调整部分配置与异常处理逻辑 8. 新增熔断器与相关指标统计 --- .../yuxi/channels/adapters/line/adapter.py | 158 +++++++++++++----- .../yuxi/channels/adapters/line/approval.py | 6 +- .../yuxi/channels/adapters/line/directives.py | 15 +- .../adapters/line/flex_templates/common.py | 6 +- .../channels/adapters/line/quick_reply.py | 77 +++++++-- .../yuxi/channels/adapters/line/send.py | 91 +++++++++- 6 files changed, 283 insertions(+), 70 deletions(-) diff --git a/backend/package/yuxi/channels/adapters/line/adapter.py b/backend/package/yuxi/channels/adapters/line/adapter.py index f8ffd890..986a4327 100644 --- a/backend/package/yuxi/channels/adapters/line/adapter.py +++ b/backend/package/yuxi/channels/adapters/line/adapter.py @@ -9,10 +9,10 @@ from collections import deque from collections.abc import AsyncIterator from typing import Any, ClassVar +from yuxi.channels.adapters.line.approval import LINEApprovalAdapter from yuxi.channels.adapters.line.formatter import LINEMessageFormatter from yuxi.channels.adapters.line.markdown_to_line import extract_flex_messages_from_markdown from yuxi.channels.adapters.line.normalizer import LINEEventNormalizer -from yuxi.channels.adapters.line.approval import LINEApprovalAdapter from yuxi.channels.adapters.line.probe import probe_line_bot from yuxi.channels.adapters.line.send import LINESender from yuxi.channels.adapters.line.webhook import ( @@ -26,6 +26,7 @@ from yuxi.channels.capabilities import ChannelCapabilities from yuxi.channels.exceptions import ( ChannelAuthenticationError, ) +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( ChannelAccountSnapshot, @@ -113,8 +114,8 @@ class LINEAdapter(BaseChannelAdapter): capabilities = ChannelCapabilities( chat_types=["direct", "group"], - polls=False, - reactions=False, + polls=True, + reactions=True, edit=False, unsend=False, reply=True, @@ -127,6 +128,7 @@ class LINEAdapter(BaseChannelAdapter): block_streaming=True, text_chunk_limit=5000, max_media_size_mb=10, + narrowcast=True, ) meta = ChannelMeta( id="line", @@ -155,7 +157,7 @@ class LINEAdapter(BaseChannelAdapter): self._token_source: str = "unknown" self._last_reply_token: str | None = None self._last_reply_token_at: float | None = None - self._token_lock: asyncio.Lock | None = None + self._token_lock: asyncio.Lock = asyncio.Lock() self.dm_policy: str = config.get("dm_policy", "open") if config else "open" self.group_policy: str = config.get("group_policy", "open") if config else "open" @@ -200,6 +202,7 @@ class LINEAdapter(BaseChannelAdapter): self._replay_guard = WebhookReplayGuard() self._signature_router = MultiAccountSignatureRouter() + self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60, channel_id="line") self.thread_bindings_enabled: bool = ( config.get("thread_bindings", {}).get("enabled", False) if config else False @@ -293,8 +296,6 @@ class LINEAdapter(BaseChannelAdapter): if not secret: raise ChannelAuthenticationError("LINE Channel Secret not configured") - self._token_lock = asyncio.Lock() - proxy_url = self.config.get("proxy") self._sender = LINESender(token, proxy=proxy_url) await self._sender.__aenter__() @@ -331,7 +332,7 @@ class LINEAdapter(BaseChannelAdapter): self._queue_task.cancel() try: await asyncio.wait_for(self._queue_task, timeout=5.0) - except (asyncio.CancelledError, asyncio.TimeoutError, Exception): + except (TimeoutError, asyncio.CancelledError, Exception): pass self._queue_task = None if self._sender: @@ -350,7 +351,7 @@ class LINEAdapter(BaseChannelAdapter): for task in self._loading_animation_tasks.values(): try: await asyncio.wait_for(task, timeout=3.0) - except (asyncio.CancelledError, asyncio.TimeoutError, Exception): + except (TimeoutError, asyncio.CancelledError, Exception): pass self._loading_animation_tasks.clear() if self._stream_cleanup_task: @@ -393,44 +394,53 @@ class LINEAdapter(BaseChannelAdapter): messages = self._build_send_payload(response) - reply_token = response.metadata.get("reply_token") - if reply_token: - result = ( - await self._sender.reply_message(reply_token, messages) if messages else DeliveryResult(success=True) - ) - if not result.success and _is_reply_token_expired(result.error): - logger.info("[LINE] replyToken expired, falling back to push") + async def _do_send() -> DeliveryResult: + reply_token = response.metadata.get("reply_token") + if reply_token: result = ( - await self._sender.push_message(chat_id, messages) if messages else DeliveryResult(success=True) + await self._sender.reply_message(reply_token, messages) + if messages + else DeliveryResult(success=True) ) + if not result.success and _is_reply_token_expired(result.error): + logger.info("[LINE] replyToken expired, falling back to push") + result = ( + await self._sender.push_message(chat_id, messages) if messages else DeliveryResult(success=True) + ) + self._record_result(result) + if _is_auth_error(result.error): + await self._auto_disable_graduated("LINE auth error during reply") + if _is_comm_channel_disabled(result.error): + await self._auto_disable_graduated("LINE communication channel disabled") + if result.success and result.message_id: + self._track_sent_message(result.message_id, chat_id, "reply") + self._last_message_at = time.time() + return result + + if not messages: + self._last_message_at = time.time() + return DeliveryResult(success=True) + result = await self._sender.push_message(chat_id, messages) self._record_result(result) if _is_auth_error(result.error): - await self._auto_disable_graduated("LINE auth error during reply") + await self._auto_disable_graduated("LINE auth error during push") if _is_comm_channel_disabled(result.error): await self._auto_disable_graduated("LINE communication channel disabled") + if _is_rate_limited(result.error): + logger.warning(f"[LINE] rate limited during push to {chat_id}") + if result.error: + error_category = _classify_send_error(result.error) + logger.debug(f"[LINE] push error category={error_category}: {result.error}") if result.success and result.message_id: - self._track_sent_message(result.message_id, chat_id, "reply") + self._track_sent_message(result.message_id, chat_id, "push") self._last_message_at = time.time() return result - if not messages: - self._last_message_at = time.time() - return DeliveryResult(success=True) - result = await self._sender.push_message(chat_id, messages) - self._record_result(result) - if _is_auth_error(result.error): - await self._auto_disable_graduated("LINE auth error during push") - if _is_comm_channel_disabled(result.error): - await self._auto_disable_graduated("LINE communication channel disabled") - if _is_rate_limited(result.error): - logger.warning(f"[LINE] rate limited during push to {chat_id}") - if result.error: - error_category = _classify_send_error(result.error) - logger.debug(f"[LINE] push error category={error_category}: {result.error}") - if result.success and result.message_id: - self._track_sent_message(result.message_id, chat_id, "push") - self._last_message_at = time.time() - return result + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + self._metrics["messages_failed"] += 1 + return DeliveryResult(success=False, error="Circuit breaker open") def _build_send_payload(self, response: ChannelResponse) -> list[dict]: metadata = response.metadata @@ -722,6 +732,40 @@ class LINEAdapter(BaseChannelAdapter): messages = self._formatter.format(response) return await self._sender.broadcast_message(messages) + async def send_narrowcast( + self, + messages: list[dict], + recipient: dict | None = None, + demographic_filter: dict | None = None, + limit: dict | None = None, + notification_disabled: bool = False, + retry_key: str | None = None, + ) -> DeliveryResult: + if self._status != ChannelStatus.CONNECTED or not self._sender: + return DeliveryResult(success=False, error="LINE not connected") + + result = await self._sender.narrowcast_message( + messages=messages, + recipient=recipient, + demographic_filter=demographic_filter, + limit=limit, + notification_disabled=notification_disabled, + retry_key=retry_key, + ) + if result.success: + self._metrics["messages_sent"] += 1 + else: + self._metrics["messages_failed"] += 1 + self._metrics["error_counts"][_classify_send_error(result.error)] = ( + self._metrics["error_counts"].get(_classify_send_error(result.error), 0) + 1 + ) + return result + + async def get_narrowcast_progress(self, request_id: str) -> dict | None: + if self._status != ChannelStatus.CONNECTED or not self._sender: + return None + return await self._sender.get_narrowcast_progress(request_id) + async def mark_as_read(self, chat_id: str) -> bool: if self._status != ChannelStatus.CONNECTED or not self._sender: return False @@ -767,11 +811,6 @@ class LINEAdapter(BaseChannelAdapter): return False return await self._sender.link_rich_menu_to_user(user_id, rich_menu_id) - async def link_rich_menu_bulk(self, user_ids: list[str], rich_menu_id: str) -> bool: - if self._status != ChannelStatus.CONNECTED or not self._sender: - return False - return await self._sender.link_rich_menu_to_users(user_ids, rich_menu_id) - async def unlink_rich_menu(self, user_id: str) -> bool: if self._status != ChannelStatus.CONNECTED or not self._sender: return False @@ -1227,9 +1266,10 @@ class LINEAdapter(BaseChannelAdapter): @staticmethod def _trim_set(id_set: set, order: deque, max_size: int = 10000, keep: int = 5000): - while len(order) > max_size: - old = order.popleft() - id_set.discard(old) + if len(order) > max_size: + while len(order) > keep: + old = order.popleft() + id_set.discard(old) @staticmethod def _ensure_https_url(url: str) -> str | None: @@ -1266,7 +1306,7 @@ class LINEAdapter(BaseChannelAdapter): async def receive(self) -> AsyncIterator[ChannelMessage]: return - yield # pragma: no cover + yield # type: ignore[misc] def normalize_inbound(self, raw: bytes) -> ChannelMessage: body_str = raw.decode("utf-8") if isinstance(raw, bytes) else raw @@ -1552,6 +1592,34 @@ class LINEAdapter(BaseChannelAdapter): return peers + @staticmethod + def build_text_v2_message(text: str, substitutions: list[dict] | None = None) -> dict: + msg: dict = {"type": "textV2", "text": text} + if substitutions: + msg["substitutions"] = substitutions + return msg + + @staticmethod + def build_imagemap_message( + base_url: str, + alt_text: str, + base_width: int = 1040, + base_height: int = 1040, + actions: list[dict] | None = None, + video: dict | None = None, + ) -> dict: + msg: dict = { + "type": "imagemap", + "baseUrl": base_url, + "altText": alt_text[:400], + "baseSize": {"width": base_width, "height": base_height}, + } + if actions: + msg["actions"] = actions + if video: + msg["video"] = video + return msg + def build_channel_summary(self) -> dict: return { "channel": "line", diff --git a/backend/package/yuxi/channels/adapters/line/approval.py b/backend/package/yuxi/channels/adapters/line/approval.py index 0f35b701..a524fdf3 100644 --- a/backend/package/yuxi/channels/adapters/line/approval.py +++ b/backend/package/yuxi/channels/adapters/line/approval.py @@ -1,9 +1,13 @@ from __future__ import annotations +import itertools + from yuxi.utils.logging_config import logger class LINEApprovalAdapter: + _id_counter = itertools.count() + def __init__(self, adapter): self._adapter = adapter self._pending_approvals: dict[str, dict] = {} @@ -19,7 +23,7 @@ class LINEApprovalAdapter: ) -> dict | None: import time - approval_id = f"apr_{int(time.time() * 1000)}_{chat_id[:10]}" + approval_id = f"apr_{int(time.time() * 1000)}_{next(self._id_counter)}_{chat_id[:10]}" if actions is None: actions = [ diff --git a/backend/package/yuxi/channels/adapters/line/directives.py b/backend/package/yuxi/channels/adapters/line/directives.py index 34940441..fb4e6f27 100644 --- a/backend/package/yuxi/channels/adapters/line/directives.py +++ b/backend/package/yuxi/channels/adapters/line/directives.py @@ -2,6 +2,9 @@ from __future__ import annotations import re +from yuxi.channels.adapters.line.flex_templates.common import truncate_alt_text +from yuxi.channels.adapters.line.quick_reply import build_quick_reply_items + QUICK_REPLIES_RE = re.compile(r"\[\[quick_replies:\s*(.+?)\]\]") CONFIRM_RE = re.compile(r"\[\[confirm:\s*(.+?)\|\s*(.+?)\|\s*(.+?)\]\]") BUTTONS_RE = re.compile(r"\[\[buttons:\s*(.+?)\]\]") @@ -20,13 +23,7 @@ def parse_line_directives(text: str) -> list[dict] | None: if qr_match and len(result) < _MAX_DIRECTIVES: labels = [lbl.strip() for lbl in qr_match.group(1).split(",") if lbl.strip()] main_text = QUICK_REPLIES_RE.sub("", text).strip() - items = [ - { - "type": "action", - "action": {"type": "message", "label": label[:20], "text": label}, - } - for label in labels[:13] - ] + items = build_quick_reply_items(labels) if items: result.append( { @@ -48,7 +45,7 @@ def parse_line_directives(text: str) -> list[dict] | None: result.append( { "type": "template", - "altText": question[:400], + "altText": truncate_alt_text(question), "template": { "type": "confirm", "text": question[:240], @@ -88,7 +85,7 @@ def parse_line_directives(text: str) -> list[dict] | None: result.append( { "type": "template", - "altText": title[:400], + "altText": truncate_alt_text(title), "template": { "type": "buttons", "title": title[:40], diff --git a/backend/package/yuxi/channels/adapters/line/flex_templates/common.py b/backend/package/yuxi/channels/adapters/line/flex_templates/common.py index a21416c6..8afce1a8 100644 --- a/backend/package/yuxi/channels/adapters/line/flex_templates/common.py +++ b/backend/package/yuxi/channels/adapters/line/flex_templates/common.py @@ -4,7 +4,7 @@ from __future__ import annotations def to_flex_message(contents: dict, alt_text: str = "Flex Message") -> dict: return { "type": "flex", - "altText": alt_text[:400], + "altText": truncate_alt_text(alt_text), "contents": contents, } @@ -59,3 +59,7 @@ def _hero_image(url: str, aspect_ratio: str = "20:13") -> dict: "aspectRatio": aspect_ratio, "aspectMode": "cover", } + + +def truncate_alt_text(text: str, max_len: int = 400) -> str: + return text[:max_len] diff --git a/backend/package/yuxi/channels/adapters/line/quick_reply.py b/backend/package/yuxi/channels/adapters/line/quick_reply.py index 49bb083c..5dbf518a 100644 --- a/backend/package/yuxi/channels/adapters/line/quick_reply.py +++ b/backend/package/yuxi/channels/adapters/line/quick_reply.py @@ -1,22 +1,75 @@ from __future__ import annotations -def build_quick_reply_items(options: list[str]) -> list[dict]: +def build_quick_reply_items(options: list[str | dict]) -> list[dict]: items = [] - for label in options[:13]: - if not label or not label.strip(): - continue - safe_label = label.strip()[:20] - items.append( - { - "type": "action", - "action": {"type": "message", "label": safe_label, "text": label}, - } - ) + for opt in options[:13]: + if isinstance(opt, str): + label = opt.strip() + if not label: + continue + safe_label = label[:20] + items.append( + { + "type": "action", + "action": {"type": "message", "label": safe_label, "text": label}, + } + ) + elif isinstance(opt, dict): + label = opt.get("label", "").strip() + if not label: + continue + safe_label = label[:20] + action_type = opt.get("action_type", "message") + + if action_type == "postback": + action = { + "type": "postback", + "label": safe_label, + "data": opt.get("data", ""), + "display_text": opt.get("display_text", safe_label), + } + if "input_option" in opt: + action["inputOption"] = opt["input_option"] + if "fill_in_text" in opt: + action["fillInText"] = opt["fill_in_text"] + elif action_type == "uri": + action = { + "type": "uri", + "label": safe_label, + "uri": opt.get("uri", ""), + } + if opt.get("alt_uri"): + alt_uri = opt["alt_uri"] + if isinstance(alt_uri, dict): + action["altUri"] = alt_uri + elif action_type == "camera": + action = { + "type": "camera", + "label": safe_label, + } + elif action_type == "camera_roll": + action = { + "type": "cameraRoll", + "label": safe_label, + } + elif action_type == "location": + action = { + "type": "location", + "label": safe_label, + } + else: + action = { + "type": "message", + "label": safe_label, + "text": opt.get("text", label), + } + + items.append({"type": "action", "action": action}) return items -def build_text_with_quick_reply(text: str, options: list[str]) -> dict: +def build_text_with_quick_reply(text: str, options: list[str | dict]) -> dict: if not options: return {"type": "text", "text": text[:5000]} items = build_quick_reply_items(options) diff --git a/backend/package/yuxi/channels/adapters/line/send.py b/backend/package/yuxi/channels/adapters/line/send.py index 1ce49e69..285b7b52 100644 --- a/backend/package/yuxi/channels/adapters/line/send.py +++ b/backend/package/yuxi/channels/adapters/line/send.py @@ -52,8 +52,15 @@ class LINESender: payload = {"messages": messages[:5]} return await self._post_with_retry("/message/broadcast", payload, api_name="broadcast") - async def reply_message(self, reply_token: str, messages: list[dict]) -> DeliveryResult: - payload = {"replyToken": reply_token, "messages": messages[:5]} + async def reply_message( + self, reply_token: str, messages: list[dict], quote_token: str | None = None + ) -> DeliveryResult: + payload: dict = {"replyToken": reply_token, "messages": messages[:5]} + if quote_token: + messages_for_send = list(payload["messages"]) + if messages_for_send: + messages_for_send[0] = {**messages_for_send[0], "quoteToken": quote_token} + payload["messages"] = messages_for_send return await self._post_with_retry("/message/reply", payload, api_name="reply") async def show_loading_animation(self, chat_id: str, seconds: int = 20) -> None: @@ -183,6 +190,32 @@ class LINESender: except Exception: return None + async def get_number_of_followers(self, date: str | None = None) -> dict | None: + if not self._client: + return None + try: + params = {"date": date} if date else {} + resp = await self._client.get( + "/insight/followers", + params=params, + ) + if resp.status_code == 200: + return resp.json() + return None + except Exception: + return None + + async def get_friend_demographics(self) -> dict | None: + if not self._client: + return None + try: + resp = await self._client.get("/insight/message/demographic") + if resp.status_code == 200: + return resp.json() + return None + except Exception: + return None + async def get_delivery_status(self, date: str, msg_type: str = "push") -> dict | None: if not self._client: return None @@ -257,6 +290,60 @@ class LINESender: async def validate_broadcast(self, messages: list[dict]) -> bool: return await self._validate_messages("/message/validate/broadcast", {"messages": messages[:5]}) + async def narrowcast_message( + self, + messages: list[dict], + recipient: dict | None = None, + demographic_filter: dict | None = None, + limit: dict | None = None, + notification_disabled: bool = False, + retry_key: str | None = None, + ) -> DeliveryResult: + if not self._client: + return DeliveryResult(success=False, error="Client not initialized") + + payload: dict = {"messages": messages[:5]} + if recipient: + payload["recipient"] = recipient + if demographic_filter: + payload["filter"] = {"demographic": demographic_filter} + if limit: + payload["limit"] = limit + if notification_disabled: + payload["notificationDisabled"] = True + + headers = {} + if retry_key: + headers["X-Line-Retry-Key"] = retry_key + + try: + resp = await self._client.post( + "/message/narrowcast", + json=payload, + headers=headers, + ) + if resp.status_code == 202: + return DeliveryResult(success=True, message_id="narrowcast_accepted") + if resp.status_code == 200: + return DeliveryResult(success=True) + return DeliveryResult( + success=False, + error=f"Narrowcast failed: {resp.status_code}", + ) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def get_narrowcast_progress(self, request_id: str) -> dict | None: + if not self._client: + return None + try: + resp = await self._client.get(f"/message/progress/narrowcast?requestId={request_id}") + if resp.status_code == 200: + return resp.json() + return None + except Exception: + return None + async def _validate_messages(self, path: str, payload: dict) -> bool: if not self._client: return False