Coverage for modules\email\imap_service.py: 11%

1""" 

2IMAP Service - Fetch emails like Outlook/Thunderbird 

3""" 

4import imaplib 

5import email 

6from email.header import decode_header 

7from datetime import datetime 

8from typing import Optional, List, Tuple 

9import base64 

10 

11 

12class IMAPService: 

13 """Service for connecting to IMAP email servers.""" 

14 

15 def __init__(self, host: str, port: int, username: str, password: str, ssl: bool = True): 

16 self.host = host 

17 self.port = port 

18 self.username = username 

19 self.password = password 

20 self.ssl = ssl 

21 self.connection: Optional[imaplib.IMAP4_SSL] = None 

22 

23 def connect(self) -> bool: 

24 """Establish IMAP connection.""" 

25 try: 

26 if self.ssl: 

27 self.connection = imaplib.IMAP4_SSL(self.host, self.port) 

28 else: 

29 self.connection = imaplib.IMAP4(self.host, self.port) 

30 

31 result = self.connection.login(self.username, self.password) 

32 return result[0] == 'OK' 

33 except Exception as e: 

34 print(f"IMAP connection failed: {e}") 

35 return False 

36 

37 def disconnect(self): 

38 """Close IMAP connection.""" 

39 if self.connection: 

40 try: 

41 self.connection.logout() 

42 except: 

43 pass 

44 self.connection = None 

45 

46 def list_folders(self) -> List[dict]: 

47 """List all available folders.""" 

48 if not self.connection: 

49 return [] 

50 

51 folders = [] 

52 try: 

53 status, folder_list = self.connection.list() 

54 if status == 'OK': 

55 for folder in folder_list: 

56 # Parse folder info 

57 parts = folder.decode('utf-8', errors='replace').split('"') 

58 if len(parts) >= 2: 

59 folder_name = parts[-2] if parts[-1].strip() == '' else parts[1] 

60 flags = parts[0].strip('()').split() 

61 folders.append({ 

62 'name': folder_name, 

63 'flags': flags, 

64 }) 

65 except Exception as e: 

66 print(f"Error listing folders: {e}") 

67 

68 return folders 

69 

70 def select_folder(self, folder: str = 'INBOX') -> Tuple[bool, int]: 

71 """Select a folder and return (success, message_count).""" 

72 if not self.connection: 

73 return False, 0 

74 

75 try: 

76 status, data = self.connection.select(f'"{folder}"') 

77 if status == 'OK': 

78 # Get message count from data like ['7234'] 

79 msg_count = int(data[0]) if data and data[0] else 0 

80 return True, msg_count 

81 return False, 0 

82 except Exception as e: 

83 print(f"Error selecting folder {folder}: {e}") 

84 return False, 0 

85 

86 def fetch_messages(self, folder: str = 'INBOX', page: int = 1, per_page: int = 50) -> List[dict]: 

87 """Fetch message headers from a folder.""" 

88 if not self.connection: 

89 return [] 

90 

91 messages = [] 

92 try: 

93 # Select folder 

94 success, total = self.select_folder(folder) 

95 if not success: 

96 return [] 

97 

98 # Calculate range 

99 start = max(1, total - (page * per_page) + 1) 

100 end = total - ((page - 1) * per_page) 

101 

102 if start > end: 

103 return [] 

104 

105 # Fetch message headers 

106 status, data = self.connection.fetch(f'{start}:{end}', '(RFC822.HEADER)') 

107 if status != 'OK': 

108 return [] 

109 

110 for response_part in data: 

111 if isinstance(response_part, tuple): 

112 msg_num = response_part[0].decode('utf-8', errors='replace').split()[0] 

113 raw_headers = response_part[1] 

114 

115 msg = email.message_from_bytes(raw_headers) 

116 headers = self._parse_headers(msg) 

117 headers['message_num'] = msg_num 

118 messages.append(headers) 

119 

120 except Exception as e: 

