Coverage for integrations / channels / extensions / zalo_adapter.py: 28.9%

1""" 

2Zalo Channel Adapter 

3 

4Implements Zalo Official Account (OA) API integration. 

5Based on HevolveBot extension patterns for Vietnamese messaging platform. 

6 

7Features: 

8- Official Account (OA) API integration 

9- Text, image, file messaging 

10- Quick reply buttons 

11- List templates 

12- Request user info 

13- Broadcast messaging 

14- Webhook signature validation 

15- User interest tagging 

16- Follower management 

17- Reconnection with exponential backoff 

18""" 

19 

20from __future__ import annotations 

21 

22import asyncio 

23import logging 

24import os 

25import json 

26import hmac 

27import hashlib 

28import time 

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

30from datetime import datetime 

31from dataclasses import dataclass, field 

32from enum import Enum 

33 

34try: 

35 import aiohttp 

36 HAS_ZALO = True 

37except ImportError: 

38 HAS_ZALO = False 

39 

40from ..base import ( 

41 ChannelAdapter, 

42 ChannelConfig, 

43 ChannelStatus, 

44 Message, 

45 MessageType, 

46 MediaAttachment, 

47 SendResult, 

48 ChannelConnectionError, 

49 ChannelSendError, 

50 ChannelRateLimitError, 

51) 

52 

53logger = logging.getLogger(__name__) 

54 

55# Zalo API endpoints 

56ZALO_OA_API_URL = "https://openapi.zalo.me/v2.0/oa" 

57ZALO_GRAPH_URL = "https://graph.zalo.me/v2.0" 

58ZALO_UPLOAD_URL = "https://openapi.zalo.me/v2.0/oa/upload" 

59 

60 

61class ZaloMessageType(Enum): 

62 """Zalo message types.""" 

63 TEXT = "text" 

64 IMAGE = "image" 

65 FILE = "file" 

66 STICKER = "sticker" 

67 GIF = "gif" 

68 LIST = "list" 

69 REQUEST_USER_INFO = "request_user_info" 

70 

71 

72class ZaloEventType(Enum): 

73 """Zalo webhook event types.""" 

74 USER_SEND_TEXT = "user_send_text" 

75 USER_SEND_IMAGE = "user_send_image" 

76 USER_SEND_FILE = "user_send_file" 

77 USER_SEND_STICKER = "user_send_sticker" 

78 USER_SEND_GIF = "user_send_gif" 

79 USER_SEND_LOCATION = "user_send_location" 

80 FOLLOW = "follow" 

81 UNFOLLOW = "unfollow" 

82 USER_CLICK_BUTTON = "user_click_button" 

83 USER_SUBMIT_INFO = "user_submit_info" 

84 

85 

86@dataclass 

87class ZaloConfig(ChannelConfig): 

88 """Zalo-specific configuration.""" 

89 app_id: str = "" 

90 app_secret: str = "" 

91 access_token: str = "" 

92 refresh_token: str = "" 

93 oa_id: str = "" # Official Account ID 

94 webhook_secret: str = "" 

95 enable_user_info_request: bool = True 

96 enable_broadcast: bool = False 

97 reconnect_attempts: int = 5 

98 reconnect_delay: float = 1.0 

99 

100 

101@dataclass 

102class ZaloUser: 

103 """Zalo user information.""" 

104 user_id: str 

105 display_name: Optional[str] = None 

106 avatar: Optional[str] = None 

107 phone: Optional[str] = None 

108 is_follower: bool = True 

109 user_id_by_app: Optional[str] = None 

110 shared_info: Dict[str, Any] = field(default_factory=dict) 

111 

112 

113@dataclass 

114class ZaloQuickReply: 

115 """Quick reply button.""" 

116 title: str 

117 payload: str 

118 image_url: Optional[str] = None 

119 

120 

121@dataclass 

122class ZaloListElement: 

123 """Element in a list template.""" 

124 title: str 

125 subtitle: Optional[str] = None 

126 image_url: Optional[str] = None 

127 default_action: Optional[Dict[str, Any]] = None 

128 

129 

130class ZaloAdapter(ChannelAdapter): 

131 """ 

132 Zalo Official Account API adapter. 

133 

134 Usage: 

135 config = ZaloConfig( 

136 app_id="your-app-id", 

137 app_secret="your-secret", 

138 access_token="your-token", 

139 oa_id="your-oa-id", 

140 ) 

141 adapter = ZaloAdapter(config) 

142 adapter.on_message(my_handler) 

143 # Use with webhook endpoint 

144 """ 

