Hearth/hearth/watcher.py

"""IMAP polling for full_duplex mailboxes.

Only ever looks at mail from senders already on the allow-list (matched by
exact email address via `Config.is_known_email`). Anything from an address
that isn't a configured participant is left completely untouched — Hearth
does not read, move, or react to a stranger's email.
"""
from __future__ import annotations

import imaplib
import logging
import random
from datetime import datetime, timedelta, timezone
from email import message_from_bytes
from email.utils import parseaddr

from hearth import mailer
from hearth.composer import compose_reply
from hearth.config import Config
from hearth.scheduler import in_business_hours, now_local
from hearth.settings import settings
from hearth.store import Store

log = logging.getLogger("hearth.watcher")

SPAM_FOLDER_CANDIDATES = ["[Gmail]/Spam", "Spam", "Junk", "INBOX.Junk", "INBOX.Spam"]


def _decode(value) -> str:
    if value is None:
        return ""
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")
    return str(value)


def _random_reply_delay() -> timedelta:
    minutes = random.randint(settings.min_reply_delay_min, settings.max_reply_delay_min)
    return timedelta(minutes=minutes)


def _open_imap(participant) -> imaplib.IMAP4:
    imap = imaplib.IMAP4_SSL(participant.imap_host, participant.imap_port)
    imap.login(participant.username, participant.password)
    return imap


def _fetch_unseen(imap: imaplib.IMAP4, folder: str) -> list[bytes]:
    typ, _ = imap.select(folder, readonly=False)
    if typ != "OK":
        return []
    typ, data = imap.search(None, "UNSEEN")
    if typ != "OK" or not data or not data[0]:
        return []
    return data[0].split()


def _handle_message(config: Config, store: Store, mailbox_key: str, raw: bytes, *, from_spam: bool) -> None:
    msg = message_from_bytes(raw)
    _, from_addr = parseaddr(_decode(msg.get("From")))
    sender_key = config.is_known_email(from_addr)
    if not sender_key:
        return  # not a pool participant — never touched
    if sender_key == mailbox_key:
        return  # our own mailbox talking to itself, ignore

    message_id = mailer.sanitize_header(_decode(msg.get("Message-ID")))
    if not message_id:
        return
    if store.already_seen(message_id) or store.is_reply_queued(message_id):
        return

    in_reply_to = mailer.sanitize_header(_decode(msg.get("In-Reply-To"))) or None
    subject = mailer.sanitize_header(_decode(msg.get("Subject"))) or "(no subject)"
    references = mailer.sanitize_header(_decode(msg.get("References"))) or in_reply_to

    thread_id = store.thread_id_for(in_reply_to)
    store.record_message(
        thread_id=thread_id,
        from_key=sender_key,
        to_key=mailbox_key,
        direction="received",
        message_id_header=message_id,
        in_reply_to=in_reply_to,
        subject=subject,
        status="unspammed" if from_spam else "received",
    )

    if store.already_replied(message_id):
        return

    # Auto-replies only help domain warmup when at least one side is on a
    # warmup domain (default: levkine.ca). Skip Gmail↔Gmail chatter.
    allowed_domains = [
        d.strip() for d in settings.warmup_sender_domains.split(",") if d.strip()
    ]
    involves_warmup = config.is_warmup_domain(mailbox_key, allowed_domains) or config.is_warmup_domain(
        sender_key, allowed_domains
    )
    if not involves_warmup:
        log.info(
            "skipping auto-reply %s -> %s (neither side is on a warmup domain)",
            mailbox_key,
            sender_key,
        )
        return

    scheduled_at = (datetime.now(timezone.utc) + _random_reply_delay()).isoformat()
    store.queue_reply(
        message_id_header=message_id,
        thread_id=thread_id,
        mailbox_key=mailbox_key,
        sender_key=sender_key,
        subject=subject,
        references_header=references,
        scheduled_at=scheduled_at,
    )
    log.info(
        "queued reply: %s -> %s (subject=%r, in %s)",
        mailbox_key,
        sender_key,
        subject,
        "spam->inbox" if from_spam else "inbox",
    )


def poll_mailbox(config: Config, store: Store, mailbox_key: str) -> None:
    participant = config.get(mailbox_key)
    if not participant.can_authenticate:
        return

    try:
        imap = _open_imap(participant)
    except Exception:
        log.exception("IMAP connect failed for %s", mailbox_key)
        return

    try:
        for uid in _fetch_unseen(imap, "INBOX"):
            typ, data = imap.fetch(uid, "(RFC822)")
            if typ == "OK" and data and data[0]:
                _handle_message(config, store, mailbox_key, data[0][1], from_spam=False)

        for folder in SPAM_FOLDER_CANDIDATES:
            try:
                uids = _fetch_unseen(imap, folder)
            except Exception:
                continue
            for uid in uids:
                typ, data = imap.fetch(uid, "(RFC822)")
                if typ != "OK" or not data or not data[0]:
                    continue
                raw = data[0][1]
                _handle_message(config, store, mailbox_key, raw, from_spam=True)
                if not settings.dry_run and settings.enabled:
                    try:
                        imap.copy(uid, "INBOX")
                        imap.store(uid, "+FLAGS", "\\Deleted")
                    except Exception:
                        log.exception("failed to move message out of %s for %s", folder, mailbox_key)
            if uids:
                imap.expunge()
    finally:
        try:
            imap.close()
            imap.logout()
        except Exception:
            pass


def poll_all(config: Config, store: Store) -> None:
    for mailbox_key in config.full_duplex_keys():
        poll_mailbox(config, store, mailbox_key)


def send_due_replies(config: Config, store: Store) -> int:
    # Replies wait for the same business-hours window as new threads: a due
    # reply stays queued (due_replies keeps returning it) until the window
    # opens, so nothing is lost — just deferred to a human-plausible hour.
    if not in_business_hours(now_local()):
        return 0

    sent = 0
    for row in store.due_replies():
        mailbox = config.get(row["mailbox_key"])
        sender = config.get(row["sender_key"])
        subject, body = compose_reply(row["subject"], sender.display_name, sender.email)
        try:
            result = mailer.send(
                config,
                sender_key=row["mailbox_key"],
                recipient_key=row["sender_key"],
                subject=subject,
                body=body,
                in_reply_to=row["message_id_header"],
                references=row["references_header"],
            )
        except Exception:
            log.exception("failed to send queued reply %s -> %s", mailbox.email, sender.email)
            continue

        store.record_message(
            thread_id=row["thread_id"],
            from_key=row["mailbox_key"],
            to_key=row["sender_key"],
            direction="sent",
            message_id_header=result.message_id,
            in_reply_to=row["message_id_header"],
            subject=subject,
        )
        store.clear_queued_reply(row["message_id_header"])
        sent += 1
    return sent