121 print(f"Error fetching messages: {e}") 

122 

123 return messages 

124 

125 def fetch_message(self, message_num: str) -> Optional[dict]: 

126 """Fetch full message by number.""" 

127 if not self.connection: 

128 return None 

129 

130 try: 

131 # Fetch full message 

132 status, data = self.connection.fetch(message_num, '(RFC822)') 

133 if status != 'OK': 

134 return None 

135 

136 if not data or not isinstance(data[0], tuple): 

137 return None 

138 

139 raw_msg = data[0][1] 

140 msg = email.message_from_bytes(raw_msg) 

141 

142 return self._parse_full_message(msg) 

143 

144 except Exception as e: 

145 print(f"Error fetching message {message_num}: {e}") 

146 return None 

147 

148 def mark_as_read(self, message_num: str) -> bool: 

149 """Mark message as read.""" 

150 if not self.connection: 

151 return False 

152 

153 try: 

154 status, _ = self.connection.store(message_num, '+FLAGS', '\\Seen') 

155 return status == 'OK' 

156 except Exception as e: 

157 print(f"Error marking as read: {e}") 

158 return False 

159 

160 def mark_as_unread(self, message_num: str) -> bool: 

161 """Mark message as unread.""" 

162 if not self.connection: 

163 return False 

164 

165 try: 

166 status, _ = self.connection.store(message_num, '-FLAGS', '\\Seen') 

167 return status == 'OK' 

168 except Exception as e: 

169 print(f"Error marking as unread: {e}") 

170 return False 

171 

172 def delete_message(self, message_num: str) -> bool: 

173 """Move message to trash.""" 

174 if not self.connection: 

175 return False 

176 

177 try: 

178 status, _ = self.connection.store(message_num, '+FLAGS', '\\Deleted') 

179 return status == 'OK' 

180 except Exception as e: 

181 print(f"Error deleting message: {e}") 

182 return False 

183 

184 def move_message(self, message_num: str, target_folder: str) -> bool: 

185 """Move message to another folder.""" 

186 if not self.connection: 

187 return False 

188 

189 try: 

190 # Copy to target, then delete from current 

191 status, _ = self.connection.copy(message_num, f'"{target_folder}"') 

192 if status == 'OK': 

193 status, _ = self.connection.store(message_num, '+FLAGS', '\\Deleted') 

194 return status == 'OK' 

195 return False 

196 except Exception as e: 

197 print(f"Error moving message: {e}") 

198 return False 

199 

200 def get_unread_count(self, folder: str = 'INBOX') -> int: 

201 """Get unread message count for a folder.""" 

202 if not self.connection: 

203 return 0 

204 

205 try: 

206 # Search for unread messages 

207 status, data = self.connection.select(f'"{folder}"') 

208 if status != 'OK': 

209 return 0 

210 

211 status, data = self.connection.search(None, 'UNSEEN') 

212 if status == 'OK': 

213 uids = data[0].split() 

214 return len(uids) if uids else 0 

215 return 0 

216 except Exception as e: 

217 print(f"Error getting unread count: {e}") 

218 return 0 

219 

220 def _parse_headers(self, msg) -> dict: 

221 """Parse email headers to extract metadata.""" 

222 subject = self._decode_header_value(msg.get('Subject', '')) 

223 sender = self._decode_header_value(msg.get('From', '')) 

224 date_str = msg.get('Date', '') 

225 

226 # Parse sender 

227 sender_name, sender_email = self._parse_address(sender) 

228 

229 # Parse date 

230 sent_at = None 

231 try: 

232 from email.utils import parsedate_to_datetime 

233 sent_at = parsedate_to_datetime(date_str) 

234 except: 

235 pass 

236 

237 # Get snippet from message body if available 

238 snippet = self._get_body_preview(msg) 

239 

