Coverage for integrations / channels / extensions / line_adapter.py: 25.8%

1""" 

2LINE Channel Adapter 

3 

4Implements LINE Messaging API integration. 

5Based on HevolveBot extension patterns for LINE. 

6 

7Features: 

8- Messaging API (send/receive messages) 

9- Rich menus 

10- Flex Messages (rich UI components) 

11- LIFF (LINE Front-end Framework) integration 

12- Quick replies 

13- Image maps 

14- Stickers 

15- Location messages 

16- Push and reply messaging 

17- Webhook signature validation 

18""" 

19 

20from __future__ import annotations 

21 

22import asyncio 

23import logging 

24import os 

25import json 

26import hmac 

27import hashlib 

28import base64 

29from typing import Optional, List, Dict, Any, Callable, Union 

30from datetime import datetime 

31from dataclasses import dataclass, field 

32try: 

33 import aiohttp 

34 HAS_AIOHTTP = True 

35except ImportError: 

36 HAS_AIOHTTP = False 

37 

38try: 

39 from linebot import LineBotApi, WebhookHandler, WebhookParser 

40 from linebot.models import ( 

41 TextSendMessage, 

42 ImageSendMessage, 

43 VideoSendMessage, 

44 AudioSendMessage, 

45 LocationSendMessage, 

46 StickerSendMessage, 

47 FlexSendMessage, 

48 TemplateSendMessage, 

49 QuickReply, 

50 QuickReplyButton, 

51 MessageAction, 

52 URIAction, 

53 PostbackAction, 

54 RichMenu, 

55 RichMenuArea, 

56 RichMenuBounds, 

57 RichMenuSize, 

58 BubbleContainer, 

59 BoxComponent, 

60 TextComponent, 

61 ButtonComponent, 

62 ImageComponent, 

63 FlexMessage, 

64 ) 

65 from linebot.exceptions import LineBotApiError, InvalidSignatureError 

66 HAS_LINE = True 

67except ImportError: 

68 HAS_LINE = False 

69 

70from ..base import ( 

71 ChannelAdapter, 

72 ChannelConfig, 

73 ChannelStatus, 

74 Message, 

75 MessageType, 

76 MediaAttachment, 

77 SendResult, 

78 ChannelConnectionError, 

79 ChannelSendError, 

80 ChannelRateLimitError, 

81) 

82 

83logger = logging.getLogger(__name__) 

84 

85 

86@dataclass 

87class LINEConfig(ChannelConfig): 

88 """LINE-specific configuration.""" 

89 channel_access_token: str = "" 

90 channel_secret: str = "" 

91 liff_id: Optional[str] = None 

92 default_rich_menu_id: Optional[str] = None 

93 enable_push_messages: bool = True 

94 enable_flex_messages: bool = True 

95 

96 

97@dataclass 

98class FlexBubble: 

99 """Flex Message Bubble builder helper.""" 

100 header: Optional[Dict[str, Any]] = None 

101 hero: Optional[Dict[str, Any]] = None 

102 body: Optional[Dict[str, Any]] = None 

103 footer: Optional[Dict[str, Any]] = None 

104 size: str = "mega" 

105 

106 def set_header(self, text: str, **kwargs) -> 'FlexBubble': 

107 """Set header with text.""" 

108 self.header = { 

109 "type": "box", 

110 "layout": "vertical", 

111 "contents": [{ 

112 "type": "text", 

113 "text": text, 

114 "weight": kwargs.get("weight", "bold"), 

115 "size": kwargs.get("size", "xl"), 

116 }] 

117 } 

118 return self 

119 

120 def set_hero_image(self, url: str, **kwargs) -> 'FlexBubble': 

121 """Set hero image.""" 

122 self.hero = { 

123 "type": "image", 

124 "url": url, 

125 "size": kwargs.get("size", "full"), 

126 "aspectRatio": kwargs.get("aspect_ratio", "20:13"), 

127 "aspectMode": kwargs.get("aspect_mode", "cover"), 

128 } 

