468 lines
No EOL
15 KiB
Python
468 lines
No EOL
15 KiB
Python
"""IMAP email watch: polls UNSEEN messages and classifies them.
|
|
|
|
Uses stdlib imaplib (SSL). Enabled only when EMAIL_WATCH_ENABLED=true.
|
|
Config: IMAP_HOST, IMAP_PORT, IMAP_USER, IMAP_PASS.
|
|
|
|
Flow:
|
|
1. Connect via IMAP SSL.
|
|
2. Fetch UNSEEN messages since last poll.
|
|
3. For each message, match sender domain + subject/body keywords to
|
|
open applications (status in sent/interviewing).
|
|
4. Classify via LLM task 'email_classify' -> {classification, state_proposal, reason}.
|
|
5. Insert email_suggestion rows (skip noise and dedupe by from+subject+day).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import email
|
|
import email.utils
|
|
import hashlib
|
|
import imaplib
|
|
import logging
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Sequence
|
|
|
|
from app.db import execute, fetch_all, fetch_one
|
|
from app import llm
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
VALID_CLASSIFICATIONS = frozenset({
|
|
"interview_invite",
|
|
"rejection",
|
|
"question",
|
|
"noise",
|
|
})
|
|
|
|
# Keywords for cheap pre-matching before LLM classify
|
|
INTERVIEW_KEYWORDS = ("interview", "invite", "meeting", "schedule", "call")
|
|
REJECTION_KEYWORDS = ("regret", "unfortunately", "not moving", "rejection", "position has been filled")
|
|
QUESTION_KEYWORDS = ("question", "clarif", "additional", "could you", "please provide")
|
|
NOISE_KEYWORDS = ("newsletter", "unsubscribe", "promotion", "advert", "offer")
|
|
|
|
|
|
def is_email_watch_enabled() -> bool:
|
|
"""Check if email watch is enabled."""
|
|
return os.environ.get("EMAIL_WATCH_ENABLED", "false").lower() in (
|
|
"true",
|
|
"1",
|
|
"yes",
|
|
)
|
|
|
|
|
|
def _extract_sender_domain(from_addr: str) -> str:
|
|
"""Extract the domain from an email From header."""
|
|
parsed = email.utils.parseaddr(from_addr)
|
|
addr = parsed[1] or from_addr
|
|
parts = addr.split("@")
|
|
if len(parts) >= 2:
|
|
return parts[-1].lower().strip()
|
|
return ""
|
|
|
|
|
|
def _build_snippet(body: str, max_len: int = 300) -> str:
|
|
"""Truncate body to a snippet."""
|
|
body = body.replace("\r", " ").replace("\n", " ").strip()
|
|
if len(body) > max_len:
|
|
return body[:max_len] + "..."
|
|
return body
|
|
|
|
|
|
def _parse_email_message(raw_bytes: bytes) -> dict[str, str]:
|
|
"""Parse raw email bytes into a dict with from, subject, body."""
|
|
msg = email.message_from_bytes(raw_bytes)
|
|
from_addr = msg.get("From", "")
|
|
subject = msg.get("Subject", "")
|
|
date_str = msg.get("Date", "")
|
|
|
|
# Extract body (prefer plain text)
|
|
body = ""
|
|
if msg.is_multipart():
|
|
for part in msg.walk():
|
|
ct = part.get_content_type()
|
|
if ct == "text/plain":
|
|
payload = part.get_payload(decode=True)
|
|
if payload:
|
|
body = payload.decode("utf-8", errors="replace")
|
|
break
|
|
if not body:
|
|
for part in msg.walk():
|
|
if part.get_content_type().startswith("text/"):
|
|
payload = part.get_payload(decode=True)
|
|
if payload:
|
|
body = payload.decode("utf-8", errors="replace")
|
|
break
|
|
else:
|
|
payload = msg.get_payload(decode=True)
|
|
if payload:
|
|
body = payload.decode("utf-8", errors="replace")
|
|
|
|
return {
|
|
"from": from_addr,
|
|
"subject": subject,
|
|
"body": body,
|
|
"date": date_str,
|
|
}
|
|
|
|
|
|
def _parse_date(date_str: str) -> datetime:
|
|
"""Parse RFC 2822 date string to UTC datetime. Falls back to now()."""
|
|
if date_str:
|
|
try:
|
|
parsed = email.utils.parsedate_to_datetime(date_str)
|
|
if parsed is not None:
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def match_application(
|
|
from_addr: str,
|
|
subject: str,
|
|
body: str,
|
|
applications: Sequence[dict[str, Any]],
|
|
) -> dict[str, Any] | None:
|
|
"""Match an email to an open application.
|
|
|
|
Match by:
|
|
1. Company name in subject/body
|
|
2. Sender domain in posting URL
|
|
|
|
Only matches applications with state in ('sent', 'interviewing').
|
|
"""
|
|
sender_domain = _extract_sender_domain(from_addr)
|
|
text_lower = (subject + " " + body).lower()
|
|
|
|
for app_row in applications:
|
|
state = app_row.get("state", "")
|
|
if state not in ("sent", "interviewing"):
|
|
continue
|
|
|
|
company = (app_row.get("company") or "").lower()
|
|
title = (app_row.get("title") or "").lower()
|
|
posting_url = app_row.get("url") or ""
|
|
|
|
# Match 1: company name in subject/body
|
|
if company and len(company) > 2 and company in text_lower:
|
|
return app_row
|
|
|
|
# Match 2: sender domain in posting URL
|
|
if sender_domain and sender_domain in (posting_url or "").lower():
|
|
return app_row
|
|
|
|
# Match 3: title keywords in subject (looser)
|
|
if title and len(title) > 3:
|
|
title_words = [w for w in title.split() if len(w) > 3]
|
|
matches = sum(1 for w in title_words if w in text_lower)
|
|
if matches >= 2 and len(title_words) >= 2:
|
|
return app_row
|
|
|
|
return None
|
|
|
|
|
|
def classify_email(subject: str, body: str) -> dict[str, Any]:
|
|
"""Classify an email via LLM task 'email_classify'.
|
|
|
|
Returns {classification, state_proposal, reason}.
|
|
Falls back to inline keyword heuristics if LLM fails.
|
|
"""
|
|
text_lower = (subject + " " + body).lower()
|
|
|
|
result = llm.run_task(
|
|
"email_classify",
|
|
f"Subject: {subject}\nBody: {body[:1000]}",
|
|
)
|
|
|
|
classification = result.get("classification", "noise")
|
|
state_proposal = result.get("state_proposal")
|
|
reason = result.get("reason", "")
|
|
|
|
# Validate
|
|
if classification not in VALID_CLASSIFICATIONS:
|
|
classification = "noise"
|
|
|
|
return {
|
|
"classification": classification,
|
|
"state_proposal": state_proposal,
|
|
"reason": reason,
|
|
}
|
|
|
|
|
|
def is_duplicate(from_addr: str, subject: str, received_at: datetime) -> bool:
|
|
"""Check if a similar email_suggestion already exists (same from+subject+day)."""
|
|
day_start = received_at.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
row = fetch_one(
|
|
"""
|
|
SELECT id FROM email_suggestion
|
|
WHERE mailbox_from = %s AND subject = %s
|
|
AND received_at >= %s AND received_at < %s + interval '1 day'
|
|
LIMIT 1
|
|
""",
|
|
(from_addr, subject, day_start, day_start),
|
|
)
|
|
return row is not None
|
|
|
|
|
|
def create_email_suggestion(
|
|
application_id: str | None,
|
|
mailbox_from: str,
|
|
subject: str,
|
|
snippet: str,
|
|
classification: str,
|
|
state_proposal: str | None,
|
|
received_at: datetime,
|
|
) -> dict[str, Any]:
|
|
"""Insert an email_suggestion row."""
|
|
row = execute(
|
|
"""
|
|
INSERT INTO email_suggestion
|
|
(application_id, mailbox_from, subject, snippet, classification, state_proposal, status, received_at)
|
|
VALUES
|
|
(%s, %s, %s, %s, %s, %s, 'pending', %s)
|
|
RETURNING *
|
|
""",
|
|
(application_id, mailbox_from, subject, snippet, classification, state_proposal, received_at),
|
|
)
|
|
if row is None:
|
|
raise RuntimeError("insert email_suggestion failed")
|
|
return _normalize_suggestion(row)
|
|
|
|
|
|
def _normalize_suggestion(row: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"id": str(row["id"]),
|
|
"application_id": str(row["application_id"]) if row.get("application_id") else None,
|
|
"mailbox_from": row["mailbox_from"],
|
|
"subject": row["subject"],
|
|
"snippet": row["snippet"],
|
|
"classification": row["classification"],
|
|
"state_proposal": row.get("state_proposal"),
|
|
"status": row["status"],
|
|
"received_at": row["received_at"].isoformat() if row.get("received_at") else None,
|
|
"created_at": row["created_at"].isoformat() if row.get("created_at") else None,
|
|
}
|
|
|
|
|
|
def list_pending_suggestions() -> list[dict[str, Any]]:
|
|
"""Return all pending email_suggestions, newest first."""
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT * FROM email_suggestion
|
|
WHERE status = 'pending'
|
|
ORDER BY received_at DESC
|
|
"""
|
|
)
|
|
return [_normalize_suggestion(r) for r in rows]
|
|
|
|
|
|
def get_suggestion(suggestion_id: str) -> dict[str, Any] | None:
|
|
row = fetch_one("SELECT * FROM email_suggestion WHERE id = %s", (suggestion_id,))
|
|
if row is None:
|
|
return None
|
|
return _normalize_suggestion(row)
|
|
|
|
|
|
def update_suggestion_status(
|
|
suggestion_id: str,
|
|
status: str,
|
|
) -> dict[str, Any] | None:
|
|
row = execute(
|
|
"UPDATE email_suggestion SET status = %s WHERE id = %s RETURNING *",
|
|
(status, suggestion_id),
|
|
)
|
|
if row is None:
|
|
return None
|
|
return _normalize_suggestion(row)
|
|
|
|
|
|
def get_open_applications() -> list[dict[str, Any]]:
|
|
"""Return applications in sent/interviewing state for matching."""
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT a.*, j.company, j.title, j.url
|
|
FROM application a
|
|
JOIN job_posting j ON a.job_posting_id = j.id
|
|
WHERE a.state IN ('sent', 'interviewing')
|
|
"""
|
|
)
|
|
result: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
result.append({
|
|
"id": str(row["id"]),
|
|
"state": row["state"],
|
|
"company": row.get("company", ""),
|
|
"title": row.get("title", ""),
|
|
"url": row.get("url", ""),
|
|
})
|
|
return result
|
|
|
|
|
|
# --- IMAP poll ---
|
|
|
|
class FakeImap:
|
|
"""Test double for imaplib IMAP4_SSL. No network.
|
|
|
|
Usage:
|
|
fake = FakeImap(messages=[(uid1, raw1), (uid2, raw2)])
|
|
poll_inbox(fake) # uses fake instead of real connection
|
|
"""
|
|
|
|
def __init__(self, messages: list[tuple[bytes, bytes]] | None = None) -> None:
|
|
# messages: list of (uid, raw_email_bytes)
|
|
self._messages = messages or []
|
|
self._seen_uids: set[bytes] = set()
|
|
self.selected = False
|
|
|
|
def select(self, mailbox: str = "INBOX") -> tuple[str, list[bytes]]:
|
|
self.selected = True
|
|
count = len(self._messages)
|
|
return ("OK", [str(count).encode()])
|
|
|
|
def search(self, charset: str | None, *criteria: str) -> tuple[str, list[bytes]]:
|
|
# Return uids of messages matching criteria (we keep it simple)
|
|
uids = [uid for uid, _ in self._messages]
|
|
return ("OK", [b" ".join(uids)])
|
|
|
|
def fetch(self, uid: bytes, parts: str) -> tuple[str, list[tuple[bytes, bytes]]]:
|
|
for msg_uid, raw in self._messages:
|
|
if msg_uid == uid:
|
|
return ("OK", [(uid, raw)])
|
|
return ("OK", [])
|
|
|
|
def store(self, uid: bytes, flags: str, flag_set: str) -> tuple[str, list[bytes]]:
|
|
self._seen_uids.add(uid)
|
|
return ("OK", [uid])
|
|
|
|
def close(self) -> tuple[str, list[bytes]]:
|
|
self.selected = False
|
|
return ("OK", [b""])
|
|
|
|
def logout(self) -> tuple[str, list[bytes]]:
|
|
return ("OK", [b"BYE"])
|
|
|
|
|
|
def _connect_imap() -> Any:
|
|
"""Connect to IMAP server using env config."""
|
|
host = os.environ.get("IMAP_HOST", "")
|
|
port = int(os.environ.get("IMAP_PORT", "993"))
|
|
user = os.environ.get("IMAP_USER", "")
|
|
password = os.environ.get("IMAP_PASS", "")
|
|
|
|
conn = imaplib.IMAP4_SSL(host, port)
|
|
conn.login(user, password)
|
|
return conn
|
|
|
|
|
|
def poll_inbox(conn: Any = None) -> list[dict[str, Any]]:
|
|
"""Poll the inbox for UNSEEN messages, classify, and create suggestions.
|
|
|
|
If conn is provided (e.g. FakeImap for tests), uses it instead of connecting.
|
|
Returns list of created suggestions.
|
|
|
|
No network is used when conn is a FakeImap.
|
|
"""
|
|
created: list[dict[str, Any]] = []
|
|
own_conn = False
|
|
|
|
if conn is None:
|
|
conn = _connect_imap()
|
|
own_conn = True
|
|
|
|
try:
|
|
conn.select("INBOX")
|
|
status, data = conn.search(None, "UNSEEN")
|
|
if status != "OK":
|
|
logger.warning("imap_watch: search failed: %s", status)
|
|
return created
|
|
|
|
uids = []
|
|
if data and data[0]:
|
|
uids = data[0].split()
|
|
|
|
if not uids:
|
|
return created
|
|
|
|
# Get open applications for matching
|
|
applications = get_open_applications()
|
|
|
|
for uid in uids:
|
|
status, fetch_data = conn.fetch(uid, "(RFC822)")
|
|
if status != "OK" or not fetch_data:
|
|
continue
|
|
|
|
raw_bytes = b""
|
|
for item in fetch_data:
|
|
if isinstance(item, tuple) and len(item) >= 2:
|
|
raw_bytes = item[1]
|
|
break
|
|
|
|
if not raw_bytes:
|
|
continue
|
|
|
|
parsed = _parse_email_message(raw_bytes)
|
|
from_addr = parsed["from"]
|
|
subject = parsed["subject"]
|
|
body = parsed["body"]
|
|
received_at = _parse_date(parsed["date"])
|
|
snippet = _build_snippet(body)
|
|
|
|
# Dedupe: same from+subject+day
|
|
if is_duplicate(from_addr, subject, received_at):
|
|
logger.debug("imap_watch: dedupe skip: %s / %s", from_addr, subject)
|
|
continue
|
|
|
|
# Match to application
|
|
app_row = match_application(from_addr, subject, body, applications)
|
|
|
|
# Classify
|
|
classification_result = classify_email(subject, body)
|
|
classification = classification_result["classification"]
|
|
state_proposal = classification_result["state_proposal"]
|
|
|
|
# Skip pure noise (don't create suggestion rows)
|
|
if classification == "noise":
|
|
continue
|
|
|
|
suggestion = create_email_suggestion(
|
|
application_id=app_row["id"] if app_row else None,
|
|
mailbox_from=from_addr,
|
|
subject=subject,
|
|
snippet=snippet,
|
|
classification=classification,
|
|
state_proposal=state_proposal,
|
|
received_at=received_at,
|
|
)
|
|
created.append(suggestion)
|
|
|
|
# Send notification for interview invites
|
|
if classification == "interview_invite" and app_row:
|
|
from app.notify import send_notification
|
|
send_notification(
|
|
"email_suggestion",
|
|
f"Interview invite from {app_row.get('company', 'unknown')}",
|
|
{
|
|
"suggestion_id": suggestion["id"],
|
|
"application_id": app_row["id"],
|
|
"classification": classification,
|
|
},
|
|
)
|
|
|
|
# Mark as seen
|
|
try:
|
|
conn.store(uid, "+FLAGS", "\\Seen")
|
|
except Exception:
|
|
logger.debug("imap_watch: store failed for uid %s", uid)
|
|
|
|
finally:
|
|
if own_conn:
|
|
try:
|
|
conn.close()
|
|
conn.logout()
|
|
except Exception:
|
|
pass
|
|
|
|
return created |