From 216ec6d4495e58a59a323ce677e5dfac5ff18123 Mon Sep 17 00:00:00 2001 From: Michael Wu Date: Sat, 21 Feb 2026 16:51:10 +0800 Subject: [PATCH 1/2] feat: add mailbox resume intake flow --- .env.example | 4 + README.md | 4 + .../five08/discord_bot/cogs/email_monitor.py | 260 +++---- .../src/five08/discord_bot/config.py | 4 + .../src/five08/discord_bot/utils/audit.py | 42 +- .../discord_bot/utils/resume_mail_ingest.py | 687 ++++++++++++++++++ tests/unit/test_discord_audit.py | 29 +- tests/unit/test_resume_mail_ingest.py | 152 ++++ 8 files changed, 1033 insertions(+), 149 deletions(-) create mode 100644 apps/discord_bot/src/five08/discord_bot/utils/resume_mail_ingest.py create mode 100644 tests/unit/test_resume_mail_ingest.py diff --git a/.env.example b/.env.example index 549cf6f3..7eb197e1 100644 --- a/.env.example +++ b/.env.example @@ -78,6 +78,10 @@ EMAIL_USERNAME=your_email@example.com EMAIL_PASSWORD=your_app_password IMAP_SERVER=imap.migadu.com SMTP_SERVER=smtp.migadu.com +EMAIL_RESUME_INTAKE_ENABLED=true +EMAIL_RESUME_ALLOWED_EXTENSIONS=pdf,doc,docx +EMAIL_RESUME_MAX_FILE_SIZE_MB=10 +EMAIL_REQUIRE_SENDER_AUTH_HEADERS=true # EspoCRM (required for worker integration) ESPO_API_KEY=your_key_here diff --git a/README.md b/README.md index 7e8fd3fd..89b32dfb 100644 --- a/README.md +++ b/README.md @@ -178,6 +178,10 @@ Use `.env.example` as the source of truth for defaults. - `Required`: `EMAIL_PASSWORD` - `Required`: `IMAP_SERVER` - `Required`: `SMTP_SERVER` +- `Optional`: `EMAIL_RESUME_INTAKE_ENABLED` (default: `true`; enables mailbox resume processing loop) +- `Optional`: `EMAIL_RESUME_ALLOWED_EXTENSIONS` (default: `pdf,doc,docx`) +- `Optional`: `EMAIL_RESUME_MAX_FILE_SIZE_MB` (default: `10`) +- `Optional`: `EMAIL_REQUIRE_SENDER_AUTH_HEADERS` (default: `true`; requires SPF/DKIM/DMARC pass headers) ### Discord CRM Audit Logging (Best Effort) diff --git a/apps/discord_bot/src/five08/discord_bot/cogs/email_monitor.py b/apps/discord_bot/src/five08/discord_bot/cogs/email_monitor.py index 699013a5..a9fd290e 100644 --- a/apps/discord_bot/src/five08/discord_bot/cogs/email_monitor.py +++ b/apps/discord_bot/src/five08/discord_bot/cogs/email_monitor.py @@ -1,34 +1,38 @@ -""" -Email monitoring cog for the 508.dev Discord bot. +"""Email monitoring cog for mailbox-driven resume intake workflows.""" -This cog monitors an IMAP email inbox and forwards new messages to a Discord channel. -It supports both plain text and HTML emails, automatically chunks long messages, -and provides commands to start/stop monitoring and check status. -""" +from __future__ import annotations -import imaplib +import contextlib import email +import imaplib import logging -from textwrap import wrap +from typing import Any + from discord.ext import commands, tasks -import discord from five08.discord_bot.config import settings +from five08.discord_bot.utils.resume_mail_ingest import ( + ResumeMailboxProcessor, + ResumeMailboxResult, +) logger = logging.getLogger(__name__) class EmailMonitor(commands.Cog): - """ - Email monitoring cog that polls IMAP inbox and forwards emails to Discord. - - This cog automatically starts monitoring when loaded and provides commands - for manual control of the monitoring process. - """ + """Poll an IMAP inbox and run resume intake on new messages.""" def __init__(self, bot: commands.Bot) -> None: self.bot = bot - # self.task_poll_inbox.start() + self.resume_processor = ResumeMailboxProcessor(settings) + + async def cog_load(self) -> None: + """Start polling task when this cog is loaded.""" + if ( + settings.email_resume_intake_enabled + and not self.task_poll_inbox.is_running() + ): + self.task_poll_inbox.start() async def cog_unload(self) -> None: """Cancel the background task when cog is unloaded.""" @@ -36,141 +40,105 @@ async def cog_unload(self) -> None: @tasks.loop(minutes=settings.check_email_wait) async def task_poll_inbox(self) -> None: - """Poll IMAP inbox for new emails and forward them to Discord.""" - channel = self.bot.get_channel(settings.channel_id) - if not channel or not isinstance(channel, discord.abc.Messageable): - logger.error(f"Could not find channel {settings.channel_id}") + """Poll IMAP inbox for new messages and process resume attachments.""" + logger.info("Reading inbox of %s", settings.email_username) + mail = imaplib.IMAP4_SSL(settings.imap_server) + try: + mail.login(settings.email_username, settings.email_password) + mail.select("INBOX") + retcode, message_batches = mail.search(None, "(UNSEEN)") + + if retcode != "OK" or not message_batches or not message_batches[0]: + logger.debug("Login complete, # of new messages: 0") + return + + message_ids = message_batches[0].split() + total_messages = len(message_ids) + + for index, raw_num in enumerate(message_ids, start=1): + num = raw_num.decode() + typ, data = mail.fetch(num, "(RFC822)") + if typ != "OK": + logger.warning( + "Skipping message %s due to failed fetch status=%s", num, typ + ) + continue + + result: ResumeMailboxResult + try: + result = await self._process_fetched_message(data) + except Exception as exc: + logger.exception( + "Failed processing inbound email num=%s error=%s", num, exc + ) + result = ResumeMailboxResult( + sender_email=None, + sender_name=None, + processed_attachments=0, + skipped_reason="message_processing_error", + ) + + self._log_processing_result( + result=result, index=index, total=total_messages + ) + + # Mark mail as seen to avoid reprocessing on next poll. + mail.store(num, "+FLAGS", "\\Seen") + + logger.debug("Login complete, # of new messages: %s", total_messages) + finally: + with contextlib.suppress(Exception): + mail.close() + with contextlib.suppress(Exception): + mail.logout() + + async def _process_fetched_message(self, data: list[Any]) -> ResumeMailboxResult: + for response_part in data: + if not isinstance(response_part, tuple): + continue + + raw_payload = response_part[1] + if not isinstance(raw_payload, (bytes, bytearray)): + continue + + message = email.message_from_bytes(bytes(raw_payload)) + return await self.resume_processor.process_message(message) + + return ResumeMailboxResult( + sender_email=None, + sender_name=None, + processed_attachments=0, + skipped_reason="message_payload_missing", + ) + + def _log_processing_result( + self, + *, + result: ResumeMailboxResult, + index: int, + total: int, + ) -> None: + sender = result.sender_email or "unknown" + if result.skipped_reason: + logger.info( + "Resume inbox message %s/%s sender=%s skipped reason=%s", + index, + total, + sender, + result.skipped_reason, + ) return - logger.info(f"Reading inbox of {settings.email_username}") - - # create an IMAP4 class with SSL and authenticate - mail = imaplib.IMAP4_SSL(settings.imap_server) - mail.login(settings.email_username, settings.email_password) - - status, messages = mail.select("INBOX") - # get unseen messages - (retcode, messages) = mail.search(None, "(UNSEEN)") - if retcode == "OK" and messages[0]: - for idx, num in enumerate(messages[0].split()): - typ, data = mail.fetch(num.decode(), "(RFC822)") - for response_part in data: - if isinstance(response_part, tuple): - original = email.message_from_string( - response_part[1].decode("utf-8") - ) - received = original["Received"] - if received: - received = received.split(";")[-1] - else: - received = "Unknown" - - logger.debug(f"From: {original['From']}") - logger.debug(f"Subject: {original['Subject']}") - logger.debug(f"Received: {received}") - msg_count = len(messages[0].split()) if messages[0] else 0 - await channel.send( - f"{'=' * 30} Message {idx + 1} of {msg_count} {'=' * 30}" - ) - await channel.send( - f"**FROM:** {original['From']}\n**SUBJECT:** {original['Subject']} \n**RECEIVED:** {received}" - ) - if original.is_multipart(): - # iterate over email parts - for part in original.walk(): - # extract content type of email - content_type = part.get_content_type() - content_disposition = str( - part.get("Content-Disposition") - ) - try: - # get the email body - payload = part.get_payload(decode=True) - if isinstance(payload, bytes): - body = payload.decode() - else: - continue - except Exception: - continue - if ( - content_type == "text/plain" - and "attachment" not in content_disposition - ): - # logger.debug text/plain emails and skip attachments - logger.debug(wrap(body, width=3900)) - await channel.send("**BODY**:") - for line in wrap( - body, - width=settings.discord_sendmsg_character_limit - - 1, - ): - await channel.send(line) - elif "attachment" in content_disposition: - # download attachment - logger.debug("attachment case") - else: - # extract content type of email - content_type = original.get_content_type() - # get the email body - payload = original.get_payload(decode=True) - if isinstance(payload, bytes): - body = payload.decode() - else: - continue - if content_type == "text/plain": - # logger.debug only text email parts - logger.debug(body) - await channel.send("**BODY**:") - for line in wrap( - body, - width=settings.discord_sendmsg_character_limit - 1, - ): - await channel.send(line) - if content_type == "text/html": - logger.debug("html case") - logger.debug(body) - await channel.send("**BODY**:") - for line in wrap( - body, - width=settings.discord_sendmsg_character_limit - 1, - ): - await channel.send(line) - logger.debug("=" * 100) - await channel.send("=" * 71) - - # mark the mail as seen so it doesn't come up again - typ, data = mail.store(num.decode(), "+FLAGS", "\\Seen") - - msg_count = len(messages[0].split()) if messages[0] else 0 - logger.debug("Login complete, # of new messages: " + str(msg_count)) - - # close the connection and logout - mail.close() - mail.logout() - logger.debug("end of this iteration") - - # @app_commands.command(name="start-email", description="Start email polling task") - # async def st(self, interaction: discord.Interaction) -> None: - # """Start email polling task.""" - # await interaction.response.send_message( - # f"Polling for emails every {settings.check_email_wait} minutes" - # ) - # if not self.task_poll_inbox.is_running(): - # self.task_poll_inbox.start() - - # @app_commands.command( - # name="email-status", description="Check if email polling task is running" - # ) - # async def is_running(self, interaction: discord.Interaction) -> None: - # """Check if email polling task is running.""" - # status = "is" if self.task_poll_inbox.is_running() else "isn't" - # await interaction.response.send_message( - # f"Inbox polling task *{status}* running" - # ) + logger.info( + "Resume inbox message %s/%s sender=%s processed_attachments=%s", + index, + total, + sender, + result.processed_attachments, + ) async def setup(bot: commands.Bot) -> None: """Add the EmailMonitor cog to the bot.""" cog = EmailMonitor(bot) await bot.add_cog(cog) - # Slash commands will be synced automatically in bot.py diff --git a/apps/discord_bot/src/five08/discord_bot/config.py b/apps/discord_bot/src/five08/discord_bot/config.py index 09240fea..71a89b83 100644 --- a/apps/discord_bot/src/five08/discord_bot/config.py +++ b/apps/discord_bot/src/five08/discord_bot/config.py @@ -30,6 +30,10 @@ class Settings(SharedSettings): email_password: str imap_server: str smtp_server: str + email_resume_intake_enabled: bool = True + email_resume_allowed_extensions: str = "pdf,doc,docx" + email_resume_max_file_size_mb: int = 10 + email_require_sender_auth_headers: bool = True # CRM/EspoCRM settings espo_api_key: str diff --git a/apps/discord_bot/src/five08/discord_bot/utils/audit.py b/apps/discord_bot/src/five08/discord_bot/utils/audit.py index b288f14d..f7235705 100644 --- a/apps/discord_bot/src/five08/discord_bot/utils/audit.py +++ b/apps/discord_bot/src/five08/discord_bot/utils/audit.py @@ -45,7 +45,7 @@ def log_command( if not self.enabled: return - event_payload = self._build_payload( + event_payload = self._build_discord_payload( interaction=interaction, action=action, result=result, @@ -54,6 +54,44 @@ def log_command( resource_id=resource_id, ) + self._queue_event(event_payload) + + def log_admin_sso_action( + self, + *, + action: str, + result: str, + actor_email: str, + actor_display_name: str | None = None, + metadata: dict[str, Any] | None = None, + resource_type: str | None = None, + resource_id: str | None = None, + correlation_id: str | None = None, + ) -> None: + """Queue best-effort audit write for non-Discord human actions.""" + if not self.enabled: + return + + normalized_email = actor_email.strip().lower() + if not normalized_email: + return + + event_payload = { + "source": "admin_dashboard", + "action": action, + "result": result, + "actor_provider": "admin_sso", + "actor_subject": normalized_email, + "actor_display_name": actor_display_name, + "resource_type": resource_type, + "resource_id": resource_id, + "correlation_id": correlation_id, + "metadata": metadata or {}, + } + + self._queue_event(event_payload) + + def _queue_event(self, event_payload: dict[str, Any]) -> None: task = asyncio.create_task(self._post_event(event_payload)) task.add_done_callback(self._on_task_done) @@ -97,7 +135,7 @@ def _on_task_done(self, task: asyncio.Task[None]) -> None: except Exception as exc: # pragma: no cover - defensive fallback logger.warning("Unexpected audit task failure: %s", exc) - def _build_payload( + def _build_discord_payload( self, *, interaction: discord.Interaction, diff --git a/apps/discord_bot/src/five08/discord_bot/utils/resume_mail_ingest.py b/apps/discord_bot/src/five08/discord_bot/utils/resume_mail_ingest.py new file mode 100644 index 00000000..05eb6f05 --- /dev/null +++ b/apps/discord_bot/src/five08/discord_bot/utils/resume_mail_ingest.py @@ -0,0 +1,687 @@ +"""Mailbox-based resume intake helpers for automated CRM updates.""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from email.message import Message +from email.utils import parseaddr +from typing import Any +from uuid import uuid4 + +import aiohttp + +from five08.clients.espo import EspoAPI, EspoAPIError +from five08.discord_bot.config import Settings +from five08.discord_bot.utils.audit import DiscordAuditLogger +from five08.queue import get_postgres_connection + +logger = logging.getLogger(__name__) + + +PRIVILEGED_ROLE_NAMES = {"admin", "steering committee", "owner"} + + +@dataclass(frozen=True) +class ResumeAttachment: + """One resume-like email attachment.""" + + filename: str + content: bytes + + +@dataclass(frozen=True) +class ResumeMailboxResult: + """Result metadata for one processed mailbox message.""" + + sender_email: str | None + sender_name: str | None + processed_attachments: int + skipped_reason: str | None = None + + +class ResumeMailboxProcessor: + """Process inbound mailbox messages into CRM resume extraction/apply flow.""" + + def __init__(self, settings: Settings) -> None: + self.settings = settings + api_url = settings.espo_base_url.rstrip("/") + "/api/v1" + self.espo_api = EspoAPI(api_url, settings.espo_api_key) + self.audit_logger = DiscordAuditLogger( + base_url=settings.audit_api_base_url, + shared_secret=settings.api_shared_secret, + timeout_seconds=settings.audit_api_timeout_seconds, + ) + + async def process_message(self, message: Message) -> ResumeMailboxResult: + """Process one email message and trigger resume extraction/apply jobs.""" + sender_name, sender_email = self._sender_identity(message) + correlation_id = self._mailbox_correlation_id(message) + + def finalize(result: ResumeMailboxResult) -> ResumeMailboxResult: + self._audit_mailbox_outcome( + sender_email=sender_email, + sender_name=sender_name, + correlation_id=correlation_id, + message=message, + result=result, + ) + return result + + if not sender_email: + return finalize( + ResumeMailboxResult( + sender_email=None, + sender_name=sender_name, + processed_attachments=0, + skipped_reason="missing_sender_email", + ) + ) + + if ( + self.settings.email_require_sender_auth_headers + and not self._has_authenticated_sender(message) + ): + return finalize( + ResumeMailboxResult( + sender_email=sender_email, + sender_name=sender_name, + processed_attachments=0, + skipped_reason="sender_authentication_failed", + ) + ) + + sender_is_authorized = await asyncio.to_thread( + self._sender_is_authorized, + sender_email, + ) + if not sender_is_authorized: + return finalize( + ResumeMailboxResult( + sender_email=sender_email, + sender_name=sender_name, + processed_attachments=0, + skipped_reason="sender_not_authorized", + ) + ) + + attachments = self._extract_resume_attachments(message) + if not attachments: + return finalize( + ResumeMailboxResult( + sender_email=sender_email, + sender_name=sender_name, + processed_attachments=0, + skipped_reason="no_resume_attachments", + ) + ) + + staging_contact = await asyncio.to_thread(self._find_or_create_staging_contact) + staging_contact_id = str(staging_contact.get("id", "")).strip() + if not staging_contact_id: + return finalize( + ResumeMailboxResult( + sender_email=sender_email, + sender_name=sender_name, + processed_attachments=0, + skipped_reason="staging_contact_id_missing", + ) + ) + + processed = 0 + for attachment in attachments: + if len(attachment.content) > self._max_attachment_size_bytes: + logger.warning( + "Skipping oversized resume attachment filename=%s size_bytes=%s sender=%s", + attachment.filename, + len(attachment.content), + sender_email, + ) + continue + + try: + ok = await self._process_attachment( + staging_contact_id=staging_contact_id, + attachment=attachment, + ) + except Exception as exc: + ok = False + logger.exception( + "Failed processing resume attachment staging_contact_id=%s filename=%s sender=%s error=%s", + staging_contact_id, + attachment.filename, + sender_email, + exc, + ) + + if ok: + processed += 1 + + skipped_reason = None + if processed == 0: + skipped_reason = "resume_processing_failed" + + return finalize( + ResumeMailboxResult( + sender_email=sender_email, + sender_name=sender_name, + processed_attachments=processed, + skipped_reason=skipped_reason, + ) + ) + + @property + def _max_attachment_size_bytes(self) -> int: + return max(1, self.settings.email_resume_max_file_size_mb) * 1024 * 1024 + + @property + def _allowed_resume_extensions(self) -> set[str]: + raw = self.settings.email_resume_allowed_extensions + values = {f".{item.strip().lower().lstrip('.')}" for item in raw.split(",")} + return {item for item in values if item != "."} + + def _sender_identity(self, message: Message) -> tuple[str | None, str | None]: + display_name, email_address = parseaddr(str(message.get("From", "")).strip()) + sender_name = display_name.strip() or None + sender_email = self._normalize_email(email_address) + return sender_name, sender_email + + def _has_authenticated_sender(self, message: Message) -> bool: + """Require pass results from SPF/DKIM/DMARC headers to reduce spoof risk.""" + auth_results = str(message.get("Authentication-Results", "")).lower() + received_spf = str(message.get("Received-SPF", "")).lower() + + dmarc_pass = "dmarc=pass" in auth_results + dkim_pass = "dkim=pass" in auth_results + spf_pass = "spf=pass" in auth_results or received_spf.startswith("pass") + return dmarc_pass or (dkim_pass and spf_pass) + + def _sender_is_authorized(self, sender_email: str) -> bool: + in_people_db = self._sender_has_privileged_role_in_people_db(sender_email) + if not in_people_db: + return False + in_crm = self._sender_has_privileged_role_in_crm(sender_email) + return in_crm + + def _sender_has_privileged_role_in_people_db(self, sender_email: str) -> bool: + query = """ + SELECT 1 + FROM people + WHERE sync_status = 'active' + AND (lower(email) = %s OR lower(email_508) = %s) + AND ( + COALESCE(discord_roles, '[]'::jsonb) ? 'Admin' + OR COALESCE(discord_roles, '[]'::jsonb) ? 'Steering Committee' + OR COALESCE(discord_roles, '[]'::jsonb) ? 'Owner' + ) + LIMIT 1; + """ + + with get_postgres_connection(self.settings) as conn: + with conn.cursor() as cursor: + cursor.execute(query, (sender_email, sender_email)) + row = cursor.fetchone() + return row is not None + + def _sender_has_privileged_role_in_crm(self, sender_email: str) -> bool: + sender_contact = self._find_contact_by_email(sender_email) + if sender_contact is None: + return False + + raw_roles = sender_contact.get("cDiscordRoles") + parsed_roles = self._parse_role_names(raw_roles) + return any(role in PRIVILEGED_ROLE_NAMES for role in parsed_roles) + + def _parse_role_names(self, raw_roles: Any) -> set[str]: + parsed: list[str] = [] + + if isinstance(raw_roles, list): + parsed = [str(item).strip() for item in raw_roles] + elif isinstance(raw_roles, str): + parsed = [item.strip() for item in raw_roles.split(",")] + elif isinstance(raw_roles, dict): + parsed = [str(value).strip() for value in raw_roles.values()] + + return {value.casefold() for value in parsed if value} + + def _find_contact_by_email(self, sender_email: str) -> dict[str, Any] | None: + search_params = { + "where": [ + { + "type": "or", + "value": [ + { + "type": "equals", + "attribute": "emailAddress", + "value": sender_email, + }, + { + "type": "equals", + "attribute": "c508Email", + "value": sender_email, + }, + ], + } + ], + "maxSize": 1, + "select": "id,name,emailAddress,c508Email,cDiscordRoles", + } + + try: + response = self.espo_api.request("GET", "Contact", search_params) + except EspoAPIError as exc: + logger.warning( + "CRM contact lookup by email failed email=%s error=%s", + sender_email, + exc, + ) + return None + + contacts = response.get("list", []) + if not isinstance(contacts, list) or not contacts: + return None + + first = contacts[0] + return first if isinstance(first, dict) else None + + def _create_contact_for_email( + self, + email_address: str, + display_name: str | None, + ) -> dict[str, Any]: + """Create a fallback contact when no existing CRM record can be resolved.""" + local_part = email_address.split("@", 1)[0] + fallback_name = local_part.replace(".", " ").replace("_", " ").strip().title() + payload: dict[str, Any] = { + "name": display_name or fallback_name or "Resume Intake", + } + if email_address.endswith("@508.dev"): + payload["c508Email"] = email_address + else: + payload["emailAddress"] = email_address + + return self.espo_api.request("POST", "Contact", payload) + + def _find_or_create_staging_contact(self) -> dict[str, Any]: + """Resolve a stable non-sender contact used only for initial extraction.""" + staging_email = self._normalize_email(self.settings.email_username) + if not staging_email: + raise ValueError("EMAIL_USERNAME is required for staging contact lookup") + + existing = self._find_contact_by_email(staging_email) + if existing is not None: + return existing + + return self._create_contact_for_email( + staging_email, + "Resume Intake Staging", + ) + + def _extract_resume_attachments(self, message: Message) -> list[ResumeAttachment]: + attachments: list[ResumeAttachment] = [] + allowed_extensions = self._allowed_resume_extensions + + for part in message.walk(): + filename = part.get_filename() + if not filename: + continue + + extension = self._file_extension(filename) + if extension not in allowed_extensions: + continue + + payload = part.get_payload(decode=True) + if not isinstance(payload, (bytes, bytearray)) or not payload: + continue + + attachments.append( + ResumeAttachment(filename=filename, content=bytes(payload)) + ) + + return attachments + + async def _process_attachment( + self, + *, + staging_contact_id: str, + attachment: ResumeAttachment, + ) -> bool: + staging_attachment_id = await asyncio.to_thread( + self._upload_contact_resume, + staging_contact_id, + attachment, + ) + if not staging_attachment_id: + return False + + staging_extract_job_id = await self._enqueue_resume_extract_job( + contact_id=staging_contact_id, + attachment_id=staging_attachment_id, + filename=attachment.filename, + ) + staging_extract_job = await self._wait_for_worker_job_result( + staging_extract_job_id + ) + if staging_extract_job is None: + return False + + staging_status = str(staging_extract_job.get("status", "")) + if staging_status != "succeeded": + return False + + staging_extract_result = staging_extract_job.get("result") + if not isinstance(staging_extract_result, dict): + return False + + if not bool(staging_extract_result.get("success", False)): + return False + + candidate_email = self._candidate_email_from_extract_result( + staging_extract_result + ) + if not candidate_email: + logger.info( + "Skipping resume attachment filename=%s due to missing candidate email in extraction", + attachment.filename, + ) + return False + + candidate_contact = await asyncio.to_thread( + self._find_contact_by_email, + candidate_email, + ) + if candidate_contact is None: + candidate_contact = await asyncio.to_thread( + self._create_contact_for_email, + candidate_email, + None, + ) + + candidate_contact_id = str(candidate_contact.get("id", "")).strip() + if not candidate_contact_id: + return False + + candidate_attachment_id = await asyncio.to_thread( + self._upload_contact_resume, + candidate_contact_id, + attachment, + ) + if not candidate_attachment_id: + return False + + candidate_link_ok = await asyncio.to_thread( + self._append_contact_resume, + candidate_contact_id, + candidate_attachment_id, + ) + if not candidate_link_ok: + return False + + candidate_extract_job_id = await self._enqueue_resume_extract_job( + contact_id=candidate_contact_id, + attachment_id=candidate_attachment_id, + filename=attachment.filename, + ) + candidate_extract_job = await self._wait_for_worker_job_result( + candidate_extract_job_id + ) + if candidate_extract_job is None: + return False + + candidate_status = str(candidate_extract_job.get("status", "")) + if candidate_status != "succeeded": + return False + + candidate_extract_result = candidate_extract_job.get("result") + if not isinstance(candidate_extract_result, dict): + return False + + if not bool(candidate_extract_result.get("success", False)): + return False + + proposed_updates_raw = candidate_extract_result.get("proposed_updates") + if not isinstance(proposed_updates_raw, dict) or not proposed_updates_raw: + return True + + proposed_updates = { + str(field): str(value) + for field, value in proposed_updates_raw.items() + if value is not None and str(value).strip() + } + if not proposed_updates: + return True + + apply_job_id = await self._enqueue_resume_apply_job( + contact_id=candidate_contact_id, + updates=proposed_updates, + ) + apply_job = await self._wait_for_worker_job_result(apply_job_id) + if apply_job is None: + return False + + apply_status = str(apply_job.get("status", "")) + if apply_status != "succeeded": + return False + + apply_result = apply_job.get("result") + if not isinstance(apply_result, dict): + return False + + return bool(apply_result.get("success", False)) + + def _candidate_email_from_extract_result( + self, extract_result: dict[str, Any] + ) -> str | None: + extracted_profile_raw = extract_result.get("extracted_profile") + if isinstance(extracted_profile_raw, dict): + email_value = self._normalize_email( + str(extracted_profile_raw.get("email", "")).strip() + ) + if email_value: + return email_value + + proposed_updates = extract_result.get("proposed_updates") + if isinstance(proposed_updates, dict): + email_value = self._normalize_email( + str(proposed_updates.get("emailAddress", "")).strip() + ) + if email_value: + return email_value + + return None + + def _upload_contact_resume( + self, + contact_id: str, + attachment: ResumeAttachment, + ) -> str | None: + try: + uploaded = self.espo_api.upload_file( + file_content=attachment.content, + filename=attachment.filename, + related_type="Contact", + related_id=contact_id, + field="resume", + ) + except EspoAPIError as exc: + logger.warning( + "Failed uploading resume to CRM contact_id=%s filename=%s error=%s", + contact_id, + attachment.filename, + exc, + ) + return None + + attachment_id = uploaded.get("id") + if not isinstance(attachment_id, str) or not attachment_id.strip(): + return None + return attachment_id + + def _append_contact_resume(self, contact_id: str, attachment_id: str) -> bool: + try: + contact = self.espo_api.request("GET", f"Contact/{contact_id}") + current_resume_ids = contact.get("resumeIds", []) + if not isinstance(current_resume_ids, list): + current_resume_ids = [] + + if attachment_id not in current_resume_ids: + current_resume_ids.append(attachment_id) + + self.espo_api.request( + "PUT", + f"Contact/{contact_id}", + {"resumeIds": current_resume_ids}, + ) + return True + except EspoAPIError as exc: + logger.warning( + "Failed linking resume attachment in CRM contact_id=%s attachment_id=%s error=%s", + contact_id, + attachment_id, + exc, + ) + return False + + async def _enqueue_resume_extract_job( + self, + *, + contact_id: str, + attachment_id: str, + filename: str, + ) -> str: + payload = { + "contact_id": contact_id, + "attachment_id": attachment_id, + "filename": filename, + } + return await self._enqueue_worker_job("/jobs/resume-extract", payload) + + async def _enqueue_resume_apply_job( + self, + *, + contact_id: str, + updates: dict[str, str], + ) -> str: + payload = { + "contact_id": contact_id, + "updates": updates, + "link_discord": None, + } + return await self._enqueue_worker_job("/jobs/resume-apply", payload) + + async def _enqueue_worker_job(self, path: str, payload: dict[str, Any]) -> str: + async with aiohttp.ClientSession() as session: + async with session.post( + self._worker_url(path), + headers=self._worker_headers(), + json=payload, + timeout=aiohttp.ClientTimeout(total=30), + ) as response: + data = await response.json() + if response.status != 202: + raise ValueError(f"Worker enqueue failed path={path}: {data}") + job_id = data.get("job_id") + if not isinstance(job_id, str) or not job_id.strip(): + raise ValueError("Missing worker job_id in response") + return job_id + + def _worker_headers(self) -> dict[str, str]: + if not self.settings.api_shared_secret: + raise ValueError("API_SHARED_SECRET is required for worker API requests") + return { + "X-API-Secret": self.settings.api_shared_secret, + "Content-Type": "application/json", + } + + def _worker_url(self, path: str) -> str: + return f"{self.settings.worker_api_base_url.rstrip('/')}{path}" + + async def _wait_for_worker_job_result( + self, + job_id: str, + *, + timeout_seconds: int = 180, + poll_seconds: float = 2.0, + ) -> dict[str, Any] | None: + terminal = {"succeeded", "dead", "canceled"} + max_attempts = max(1, int(timeout_seconds / poll_seconds)) + + for _ in range(max_attempts): + job = await self._get_worker_job_status(job_id) + status = str(job.get("status", "")) + if status in terminal: + return job + await asyncio.sleep(poll_seconds) + + return None + + async def _get_worker_job_status(self, job_id: str) -> dict[str, Any]: + async with aiohttp.ClientSession() as session: + async with session.get( + self._worker_url(f"/jobs/{job_id}"), + headers=self._worker_headers(), + timeout=aiohttp.ClientTimeout(total=30), + ) as response: + data = await response.json() + if response.status != 200: + raise ValueError(f"Worker job status failed: {data}") + if not isinstance(data, dict): + raise ValueError("Worker job status response must be an object") + return data + + def _file_extension(self, filename: str) -> str: + if "." not in filename: + return "" + return "." + filename.rsplit(".", 1)[-1].lower().strip() + + def _normalize_email(self, value: str | None) -> str | None: + if not value: + return None + normalized = value.strip().lower() + return normalized or None + + def _mailbox_correlation_id(self, message: Message) -> str: + message_id = str(message.get("Message-ID", "")).strip() + if message_id: + return message_id + return f"mailbox-{uuid4()}" + + def _audit_mailbox_outcome( + self, + *, + sender_email: str | None, + sender_name: str | None, + correlation_id: str, + message: Message, + result: ResumeMailboxResult, + ) -> None: + if not sender_email: + return + + audit_result = "error" + if result.skipped_reason in { + "sender_not_authorized", + "sender_authentication_failed", + }: + audit_result = "denied" + elif result.skipped_reason in {None, "no_resume_attachments"}: + audit_result = "success" + + metadata = { + "subject": str(message.get("Subject", "")).strip() or None, + "mailbox_username": self.settings.email_username, + "processed_attachments": result.processed_attachments, + "skipped_reason": result.skipped_reason, + } + + self.audit_logger.log_admin_sso_action( + action="crm.resume_mailbox_ingest", + result=audit_result, + actor_email=sender_email, + actor_display_name=sender_name, + metadata=metadata, + resource_type="mailbox_message", + resource_id=correlation_id, + correlation_id=correlation_id, + ) diff --git a/tests/unit/test_discord_audit.py b/tests/unit/test_discord_audit.py index 5871ceca..2acf854d 100644 --- a/tests/unit/test_discord_audit.py +++ b/tests/unit/test_discord_audit.py @@ -40,7 +40,7 @@ def test_send_event_sync_logs_warning_on_request_error() -> None: shared_secret="secret", timeout_seconds=1.0, ) - payload = logger._build_payload( + payload = logger._build_discord_payload( interaction=_mock_interaction(), action="crm.search_members", result="success", @@ -55,3 +55,30 @@ def test_send_event_sync_logs_warning_on_request_error() -> None: logger._send_event_sync(payload) mock_warning.assert_called_once() + + +def test_log_admin_sso_action_normalizes_actor_email() -> None: + """Admin SSO audit should normalize actor email and queue the event.""" + logger = DiscordAuditLogger( + base_url="http://worker-api:8090", + shared_secret="secret", + timeout_seconds=1.0, + ) + + with patch.object(logger, "_queue_event") as mock_queue: + logger.log_admin_sso_action( + action="crm.resume_mailbox_ingest", + result="success", + actor_email=" Admin@Example.COM ", + actor_display_name="Admin User", + metadata={"processed_attachments": 1}, + resource_type="mailbox_message", + resource_id="", + correlation_id="", + ) + + mock_queue.assert_called_once() + payload = mock_queue.call_args.args[0] + assert payload["source"] == "admin_dashboard" + assert payload["actor_provider"] == "admin_sso" + assert payload["actor_subject"] == "admin@example.com" diff --git a/tests/unit/test_resume_mail_ingest.py b/tests/unit/test_resume_mail_ingest.py new file mode 100644 index 00000000..1c8b566a --- /dev/null +++ b/tests/unit/test_resume_mail_ingest.py @@ -0,0 +1,152 @@ +"""Unit tests for mailbox-driven resume ingestion.""" + +from email.message import EmailMessage +from unittest.mock import AsyncMock, Mock + +import pytest + +from five08.discord_bot.config import settings +from five08.discord_bot.utils.resume_mail_ingest import ( + ResumeAttachment, + ResumeMailboxProcessor, +) + + +def _build_message(*, include_attachment: bool = True) -> EmailMessage: + message = EmailMessage() + message["From"] = "Admin User " + message["Subject"] = "Resume upload" + message["Authentication-Results"] = "mx.example; dkim=pass; spf=pass; dmarc=pass" + message.set_content("Please process this resume.") + + if include_attachment: + message.add_attachment( + b"resume-bytes", + maintype="application", + subtype="pdf", + filename="resume.pdf", + ) + + return message + + +@pytest.mark.asyncio +async def test_process_message_happy_path_uses_existing_contact() -> None: + """Authorized senders with resume attachments should process successfully.""" + processor = ResumeMailboxProcessor(settings) + processor._sender_is_authorized = Mock(return_value=True) + processor._find_or_create_staging_contact = Mock(return_value={"id": "staging-1"}) + processor._create_contact_for_email = Mock(return_value={"id": "contact-new"}) + processor._process_attachment = AsyncMock(return_value=True) + + result = await processor.process_message(_build_message()) + + assert result.skipped_reason is None + assert result.processed_attachments == 1 + processor._find_or_create_staging_contact.assert_called_once() + processor._create_contact_for_email.assert_not_called() + processor._process_attachment.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_process_message_skips_when_sender_not_authorized() -> None: + """Unauthorized senders should be rejected before any attachment processing.""" + processor = ResumeMailboxProcessor(settings) + processor._sender_is_authorized = Mock(return_value=False) + processor._process_attachment = AsyncMock(return_value=True) + processor.audit_logger = Mock() + + result = await processor.process_message(_build_message()) + + assert result.skipped_reason == "sender_not_authorized" + assert result.processed_attachments == 0 + processor._process_attachment.assert_not_called() + processor.audit_logger.log_admin_sso_action.assert_called_once() + payload = processor.audit_logger.log_admin_sso_action.call_args.kwargs + assert payload["result"] == "denied" + + +@pytest.mark.asyncio +async def test_process_message_requires_auth_headers_when_enabled() -> None: + """Spoof-resistant mode should reject messages without pass auth headers.""" + processor = ResumeMailboxProcessor(settings) + processor._sender_is_authorized = Mock(return_value=True) + + message = _build_message() + del message["Authentication-Results"] + + result = await processor.process_message(message) + + assert result.skipped_reason == "sender_authentication_failed" + processor._sender_is_authorized.assert_not_called() + + +@pytest.mark.asyncio +async def test_process_message_creates_contact_when_lookup_misses() -> None: + """A staging contact should be created/resolved once per processed message.""" + processor = ResumeMailboxProcessor(settings) + processor._sender_is_authorized = Mock(return_value=True) + processor._find_or_create_staging_contact = Mock(return_value={"id": "staging-1"}) + processor._process_attachment = AsyncMock(return_value=True) + + result = await processor.process_message(_build_message()) + + assert result.skipped_reason is None + assert result.processed_attachments == 1 + processor._find_or_create_staging_contact.assert_called_once() + + +@pytest.mark.asyncio +async def test_process_attachment_updates_candidate_not_sender() -> None: + """Attachment processing should resolve candidate email and apply to candidate contact.""" + processor = ResumeMailboxProcessor(settings) + processor._upload_contact_resume = Mock( + side_effect=["att-staging", "att-candidate"] + ) + processor._append_contact_resume = Mock(return_value=True) + processor._enqueue_resume_extract_job = AsyncMock( + side_effect=["extract-staging", "extract-candidate"] + ) + processor._enqueue_resume_apply_job = AsyncMock(return_value="apply-candidate") + processor._wait_for_worker_job_result = AsyncMock( + side_effect=[ + { + "status": "succeeded", + "result": { + "success": True, + "extracted_profile": {"email": "candidate@example.com"}, + "proposed_updates": {}, + }, + }, + { + "status": "succeeded", + "result": { + "success": True, + "proposed_updates": {"phoneNumber": "14155551234"}, + }, + }, + { + "status": "succeeded", + "result": { + "success": True, + "updated_fields": ["phoneNumber"], + }, + }, + ] + ) + processor._find_contact_by_email = Mock(return_value=None) + processor._create_contact_for_email = Mock(return_value={"id": "candidate-1"}) + + ok = await processor._process_attachment( + staging_contact_id="staging-1", + attachment=ResumeAttachment(filename="resume.pdf", content=b"resume-bytes"), + ) + + assert ok is True + processor._create_contact_for_email.assert_called_once_with( + "candidate@example.com", None + ) + processor._enqueue_resume_apply_job.assert_awaited_once_with( + contact_id="candidate-1", + updates={"phoneNumber": "14155551234"}, + ) From 567a39a4030ee31c999d7b9b0996041dc8ee16bc Mon Sep 17 00:00:00 2001 From: Michael Wu Date: Sat, 21 Feb 2026 17:00:00 +0800 Subject: [PATCH 2/2] refactor: move mailbox resume intake to worker --- .env.example | 7 +- README.md | 18 +- .../five08/discord_bot/cogs/email_monitor.py | 133 +------ .../src/five08/discord_bot/config.py | 11 +- apps/worker/src/five08/worker/api.py | 27 ++ apps/worker/src/five08/worker/config.py | 30 ++ .../five08/worker/mailbox_resume_ingest.py} | 364 ++++++++---------- tests/integration/test_email_monitor.py | 85 ++-- tests/unit/test_resume_mail_ingest.py | 152 -------- tests/unit/test_worker_config.py | 31 ++ .../unit/test_worker_mailbox_resume_ingest.py | 118 ++++++ 11 files changed, 401 insertions(+), 575 deletions(-) rename apps/{discord_bot/src/five08/discord_bot/utils/resume_mail_ingest.py => worker/src/five08/worker/mailbox_resume_ingest.py} (65%) delete mode 100644 tests/unit/test_resume_mail_ingest.py create mode 100644 tests/unit/test_worker_config.py create mode 100644 tests/unit/test_worker_mailbox_resume_ingest.py diff --git a/.env.example b/.env.example index 7eb197e1..a69802cc 100644 --- a/.env.example +++ b/.env.example @@ -66,19 +66,18 @@ CRM_SYNC_PAGE_SIZE=200 DISCORD_BOT_TOKEN=your_bot_token_here HEALTHCHECK_PORT=3000 DISCORD_SENDMSG_CHARACTER_LIMIT=2000 -CHECK_EMAIL_WAIT=2 WORKER_API_BASE_URL=http://worker-api:8090 AUDIT_API_BASE_URL= AUDIT_API_TIMEOUT_SECONDS=2.0 # Required for Discord bot commands that post to channels CHANNEL_ID=1391742724666822798 -# Email monitoring (required if email features are enabled) +# Worker mailbox resume intake (required if email intake is enabled) +CHECK_EMAIL_WAIT=2 EMAIL_USERNAME=your_email@example.com EMAIL_PASSWORD=your_app_password IMAP_SERVER=imap.migadu.com -SMTP_SERVER=smtp.migadu.com -EMAIL_RESUME_INTAKE_ENABLED=true +EMAIL_RESUME_INTAKE_ENABLED=false EMAIL_RESUME_ALLOWED_EXTENSIONS=pdf,doc,docx EMAIL_RESUME_MAX_FILE_SIZE_MB=10 EMAIL_REQUIRE_SENDER_AUTH_HEADERS=true diff --git a/README.md b/README.md index 89b32dfb..2e9cf297 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,7 @@ Use `.env.example` as the source of truth for defaults. - `Optional`: `CRM_SYNC_ENABLED` (default: `true`) - `Optional`: `CRM_SYNC_INTERVAL_SECONDS` (default: `900`) - `Optional`: `CRM_SYNC_PAGE_SIZE` (default: `200`) +- `Optional`: `CHECK_EMAIL_WAIT` (default: `2`; minutes between mailbox polls) - `Optional`: `CRM_LINKEDIN_FIELD` (default: `cLinkedInUrl`) - `Optional`: `MAX_ATTACHMENTS_PER_CONTACT` (default: `3`) - `Optional`: `MAX_FILE_SIZE_MB` (default: `10`) @@ -162,6 +163,11 @@ Use `.env.example` as the source of truth for defaults. - `Optional`: `RESUME_AI_MODEL` (default: `gpt-4o-mini`; use plain names like `gpt-4o-mini`, OpenRouter gets auto-prefixed to `openai/`) - `Optional`: `OPENAI_MODEL` (default: `gpt-4o-mini`; fallback/legacy model setting) - `Optional`: `RESUME_EXTRACTOR_VERSION` (default: `v1`; used in resume processing idempotency/ledger keys) +- `Optional`: `EMAIL_RESUME_INTAKE_ENABLED` (default: `false`; enables worker-side mailbox resume processing loop) +- `Optional`: `EMAIL_RESUME_ALLOWED_EXTENSIONS` (default: `pdf,doc,docx`) +- `Optional`: `EMAIL_RESUME_MAX_FILE_SIZE_MB` (default: `10`) +- `Optional`: `EMAIL_REQUIRE_SENDER_AUTH_HEADERS` (default: `true`; requires SPF/DKIM/DMARC pass headers) +- `Required when EMAIL_RESUME_INTAKE_ENABLED=true`: `EMAIL_USERNAME`, `EMAIL_PASSWORD`, `IMAP_SERVER` ### Discord Bot Core @@ -170,18 +176,6 @@ Use `.env.example` as the source of truth for defaults. - `Optional`: `WORKER_API_BASE_URL` (default: `http://worker-api:8090`) - `Optional`: `HEALTHCHECK_PORT` (default: `3000`) - `Optional`: `DISCORD_SENDMSG_CHARACTER_LIMIT` (default: `2000`) -- `Optional`: `CHECK_EMAIL_WAIT` (default: `2`) - -### Discord Email Monitoring - -- `Required`: `EMAIL_USERNAME` -- `Required`: `EMAIL_PASSWORD` -- `Required`: `IMAP_SERVER` -- `Required`: `SMTP_SERVER` -- `Optional`: `EMAIL_RESUME_INTAKE_ENABLED` (default: `true`; enables mailbox resume processing loop) -- `Optional`: `EMAIL_RESUME_ALLOWED_EXTENSIONS` (default: `pdf,doc,docx`) -- `Optional`: `EMAIL_RESUME_MAX_FILE_SIZE_MB` (default: `10`) -- `Optional`: `EMAIL_REQUIRE_SENDER_AUTH_HEADERS` (default: `true`; requires SPF/DKIM/DMARC pass headers) ### Discord CRM Audit Logging (Best Effort) diff --git a/apps/discord_bot/src/five08/discord_bot/cogs/email_monitor.py b/apps/discord_bot/src/five08/discord_bot/cogs/email_monitor.py index a9fd290e..12744cc7 100644 --- a/apps/discord_bot/src/five08/discord_bot/cogs/email_monitor.py +++ b/apps/discord_bot/src/five08/discord_bot/cogs/email_monitor.py @@ -1,144 +1,29 @@ -"""Email monitoring cog for mailbox-driven resume intake workflows.""" +"""Compatibility cog for legacy bot-side email monitoring. + +Mailbox resume ingestion now runs in the worker service. +""" from __future__ import annotations -import contextlib -import email -import imaplib import logging -from typing import Any - -from discord.ext import commands, tasks -from five08.discord_bot.config import settings -from five08.discord_bot.utils.resume_mail_ingest import ( - ResumeMailboxProcessor, - ResumeMailboxResult, -) +from discord.ext import commands logger = logging.getLogger(__name__) class EmailMonitor(commands.Cog): - """Poll an IMAP inbox and run resume intake on new messages.""" + """Deprecated placeholder cog kept for compatibility/visibility in health checks.""" def __init__(self, bot: commands.Bot) -> None: self.bot = bot - self.resume_processor = ResumeMailboxProcessor(settings) async def cog_load(self) -> None: - """Start polling task when this cog is loaded.""" - if ( - settings.email_resume_intake_enabled - and not self.task_poll_inbox.is_running() - ): - self.task_poll_inbox.start() - - async def cog_unload(self) -> None: - """Cancel the background task when cog is unloaded.""" - self.task_poll_inbox.cancel() - - @tasks.loop(minutes=settings.check_email_wait) - async def task_poll_inbox(self) -> None: - """Poll IMAP inbox for new messages and process resume attachments.""" - logger.info("Reading inbox of %s", settings.email_username) - mail = imaplib.IMAP4_SSL(settings.imap_server) - try: - mail.login(settings.email_username, settings.email_password) - mail.select("INBOX") - retcode, message_batches = mail.search(None, "(UNSEEN)") - - if retcode != "OK" or not message_batches or not message_batches[0]: - logger.debug("Login complete, # of new messages: 0") - return - - message_ids = message_batches[0].split() - total_messages = len(message_ids) - - for index, raw_num in enumerate(message_ids, start=1): - num = raw_num.decode() - typ, data = mail.fetch(num, "(RFC822)") - if typ != "OK": - logger.warning( - "Skipping message %s due to failed fetch status=%s", num, typ - ) - continue - - result: ResumeMailboxResult - try: - result = await self._process_fetched_message(data) - except Exception as exc: - logger.exception( - "Failed processing inbound email num=%s error=%s", num, exc - ) - result = ResumeMailboxResult( - sender_email=None, - sender_name=None, - processed_attachments=0, - skipped_reason="message_processing_error", - ) - - self._log_processing_result( - result=result, index=index, total=total_messages - ) - - # Mark mail as seen to avoid reprocessing on next poll. - mail.store(num, "+FLAGS", "\\Seen") - - logger.debug("Login complete, # of new messages: %s", total_messages) - finally: - with contextlib.suppress(Exception): - mail.close() - with contextlib.suppress(Exception): - mail.logout() - - async def _process_fetched_message(self, data: list[Any]) -> ResumeMailboxResult: - for response_part in data: - if not isinstance(response_part, tuple): - continue - - raw_payload = response_part[1] - if not isinstance(raw_payload, (bytes, bytearray)): - continue - - message = email.message_from_bytes(bytes(raw_payload)) - return await self.resume_processor.process_message(message) - - return ResumeMailboxResult( - sender_email=None, - sender_name=None, - processed_attachments=0, - skipped_reason="message_payload_missing", - ) - - def _log_processing_result( - self, - *, - result: ResumeMailboxResult, - index: int, - total: int, - ) -> None: - sender = result.sender_email or "unknown" - if result.skipped_reason: - logger.info( - "Resume inbox message %s/%s sender=%s skipped reason=%s", - index, - total, - sender, - result.skipped_reason, - ) - return - logger.info( - "Resume inbox message %s/%s sender=%s processed_attachments=%s", - index, - total, - sender, - result.processed_attachments, + "EmailMonitor cog loaded (mailbox intake now runs in worker service)" ) async def setup(bot: commands.Bot) -> None: - """Add the EmailMonitor cog to the bot.""" - cog = EmailMonitor(bot) - await bot.add_cog(cog) + """Add the EmailMonitor compatibility cog to the bot.""" + await bot.add_cog(EmailMonitor(bot)) diff --git a/apps/discord_bot/src/five08/discord_bot/config.py b/apps/discord_bot/src/five08/discord_bot/config.py index 71a89b83..b5374cee 100644 --- a/apps/discord_bot/src/five08/discord_bot/config.py +++ b/apps/discord_bot/src/five08/discord_bot/config.py @@ -23,17 +23,8 @@ class Settings(SharedSettings): # Healthcheck Configuration healthcheck_port: int = 3000 - # Email Monitoring Configuration + # Core channel configuration channel_id: int - check_email_wait: int = 2 - email_username: str - email_password: str - imap_server: str - smtp_server: str - email_resume_intake_enabled: bool = True - email_resume_allowed_extensions: str = "pdf,doc,docx" - email_resume_max_file_size_mb: int = 10 - email_require_sender_auth_headers: bool = True # CRM/EspoCRM settings espo_api_key: str diff --git a/apps/worker/src/five08/worker/api.py b/apps/worker/src/five08/worker/api.py index 2eddb023..de2e9cbb 100644 --- a/apps/worker/src/five08/worker/api.py +++ b/apps/worker/src/five08/worker/api.py @@ -33,6 +33,7 @@ from five08.worker.config import settings from five08.worker.db_migrations import run_job_migrations from five08.worker.dispatcher import build_queue_client +from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor from five08.worker.jobs import ( apply_resume_profile_job, extract_resume_profile_job, @@ -47,6 +48,7 @@ REDIS_CONN_KEY = web.AppKey("redis_conn", Redis) QUEUE_KEY = web.AppKey("queue", QueueClient) CRM_SYNC_TASK_KEY = web.AppKey("crm_sync_task", asyncio.Task) +EMAIL_RESUME_TASK_KEY = web.AppKey("email_resume_task", asyncio.Task) POSTGRES_CONN_KEY = web.AppKey("postgres_conn", Connection) POSTGRES_CONN_LOCK_KEY = web.AppKey("postgres_conn_lock", asyncio.Lock) @@ -125,6 +127,22 @@ async def _crm_sync_scheduler(app: web.Application) -> None: await asyncio.sleep(interval_seconds) +async def _email_resume_scheduler() -> None: + """Run periodic mailbox polling for resume ingestion.""" + poller = ResumeMailboxProcessor(settings) + interval_seconds = max(1, settings.check_email_wait) * 60 + while True: + try: + processed_count = await asyncio.to_thread(poller.poll_inbox) + logger.debug( + "Completed mailbox resume poll processed_attachments=%s", + processed_count, + ) + except Exception: + logger.exception("Failed mailbox resume poll iteration") + await asyncio.sleep(interval_seconds) + + def _check_postgres_connection(connection: Connection) -> bool: try: with connection.cursor() as cursor: @@ -594,6 +612,10 @@ async def on_startup(app: web.Application) -> None: app[CRM_SYNC_TASK_KEY] = asyncio.create_task(_crm_sync_scheduler(app)) else: logger.info("CRM sync scheduler disabled by config") + if settings.email_resume_intake_enabled: + app[EMAIL_RESUME_TASK_KEY] = asyncio.create_task(_email_resume_scheduler()) + else: + logger.info("Mailbox resume intake scheduler disabled by config") async def on_cleanup(app: web.Application) -> None: @@ -607,6 +629,11 @@ async def on_cleanup(app: web.Application) -> None: task.cancel() with contextlib.suppress(asyncio.CancelledError): await task + if EMAIL_RESUME_TASK_KEY in app: + task = app[EMAIL_RESUME_TASK_KEY] + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task def create_app() -> web.Application: diff --git a/apps/worker/src/five08/worker/config.py b/apps/worker/src/five08/worker/config.py index 17284d0e..bc25f748 100644 --- a/apps/worker/src/five08/worker/config.py +++ b/apps/worker/src/five08/worker/config.py @@ -2,6 +2,8 @@ from urllib.parse import urlparse +from pydantic import model_validator + from five08.settings import SharedSettings @@ -29,6 +31,34 @@ class WorkerSettings(SharedSettings): crm_sync_enabled: bool = True crm_sync_interval_seconds: int = 900 crm_sync_page_size: int = 200 + email_resume_intake_enabled: bool = False + check_email_wait: int = 2 + email_username: str | None = None + email_password: str | None = None + imap_server: str | None = None + email_resume_allowed_extensions: str = "pdf,doc,docx" + email_resume_max_file_size_mb: int = 10 + email_require_sender_auth_headers: bool = True + + @model_validator(mode="after") + def validate_email_resume_intake_settings(self) -> "WorkerSettings": + """Require mailbox settings when worker-side email intake is enabled.""" + if not self.email_resume_intake_enabled: + return self + + if not (self.email_username or "").strip(): + raise ValueError( + "EMAIL_USERNAME must be set when EMAIL_RESUME_INTAKE_ENABLED=true" + ) + if not (self.email_password or "").strip(): + raise ValueError( + "EMAIL_PASSWORD must be set when EMAIL_RESUME_INTAKE_ENABLED=true" + ) + if not (self.imap_server or "").strip(): + raise ValueError( + "IMAP_SERVER must be set when EMAIL_RESUME_INTAKE_ENABLED=true" + ) + return self @property def allowed_file_extensions(self) -> set[str]: diff --git a/apps/discord_bot/src/five08/discord_bot/utils/resume_mail_ingest.py b/apps/worker/src/five08/worker/mailbox_resume_ingest.py similarity index 65% rename from apps/discord_bot/src/five08/discord_bot/utils/resume_mail_ingest.py rename to apps/worker/src/five08/worker/mailbox_resume_ingest.py index 05eb6f05..341db85d 100644 --- a/apps/discord_bot/src/five08/discord_bot/utils/resume_mail_ingest.py +++ b/apps/worker/src/five08/worker/mailbox_resume_ingest.py @@ -1,8 +1,10 @@ -"""Mailbox-based resume intake helpers for automated CRM updates.""" +"""Worker-side IMAP resume ingestion pipeline.""" from __future__ import annotations -import asyncio +import contextlib +import email +import imaplib import logging from dataclasses import dataclass from email.message import Message @@ -10,22 +12,26 @@ from typing import Any from uuid import uuid4 -import aiohttp - +from five08.audit import ( + ActorProvider, + AuditEventInput, + AuditResult, + AuditSource, + insert_audit_event, +) from five08.clients.espo import EspoAPI, EspoAPIError -from five08.discord_bot.config import Settings -from five08.discord_bot.utils.audit import DiscordAuditLogger from five08.queue import get_postgres_connection +from five08.worker.config import WorkerSettings +from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor logger = logging.getLogger(__name__) - PRIVILEGED_ROLE_NAMES = {"admin", "steering committee", "owner"} @dataclass(frozen=True) class ResumeAttachment: - """One resume-like email attachment.""" + """One resume-like email attachment payload.""" filename: str content: bytes @@ -33,7 +39,7 @@ class ResumeAttachment: @dataclass(frozen=True) class ResumeMailboxResult: - """Result metadata for one processed mailbox message.""" + """Result metadata for one mailbox message.""" sender_email: str | None sender_name: str | None @@ -42,20 +48,89 @@ class ResumeMailboxResult: class ResumeMailboxProcessor: - """Process inbound mailbox messages into CRM resume extraction/apply flow.""" + """Poll mailbox and apply resume extraction updates to candidate contacts.""" - def __init__(self, settings: Settings) -> None: + def __init__(self, settings: WorkerSettings) -> None: self.settings = settings api_url = settings.espo_base_url.rstrip("/") + "/api/v1" self.espo_api = EspoAPI(api_url, settings.espo_api_key) - self.audit_logger = DiscordAuditLogger( - base_url=settings.audit_api_base_url, - shared_secret=settings.api_shared_secret, - timeout_seconds=settings.audit_api_timeout_seconds, + self.resume_processor = ResumeProfileProcessor() + + def poll_inbox(self) -> int: + """Process one IMAP poll cycle and return successfully processed attachment count.""" + email_username = (self.settings.email_username or "").strip() + email_password = (self.settings.email_password or "").strip() + imap_server = (self.settings.imap_server or "").strip() + + if not email_username or not email_password or not imap_server: + logger.warning( + "Skipping mailbox poll because mailbox settings are incomplete" + ) + return 0 + + processed_total = 0 + mail = imaplib.IMAP4_SSL(imap_server) + try: + mail.login(email_username, email_password) + mail.select("INBOX") + retcode, message_batches = mail.search(None, "(UNSEEN)") + if retcode != "OK" or not message_batches or not message_batches[0]: + logger.debug("Mailbox poll complete, no unseen messages") + return 0 + + for raw_num in message_batches[0].split(): + num = raw_num.decode() + typ, data = mail.fetch(num, "(RFC822)") + if typ != "OK": + logger.warning( + "Skipping mailbox message %s due to fetch status=%s", num, typ + ) + continue + + try: + result = self._process_fetched_message(data) + except Exception as exc: + logger.exception( + "Failed processing mailbox message num=%s error=%s", num, exc + ) + result = ResumeMailboxResult( + sender_email=None, + sender_name=None, + processed_attachments=0, + skipped_reason="message_processing_error", + ) + + processed_total += result.processed_attachments + mail.store(num, "+FLAGS", "\\Seen") + + return processed_total + finally: + with contextlib.suppress(Exception): + mail.close() + with contextlib.suppress(Exception): + mail.logout() + + def _process_fetched_message(self, data: list[Any]) -> ResumeMailboxResult: + for response_part in data: + if not isinstance(response_part, tuple): + continue + + raw_payload = response_part[1] + if not isinstance(raw_payload, (bytes, bytearray)): + continue + + message = email.message_from_bytes(bytes(raw_payload)) + return self.process_message(message) + + return ResumeMailboxResult( + sender_email=None, + sender_name=None, + processed_attachments=0, + skipped_reason="message_payload_missing", ) - async def process_message(self, message: Message) -> ResumeMailboxResult: - """Process one email message and trigger resume extraction/apply jobs.""" + def process_message(self, message: Message) -> ResumeMailboxResult: + """Process one email message and apply candidate CRM updates.""" sender_name, sender_email = self._sender_identity(message) correlation_id = self._mailbox_correlation_id(message) @@ -92,11 +167,7 @@ def finalize(result: ResumeMailboxResult) -> ResumeMailboxResult: ) ) - sender_is_authorized = await asyncio.to_thread( - self._sender_is_authorized, - sender_email, - ) - if not sender_is_authorized: + if not self._sender_is_authorized(sender_email): return finalize( ResumeMailboxResult( sender_email=sender_email, @@ -117,7 +188,7 @@ def finalize(result: ResumeMailboxResult) -> ResumeMailboxResult: ) ) - staging_contact = await asyncio.to_thread(self._find_or_create_staging_contact) + staging_contact = self._find_or_create_staging_contact() staging_contact_id = str(staging_contact.get("id", "")).strip() if not staging_contact_id: return finalize( @@ -133,7 +204,7 @@ def finalize(result: ResumeMailboxResult) -> ResumeMailboxResult: for attachment in attachments: if len(attachment.content) > self._max_attachment_size_bytes: logger.warning( - "Skipping oversized resume attachment filename=%s size_bytes=%s sender=%s", + "Skipping oversized resume attachment filename=%s size=%s sender=%s", attachment.filename, len(attachment.content), sender_email, @@ -141,7 +212,7 @@ def finalize(result: ResumeMailboxResult) -> ResumeMailboxResult: continue try: - ok = await self._process_attachment( + ok = self._process_attachment( staging_contact_id=staging_contact_id, attachment=attachment, ) @@ -188,7 +259,7 @@ def _sender_identity(self, message: Message) -> tuple[str | None, str | None]: return sender_name, sender_email def _has_authenticated_sender(self, message: Message) -> bool: - """Require pass results from SPF/DKIM/DMARC headers to reduce spoof risk.""" + """Require pass SPF/DKIM/DMARC headers to reduce spoofed sender risk.""" auth_results = str(message.get("Authentication-Results", "")).lower() received_spf = str(message.get("Received-SPF", "")).lower() @@ -201,8 +272,7 @@ def _sender_is_authorized(self, sender_email: str) -> bool: in_people_db = self._sender_has_privileged_role_in_people_db(sender_email) if not in_people_db: return False - in_crm = self._sender_has_privileged_role_in_crm(sender_email) - return in_crm + return self._sender_has_privileged_role_in_crm(sender_email) def _sender_has_privileged_role_in_people_db(self, sender_email: str) -> bool: query = """ @@ -245,7 +315,7 @@ def _parse_role_names(self, raw_roles: Any) -> set[str]: return {value.casefold() for value in parsed if value} - def _find_contact_by_email(self, sender_email: str) -> dict[str, Any] | None: + def _find_contact_by_email(self, email_address: str) -> dict[str, Any] | None: search_params = { "where": [ { @@ -254,12 +324,12 @@ def _find_contact_by_email(self, sender_email: str) -> dict[str, Any] | None: { "type": "equals", "attribute": "emailAddress", - "value": sender_email, + "value": email_address, }, { "type": "equals", "attribute": "c508Email", - "value": sender_email, + "value": email_address, }, ], } @@ -273,7 +343,7 @@ def _find_contact_by_email(self, sender_email: str) -> dict[str, Any] | None: except EspoAPIError as exc: logger.warning( "CRM contact lookup by email failed email=%s error=%s", - sender_email, + email_address, exc, ) return None @@ -290,7 +360,6 @@ def _create_contact_for_email( email_address: str, display_name: str | None, ) -> dict[str, Any]: - """Create a fallback contact when no existing CRM record can be resolved.""" local_part = email_address.split("@", 1)[0] fallback_name = local_part.replace(".", " ").replace("_", " ").strip().title() payload: dict[str, Any] = { @@ -304,7 +373,6 @@ def _create_contact_for_email( return self.espo_api.request("POST", "Contact", payload) def _find_or_create_staging_contact(self) -> dict[str, Any]: - """Resolve a stable non-sender contact used only for initial extraction.""" staging_email = self._normalize_email(self.settings.email_username) if not staging_email: raise ValueError("EMAIL_USERNAME is required for staging contact lookup") @@ -313,10 +381,7 @@ def _find_or_create_staging_contact(self) -> dict[str, Any]: if existing is not None: return existing - return self._create_contact_for_email( - staging_email, - "Resume Intake Staging", - ) + return self._create_contact_for_email(staging_email, "Resume Intake Staging") def _extract_resume_attachments(self, message: Message) -> list[ResumeAttachment]: attachments: list[ResumeAttachment] = [] @@ -341,134 +406,81 @@ def _extract_resume_attachments(self, message: Message) -> list[ResumeAttachment return attachments - async def _process_attachment( + def _process_attachment( self, *, staging_contact_id: str, attachment: ResumeAttachment, ) -> bool: - staging_attachment_id = await asyncio.to_thread( - self._upload_contact_resume, - staging_contact_id, - attachment, + staging_attachment_id = self._upload_contact_resume( + staging_contact_id, attachment ) if not staging_attachment_id: return False - staging_extract_job_id = await self._enqueue_resume_extract_job( + staging_extract = self.resume_processor.extract_profile_proposal( contact_id=staging_contact_id, attachment_id=staging_attachment_id, filename=attachment.filename, ) - staging_extract_job = await self._wait_for_worker_job_result( - staging_extract_job_id - ) - if staging_extract_job is None: - return False - - staging_status = str(staging_extract_job.get("status", "")) - if staging_status != "succeeded": - return False - - staging_extract_result = staging_extract_job.get("result") - if not isinstance(staging_extract_result, dict): - return False - - if not bool(staging_extract_result.get("success", False)): + if not staging_extract.success: return False candidate_email = self._candidate_email_from_extract_result( - staging_extract_result + { + "extracted_profile": staging_extract.extracted_profile.model_dump(), + "proposed_updates": staging_extract.proposed_updates, + } ) if not candidate_email: logger.info( - "Skipping resume attachment filename=%s due to missing candidate email in extraction", + "Skipping resume attachment filename=%s due to missing candidate email", attachment.filename, ) return False - candidate_contact = await asyncio.to_thread( - self._find_contact_by_email, - candidate_email, - ) + candidate_contact = self._find_contact_by_email(candidate_email) if candidate_contact is None: - candidate_contact = await asyncio.to_thread( - self._create_contact_for_email, - candidate_email, - None, - ) + candidate_contact = self._create_contact_for_email(candidate_email, None) candidate_contact_id = str(candidate_contact.get("id", "")).strip() if not candidate_contact_id: return False - candidate_attachment_id = await asyncio.to_thread( - self._upload_contact_resume, + candidate_attachment_id = self._upload_contact_resume( candidate_contact_id, attachment, ) if not candidate_attachment_id: return False - candidate_link_ok = await asyncio.to_thread( - self._append_contact_resume, - candidate_contact_id, - candidate_attachment_id, - ) - if not candidate_link_ok: + if not self._append_contact_resume( + candidate_contact_id, candidate_attachment_id + ): return False - candidate_extract_job_id = await self._enqueue_resume_extract_job( + candidate_extract = self.resume_processor.extract_profile_proposal( contact_id=candidate_contact_id, attachment_id=candidate_attachment_id, filename=attachment.filename, ) - candidate_extract_job = await self._wait_for_worker_job_result( - candidate_extract_job_id - ) - if candidate_extract_job is None: - return False - - candidate_status = str(candidate_extract_job.get("status", "")) - if candidate_status != "succeeded": - return False - - candidate_extract_result = candidate_extract_job.get("result") - if not isinstance(candidate_extract_result, dict): + if not candidate_extract.success: return False - if not bool(candidate_extract_result.get("success", False)): - return False - - proposed_updates_raw = candidate_extract_result.get("proposed_updates") - if not isinstance(proposed_updates_raw, dict) or not proposed_updates_raw: - return True - proposed_updates = { str(field): str(value) - for field, value in proposed_updates_raw.items() + for field, value in candidate_extract.proposed_updates.items() if value is not None and str(value).strip() } if not proposed_updates: return True - apply_job_id = await self._enqueue_resume_apply_job( + apply_result = self.resume_processor.apply_profile_updates( contact_id=candidate_contact_id, updates=proposed_updates, + link_discord=None, ) - apply_job = await self._wait_for_worker_job_result(apply_job_id) - if apply_job is None: - return False - - apply_status = str(apply_job.get("status", "")) - if apply_status != "succeeded": - return False - - apply_result = apply_job.get("result") - if not isinstance(apply_result, dict): - return False - - return bool(apply_result.get("success", False)) + return bool(apply_result.success) def _candidate_email_from_extract_result( self, extract_result: dict[str, Any] @@ -543,93 +555,6 @@ def _append_contact_resume(self, contact_id: str, attachment_id: str) -> bool: ) return False - async def _enqueue_resume_extract_job( - self, - *, - contact_id: str, - attachment_id: str, - filename: str, - ) -> str: - payload = { - "contact_id": contact_id, - "attachment_id": attachment_id, - "filename": filename, - } - return await self._enqueue_worker_job("/jobs/resume-extract", payload) - - async def _enqueue_resume_apply_job( - self, - *, - contact_id: str, - updates: dict[str, str], - ) -> str: - payload = { - "contact_id": contact_id, - "updates": updates, - "link_discord": None, - } - return await self._enqueue_worker_job("/jobs/resume-apply", payload) - - async def _enqueue_worker_job(self, path: str, payload: dict[str, Any]) -> str: - async with aiohttp.ClientSession() as session: - async with session.post( - self._worker_url(path), - headers=self._worker_headers(), - json=payload, - timeout=aiohttp.ClientTimeout(total=30), - ) as response: - data = await response.json() - if response.status != 202: - raise ValueError(f"Worker enqueue failed path={path}: {data}") - job_id = data.get("job_id") - if not isinstance(job_id, str) or not job_id.strip(): - raise ValueError("Missing worker job_id in response") - return job_id - - def _worker_headers(self) -> dict[str, str]: - if not self.settings.api_shared_secret: - raise ValueError("API_SHARED_SECRET is required for worker API requests") - return { - "X-API-Secret": self.settings.api_shared_secret, - "Content-Type": "application/json", - } - - def _worker_url(self, path: str) -> str: - return f"{self.settings.worker_api_base_url.rstrip('/')}{path}" - - async def _wait_for_worker_job_result( - self, - job_id: str, - *, - timeout_seconds: int = 180, - poll_seconds: float = 2.0, - ) -> dict[str, Any] | None: - terminal = {"succeeded", "dead", "canceled"} - max_attempts = max(1, int(timeout_seconds / poll_seconds)) - - for _ in range(max_attempts): - job = await self._get_worker_job_status(job_id) - status = str(job.get("status", "")) - if status in terminal: - return job - await asyncio.sleep(poll_seconds) - - return None - - async def _get_worker_job_status(self, job_id: str) -> dict[str, Any]: - async with aiohttp.ClientSession() as session: - async with session.get( - self._worker_url(f"/jobs/{job_id}"), - headers=self._worker_headers(), - timeout=aiohttp.ClientTimeout(total=30), - ) as response: - data = await response.json() - if response.status != 200: - raise ValueError(f"Worker job status failed: {data}") - if not isinstance(data, dict): - raise ValueError("Worker job status response must be an object") - return data - def _file_extension(self, filename: str) -> str: if "." not in filename: return "" @@ -659,14 +584,14 @@ def _audit_mailbox_outcome( if not sender_email: return - audit_result = "error" + audit_result = AuditResult.ERROR if result.skipped_reason in { "sender_not_authorized", "sender_authentication_failed", }: - audit_result = "denied" + audit_result = AuditResult.DENIED elif result.skipped_reason in {None, "no_resume_attachments"}: - audit_result = "success" + audit_result = AuditResult.SUCCESS metadata = { "subject": str(message.get("Subject", "")).strip() or None, @@ -675,13 +600,26 @@ def _audit_mailbox_outcome( "skipped_reason": result.skipped_reason, } - self.audit_logger.log_admin_sso_action( - action="crm.resume_mailbox_ingest", - result=audit_result, - actor_email=sender_email, - actor_display_name=sender_name, - metadata=metadata, - resource_type="mailbox_message", - resource_id=correlation_id, - correlation_id=correlation_id, - ) + try: + insert_audit_event( + self.settings, + AuditEventInput( + source=AuditSource.ADMIN_DASHBOARD, + action="crm.resume_mailbox_ingest", + result=audit_result, + actor_provider=ActorProvider.ADMIN_SSO, + actor_subject=sender_email, + actor_display_name=sender_name, + resource_type="mailbox_message", + resource_id=correlation_id, + correlation_id=correlation_id, + metadata=metadata, + ), + ) + except Exception as exc: + logger.warning( + "Failed writing mailbox audit event correlation_id=%s sender=%s error=%s", + correlation_id, + sender_email, + exc, + ) diff --git a/tests/integration/test_email_monitor.py b/tests/integration/test_email_monitor.py index 78b2d0b7..e5309acd 100644 --- a/tests/integration/test_email_monitor.py +++ b/tests/integration/test_email_monitor.py @@ -1,80 +1,45 @@ -""" -Integration tests for email monitoring feature. -""" +"""Integration tests for legacy EmailMonitor compatibility cog.""" + +from unittest.mock import patch import pytest -from unittest.mock import Mock, AsyncMock, patch -import imaplib from five08.discord_bot.cogs.email_monitor import EmailMonitor class TestEmailMonitorIntegration: - """Integration tests for EmailMonitor feature.""" - - @pytest.fixture - def email_monitor(self, mock_bot): - """Create an EmailMonitor instance for testing.""" - # Mock the task starting to avoid actual background task - with patch.object( - EmailMonitor, "__init__", lambda self, bot: setattr(self, "bot", bot) - ): - monitor = EmailMonitor(mock_bot) - monitor.task_poll_inbox = AsyncMock() - monitor.task_poll_inbox.start = Mock() - monitor.task_poll_inbox.cancel = Mock() - monitor.task_poll_inbox.is_running = Mock(return_value=False) - return monitor - - @pytest.fixture - def email_monitor_real_poll(self, mock_bot): - """Create an EmailMonitor instance with the real poll coroutine.""" - return EmailMonitor(mock_bot) + """Integration tests for EmailMonitor compatibility behavior.""" @pytest.mark.asyncio - async def test_poll_inbox_handles_imap_errors( - self, email_monitor_real_poll, mock_discord_channel - ): - """Test that IMAP errors are handled gracefully.""" - email_monitor_real_poll.bot.get_channel.return_value = mock_discord_channel + async def test_cog_load_logs_deprecation_notice(self, mock_bot) -> None: + monitor = EmailMonitor(mock_bot) - with patch( - "imaplib.IMAP4_SSL", side_effect=imaplib.IMAP4.error("Connection failed") - ): - # IMAP transport errors currently bubble from the poll loop. - with pytest.raises(imaplib.IMAP4.error): - await EmailMonitor.task_poll_inbox.coro(email_monitor_real_poll) + with patch("five08.discord_bot.cogs.email_monitor.logger.info") as mock_info: + await monitor.cog_load() - @pytest.mark.asyncio - async def test_poll_inbox_handles_email_parsing_errors( - self, email_monitor_real_poll, mock_discord_channel, mock_imap_server - ): - """Test handling of malformed email messages.""" - email_monitor_real_poll.bot.get_channel.return_value = mock_discord_channel - mock_imap_server.search.return_value = ("OK", [b"1"]) + mock_info.assert_called_once() - # Malformed email data - mock_imap_server.fetch.return_value = ("OK", [(None, b"malformed email data")]) + @pytest.mark.asyncio + async def test_setup_function_adds_cog(self, mock_bot) -> None: + from five08.discord_bot.cogs.email_monitor import setup - with patch("imaplib.IMAP4_SSL", return_value=mock_imap_server): - await EmailMonitor.task_poll_inbox.coro(email_monitor_real_poll) + await setup(mock_bot) - @pytest.mark.asyncio - async def test_cog_unload_cancels_task(self, email_monitor): - """Test that cog_unload properly cancels the background task.""" - await email_monitor.cog_unload() - email_monitor.task_poll_inbox.cancel.assert_called_once() + mock_bot.add_cog.assert_called_once() + added_cog = mock_bot.add_cog.call_args[0][0] + assert isinstance(added_cog, EmailMonitor) @pytest.mark.asyncio - async def test_setup_function(self, mock_bot): - """Test the setup function adds the feature to the bot.""" + async def test_setup_constructs_email_monitor(self, mock_bot) -> None: from five08.discord_bot.cogs.email_monitor import setup - with patch.object( - EmailMonitor, "__init__", lambda self, bot: setattr(self, "bot", bot) - ): + with patch( + "five08.discord_bot.cogs.email_monitor.EmailMonitor", wraps=EmailMonitor + ) as mock_class: await setup(mock_bot) - mock_bot.add_cog.assert_called_once() - added_cog = mock_bot.add_cog.call_args[0][0] - assert isinstance(added_cog, EmailMonitor) + mock_class.assert_called_once_with(mock_bot) + + def test_cog_stores_bot_reference(self, mock_bot) -> None: + monitor = EmailMonitor(mock_bot) + assert monitor.bot is mock_bot diff --git a/tests/unit/test_resume_mail_ingest.py b/tests/unit/test_resume_mail_ingest.py deleted file mode 100644 index 1c8b566a..00000000 --- a/tests/unit/test_resume_mail_ingest.py +++ /dev/null @@ -1,152 +0,0 @@ -"""Unit tests for mailbox-driven resume ingestion.""" - -from email.message import EmailMessage -from unittest.mock import AsyncMock, Mock - -import pytest - -from five08.discord_bot.config import settings -from five08.discord_bot.utils.resume_mail_ingest import ( - ResumeAttachment, - ResumeMailboxProcessor, -) - - -def _build_message(*, include_attachment: bool = True) -> EmailMessage: - message = EmailMessage() - message["From"] = "Admin User " - message["Subject"] = "Resume upload" - message["Authentication-Results"] = "mx.example; dkim=pass; spf=pass; dmarc=pass" - message.set_content("Please process this resume.") - - if include_attachment: - message.add_attachment( - b"resume-bytes", - maintype="application", - subtype="pdf", - filename="resume.pdf", - ) - - return message - - -@pytest.mark.asyncio -async def test_process_message_happy_path_uses_existing_contact() -> None: - """Authorized senders with resume attachments should process successfully.""" - processor = ResumeMailboxProcessor(settings) - processor._sender_is_authorized = Mock(return_value=True) - processor._find_or_create_staging_contact = Mock(return_value={"id": "staging-1"}) - processor._create_contact_for_email = Mock(return_value={"id": "contact-new"}) - processor._process_attachment = AsyncMock(return_value=True) - - result = await processor.process_message(_build_message()) - - assert result.skipped_reason is None - assert result.processed_attachments == 1 - processor._find_or_create_staging_contact.assert_called_once() - processor._create_contact_for_email.assert_not_called() - processor._process_attachment.assert_awaited_once() - - -@pytest.mark.asyncio -async def test_process_message_skips_when_sender_not_authorized() -> None: - """Unauthorized senders should be rejected before any attachment processing.""" - processor = ResumeMailboxProcessor(settings) - processor._sender_is_authorized = Mock(return_value=False) - processor._process_attachment = AsyncMock(return_value=True) - processor.audit_logger = Mock() - - result = await processor.process_message(_build_message()) - - assert result.skipped_reason == "sender_not_authorized" - assert result.processed_attachments == 0 - processor._process_attachment.assert_not_called() - processor.audit_logger.log_admin_sso_action.assert_called_once() - payload = processor.audit_logger.log_admin_sso_action.call_args.kwargs - assert payload["result"] == "denied" - - -@pytest.mark.asyncio -async def test_process_message_requires_auth_headers_when_enabled() -> None: - """Spoof-resistant mode should reject messages without pass auth headers.""" - processor = ResumeMailboxProcessor(settings) - processor._sender_is_authorized = Mock(return_value=True) - - message = _build_message() - del message["Authentication-Results"] - - result = await processor.process_message(message) - - assert result.skipped_reason == "sender_authentication_failed" - processor._sender_is_authorized.assert_not_called() - - -@pytest.mark.asyncio -async def test_process_message_creates_contact_when_lookup_misses() -> None: - """A staging contact should be created/resolved once per processed message.""" - processor = ResumeMailboxProcessor(settings) - processor._sender_is_authorized = Mock(return_value=True) - processor._find_or_create_staging_contact = Mock(return_value={"id": "staging-1"}) - processor._process_attachment = AsyncMock(return_value=True) - - result = await processor.process_message(_build_message()) - - assert result.skipped_reason is None - assert result.processed_attachments == 1 - processor._find_or_create_staging_contact.assert_called_once() - - -@pytest.mark.asyncio -async def test_process_attachment_updates_candidate_not_sender() -> None: - """Attachment processing should resolve candidate email and apply to candidate contact.""" - processor = ResumeMailboxProcessor(settings) - processor._upload_contact_resume = Mock( - side_effect=["att-staging", "att-candidate"] - ) - processor._append_contact_resume = Mock(return_value=True) - processor._enqueue_resume_extract_job = AsyncMock( - side_effect=["extract-staging", "extract-candidate"] - ) - processor._enqueue_resume_apply_job = AsyncMock(return_value="apply-candidate") - processor._wait_for_worker_job_result = AsyncMock( - side_effect=[ - { - "status": "succeeded", - "result": { - "success": True, - "extracted_profile": {"email": "candidate@example.com"}, - "proposed_updates": {}, - }, - }, - { - "status": "succeeded", - "result": { - "success": True, - "proposed_updates": {"phoneNumber": "14155551234"}, - }, - }, - { - "status": "succeeded", - "result": { - "success": True, - "updated_fields": ["phoneNumber"], - }, - }, - ] - ) - processor._find_contact_by_email = Mock(return_value=None) - processor._create_contact_for_email = Mock(return_value={"id": "candidate-1"}) - - ok = await processor._process_attachment( - staging_contact_id="staging-1", - attachment=ResumeAttachment(filename="resume.pdf", content=b"resume-bytes"), - ) - - assert ok is True - processor._create_contact_for_email.assert_called_once_with( - "candidate@example.com", None - ) - processor._enqueue_resume_apply_job.assert_awaited_once_with( - contact_id="candidate-1", - updates={"phoneNumber": "14155551234"}, - ) diff --git a/tests/unit/test_worker_config.py b/tests/unit/test_worker_config.py new file mode 100644 index 00000000..60085e1a --- /dev/null +++ b/tests/unit/test_worker_config.py @@ -0,0 +1,31 @@ +"""Unit tests for worker settings email intake validation.""" + +import pytest +from pydantic import ValidationError + +from five08.worker.config import WorkerSettings + + +def test_email_intake_requires_mailbox_credentials() -> None: + with pytest.raises(ValidationError, match="EMAIL_PASSWORD must be set"): + WorkerSettings( + espo_base_url="https://crm.test.com", + espo_api_key="test-key", + email_resume_intake_enabled=True, + email_username="workflows@508.dev", + email_password=" ", + imap_server="imap.test.com", + ) + + +def test_email_intake_validation_passes_with_required_fields() -> None: + settings = WorkerSettings( + espo_base_url="https://crm.test.com", + espo_api_key="test-key", + email_resume_intake_enabled=True, + email_username="workflows@508.dev", + email_password="password", + imap_server="imap.test.com", + ) + + assert settings.email_resume_intake_enabled is True diff --git a/tests/unit/test_worker_mailbox_resume_ingest.py b/tests/unit/test_worker_mailbox_resume_ingest.py new file mode 100644 index 00000000..4d3c748e --- /dev/null +++ b/tests/unit/test_worker_mailbox_resume_ingest.py @@ -0,0 +1,118 @@ +"""Unit tests for worker-side mailbox resume ingestion.""" + +from __future__ import annotations + +from email.message import EmailMessage +from types import SimpleNamespace +from unittest.mock import Mock + +from five08.worker.mailbox_resume_ingest import ResumeAttachment, ResumeMailboxProcessor + + +class _FakeProfile: + def __init__(self, email: str | None) -> None: + self._email = email + + def model_dump(self) -> dict[str, str | None]: + return {"email": self._email} + + +def _build_settings() -> SimpleNamespace: + return SimpleNamespace( + espo_base_url="https://crm.test.com", + espo_api_key="test_key", + email_username="workflows@508.dev", + email_password="test_password", + imap_server="imap.test.com", + email_resume_allowed_extensions="pdf,doc,docx", + email_resume_max_file_size_mb=10, + email_require_sender_auth_headers=True, + ) + + +def _build_message(*, include_attachment: bool = True) -> EmailMessage: + message = EmailMessage() + message["From"] = "Admin User " + message["Subject"] = "Resume upload" + message["Authentication-Results"] = "mx.example; dkim=pass; spf=pass; dmarc=pass" + message.set_content("Please process this resume.") + + if include_attachment: + message.add_attachment( + b"resume-bytes", + maintype="application", + subtype="pdf", + filename="resume.pdf", + ) + + return message + + +def test_process_message_happy_path() -> None: + processor = ResumeMailboxProcessor(_build_settings()) + processor._audit_mailbox_outcome = Mock() + processor._sender_is_authorized = Mock(return_value=True) + processor._find_or_create_staging_contact = Mock(return_value={"id": "staging-1"}) + processor._process_attachment = Mock(return_value=True) + + result = processor.process_message(_build_message()) + + assert result.skipped_reason is None + assert result.processed_attachments == 1 + processor._find_or_create_staging_contact.assert_called_once() + processor._process_attachment.assert_called_once() + + +def test_process_message_denies_unauthorized_sender() -> None: + processor = ResumeMailboxProcessor(_build_settings()) + processor._audit_mailbox_outcome = Mock() + processor._sender_is_authorized = Mock(return_value=False) + + result = processor.process_message(_build_message()) + + assert result.skipped_reason == "sender_not_authorized" + assert result.processed_attachments == 0 + + +def test_process_attachment_updates_candidate_contact() -> None: + processor = ResumeMailboxProcessor(_build_settings()) + processor._upload_contact_resume = Mock( + side_effect=["att-staging", "att-candidate"] + ) + processor._append_contact_resume = Mock(return_value=True) + processor._find_contact_by_email = Mock(return_value=None) + processor._create_contact_for_email = Mock(return_value={"id": "candidate-1"}) + + staging_extract = SimpleNamespace( + success=True, + extracted_profile=_FakeProfile("candidate@example.com"), + proposed_updates={}, + ) + candidate_extract = SimpleNamespace( + success=True, + extracted_profile=_FakeProfile(None), + proposed_updates={"phoneNumber": "14155551234"}, + ) + apply_result = SimpleNamespace(success=True) + + processor.resume_processor = Mock() + processor.resume_processor.extract_profile_proposal.side_effect = [ + staging_extract, + candidate_extract, + ] + processor.resume_processor.apply_profile_updates.return_value = apply_result + + ok = processor._process_attachment( + staging_contact_id="staging-1", + attachment=ResumeAttachment(filename="resume.pdf", content=b"resume-bytes"), + ) + + assert ok is True + processor._create_contact_for_email.assert_called_once_with( + "candidate@example.com", None + ) + processor.resume_processor.apply_profile_updates.assert_called_once_with( + contact_id="candidate-1", + updates={"phoneNumber": "14155551234"}, + link_discord=None, + )