145 

146 def __init__(self, config: ZaloConfig): 

147 if not HAS_ZALO: 

148 raise ImportError( 

149 "aiohttp not installed. " 

150 "Install with: pip install aiohttp" 

151 ) 

152 

153 super().__init__(config) 

154 self.zalo_config: ZaloConfig = config 

155 self._session: Optional[aiohttp.ClientSession] = None 

156 self._user_cache: Dict[str, ZaloUser] = {} 

157 self._follow_handlers: List[Callable] = [] 

158 self._unfollow_handlers: List[Callable] = [] 

159 self._button_handlers: Dict[str, Callable] = {} 

160 self._user_info_handlers: List[Callable] = [] 

161 self._reconnect_count: int = 0 

162 

163 @property 

164 def name(self) -> str: 

165 return "zalo" 

166 

167 async def connect(self) -> bool: 

168 """Initialize Zalo OA API client.""" 

169 if not self.zalo_config.access_token: 

170 logger.error("Zalo access token required") 

171 return False 

172 

173 if not self.zalo_config.oa_id: 

174 logger.error("Zalo OA ID required") 

175 return False 

176 

177 try: 

178 # Create aiohttp session 

179 self._session = aiohttp.ClientSession() 

180 

181 # Verify token by getting OA info 

182 oa_info = await self._get_oa_info() 

183 if oa_info: 

184 logger.info(f"Zalo OA connected: {oa_info.get('name', 'Unknown')}") 

185 self.status = ChannelStatus.CONNECTED 

186 self._reconnect_count = 0 

187 return True 

188 else: 

189 logger.error("Failed to verify Zalo OA token") 

190 self.status = ChannelStatus.ERROR 

191 return False 

192 

193 except Exception as e: 

194 logger.error(f"Failed to connect to Zalo: {e}") 

195 self.status = ChannelStatus.ERROR 

196 return False 

197 

198 async def disconnect(self) -> None: 

199 """Disconnect Zalo adapter.""" 

200 if self._session: 

201 await self._session.close() 

202 self._session = None 

203 

204 self._user_cache.clear() 

205 self.status = ChannelStatus.DISCONNECTED 

206 

207 async def _get_oa_info(self) -> Optional[Dict[str, Any]]: 

208 """Get OA information to verify connection.""" 

209 if not self._session: 

210 return None 

211 

212 try: 

213 headers = { 

214 "access_token": self.zalo_config.access_token, 

215 } 

216 

217 async with self._session.get( 

218 f"{ZALO_OA_API_URL}/getoa", 

219 headers=headers, 

220 ) as resp: 

221 if resp.status == 200: 

222 data = await resp.json() 

223 if data.get("error") == 0: 

224 return data.get("data", {}) 

225 else: 

226 logger.error(f"Zalo API error: {data.get('message')}") 

227 

228 except Exception as e: 

229 logger.error(f"Failed to get OA info: {e}") 

230 

231 return None 

232 

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

234 """Validate webhook signature.""" 

235 if not self.zalo_config.webhook_secret: 

236 return True # No secret configured, skip validation 

237 

238 try: 

239 # Zalo uses: sha256(app_id + body + timestamp + secret) 

240 data = f"{self.zalo_config.app_id}{body}{timestamp}{self.zalo_config.webhook_secret}" 

241 expected = hashlib.sha256(data.encode()).hexdigest() 

242 return hmac.compare_digest(signature, expected) 

243 except Exception: 

244 return False 

245 

246 async def handle_webhook(self, body: Dict[str, Any]) -> None: 

247 """ 

248 Handle incoming webhook request from Zalo. 

249 Should be called from your webhook endpoint. 

250 """ 

251 event_name = body.get("event_name") 

252 sender = body.get("sender", {}) 

253 message = body.get("message", {}) 

254 timestamp = body.get("timestamp") 

255 

256 if not event_name: 

257 logger.warning("No event_name in Zalo webhook") 

258 return 

259 

260 try: 

261 event_type = ZaloEventType(event_name) 

262 except ValueError: 

263 logger.warning(f"Unknown Zalo event: {event_name}") 

264 return 

265 

266 # Handle different event types 

267 if event_type == ZaloEventType.USER_SEND_TEXT: 