129 return self 

130 

131 def set_body(self, contents: List[Dict[str, Any]]) -> 'FlexBubble': 

132 """Set body contents.""" 

133 self.body = { 

134 "type": "box", 

135 "layout": "vertical", 

136 "contents": contents, 

137 } 

138 return self 

139 

140 def add_body_text(self, text: str, **kwargs) -> 'FlexBubble': 

141 """Add text to body.""" 

142 if not self.body: 

143 self.body = {"type": "box", "layout": "vertical", "contents": []} 

144 self.body["contents"].append({ 

145 "type": "text", 

146 "text": text, 

147 "wrap": kwargs.get("wrap", True), 

148 "size": kwargs.get("size", "md"), 

149 }) 

150 return self 

151 

152 def set_footer(self, buttons: List[Dict[str, Any]]) -> 'FlexBubble': 

153 """Set footer with buttons.""" 

154 self.footer = { 

155 "type": "box", 

156 "layout": "vertical", 

157 "spacing": "sm", 

158 "contents": buttons, 

159 } 

160 return self 

161 

162 def add_button( 

163 self, 

164 label: str, 

165 action_type: str = "message", 

166 data: str = None, 

167 **kwargs 

168 ) -> 'FlexBubble': 

169 """Add a button to footer.""" 

170 if not self.footer: 

171 self.footer = {"type": "box", "layout": "vertical", "spacing": "sm", "contents": []} 

172 

173 if action_type == "uri": 

174 action = {"type": "uri", "label": label, "uri": data} 

175 elif action_type == "postback": 

176 action = {"type": "postback", "label": label, "data": data} 

177 else: 

178 action = {"type": "message", "label": label, "text": data or label} 

179 

180 self.footer["contents"].append({ 

181 "type": "button", 

182 "style": kwargs.get("style", "primary"), 

183 "action": action, 

184 }) 

185 return self 

186 

187 def to_dict(self) -> Dict[str, Any]: 

188 """Convert to Flex Message JSON.""" 

189 bubble = { 

190 "type": "bubble", 

191 "size": self.size, 

192 } 

193 if self.header: 

194 bubble["header"] = self.header 

195 if self.hero: 

196 bubble["hero"] = self.hero 

197 if self.body: 

198 bubble["body"] = self.body 

199 if self.footer: 

200 bubble["footer"] = self.footer 

201 return bubble 

202 

203 

204@dataclass 

205class QuickReplyItem: 

206 """Quick reply button item.""" 

207 label: str 

208 action_type: str = "message" # message, postback, uri, datetime, camera, cameraRoll, location 

209 text: Optional[str] = None 

210 data: Optional[str] = None 

211 image_url: Optional[str] = None 

212 

213 

214class LINEAdapter(ChannelAdapter): 

215 """ 

216 LINE Messaging API adapter. 

217 

218 Usage: 

219 config = LINEConfig( 

220 channel_access_token="your-token", 

221 channel_secret="your-secret", 

222 ) 

223 adapter = LINEAdapter(config) 

224 adapter.on_message(my_handler) 

225 # Use with webhook endpoint 

226 """ 

227 

228 def __init__(self, config: LINEConfig): 

229 if not HAS_LINE: 

230 raise ImportError( 

231 "line-bot-sdk not installed. " 

232 "Install with: pip install line-bot-sdk" 

233 ) 

234 

235 super().__init__(config) 

236 self.line_config: LINEConfig = config 

237 self._api: Optional[LineBotApi] = None 

238 self._parser: Optional[WebhookParser] = None 

239 self._postback_handlers: Dict[str, Callable] = {} 

240 self._rich_menus: Dict[str, str] = {} # alias -> rich_menu_id 

241 

242 @property 

243 def name(self) -> str: 

244 return "line" 

245 

246 async def connect(self) -> bool: 

247 """Initialize LINE Bot API client.""" 

248 if not self.line_config.channel_access_token: 