240 return { 

241 'message_id': msg.get('Message-ID', ''), 

242 'subject': subject, 

243 'from_address': sender_email, 

244 'from_name': sender_name, 

245 'sent_at': sent_at, 

246 'snippet': snippet, 

247 } 

248 

249 def _parse_full_message(self, msg) -> dict: 

250 """Parse full email message.""" 

251 headers = self._parse_headers(msg) 

252 

253 # Get body 

254 body_text, body_html = self._get_body(msg) 

255 

256 # Get all recipients 

257 to_addresses = self._parse_address_list(msg.get('To', '')) 

258 cc_addresses = self._parse_address_list(msg.get('Cc', '')) 

259 

260 headers['body_text'] = body_text 

261 headers['body_html'] = body_html 

262 headers['to_addresses'] = to_addresses 

263 headers['cc_addresses'] = cc_addresses 

264 headers['size_bytes'] = len(msg.as_bytes()) 

265 

266 return headers 

267 

268 def _decode_header_value(self, value: str) -> str: 

269 """Decode email header value.""" 

270 if not value: 

271 return '' 

272 

273 try: 

274 parts = decode_header(value) 

275 result = [] 

276 for part, encoding in parts: 

277 if isinstance(part, bytes): 

278 encoding = encoding or 'utf-8' 

279 try: 

280 result.append(part.decode(encoding, errors='replace')) 

281 except: 

282 result.append(part.decode('utf-8', errors='replace')) 

283 else: 

284 result.append(part) 

285 return ' '.join(result) 

286 except: 

287 return value 

288 

289 def _parse_address(self, address: str) -> tuple: 

290 """Parse email address to (name, email).""" 

291 if not address: 

292 return None, None 

293 

294 try: 

295 from email.utils import parseaddr 

296 name, email = parseaddr(address) 

297 if not name: 

298 name = None 

299 if not email: 

300 email = address 

301 return name, email 

302 except: 

303 return None, address 

304 

305 def _parse_address_list(self, address_list: str) -> List[dict]: 

306 """Parse comma-separated address list.""" 

307 if not address_list: 

308 return [] 

309 

310 addresses = [] 

311 for addr in address_list.split(','): 

312 name, email = self._parse_address(addr.strip()) 

313 if email: 

314 addresses.append({'name': name, 'email': email}) 

315 return addresses 

316 

317 def _get_body_preview(self, msg, max_length: int = 200) -> str: 

318 """Get preview of message body.""" 

319 body_text, _ = self._get_body(msg) 

320 if body_text: 

321 return body_text[:max_length].replace('\n', ' ').strip() 

322 return '' 

323 

324 def _get_body(self, msg) -> tuple: 

325 """Extract text and HTML body from message.""" 

326 text_body = '' 

327 html_body = '' 

328 

329 if msg.is_multipart(): 

330 for part in msg.walk(): 

331 content_type = part.get_content_type() 

332 payload = part.get_payload(decode=True) 

333 

334 if payload: 

335 charset = part.get_content_charset() or 'utf-8' 

336 try: 

337 decoded = payload.decode(charset, errors='replace') 

338 except: 

339 decoded = payload.decode('utf-8', errors='replace') 

340 

341 if content_type == 'text/plain': 

342 text_body = decoded 

343 elif content_type == 'text/html': 

344 html_body = decoded 

345 else: 

346 payload = msg.get_payload(decode=True) 

347 if payload: 

348 charset = msg.get_content_charset() or 'utf-8' 

349 try: 

350 decoded = payload.decode(charset, errors='replace') 

351 except: 

352 decoded = payload.decode('utf-8', errors='replace') 

353 

354 if msg.get_content_type() == 'text/plain': 

355 text_body = decoded 

356 elif msg.get_content_type() == 'text/html': 

357 html_body = decoded 

358 

359 return text_body.strip(), html_body.strip() 

360 

361 def __enter__(self): 

362 self.connect() 

363 return self 

364 

365 def __exit__(self, exc_type, exc_val, exc_tb): 

366 self.disconnect()