268 await self._handle_text_message(sender, message, timestamp) 

269 elif event_type == ZaloEventType.USER_SEND_IMAGE: 

270 await self._handle_image_message(sender, message, timestamp) 

271 elif event_type == ZaloEventType.USER_SEND_FILE: 

272 await self._handle_file_message(sender, message, timestamp) 

273 elif event_type == ZaloEventType.USER_SEND_STICKER: 

274 await self._handle_sticker_message(sender, message, timestamp) 

275 elif event_type == ZaloEventType.USER_SEND_GIF: 

276 await self._handle_gif_message(sender, message, timestamp) 

277 elif event_type == ZaloEventType.USER_SEND_LOCATION: 

278 await self._handle_location_message(sender, message, timestamp) 

279 elif event_type == ZaloEventType.FOLLOW: 

280 await self._handle_follow(sender, timestamp) 

281 elif event_type == ZaloEventType.UNFOLLOW: 

282 await self._handle_unfollow(sender, timestamp) 

283 elif event_type == ZaloEventType.USER_CLICK_BUTTON: 

284 await self._handle_button_click(sender, message, timestamp) 

285 elif event_type == ZaloEventType.USER_SUBMIT_INFO: 

286 await self._handle_user_info_submit(sender, body.get("info", {}), timestamp) 

287 

288 async def _handle_text_message( 

289 self, 

290 sender: Dict[str, Any], 

291 message: Dict[str, Any], 

292 timestamp: int, 

293 ) -> None: 

294 """Handle text message.""" 

295 msg = self._convert_message(sender, message, timestamp) 

296 await self._dispatch_message(msg) 

297 

298 async def _handle_image_message( 

299 self, 

300 sender: Dict[str, Any], 

301 message: Dict[str, Any], 

302 timestamp: int, 

303 ) -> None: 

304 """Handle image message.""" 

305 msg = self._convert_message(sender, message, timestamp) 

306 msg.media.append(MediaAttachment( 

307 type=MessageType.IMAGE, 

308 url=message.get("url"), 

309 file_id=message.get("msg_id"), 

310 )) 

311 await self._dispatch_message(msg) 

312 

313 async def _handle_file_message( 

314 self, 

315 sender: Dict[str, Any], 

316 message: Dict[str, Any], 

317 timestamp: int, 

318 ) -> None: 

319 """Handle file message.""" 

320 msg = self._convert_message(sender, message, timestamp) 

321 msg.media.append(MediaAttachment( 

322 type=MessageType.DOCUMENT, 

323 url=message.get("url"), 

324 file_name=message.get("name"), 

325 file_size=message.get("size"), 

326 )) 

327 await self._dispatch_message(msg) 

328 

329 async def _handle_sticker_message( 

330 self, 

331 sender: Dict[str, Any], 

332 message: Dict[str, Any], 

333 timestamp: int, 

334 ) -> None: 

335 """Handle sticker message.""" 

336 msg = self._convert_message(sender, message, timestamp) 

337 msg.text = f"[sticker:{message.get('sticker_id')}]" 

338 await self._dispatch_message(msg) 

339 

340 async def _handle_gif_message( 

341 self, 

342 sender: Dict[str, Any], 

343 message: Dict[str, Any], 

344 timestamp: int, 

345 ) -> None: 

346 """Handle GIF message.""" 

347 msg = self._convert_message(sender, message, timestamp) 

348 msg.media.append(MediaAttachment( 

349 type=MessageType.IMAGE, 

350 url=message.get("url"), 

351 )) 

352 await self._dispatch_message(msg) 

353 

354 async def _handle_location_message( 

355 self, 

356 sender: Dict[str, Any], 

357 message: Dict[str, Any], 

358 timestamp: int, 

359 ) -> None: 

360 """Handle location message.""" 

361 msg = self._convert_message(sender, message, timestamp) 

362 lat = message.get("latitude") 

363 lon = message.get("longitude") 

364 msg.text = f"[location:{lat},{lon}]" 

365 await self._dispatch_message(msg) 

366 

367 async def _handle_follow(self, sender: Dict[str, Any], timestamp: int) -> None: 

368 """Handle follow event.""" 

369 user = ZaloUser( 

370 user_id=sender.get("id", ""), 

371 ) 

372 

373 logger.info(f"User followed: {user.user_id}") 

374 

375 for handler in self._follow_handlers: 