249 logger.error("LINE channel access token required") 

250 return False 

251 

252 if not self.line_config.channel_secret: 

253 logger.error("LINE channel secret required") 

254 return False 

255 

256 try: 

257 # Initialize API client 

258 self._api = LineBotApi(self.line_config.channel_access_token) 

259 

260 # Initialize webhook parser 

261 self._parser = WebhookParser(self.line_config.channel_secret) 

262 

263 # Verify token by getting bot info 

264 bot_info = self._api.get_bot_info() 

265 logger.info(f"LINE connected as: {bot_info.display_name}") 

266 

267 self.status = ChannelStatus.CONNECTED 

268 return True 

269 

270 except LineBotApiError as e: 

271 logger.error(f"Failed to connect to LINE: {e.message}") 

272 self.status = ChannelStatus.ERROR 

273 return False 

274 except Exception as e: 

275 logger.error(f"Failed to connect to LINE: {e}") 

276 self.status = ChannelStatus.ERROR 

277 return False 

278 

279 async def disconnect(self) -> None: 

280 """Disconnect LINE adapter.""" 

281 self._api = None 

282 self._parser = None 

283 self.status = ChannelStatus.DISCONNECTED 

284 

285 def validate_signature(self, body: str, signature: str) -> bool: 

286 """Validate webhook signature.""" 

287 if not self.line_config.channel_secret: 

288 return False 

289 

290 hash_value = hmac.new( 

291 self.line_config.channel_secret.encode('utf-8'), 

292 body.encode('utf-8'), 

293 hashlib.sha256 

294 ).digest() 

295 

296 expected_signature = base64.b64encode(hash_value).decode('utf-8') 

297 return hmac.compare_digest(signature, expected_signature) 

298 

299 async def handle_webhook(self, body: str, signature: str) -> None: 

300 """ 

301 Handle incoming webhook request from LINE. 

302 Should be called from your webhook endpoint. 

303 """ 

304 if not self._parser: 

305 raise ChannelConnectionError("Adapter not initialized") 

306 

307 # Validate signature 

308 if not self.validate_signature(body, signature): 

309 raise InvalidSignatureError("Invalid signature") 

310 

311 # Parse events 

312 events = self._parser.parse(body, signature) 

313 

314 for event in events: 

315 await self._handle_event(event) 

316 

317 async def _handle_event(self, event: Any) -> None: 

318 """Handle LINE webhook event.""" 

319 event_type = event.type 

320 

321 if event_type == "message": 

322 await self._handle_message_event(event) 

323 elif event_type == "postback": 

324 await self._handle_postback_event(event) 

325 elif event_type == "follow": 

326 await self._handle_follow_event(event) 

327 elif event_type == "unfollow": 

328 await self._handle_unfollow_event(event) 

329 elif event_type == "join": 

330 await self._handle_join_event(event) 

331 elif event_type == "leave": 

332 await self._handle_leave_event(event) 

333 

334 async def _handle_message_event(self, event: Any) -> None: 

335 """Handle message event.""" 

336 message = self._convert_message(event) 

337 await self._dispatch_message(message) 

338 

339 async def _handle_postback_event(self, event: Any) -> None: 

340 """Handle postback event.""" 

341 data = event.postback.data 

342 

343 # Check for registered handler 

344 if data in self._postback_handlers: 

345 handler = self._postback_handlers[data] 

346 await handler(event) 

347 else: 

348 # Convert to message-like event 

349 message = Message( 

350 id=event.timestamp, 

351 channel=self.name, 

352 sender_id=event.source.user_id, 

353 chat_id=self._get_chat_id(event.source), 

354 text=f"[postback:{data}]", 

355 timestamp=datetime.fromtimestamp(event.timestamp / 1000), 

356 is_group=event.source.type != "user", 

357 raw={'postback': {'data': data}}, 

358 ) 

359 await self._dispatch_message(message) 

360 

