| 1 | """ |
| 2 | IMAP/Exchange email reader. |
| 3 | |
| 4 | No agent/tool dependencies. |
| 5 | """ |
| 6 | |
| 7 | import asyncio |
| 8 | import email |
| 9 | import os |
| 10 | import re |
| 11 | import uuid |
| 12 | from dataclasses import dataclass, field |
| 13 | from datetime import datetime, timedelta |
| 14 | from email.header import decode_header |
| 15 | from email.message import Message as EmailMessage |
| 16 | from fnmatch import fnmatch |
| 17 | |
| 18 | import html2text |
| 19 | from bs4 import BeautifulSoup |
| 20 | from imapclient import IMAPClient |
| 21 | |
| 22 | from helpers import files |
| 23 | from helpers.errors import format_error |
| 24 | from helpers.print_style import PrintStyle |
| 25 | |
| 26 | |
| 27 | # ------------------------------------------------------------------ |
| 28 | # Data models |
| 29 | # ------------------------------------------------------------------ |
| 30 | |
| 31 | @dataclass |
| 32 | class InboundMessage: |
| 33 | sender: str |
| 34 | subject: str |
| 35 | body: str |
| 36 | attachments: list[str] = field(default_factory=list) |
| 37 | message_id: str = "" |
| 38 | in_reply_to: str = "" |
| 39 | references: str = "" |
| 40 | |
| 41 | |
| 42 | # ------------------------------------------------------------------ |
| 43 | # IMAP connection |
| 44 | # ------------------------------------------------------------------ |
| 45 | |
| 46 | async def connect_imap( |
| 47 | server: str, |
| 48 | port: int = 993, |
| 49 | username: str = "", |
| 50 | password: str = "", |
| 51 | ssl: bool = True, |
| 52 | timeout: int = 30, |
| 53 | ) -> IMAPClient: |
| 54 | loop = asyncio.get_event_loop() |
| 55 | |
| 56 | def _sync(): |
| 57 | client = IMAPClient(server, port=port, ssl=ssl, timeout=timeout) |
| 58 | client._imap._maxline = 100000 # type: ignore[attr-defined] |
| 59 | client.login(username, password) |
| 60 | return client |
| 61 | |
| 62 | return await loop.run_in_executor(None, _sync) |
| 63 | |
| 64 | |
| 65 | async def disconnect_imap(client: IMAPClient) -> None: |
| 66 | try: |
| 67 | loop = asyncio.get_event_loop() |
| 68 | await loop.run_in_executor(None, client.logout) |
| 69 | except Exception as e: |
| 70 | PrintStyle.error(f"IMAP disconnect error: {format_error(e)}") |
| 71 | |
| 72 | |
| 73 | # ------------------------------------------------------------------ |
| 74 | # Fetch messages |
| 75 | # ------------------------------------------------------------------ |
| 76 | |
| 77 | async def fetch_new( |
| 78 | client: IMAPClient, |
| 79 | download_folder: str, |
| 80 | last_uid: int = 0, |
| 81 | sender_whitelist: list[str] | None = None, |
| 82 | max_messages: int = 10, |
| 83 | ) -> tuple[list[InboundMessage], int]: |
| 84 | """Fetch emails newer than last_uid. Returns (messages, new_last_uid).""" |
| 85 | loop = asyncio.get_event_loop() |
| 86 | |
| 87 | def _search(): |
| 88 | client.select_folder("INBOX") |
| 89 | # Use Gmail category filter if supported, otherwise plain UNSEEN |
| 90 | try: |
| 91 | return client.gmail_search("category:primary is:unread") |
| 92 | except Exception: |
| 93 | return client.search(["UNSEEN"]) # type: ignore[arg-type] |
| 94 | |
| 95 | msg_ids = await loop.run_in_executor(None, _search) |
| 96 | if not msg_ids: |
| 97 | return [], last_uid |
| 98 | |
| 99 | # Filter out already-seen UIDs |
| 100 | if last_uid > 0: |
| 101 | msg_ids = [uid for uid in msg_ids if uid > last_uid] |
| 102 | |
| 103 | if not msg_ids: |
| 104 | return [], last_uid |
| 105 | |
| 106 | new_last_uid = max(msg_ids) |
| 107 | |
| 108 | # Cap to most recent |
| 109 | if len(msg_ids) > max_messages: |
| 110 | PrintStyle.standard( |
| 111 | f"Email: {len(msg_ids)} new, processing latest {max_messages}" |
| 112 | ) |
| 113 | msg_ids = msg_ids[-max_messages:] |
| 114 | else: |
| 115 | PrintStyle.standard(f"Email: found {len(msg_ids)} new messages") |
| 116 | |
| 117 | results: list[InboundMessage] = [] |
| 118 | |
| 119 | for msg_id in msg_ids: |
| 120 | try: |
| 121 | msg = await _fetch_single(client, msg_id, download_folder, sender_whitelist) |
| 122 | if msg: |
| 123 | results.append(msg) |
| 124 | except Exception as e: |
| 125 | PrintStyle.error(f"Email: error processing message {msg_id}: {format_error(e)}") |
| 126 | return results, new_last_uid |
| 127 | |
| 128 | |
| 129 | async def get_highest_uid(client: IMAPClient) -> int: |
| 130 | """Get the highest UID in inbox without fetching any messages.""" |
| 131 | loop = asyncio.get_event_loop() |
| 132 | |
| 133 | def _search(): |
| 134 | client.select_folder("INBOX") |
| 135 | uids = client.search(["ALL"]) # type: ignore[arg-type] |
| 136 | return max(uids) if uids else 0 |
| 137 | |
| 138 | return await loop.run_in_executor(None, _search) |
| 139 | |
| 140 | |
| 141 | async def fetch_unread_since( |
| 142 | client: IMAPClient, |
| 143 | download_folder: str, |
| 144 | days: int, |
| 145 | sender_whitelist: list[str] | None = None, |
| 146 | max_messages: int = 10, |
| 147 | ) -> tuple[list[InboundMessage], int]: |
| 148 | """Fetch unread emails from the last N days. Returns (messages, highest_uid).""" |
| 149 | loop = asyncio.get_event_loop() |
| 150 | since_date = datetime.now() - timedelta(days=days) |
| 151 | |
| 152 | def _search(): |
| 153 | client.select_folder("INBOX") |
| 154 | try: |
| 155 | return client.gmail_search( |
| 156 | f"category:primary is:unread after:{since_date.strftime('%Y/%m/%d')}" |
| 157 | ) |
| 158 | except Exception: |
| 159 | return client.search(["UNSEEN", "SINCE", since_date.date()]) # type: ignore[arg-type] |
| 160 | |
| 161 | msg_ids = await loop.run_in_executor(None, _search) |
| 162 | if not msg_ids: |
| 163 | return [], 0 |
| 164 | |
| 165 | highest_uid = max(msg_ids) |
| 166 | |
| 167 | if len(msg_ids) > max_messages: |
| 168 | PrintStyle.standard( |
| 169 | f"Email: {len(msg_ids)} unread, processing latest {max_messages}" |
| 170 | ) |
| 171 | msg_ids = msg_ids[-max_messages:] |
| 172 | else: |
| 173 | PrintStyle.standard( |
| 174 | f"Email: found {len(msg_ids)} unread messages from last {days} days" |
| 175 | ) |
| 176 | |
| 177 | results: list[InboundMessage] = [] |
| 178 | for msg_id in msg_ids: |
| 179 | try: |
| 180 | msg = await _fetch_single(client, msg_id, download_folder, sender_whitelist) |
| 181 | if msg: |
| 182 | results.append(msg) |
| 183 | except Exception as e: |
| 184 | PrintStyle.error( |
| 185 | f"Email: error processing message {msg_id}: {format_error(e)}" |
| 186 | ) |
| 187 | |
| 188 | return results, highest_uid |
| 189 | |
| 190 | |
| 191 | async def _fetch_single( |
| 192 | client: IMAPClient, |
| 193 | msg_id: int, |
| 194 | download_folder: str, |
| 195 | sender_whitelist: list[str] | None, |
| 196 | ) -> InboundMessage | None: |
| 197 | loop = asyncio.get_event_loop() |
| 198 | |
| 199 | def _sync_fetch(): |
| 200 | data = client.fetch([msg_id], ["RFC822"])[msg_id] |
| 201 | # Explicitly mark as read — RFC822 fetch doesn't always set \Seen on all servers |
| 202 | client.add_flags([msg_id], [b"\\Seen"]) |
| 203 | return data |
| 204 | |
| 205 | raw = await loop.run_in_executor(None, _sync_fetch) |
| 206 | email_data = raw.get(b"RFC822") |
| 207 | if not email_data: |
| 208 | return None |
| 209 | |
| 210 | email_msg = email.message_from_bytes(email_data) # type: ignore[arg-type] |
| 211 | |
| 212 | sender = _decode_header(email_msg.get("From", "")) |
| 213 | if _is_noreply(sender): |
| 214 | return None |
| 215 | if sender_whitelist and not _matches_whitelist(sender, sender_whitelist): |
| 216 | return None |
| 217 | |
| 218 | subject = _decode_header(email_msg.get("Subject", "")) |
| 219 | message_id = email_msg.get("Message-ID", "") |
| 220 | in_reply_to = email_msg.get("In-Reply-To", "") |
| 221 | references = email_msg.get("References", "") |
| 222 | |
| 223 | body, attachments = await _parse_body(email_msg, download_folder) |
| 224 | |
| 225 | return InboundMessage( |
| 226 | sender=sender, |
| 227 | subject=subject, |
| 228 | body=body, |
| 229 | attachments=attachments, |
| 230 | message_id=message_id, |
| 231 | in_reply_to=in_reply_to, |
| 232 | references=references, |
| 233 | ) |
| 234 | |
| 235 | |
| 236 | # ------------------------------------------------------------------ |
| 237 | # Exchange connection |
| 238 | # ------------------------------------------------------------------ |
| 239 | |
| 240 | async def connect_exchange( |
| 241 | server: str, |
| 242 | username: str, |
| 243 | password: str, |
| 244 | ): |
| 245 | from exchangelib import Account, Configuration, Credentials, DELEGATE |
| 246 | |
| 247 | loop = asyncio.get_event_loop() |
| 248 | |
| 249 | def _sync(): |
| 250 | creds = Credentials(username=username, password=password) |
| 251 | config = Configuration(server=server, credentials=creds) |
| 252 | return Account( |
| 253 | primary_smtp_address=username, |
| 254 | config=config, |
| 255 | autodiscover=False, |
| 256 | access_type=DELEGATE, |
| 257 | ) |
| 258 | |
| 259 | return await loop.run_in_executor(None, _sync) |
| 260 | |
| 261 | |
| 262 | async def fetch_unread_exchange( |
| 263 | account, |
| 264 | download_folder: str, |
| 265 | sender_whitelist: list[str] | None = None, |
| 266 | since_days: int = 0, |
| 267 | ) -> list[InboundMessage]: |
| 268 | from exchangelib import Q |
| 269 | |
| 270 | loop = asyncio.get_event_loop() |
| 271 | |
| 272 | def _sync(): |
| 273 | q = Q(is_read=False) |
| 274 | if since_days > 0: |
| 275 | since = datetime.now(tz=account.default_timezone) - timedelta(days=since_days) |
| 276 | q &= Q(datetime_received__gte=since) |
| 277 | return list(account.inbox.filter(q)) |
| 278 | |
| 279 | items = await loop.run_in_executor(None, _sync) |
| 280 | results: list[InboundMessage] = [] |
| 281 | |
| 282 | for item in items: |
| 283 | sender = str(item.sender.email_address) if item.sender else "" |
| 284 | if _is_noreply(sender): |
| 285 | continue |
| 286 | if sender_whitelist and not _matches_whitelist(sender, sender_whitelist): |
| 287 | continue |
| 288 | |
| 289 | body = str(item.text_body or item.body or "") |
| 290 | if item.body and str(item.body).strip().startswith("<"): |
| 291 | body = _html_to_text(str(item.body)) |
| 292 | |
| 293 | attachment_paths: list[str] = [] |
| 294 | if item.attachments: |
| 295 | for att in item.attachments: |
| 296 | if hasattr(att, "content") and att.name: |
| 297 | path = await _save_attachment(att.name, att.content, download_folder) |
| 298 | attachment_paths.append(path) |
| 299 | |
| 300 | results.append(InboundMessage( |
| 301 | sender=sender, |
| 302 | subject=str(item.subject or ""), |
| 303 | body=body, |
| 304 | attachments=attachment_paths, |
| 305 | message_id=str(getattr(item, "message_id", "") or ""), |
| 306 | in_reply_to=str(getattr(item, "in_reply_to", "") or ""), |
| 307 | references="", |
| 308 | )) |
| 309 | |
| 310 | return results |
| 311 | |
| 312 | |
| 313 | # ------------------------------------------------------------------ |
| 314 | # Parsing helpers |
| 315 | # ------------------------------------------------------------------ |
| 316 | |
| 317 | async def _parse_body( |
| 318 | email_msg: EmailMessage, |
| 319 | download_folder: str, |
| 320 | ) -> tuple[str, list[str]]: |
| 321 | body = "" |
| 322 | attachments: list[str] = [] |
| 323 | cid_map: dict[str, str] = {} |
| 324 | body_parts: list[str] = [] |
| 325 | |
| 326 | if email_msg.is_multipart(): |
| 327 | for part in email_msg.walk(): |
| 328 | content_type = part.get_content_type() |
| 329 | disposition = str(part.get("Content-Disposition", "")) |
| 330 | |
| 331 | if part.get_content_maintype() == "multipart": |
| 332 | continue |
| 333 | |
| 334 | if "attachment" in disposition or part.get("Content-ID"): |
| 335 | filename = part.get_filename() |
| 336 | if filename: |
| 337 | filename = _decode_header(filename) |
| 338 | content = part.get_payload(decode=True) |
| 339 | if isinstance(content, bytes): |
| 340 | path = await _save_attachment(filename, content, download_folder) |
| 341 | attachments.append(path) |
| 342 | cid = part.get("Content-ID") |
| 343 | if cid: |
| 344 | cid_map[cid.strip("<>")] = path |
| 345 | |
| 346 | if not cid: |
| 347 | body_parts.append(f"\n[attachment://{path}]\n") |
| 348 | |
| 349 | elif content_type == "text/plain": |
| 350 | if not body: |
| 351 | charset = part.get_content_charset() or "utf-8" |
| 352 | payload = part.get_payload(decode=True) |
| 353 | body = payload.decode(charset, errors="ignore") if isinstance(payload, bytes) else "" |
| 354 | body_parts.append(body) |
| 355 | |
| 356 | elif content_type == "text/html": |
| 357 | if not body: |
| 358 | charset = part.get_content_charset() or "utf-8" |
| 359 | payload = part.get_payload(decode=True) |
| 360 | html = payload.decode(charset, errors="ignore") if isinstance(payload, bytes) else "" |
| 361 | body = _html_to_text(html, cid_map) |
| 362 | body_parts.append(body) |
| 363 | |
| 364 | if len(body_parts) > 1: |
| 365 | body = "".join(body_parts) |
| 366 | else: |
| 367 | content_type = email_msg.get_content_type() |
| 368 | charset = email_msg.get_content_charset() or "utf-8" |
| 369 | content = email_msg.get_payload(decode=True) |
| 370 | if isinstance(content, bytes): |
| 371 | if content_type == "text/html": |
| 372 | body = _html_to_text(content.decode(charset, errors="ignore")) |
| 373 | else: |
| 374 | body = content.decode(charset, errors="ignore") |
| 375 | |
| 376 | body = _strip_quoted_reply(body) |
| 377 | return body, attachments |
| 378 | |
| 379 | |
| 380 | def _strip_quoted_reply(text: str) -> str: |
| 381 | """Remove quoted reply chains (e.g. 'On ... wrote:' + '>' lines).""" |
| 382 | if not text: |
| 383 | return text |
| 384 | lines = text.splitlines() |
| 385 | cut = len(lines) |
| 386 | for i, line in enumerate(lines): |
| 387 | # Match "On <date> <someone> wrote:" pattern |
| 388 | if re.match(r"^On .+ wrote:\s*$", line.strip()): |
| 389 | # Verify next non-empty lines are quoted |
| 390 | rest = [l for l in lines[i + 1:] if l.strip()] |
| 391 | if not rest or rest[0].strip().startswith(">"): |
| 392 | cut = i |
| 393 | break |
| 394 | # Also strip trailing blank lines before the cut |
| 395 | while cut > 0 and not lines[cut - 1].strip(): |
| 396 | cut -= 1 |
| 397 | return "\n".join(lines[:cut]).strip() |
| 398 | |
| 399 | |
| 400 | def _html_to_text(html_content: str, cid_map: dict[str, str] | None = None) -> str: |
| 401 | if cid_map: |
| 402 | soup = BeautifulSoup(html_content, "html.parser") |
| 403 | for img in soup.find_all("img"): |
| 404 | src = str(img.get("src", "")) # type: ignore |
| 405 | if src.startswith("cid:"): |
| 406 | cid = src[4:] |
| 407 | if cid in cid_map: |
| 408 | img.replace_with(soup.new_string(f"[attachment://{cid_map[cid]}]")) |
| 409 | html_content = str(soup) |
| 410 | |
| 411 | h = html2text.HTML2Text() |
| 412 | h.ignore_links = False |
| 413 | h.ignore_images = False |
| 414 | h.ignore_emphasis = False |
| 415 | h.body_width = 0 |
| 416 | text = h.handle(html_content) |
| 417 | text = re.sub(r"\n{3,}", "\n\n", text).strip() |
| 418 | return text |
| 419 | |
| 420 | |
| 421 | async def _save_attachment(filename: str, content: bytes, download_folder: str) -> str: |
| 422 | filename = files.safe_file_name(filename) |
| 423 | name, ext = os.path.splitext(filename) |
| 424 | unique = f"{name}_{uuid.uuid4().hex[:8]}{ext}" |
| 425 | rel_path = os.path.join(download_folder, unique) |
| 426 | from helpers import runtime |
| 427 | from plugins._email_integration.helpers.attachment_writer import write_attachment |
| 428 | |
| 429 | import base64 |
| 430 | content_b64 = base64.b64encode(content).decode() |
| 431 | |
| 432 | result = await runtime.call_development_function( |
| 433 | write_attachment, rel_path, content_b64 |
| 434 | ) |
| 435 | if result.get("error"): |
| 436 | from helpers.print_style import PrintStyle |
| 437 | PrintStyle.error(f"Failed to save attachment {filename}: {result['error']}") |
| 438 | |
| 439 | return result.get("path", files.get_abs_path(rel_path)) |
| 440 | |
| 441 | |
| 442 | def _decode_header(header: str) -> str: |
| 443 | if not header: |
| 444 | return "" |
| 445 | parts = [] |
| 446 | for part, encoding in decode_header(header): |
| 447 | if isinstance(part, bytes): |
| 448 | parts.append(part.decode(encoding or "utf-8", errors="ignore")) |
| 449 | else: |
| 450 | parts.append(str(part)) |
| 451 | return " ".join(parts) |
| 452 | |
| 453 | |
| 454 | def _is_noreply(sender: str) -> bool: |
| 455 | addr = sender.lower() |
| 456 | match = re.search(r"<([^>]+)>", addr) |
| 457 | if match: |
| 458 | addr = match.group(1) |
| 459 | local = addr.split("@")[0] if "@" in addr else addr |
| 460 | return local in ( |
| 461 | "noreply", "no-reply", "no_reply", |
| 462 | "donotreply", "do-not-reply", "do_not_reply", |
| 463 | "mailer-daemon", "postmaster", |
| 464 | ) |
| 465 | |
| 466 | |
| 467 | def _matches_whitelist(sender: str, whitelist: list[str]) -> bool: |
| 468 | sender_email = _extract_email_from_sender(sender.lower()) |
| 469 | for pattern in whitelist: |
| 470 | if fnmatch(sender_email, pattern.lower()): |
| 471 | return True |
| 472 | return False |
| 473 | |
| 474 | |
| 475 | def _extract_email_from_sender(sender: str) -> str: |
| 476 | """Extract email address from sender string. |
| 477 | |
| 478 | Handles formats like: |
| 479 | - "email@example.com" |
| 480 | - "Name <email@example.com>" |
| 481 | - "\"Display Name\" <email@example.com>" |
| 482 | |
| 483 | Uses content inside angle brackets as authoritative to prevent spoofing |
| 484 | by fake emails in the display name (e.g., "John ceo@company.com <real@email.com>"). |
| 485 | """ |
| 486 | import re |
| 487 | # Look for email inside angle brackets - this is the authoritative source |
| 488 | match = re.search(r"<([^>]+)>", sender) |
| 489 | if match: |
| 490 | email = match.group(1).strip() |
| 491 | # Validate it looks like an email |
| 492 | if re.match(r"^[^\s<>@]+@[^\s<>@]+\.[^\s<>@]+$", email): |
| 493 | return email |
| 494 | |
| 495 | # No angle brackets - extract email from the whole string |
| 496 | # This handles plain "email@example.com" or malformed input |
| 497 | email_match = re.search(r"[^\s<>]+@[^\s<>]+\.[^\s<>]+", sender) |
| 498 | if email_match: |
| 499 | return email_match.group(0) |
| 500 | |
| 501 | # Fallback: return the whole string (will likely fail pattern match) |
| 502 | return sender |