376 try: 

377 result = handler(user) 

378 if asyncio.iscoroutine(result): 

379 await result 

380 except Exception as e: 

381 logger.error(f"Follow handler error: {e}") 

382 

383 async def _handle_unfollow(self, sender: Dict[str, Any], timestamp: int) -> None: 

384 """Handle unfollow event.""" 

385 user_id = sender.get("id", "") 

386 logger.info(f"User unfollowed: {user_id}") 

387 

388 for handler in self._unfollow_handlers: 

389 try: 

390 result = handler(user_id) 

391 if asyncio.iscoroutine(result): 

392 await result 

393 except Exception as e: 

394 logger.error(f"Unfollow handler error: {e}") 

395 

396 async def _handle_button_click( 

397 self, 

398 sender: Dict[str, Any], 

399 message: Dict[str, Any], 

400 timestamp: int, 

401 ) -> None: 

402 """Handle quick reply button click.""" 

403 payload = message.get("payload", "") 

404 

405 if payload in self._button_handlers: 

406 handler = self._button_handlers[payload] 

407 try: 

408 result = handler(sender, message) 

409 if asyncio.iscoroutine(result): 

410 await result 

411 except Exception as e: 

412 logger.error(f"Button handler error: {e}") 

413 else: 

414 # Convert to regular message 

415 msg = self._convert_message(sender, message, timestamp) 

416 msg.text = payload 

417 msg.raw["button_payload"] = payload 

418 await self._dispatch_message(msg) 

419 

420 async def _handle_user_info_submit( 

421 self, 

422 sender: Dict[str, Any], 

423 info: Dict[str, Any], 

424 timestamp: int, 

425 ) -> None: 

426 """Handle user info submission.""" 

427 user = ZaloUser( 

428 user_id=sender.get("id", ""), 

429 display_name=info.get("name"), 

430 phone=info.get("phone"), 

431 avatar=info.get("avatar"), 

432 shared_info=info, 

433 ) 

434 

435 self._user_cache[user.user_id] = user 

436 

437 for handler in self._user_info_handlers: 

438 try: 

439 result = handler(user) 

440 if asyncio.iscoroutine(result): 

441 await result 

442 except Exception as e: 

443 logger.error(f"User info handler error: {e}") 

444 

445 def _convert_message( 

446 self, 

447 sender: Dict[str, Any], 

448 message: Dict[str, Any], 

449 timestamp: int, 

450 ) -> Message: 

451 """Convert Zalo event to unified Message format.""" 

452 user_id = sender.get("id", "") 

453 

454 return Message( 

455 id=message.get("msg_id", str(timestamp)), 

456 channel=self.name, 

457 sender_id=user_id, 

458 sender_name=sender.get("name"), 

459 chat_id=user_id, # 1:1 chat with user 

460 text=message.get("text", ""), 

461 timestamp=datetime.fromtimestamp(timestamp / 1000) if timestamp else datetime.now(), 

462 is_group=False, 

463 raw={ 

464 "sender": sender, 

465 "message": message, 

466 }, 

467 ) 

468 

469 async def send_message( 

470 self, 

471 chat_id: str, 

472 text: str, 

473 reply_to: Optional[str] = None, 

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

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

476 ) -> SendResult: 

477 """Send a message to a Zalo user.""" 

478 if not self._session: 

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

480 

481 try: 

482 # Handle quick reply buttons 

483 if buttons: 

484 return await self._send_with_buttons(chat_id, text, buttons) 

485 

486 # Handle media 

487 if media and len(media) > 0: 

488 for m in media: 

489 if m.type == MessageType.IMAGE: 

490 await self._send_image(chat_id, m.url, text) 

491 elif m.type == MessageType.DOCUMENT: 

492 await self._send_file(chat_id, m.url, m.file_name) 

493 return SendResult(success=True) 

494 

495 # Send text message 

496 return await self._send_text(chat_id, text) 

497 

498 except Exception as e: 

499 logger.error(f"Failed to send Zalo message: {e}") 

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

501 

502 async def _send_text(self, user_id: str, text: str) -> SendResult: 

503 """Send text message.""" 

504 if not self._session: 

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

506 

507 try: 

508 headers = { 

509 "access_token": self.zalo_config.access_token, 

510 "Content-Type": "application/json", 

511 } 

512 

513 data = { 

514 "recipient": {"user_id": user_id}, 

515 "message": {"text": text}, 

516 } 