361 async def _handle_follow_event(self, event: Any) -> None: 

362 """Handle follow event (user added bot).""" 

363 logger.info(f"User followed: {event.source.user_id}") 

364 

365 async def _handle_unfollow_event(self, event: Any) -> None: 

366 """Handle unfollow event (user blocked bot).""" 

367 logger.info(f"User unfollowed: {event.source.user_id}") 

368 

369 async def _handle_join_event(self, event: Any) -> None: 

370 """Handle join event (bot added to group/room).""" 

371 chat_id = self._get_chat_id(event.source) 

372 logger.info(f"Bot joined: {chat_id}") 

373 

374 async def _handle_leave_event(self, event: Any) -> None: 

375 """Handle leave event (bot removed from group/room).""" 

376 chat_id = self._get_chat_id(event.source) 

377 logger.info(f"Bot left: {chat_id}") 

378 

379 def _get_chat_id(self, source: Any) -> str: 

380 """Get chat ID from event source.""" 

381 if source.type == "user": 

382 return source.user_id 

383 elif source.type == "group": 

384 return source.group_id 

385 elif source.type == "room": 

386 return source.room_id 

387 return "" 

388 

389 def _convert_message(self, event: Any) -> Message: 

390 """Convert LINE event to unified Message format.""" 

391 source = event.source 

392 msg = event.message 

393 

394 # Get text content 

395 text = "" 

396 media = [] 

397 

398 if msg.type == "text": 

399 text = msg.text 

400 elif msg.type == "image": 

401 media.append(MediaAttachment( 

402 type=MessageType.IMAGE, 

403 file_id=msg.id, 

404 )) 

405 elif msg.type == "video": 

406 media.append(MediaAttachment( 

407 type=MessageType.VIDEO, 

408 file_id=msg.id, 

409 )) 

410 elif msg.type == "audio": 

411 media.append(MediaAttachment( 

412 type=MessageType.AUDIO, 

413 file_id=msg.id, 

414 )) 

415 elif msg.type == "location": 

416 text = f"[location:{msg.latitude},{msg.longitude}]" 

417 elif msg.type == "sticker": 

418 text = f"[sticker:{msg.package_id}/{msg.sticker_id}]" 

419 

420 # Determine chat type 

421 is_group = source.type != "user" 

422 chat_id = self._get_chat_id(source) 

423 

424 return Message( 

425 id=msg.id, 

426 channel=self.name, 

427 sender_id=source.user_id if hasattr(source, 'user_id') else "", 

428 chat_id=chat_id, 

429 text=text, 

430 media=media, 

431 timestamp=datetime.fromtimestamp(event.timestamp / 1000), 

432 is_group=is_group, 

433 raw={ 

434 'reply_token': event.reply_token, 

435 'source_type': source.type, 

436 'message_type': msg.type, 

437 }, 

438 ) 

439 

440 async def send_message( 

441 self, 

442 chat_id: str, 

443 text: str, 

444 reply_to: Optional[str] = None, 

445 media: Optional[List[MediaAttachment]] = None, 

446 buttons: Optional[List[Dict]] = None, 

447 ) -> SendResult: 

448 """Send a message to a LINE chat.""" 

449 if not self._api: 

450 return SendResult(success=False, error="Not connected") 

451 

452 try: 

453 messages = [] 

454 

455 # Build message based on content 

456 if buttons and self.line_config.enable_flex_messages: 

457 # Use flex message for buttons 

458 flex = self._build_flex_message(text, buttons) 

459 messages.append(FlexSendMessage(alt_text=text, contents=flex)) 

460 elif media and len(media) > 0: 

461 # Send media 

462 for m in media: 

463 if m.type == MessageType.IMAGE: 

464 messages.append(ImageSendMessage( 

465 original_content_url=m.url, 

466 preview_image_url=m.url, 

467 )) 

468 elif m.type == MessageType.VIDEO: 

469 messages.append(VideoSendMessage( 

470 original_content_url=m.url, 

471 preview_image_url=m.url, 

472 )) 

