Coverage for agentos/channels/adapters/line.py: 0%

1""" 

2LINE Channel Adapter — LINE Messaging API. 

3 

4LINE Developers Console → Channel Access Token + Channel Secret → webhook → ChannelMessage. 

5""" 

6 

7from __future__ import annotations 

8 

9import base64 

10import hashlib 

11import hmac 

12import json 

13 

14from agentos.channels.base import BaseChannelAdapter, ChannelConfig, ReplyResult 

15from agentos.channels.message import ChannelMessage, ChannelType, MessageType 

16 

17 

18class LINEAdapter(BaseChannelAdapter): 

19 """LINE Messaging API adapter. 

20 

21 Config fields: 

22 channel_access_token: LINE channel access token (long-lived) 

23 channel_secret: LINE channel secret (for signature verification) 

24 reply_retry_limit: max reply attempts (default 1) 

25 """ 

26 

27 CHANNEL_TYPE = ChannelType.LINE 

28 API_BASE = "https://api.line.me/v2" 

29 API_DATA = "https://api-data.line.me/v2" 

30 

31 def __init__(self, config: ChannelConfig): 

32 super().__init__(config) 

33 self._access_token = config.extra.get("channel_access_token", "") 

34 self._channel_secret = config.extra.get("channel_secret", "") 

35 self._retry_limit = config.extra.get("reply_retry_limit", 1) 

36 

37 @property 

38 def _headers(self) -> dict: 

39 return {"Authorization": f"Bearer {self._access_token}"} 

40 

41 # ── Signature verification ── 

42 

43 def verify_signature(self, body: bytes, signature: str) -> bool: 

44 """Verify LINE webhook signature (HMAC-SHA256 base64).""" 

45 computed = base64.b64encode( 

46 hmac.new( 

47 self._channel_secret.encode(), 

48 body, 

49 hashlib.sha256, 

50 ).digest() 

51 ).decode() 

52 return hmac.compare_digest(computed, signature) 

53 

54 # ── Message parsing ── 

55 

56 async def parse_incoming(self, payload: dict) -> ChannelMessage | None: 

57 """Parse LINE webhook events into ChannelMessage.""" 

58 events = payload.get("events", []) 

59 if not events: 

60 return None 

61 

62 event = events[0] 

63 event_type = event.get("type", "") 

64 

65 if event_type == "message": 

66 return self._parse_message(event) 

67 elif event_type == "postback": 

68 return self._parse_postback(event) 

69 elif event_type == "follow": 

70 return self._parse_follow(event) 

71 elif event_type == "unfollow": 

72 return ChannelMessage( 

73 channel_type=ChannelType.LINE, 

74 channel_id=event.get("source", {}).get("userId", ""), 

75 user_id=event.get("source", {}).get("userId", ""), 

76 content="unfollow", 

77 message_type=MessageType.SYSTEM, 

78 raw=event, 

79 ) 

80 

81 return None 

82 

83 def _parse_message(self, event: dict) -> ChannelMessage | None: 

84 """Parse a LINE message event.""" 

85 source = event.get("source", {}) 

86 user_id = source.get("userId", "") 

87 group_id = source.get("groupId", "") 

88 room_id = source.get("roomId", "") 

89 channel_id = group_id or room_id or user_id 

90 

91 msg = event.get("message", {}) 

92 msg_type = msg.get("type", "text") 

93 

94 content = "" 

95 mtype = MessageType.TEXT 

96 metadata = { 

97 "source_type": source.get("type", "user"), 

98 "group_id": group_id, 

99 "room_id": room_id, 

100 "display_name": "", # Filled via profile API if needed 

101 } 

102 

103 if msg_type == "text": 

104 content = msg.get("text", "") 

105 elif msg_type == "image": 

106 content = "[Image]" 

107 mtype = MessageType.IMAGE 

108 metadata["message_id"] = msg.get("id", "") 

109 elif msg_type == "video": 

110 content = "[Video]" 

111 mtype = MessageType.VIDEO 

112 elif msg_type == "audio": 

113 content = "[Voice message]" 

114 mtype = MessageType.VOICE 

115 elif msg_type == "file": 

116 content = f"[File: {msg.get('fileName', 'unknown')}]" 

117 mtype = MessageType.FILE 

118 metadata["file_name"] = msg.get("fileName", "") 

119 metadata["file_size"] = msg.get("fileSize", 0) 

120 elif msg_type == "location": 

121 content = f"[Location: {msg.get('title', '')} {msg.get('address', '')}]" 

122 mtype = MessageType.LOCATION 

123 elif msg_type == "sticker": 

124 content = f"[Sticker: {msg.get('packageId')}/{msg.get('stickerId')}]" 

125 else: 

126 content = f"[{msg_type}]" 

127 

128 return ChannelMessage( 

129 channel_type=ChannelType.LINE, 

130 channel_id=channel_id, 

131 user_id=user_id, 

132 content=content, 

133 message_type=mtype, 

134 raw=event, 

135 reply_token=event.get("replyToken", ""), 

136 metadata=metadata, 

137 ) 

138 

139 def _parse_postback(self, event: dict) -> ChannelMessage | None: 

140 """Parse LINE postback event (rich menu, button tap).""" 

141 source = event.get("source", {}) 

142 data = event.get("postback", {}).get("data", "") 

143 params = event.get("postback", {}).get("params", {}) 

144 

145 return ChannelMessage( 

146 channel_type=ChannelType.LINE, 

147 channel_id=source.get("userId", ""), 

148 user_id=source.get("userId", ""), 

149 content=data, 

150 message_type=MessageType.INTERACTIVE, 

151 raw=event, 

152 reply_token=event.get("replyToken", ""), 

153 metadata={"postback_params": params}, 

154 ) 