517 

518 async with self._session.post( 

519 f"{ZALO_OA_API_URL}/message", 

520 headers=headers, 

521 json=data, 

522 ) as resp: 

523 result = await resp.json() 

524 if result.get("error") == 0: 

525 return SendResult( 

526 success=True, 

527 message_id=result.get("data", {}).get("message_id"), 

528 ) 

529 else: 

530 error_msg = result.get("message", "Unknown error") 

531 if result.get("error") == -201: 

532 raise ChannelRateLimitError() 

533 return SendResult(success=False, error=error_msg) 

534 

535 except ChannelRateLimitError: 

536 raise 

537 except Exception as e: 

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

539 

540 async def _send_with_buttons( 

541 self, 

542 user_id: str, 

543 text: str, 

544 buttons: List[Dict], 

545 ) -> SendResult: 

546 """Send message with quick reply buttons.""" 

547 if not self._session: 

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

549 

550 try: 

551 headers = { 

552 "access_token": self.zalo_config.access_token, 

553 "Content-Type": "application/json", 

554 } 

555 

556 # Build quick replies 

557 quick_replies = [] 

558 for btn in buttons: 

559 qr = { 

560 "content_type": "text", 

561 "title": btn.get("text", ""), 

562 "payload": btn.get("callback_data", btn.get("text", "")), 

563 } 

564 if btn.get("image_url"): 

565 qr["image_url"] = btn["image_url"] 

566 quick_replies.append(qr) 

567 

568 data = { 

569 "recipient": {"user_id": user_id}, 

570 "message": { 

571 "text": text, 

572 "quick_replies": quick_replies, 

573 }, 

574 } 

575 

576 async with self._session.post( 

577 f"{ZALO_OA_API_URL}/message", 

578 headers=headers, 

579 json=data, 

580 ) as resp: 

581 result = await resp.json() 

582 if result.get("error") == 0: 

583 return SendResult( 

584 success=True, 

585 message_id=result.get("data", {}).get("message_id"), 

586 ) 

587 else: 

588 return SendResult(success=False, error=result.get("message")) 

589 

590 except Exception as e: 

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

592 

593 async def _send_image( 

594 self, 

595 user_id: str, 

596 image_url: str, 

597 caption: str = "", 

598 ) -> SendResult: 

599 """Send image message.""" 

600 if not self._session: 

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

602 

603 try: 

604 headers = { 

605 "access_token": self.zalo_config.access_token, 

606 "Content-Type": "application/json", 

607 } 

608 

609 data = { 

610 "recipient": {"user_id": user_id}, 

611 "message": { 

612 "attachment": { 

613 "type": "template", 

614 "payload": { 

615 "template_type": "media", 

616 "elements": [{ 

617 "media_type": "image", 

618 "url": image_url, 

619 }], 

620 }, 

621 }, 

622 }, 

623 } 

624 

625 async with self._session.post( 

626 f"{ZALO_OA_API_URL}/message", 

627 headers=headers, 

628 json=data, 

629 ) as resp: 

630 result = await resp.json() 

631 if result.get("error") == 0: 

632 # Send caption as separate message if provided 

633 if caption: 

634 await self._send_text(user_id, caption) 

635 return SendResult( 

636 success=True, 

637 message_id=result.get("data", {}).get("message_id"), 

638 ) 

639 else: 

640 return SendResult(success=False, error=result.get("message")) 

641 

642 except Exception as e: 

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

644 

645 async def _send_file( 

646 self, 

647 user_id: str, 

648 file_url: str, 

649 file_name: str = "", 

650 ) -> SendResult: 

651 """Send file message.""" 

652 if not self._session: 

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

654 

655 try: 

656 headers = { 

657 "access_token": self.zalo_config.access_token, 

658 "Content-Type": "application/json", 

659 } 

660 

661 data = { 

662 "recipient": {"user_id": user_id}, 

663 "message": { 

664 "attachment": { 

665 "type": "file", 

666 "payload": { 

667 "url": file_url, 

668 }, 

669 }, 

670 }, 

671 } 

672 

673 async with self._session.post( 

674 f"{ZALO_OA_API_URL}/message", 

675 headers=headers, 

676 json=data, 

677 ) as resp: 

678 result = await resp.json() 

679 if result.get("error") == 0: 

680 return SendResult( 

681 success=True, 

682 message_id=result.get("data", {}).get("message_id"), 

683 ) 

