diff --git a/tests/test_gmail_client.py b/tests/test_gmail_client.py new file mode 100644 index 0000000..3f48150 --- /dev/null +++ b/tests/test_gmail_client.py @@ -0,0 +1,55 @@ +import base64 + +from zemail.gmail_client import _decode_body + + +def _encoded(text: str) -> dict: + return {"body": {"data": base64.urlsafe_b64encode(text.encode()).decode()}} + + +def test_decode_body_uses_html_when_plain_text_is_absent(): + payload = { + "mimeType": "text/html", + **_encoded( + """ + + + +

Quarterly planning & budget notes

+ +
Follow up with finance.
+ + + """ + ), + } + + assert _decode_body(payload) == "Quarterly planning & budget notes\nFollow up with finance." + + +def test_decode_body_prefers_plain_text_over_html_alternative(): + payload = { + "mimeType": "multipart/alternative", + "parts": [ + {"mimeType": "text/html", **_encoded("

HTML-only fallback

")}, + {"mimeType": "text/plain", **_encoded("Plain text body")}, + ], + } + + assert _decode_body(payload) == "Plain text body" + + +def test_decode_body_finds_nested_html_fallback(): + payload = { + "mimeType": "multipart/mixed", + "parts": [ + { + "mimeType": "multipart/alternative", + "parts": [ + {"mimeType": "text/html", **_encoded("
Nested invoice thread
")}, + ], + } + ], + } + + assert _decode_body(payload) == "Nested invoice thread" diff --git a/zemail/gmail_client.py b/zemail/gmail_client.py index c7921b1..4ca8a9a 100644 --- a/zemail/gmail_client.py +++ b/zemail/gmail_client.py @@ -5,6 +5,8 @@ import string import sys import time +from html import unescape +from html.parser import HTMLParser from google.auth.transport.requests import Request from google_auth_oauthlib.flow import InstalledAppFlow @@ -63,6 +65,88 @@ def __init__(self, auth_url: str): """) +class _HTMLBodyParser(HTMLParser): + """Extract readable text from simple email HTML without adding dependencies.""" + + _BLOCK_TAGS = { + "address", + "blockquote", + "br", + "div", + "h1", + "h2", + "h3", + "h4", + "h5", + "h6", + "li", + "ol", + "p", + "pre", + "table", + "td", + "th", + "tr", + "ul", + } + + def __init__(self): + super().__init__() + self._parts = [] + self._skip_depth = 0 + + def handle_starttag(self, tag, attrs): + tag = tag.lower() + if tag in {"script", "style"}: + self._skip_depth += 1 + elif tag in self._BLOCK_TAGS: + self._parts.append("\n") + + def handle_endtag(self, tag): + tag = tag.lower() + if tag in {"script", "style"} and self._skip_depth: + self._skip_depth -= 1 + elif tag in self._BLOCK_TAGS: + self._parts.append("\n") + + def handle_data(self, data): + if not self._skip_depth: + self._parts.append(data) + + def text(self) -> str: + lines = [] + for raw_line in unescape("".join(self._parts)).splitlines(): + line = " ".join(raw_line.split()) + if line: + lines.append(line) + return "\n".join(lines) + + +def _html_to_text(markup: str) -> str: + parser = _HTMLBodyParser() + parser.feed(markup) + parser.close() + return parser.text() + + +def _decode_part_data(payload: dict) -> str: + data = payload.get("body", {}).get("data") + if not data: + return "" + return base64.urlsafe_b64decode(data).decode("utf-8", errors="replace") + + +def _find_body_by_mime(payload: dict, mime_type: str) -> str: + if payload.get("mimeType") == mime_type: + return _decode_part_data(payload) + + for part in payload.get("parts", []): + result = _find_body_by_mime(part, mime_type) + if result: + return result + return "" + + def _start_oauth_server(): """Start a background OAuth server. Returns the auth URL. @@ -72,7 +156,7 @@ def _start_oauth_server(): import socket import threading import wsgiref.simple_server - from urllib.parse import parse_qs, urlparse + from urllib.parse import parse_qs global _oauth_server_thread, _oauth_auth_url, _oauth_redirect_uri @@ -194,18 +278,15 @@ def complete_oauth(auth_code: str): def _decode_body(payload: dict) -> str: - """Extract plain text body from a Gmail message payload.""" - if payload.get("mimeType") == "text/plain" and payload.get("body", {}).get("data"): - return base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", errors="replace") - - parts = payload.get("parts", []) - for part in parts: - if part.get("mimeType") == "text/plain" and part.get("body", {}).get("data"): - return base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8", errors="replace") - if part.get("parts"): - result = _decode_body(part) - if result: - return result + """Extract searchable body text from a Gmail message payload.""" + plain_text = _find_body_by_mime(payload, "text/plain") + if plain_text: + return plain_text + + html_text = _find_body_by_mime(payload, "text/html") + if html_text: + return _html_to_text(html_text) + return ""