155 

156 def _parse_follow(self, event: dict) -> ChannelMessage: 

157 """Parse LINE follow event.""" 

158 source = event.get("source", {}) 

159 return ChannelMessage( 

160 channel_type=ChannelType.LINE, 

161 channel_id=source.get("userId", ""), 

162 user_id=source.get("userId", ""), 

163 content="follow", 

164 message_type=MessageType.SYSTEM, 

165 raw=event, 

166 ) 

167 

168 # ── Reply ── 

169 

170 async def reply(self, channel_id: str, content: str, **kwargs) -> ReplyResult: 

171 """Send a reply text message.""" 

172 reply_token = kwargs.get("reply_token", "") 

173 if not reply_token: 

174 return ReplyResult(success=False, error="reply_token required") 

175 

176 return await self._api_reply( 

177 reply_token, 

178 [ 

179 {"type": "text", "text": content[:5000]}, 

180 ], 

181 ) 

182 

183 async def reply_flex( 

184 self, 

185 channel_id: str, 

186 alt_text: str, 

187 contents: dict, 

188 **kwargs, 

189 ) -> ReplyResult: 

190 """Send a LINE Flex Message (bubble/carousel).""" 

191 reply_token = kwargs.get("reply_token", "") 

192 if not reply_token: 

193 return ReplyResult(success=False, error="reply_token required") 

194 

195 return await self._api_reply( 

196 reply_token, 

197 [ 

198 {"type": "flex", "altText": alt_text, "contents": contents}, 

199 ], 

200 ) 

201 

202 async def reply_quick_reply( 

203 self, 

204 channel_id: str, 

205 text: str, 

206 items: list[dict], 

207 **kwargs, 

208 ) -> ReplyResult: 

209 """Send text with quick reply buttons. 

210 

211 items = [{"type": "action", "action": {"type": "message", "label": "Yes", "text": "Yes"}}, ...] 

212 """ 

213 reply_token = kwargs.get("reply_token", "") 

214 if not reply_token: 

215 return ReplyResult(success=False, error="reply_token required") 

216 

217 return await self._api_reply( 

218 reply_token, 

219 [ 

220 { 

221 "type": "text", 

222 "text": text[:5000], 

223 "quickReply": {"items": items[:13]}, 

224 }, 

225 ], 

226 ) 

227 

228 async def push_message( 

229 self, 

230 user_id: str, 

231 messages: list[dict], 

232 ) -> ReplyResult: 

233 """Push a message to a user (outside reply window).""" 

234 return await self._api_push(user_id, messages) 

235 

236 async def multicast( 

237 self, 

238 user_ids: list[str], 

239 messages: list[dict], 

240 ) -> ReplyResult: 

241 """Send the same message to up to 500 users.""" 

242 return await self._api_call( 

243 f"{self.API_BASE}/bot/message/multicast", 

244 {"to": user_ids[:500], "messages": messages}, 

245 method="POST", 

246 ) 

247 

248 # ── Profile ── 

249 

250 async def get_profile(self, user_id: str) -> dict | None: 

251 """Get LINE user profile.""" 

252 result = await self._api_call( 

253 f"{self.API_BASE}/bot/profile/{user_id}", 

254 method="GET", 

255 ) 

256 if result.success: 

257 return result.raw 

258 return None 

259 

260 # ── Rich Menu ── 

261 

262 async def set_default_rich_menu(self, rich_menu_id: str) -> bool: 

263 """Set the default rich menu for all users.""" 

264 result = await self._api_call( 

265 f"{self.API_BASE}/bot/user/all/richmenu/{rich_menu_id}", 

266 method="POST", 

267 ) 

268 return result.success 

269 

270 # ── Internal API ── 

271 

272 async def _api_reply(self, reply_token: str, messages: list) -> ReplyResult: 

273 """Send a reply via reply API.""" 

274 return await self._api_call( 

275 f"{self.API_BASE}/bot/message/reply", 

276 {"replyToken": reply_token, "messages": messages}, 

277 method="POST", 

278 ) 

279 

280 async def _api_push(self, user_id: str, messages: list) -> ReplyResult: 

281 """Send a push message.""" 

282 return await self._api_call( 

283 f"{self.API_BASE}/bot/message/push", 

284 {"to": user_id, "messages": messages}, 

285 method="POST", 

286 ) 

287 

288 async def _api_call( 

289 self, 

290 url: str, 

291 body: dict = None, 

292 method: str = "POST", 

293 ) -> ReplyResult: 

294 """Generic LINE API call.""" 

295 headers = { 

296 "Authorization": f"Bearer {self._access_token}", 

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

298 } 

299 

300 try: 

301 import aiohttp 

302 

303 async with aiohttp.ClientSession() as session: 

304 if method == "GET": 

305 async with session.get(url, headers=headers) as resp: 

306 data = await resp.json() 

307 return ReplyResult(success=True, raw=data) 

308 else: 

309 async with session.post(url, headers=headers, json=body) as resp: 

310 data = await resp.json() 

311 if resp.status == 200: 

312 return ReplyResult(success=True, message_id="ok", raw=data) 

313 return ReplyResult( 

314 success=False, 

315 error=data.get("message", "unknown"), 

316 ) 

317 except ImportError: 

318 import urllib.request 

319 

320 req = urllib.request.Request( 

321 url, 

322 data=json.dumps(body).encode() if body else None, 

323 headers=headers, 

324 ) 

325 with urllib.request.urlopen(req) as resp: 

326 data = json.loads(resp.read()) 

327 return ReplyResult(success=True, message_id="ok", raw=data)