684 else: 

685 return SendResult(success=False, error=result.get("message")) 

686 

687 except Exception as e: 

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

689 

690 async def edit_message( 

691 self, 

692 chat_id: str, 

693 message_id: str, 

694 text: str, 

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

696 ) -> SendResult: 

697 """ 

698 Edit a Zalo message. 

699 Note: Zalo doesn't support message editing. 

700 """ 

701 logger.warning("Zalo doesn't support message editing") 

702 return SendResult(success=False, error="Not supported") 

703 

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

705 """ 

706 Delete a Zalo message. 

707 Note: Zalo doesn't support message deletion. 

708 """ 

709 logger.warning("Zalo doesn't support message deletion") 

710 return False 

711 

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

713 """ 

714 Send typing indicator. 

715 Note: Zalo doesn't support typing indicators. 

716 """ 

717 pass 

718 

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

720 """Get user profile information.""" 

721 return await self.get_user_profile(chat_id) 

722 

723 # Zalo-specific methods 

724 

725 async def get_user_profile(self, user_id: str) -> Optional[Dict[str, Any]]: 

726 """Get user profile from Zalo.""" 

727 if not self._session: 

728 return None 

729 

730 # Check cache 

731 if user_id in self._user_cache: 

732 user = self._user_cache[user_id] 

733 return { 

734 "user_id": user.user_id, 

735 "display_name": user.display_name, 

736 "avatar": user.avatar, 

737 "phone": user.phone, 

738 } 

739 

740 try: 

741 headers = { 

742 "access_token": self.zalo_config.access_token, 

743 } 

744 

745 async with self._session.get( 

746 f"{ZALO_OA_API_URL}/getprofile", 

747 headers=headers, 

748 params={"user_id": user_id}, 

749 ) as resp: 

750 result = await resp.json() 

751 if result.get("error") == 0: 

752 data = result.get("data", {}) 

753 return { 

754 "user_id": user_id, 

755 "display_name": data.get("display_name"), 

756 "avatar": data.get("avatar"), 

757 "user_id_by_app": data.get("user_id_by_app"), 

758 } 

759 

760 except Exception as e: 

761 logger.error(f"Failed to get user profile: {e}") 

762 

763 return None 

764 

765 def on_follow(self, handler: Callable[[ZaloUser], Any]) -> None: 

766 """Register a follow event handler.""" 

767 self._follow_handlers.append(handler) 

768 

769 def on_unfollow(self, handler: Callable[[str], Any]) -> None: 

770 """Register an unfollow event handler.""" 

771 self._unfollow_handlers.append(handler) 

772 

773 def register_button_handler( 

774 self, 

775 payload: str, 

776 handler: Callable[[Dict, Dict], Any], 

777 ) -> None: 

778 """Register a button click handler.""" 

779 self._button_handlers[payload] = handler 

780 

781 def on_user_info(self, handler: Callable[[ZaloUser], Any]) -> None: 

782 """Register a user info submission handler.""" 

783 self._user_info_handlers.append(handler) 

784 

785 async def request_user_info( 

786 self, 

787 user_id: str, 

788 title: str = "Share your info", 

789 subtitle: str = "", 

790 ) -> SendResult: 

791 """Request user info (name, phone, etc.).""" 

792 if not self._session: 

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

794 

795 if not self.zalo_config.enable_user_info_request: 

796 return SendResult(success=False, error="User info request disabled") 

797 

798 try: 

799 headers = { 

800 "access_token": self.zalo_config.access_token, 

801 "Content-Type": "application/json", 

802 } 

803 

804 data = { 

805 "recipient": {"user_id": user_id}, 

806 "message": { 

807 "attachment": { 

808 "type": "template", 

809 "payload": { 

810 "template_type": "request_user_info", 

811 "elements": [{ 

812 "title": title, 

813 "subtitle": subtitle, 

814 "image_url": "", 

815 }], 

816 }, 

817 }, 

818 }, 

819 } 

820 

821 async with self._session.post( 

822 f"{ZALO_OA_API_URL}/message", 

823 headers=headers, 

824 json=data, 

825 ) as resp: 

826 result = await resp.json() 

827 if result.get("error") == 0: 

828 return SendResult(success=True) 

829 else: 

830 return SendResult(success=False, error=result.get("message")) 

831 

832 except Exception as e: 

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