473 elif m.type == MessageType.AUDIO: 

474 messages.append(AudioSendMessage( 

475 original_content_url=m.url, 

476 duration=60000, # Default duration 

477 )) 

478 if text: 

479 messages.append(TextSendMessage(text=text)) 

480 else: 

481 messages.append(TextSendMessage(text=text)) 

482 

483 # Use push message if no reply token 

484 if reply_to: 

485 self._api.reply_message(reply_to, messages) 

486 else: 

487 self._api.push_message(chat_id, messages) 

488 

489 return SendResult(success=True) 

490 

491 except LineBotApiError as e: 

492 if e.status_code == 429: 

493 raise ChannelRateLimitError() 

494 logger.error(f"Failed to send LINE message: {e.message}") 

495 return SendResult(success=False, error=e.message) 

496 except Exception as e: 

497 logger.error(f"Failed to send LINE message: {e}") 

498 return SendResult(success=False, error=str(e)) 

499 

500 def _build_flex_message( 

501 self, 

502 text: str, 

503 buttons: List[Dict], 

504 ) -> Dict[str, Any]: 

505 """Build a flex message with buttons.""" 

506 bubble = FlexBubble() 

507 bubble.add_body_text(text) 

508 

509 for btn in buttons: 

510 if btn.get('url'): 

511 bubble.add_button( 

512 btn['text'], 

513 action_type="uri", 

514 data=btn['url'], 

515 ) 

516 else: 

517 bubble.add_button( 

518 btn['text'], 

519 action_type="postback", 

520 data=btn.get('callback_data', btn['text']), 

521 ) 

522 

523 return bubble.to_dict() 

524 

525 async def edit_message( 

526 self, 

527 chat_id: str, 

528 message_id: str, 

529 text: str, 

530 buttons: Optional[List[Dict]] = None, 

531 ) -> SendResult: 

532 """ 

533 Edit an existing LINE message. 

534 Note: LINE doesn't support editing messages, so this sends a new message. 

535 """ 

536 logger.warning("LINE doesn't support message editing, sending new message") 

537 return await self.send_message(chat_id, text, buttons=buttons) 

538 

539 async def delete_message(self, chat_id: str, message_id: str) -> bool: 

540 """ 

541 Delete a LINE message. 

542 Note: LINE doesn't support message deletion by bots. 

543 """ 

544 logger.warning("LINE doesn't support message deletion") 

545 return False 

546 

547 async def send_typing(self, chat_id: str) -> None: 

548 """ 

549 Send typing indicator. 

550 Note: LINE doesn't have a typing indicator API. 

551 """ 

552 pass 

553 

554 async def get_chat_info(self, chat_id: str) -> Optional[Dict[str, Any]]: 

555 """Get information about a LINE chat.""" 

556 if not self._api: 

557 return None 

558 

559 try: 

560 # Try to get group info 

561 try: 

562 group = self._api.get_group_summary(chat_id) 

563 return { 

564 'id': chat_id, 

565 'type': 'group', 

566 'name': group.group_name, 

567 'picture_url': group.picture_url, 

568 } 

569 except Exception: 

570 pass 

571 

572 # Try to get room info (chat rooms don't have names) 

573 try: 

574 # Rooms don't have summary API 

575 return { 

576 'id': chat_id, 

577 'type': 'room', 

578 } 

579 except Exception: 

580 pass 

581 

582 # Assume it's a user 

583 try: 

584 profile = self._api.get_profile(chat_id) 

585 return { 

586 'id': chat_id, 

587 'type': 'user', 

588 'name': profile.display_name, 

589 'picture_url': profile.picture_url, 

590 'status_message': profile.status_message, 

591 } 

592 except Exception: 

593 pass 

594 

595 return None 

596 

597 except Exception as e: 

598 logger.error(f"Failed to get chat info: {e}") 

599 return None 

600 

601 # LINE-specific methods 

602 

