"""Inbox reader: fetch emails via IMAP, match to leads, generate follow-up replies.""" import email import imaplib import json import logging import re logger = logging.getLogger(__name__) def imap_password(settings): from app import decrypt_setting_key return decrypt_setting_key(settings, "imap_password_encrypted") def fetch_inbox(settings, limit=50): """Connect to IMAP, fetch emails, return list of dicts. Each dict: message_id, from_email, from_name, subject, body_text, received_at.""" from app import settings_value host = settings_value(settings, "imap_host", "") or "" if not host: raise RuntimeError("IMAP host is not configured in Settings.") try: port = int(settings_value(settings, "imap_port", 993) or 993) except (TypeError, ValueError): port = 993 encryption = (settings_value(settings, "imap_encryption", "ssl") or "ssl").lower() username = settings_value(settings, "imap_username", "") or settings_value(settings, "smtp_username", "") or "" password = imap_password(settings) if not username or not password: raise RuntimeError("IMAP username/password not configured.") if encryption == "ssl": conn = imaplib.IMAP4_SSL(host, port) else: conn = imaplib.IMAP4(host, port) if encryption == "tls": conn.starttls() conn.login(username, password) conn.select("INBOX") _, msg_nums = conn.search(None, "UNSEEN") results = [] for num in reversed(msg_nums[0].split()[:limit]): _, msg_data = conn.fetch(num, "(RFC822)") if not msg_data or not msg_data[0]: continue raw = msg_data[0][1] msg = email.message_from_bytes(raw) message_id = msg.get("Message-ID", "").strip() if not message_id: # Fallback: use date + from + subject hash as a unique ID. Use the RAW # headers here (from_email/subject are parsed below, not yet defined). import hashlib message_id = "gen-" + hashlib.md5( (msg.get("Date", "") + msg.get("From", "") + msg.get("Subject", "")).encode() ).hexdigest()[:16] from_header = msg.get("From", "") from_email = "" from_name = "" m = re.search(r'<([^>]+)>', from_header) if m: from_email = m.group(1).strip().lower() from_name = from_header[:m.start()].strip().strip('"') else: from_email = from_header.strip().lower() subject = msg.get("Subject", "").strip() if subject: try: from email.header import decode_header parts = decode_header(subject) subject = "".join( p[0].decode(p[1] or "utf-8", errors="replace") if isinstance(p[0], bytes) else str(p[0]) for p in parts) except Exception: pass body_text = "" if msg.is_multipart(): for part in msg.walk(): if part.get_content_type() == "text/plain": payload = part.get_payload(decode=True) if payload: charset = part.get_content_charset() or "utf-8" body_text = payload.decode(charset, errors="replace").strip() break else: payload = msg.get_payload(decode=True) if payload: charset = msg.get_content_charset() or "utf-8" body_text = payload.decode(charset, errors="replace").strip() date_str = msg.get("Date", "") received_at_iso = "" try: from email.utils import parsedate_to_datetime dt = parsedate_to_datetime(date_str) if dt: received_at_iso = dt.isoformat() except Exception: pass results.append({ "message_id": message_id, "from_email": from_email, "from_name": from_name, "subject": subject, "body_text": body_text[:10000], "received_at": date_str, "received_at_iso": received_at_iso, }) conn.logout() return results def test_connection(settings): """Connect + login + select INBOX + count UNSEEN and ALL. Returns a dict for the admin 'Test IMAP' button: {ok, unread, total, error}. This is the fastest way to see WHY the inbox pull finds nothing (auth fail, wrong host, or just 0 unread).""" from app import settings_value host = settings_value(settings, "imap_host", "") or "" if not host: return {"ok": False, "error": "IMAP host is not configured in Settings."} try: port = int(settings_value(settings, "imap_port", 993) or 993) except (TypeError, ValueError): port = 993 encryption = (settings_value(settings, "imap_encryption", "ssl") or "ssl").lower() username = settings_value(settings, "imap_username", "") or settings_value(settings, "smtp_username", "") or "" password = imap_password(settings) if not username or not password: return {"ok": False, "error": "IMAP username/password not configured (or the password did not decrypt — re-save it)."} try: if encryption == "ssl": conn = imaplib.IMAP4_SSL(host, port, timeout=20) else: conn = imaplib.IMAP4(host, port, timeout=20) if encryption == "tls": conn.starttls() try: conn.login(username, password) conn.select("INBOX") _, unseen = conn.search(None, "UNSEEN") _, allm = conn.search(None, "ALL") unread_n = len(unseen[0].split()) if unseen and unseen[0] else 0 total_n = len(allm[0].split()) if allm and allm[0] else 0 return {"ok": True, "unread": unread_n, "total": total_n, "error": None} finally: try: conn.logout() except Exception: pass except Exception as e: return {"ok": False, "error": f"{type(e).__name__}: {e}"} def match_lead(db, from_email): """Match an email address to the most recent campaign link (lead). Returns a dict with lead info or None.""" row = db.execute( """SELECT cl.*, l.year, l.make, l.model, l.trim, l.price, l.slug, lc.id AS campaign_id FROM campaign_links cl JOIN lead_campaigns lc ON lc.id = cl.campaign_id LEFT JOIN listings l ON l.id = lc.listing_id WHERE cl.lead_email = ? AND COALESCE(cl.duplicate,0) = 0 ORDER BY cl.id DESC LIMIT 1""", (from_email,), ).fetchone() if not row: return None try: price_str = "${:,}".format(int(float(str(row["price"] or "0").replace(",", "")))) except (ValueError, TypeError): price_str = "" car_line = f"{row['year']} {row['make']} {row['model']} {row['trim'] or ''} {price_str}".strip() return { "link_id": row["id"], "lead_email": row["lead_email"], "lead_name": row["lead_name"], "lead_message": row["lead_message"], "our_reply": row["generated_reply"], "car_line": car_line, "token": row["token"], } def build_followup_prompt(master_prompt, lead_info, incoming_email_body, incoming_subject): """Build the AI prompt for generating a follow-up reply.""" mp = (master_prompt or "").strip() or ( "You are a helpful assistant writing a follow-up reply to a lead who emailed back about a vehicle." ) car = lead_info.get("car_line", "") if lead_info else "" lead_msg = lead_info.get("lead_message", "") if lead_info else "" our_reply = lead_info.get("our_reply", "") if lead_info else "" lead_name = lead_info.get("lead_name", "") if lead_info else "" return ( f"{mp}\n\n" f"CONTEXT:\n" f"Vehicle: {car}\n" f"Lead name: {lead_name}\n" f"Lead's original inquiry: {lead_msg}\n" f"Our previous reply: {our_reply}\n\n" f"LEAD'S NEW EMAIL (subject: {incoming_subject}):\n{incoming_email_body}\n\n" f"Write a natural follow-up reply to the lead's new email. " f"Answer any new questions they asked. Keep it concise and human. " f"Reuse the subject of their email (with Re: if not already present).\n" f'Return JSON only: {{"subject": "the subject", "body": "the reply body"}}' )