603 def register_postback_handler( 

604 self, 

605 data: str, 

606 handler: Callable[[Any], Any], 

607 ) -> None: 

608 """Register a handler for postback actions.""" 

609 self._postback_handlers[data] = handler 

610 

611 async def send_flex_message( 

612 self, 

613 chat_id: str, 

614 bubble: FlexBubble, 

615 alt_text: str = "Flex Message", 

616 reply_token: Optional[str] = None, 

617 ) -> SendResult: 

618 """Send a flex message.""" 

619 if not self._api: 

620 return SendResult(success=False, error="Not connected") 

621 

622 try: 

623 message = FlexSendMessage( 

624 alt_text=alt_text, 

625 contents=bubble.to_dict(), 

626 ) 

627 

628 if reply_token: 

629 self._api.reply_message(reply_token, message) 

630 else: 

631 self._api.push_message(chat_id, message) 

632 

633 return SendResult(success=True) 

634 

635 except LineBotApiError as e: 

636 logger.error(f"Failed to send flex message: {e.message}") 

637 return SendResult(success=False, error=e.message) 

638 

639 async def send_quick_reply( 

640 self, 

641 chat_id: str, 

642 text: str, 

643 items: List[QuickReplyItem], 

644 reply_token: Optional[str] = None, 

645 ) -> SendResult: 

646 """Send a message with quick reply buttons.""" 

647 if not self._api: 

648 return SendResult(success=False, error="Not connected") 

649 

650 try: 

651 quick_reply_items = [] 

652 

653 for item in items: 

654 if item.action_type == "message": 

655 action = MessageAction(label=item.label, text=item.text or item.label) 

656 elif item.action_type == "postback": 

657 action = PostbackAction(label=item.label, data=item.data or item.label) 

658 elif item.action_type == "uri": 

659 action = URIAction(label=item.label, uri=item.data) 

660 else: 

661 action = MessageAction(label=item.label, text=item.text or item.label) 

662 

663 quick_reply_items.append(QuickReplyButton( 

664 action=action, 

665 image_url=item.image_url, 

666 )) 

667 

668 message = TextSendMessage( 

669 text=text, 

670 quick_reply=QuickReply(items=quick_reply_items), 

671 ) 

672 

673 if reply_token: 

674 self._api.reply_message(reply_token, message) 

675 else: 

676 self._api.push_message(chat_id, message) 

677 

678 return SendResult(success=True) 

679 

680 except LineBotApiError as e: 

681 logger.error(f"Failed to send quick reply: {e.message}") 

682 return SendResult(success=False, error=e.message) 

683 

684 async def send_sticker( 

685 self, 

686 chat_id: str, 

687 package_id: str, 

688 sticker_id: str, 

689 reply_token: Optional[str] = None, 

690 ) -> SendResult: 

691 """Send a sticker.""" 

692 if not self._api: 

693 return SendResult(success=False, error="Not connected") 

694 

695 try: 

696 message = StickerSendMessage( 

697 package_id=package_id, 

698 sticker_id=sticker_id, 

699 ) 

700 

701 if reply_token: 

702 self._api.reply_message(reply_token, message) 

703 else: 

704 self._api.push_message(chat_id, message) 

705 

706 return SendResult(success=True) 

707 

708 except LineBotApiError as e: 

709 logger.error(f"Failed to send sticker: {e.message}") 

710 return SendResult(success=False, error=e.message) 

711 

712 async def send_location( 

713 self, 

714 chat_id: str, 

715 title: str, 

716 address: str, 

717 latitude: float, 

718 longitude: float, 

719 reply_token: Optional[str] = None, 

720 ) -> SendResult: 

721 """Send a location message.""" 

722 if not self._api: 

723 return SendResult(success=False, error="Not connected") 

724 

725 try: 

726 message = LocationSendMessage( 

727 title=title, 

728 address=address, 

729 latitude=latitude, 

730 longitude=longitude, 

731 ) 

732 

733 if reply_token: 

734 self._api.reply_message(reply_token, message) 

735 else: 

736 self._api.push_message(chat_id, message) 

737 

738 return SendResult(success=True) 

739 

740 except LineBotApiError as e: 

741 logger.error(f"Failed to send location: {e.message}") 

742 return SendResult(success=False, error=e.message) 

743 

744 async def create_rich_menu( 

745 self, 

746 name: str, 

747 chat_bar_text: str, 

748 areas: List[Dict[str, Any]], 

749 size: tuple = (2500, 1686), 

750 ) -> Optional[str]: 

751 """Create a rich menu.""" 

752 if not self._api: 

753 return None 

754 

755 try: 

756 rich_menu = RichMenu( 

757 size=RichMenuSize(width=size[0], height=size[1]), 

758 selected=True, 

759 name=name, 

760 chat_bar_text=chat_bar_text, 

761 areas=[ 

762 RichMenuArea( 

763 bounds=RichMenuBounds( 

764 x=area['x'], 

765 y=area['y'], 

766 width=area['width'], 

767 height=area['height'], 

768 ), 

769 action=self._build_action(area['action']), 

770 ) 

771 for area in areas 

772 ], 

773 ) 

774 

775 rich_menu_id = self._api.create_rich_menu(rich_menu) 

776 self._rich_menus[name] = rich_menu_id 

777 return rich_menu_id 

778 

779 except LineBotApiError as e: 

780 logger.error(f"Failed to create rich menu: {e.message}") 

781 return None 

782 

783 def _build_action(self, action: Dict[str, Any]) -> Any: 

784 """Build action object from dict.""" 

785 action_type = action.get('type', 'message') 

786 

787 if action_type == 'uri': 

788 return URIAction(label=action.get('label', ''), uri=action.get('uri', '')) 

789 elif action_type == 'postback': 

790 return PostbackAction(label=action.get('label', ''), data=action.get('data', '')) 

791 else: 

792 return MessageAction(label=action.get('label', ''), text=action.get('text', '')) 

793 

794 async def set_rich_menu(self, user_id: str, rich_menu_id: str) -> bool: 

795 """Link a rich menu to a user.""" 

796 if not self._api: 

797 return False 

798 

799 try: 

800 self._api.link_rich_menu_to_user(user_id, rich_menu_id) 

801 return True 

802 except LineBotApiError as e: 

803 logger.error(f"Failed to set rich menu: {e.message}") 

804 return False 

805 

806 async def get_message_content(self, message_id: str) -> Optional[bytes]: 

807 """Get content of a media message.""" 

808 if not self._api: 

809 return None 

810 

811 try: 

812 content = self._api.get_message_content(message_id) 

813 return content.content 

814 except LineBotApiError as e: 

815 logger.error(f"Failed to get message content: {e.message}") 

816 return None 

817 

818 def get_liff_url(self, path: str = "") -> Optional[str]: 

819 """Get LIFF URL for the configured LIFF app.""" 

820 if not self.line_config.liff_id: 

821 return None 

822 return f"https://liff.line.me/{self.line_config.liff_id}{path}" 

823 

824 

825def create_line_adapter( 

826 channel_access_token: str = None, 

827 channel_secret: str = None, 

828 **kwargs 

829) -> LINEAdapter: 

830 """ 

831 Factory function to create LINE adapter. 

832 

833 Args: 

834 channel_access_token: LINE channel access token (or set LINE_CHANNEL_ACCESS_TOKEN env var) 

835 channel_secret: LINE channel secret (or set LINE_CHANNEL_SECRET env var) 

836 **kwargs: Additional config options 

837 

838 Returns: 

839 Configured LINEAdapter 

840 """ 

841 channel_access_token = channel_access_token or os.getenv("LINE_CHANNEL_ACCESS_TOKEN") 

842 channel_secret = channel_secret or os.getenv("LINE_CHANNEL_SECRET") 

843 

844 if not channel_access_token: