diff --git a/apps/api/app/db/migrate.py b/apps/api/app/db/migrate.py index 2e06910..ed642df 100644 --- a/apps/api/app/db/migrate.py +++ b/apps/api/app/db/migrate.py @@ -73,6 +73,7 @@ def reset_database(database_url: str | None = None) -> None: conn.execute( """ DROP TABLE IF EXISTS task_run, outbox, approval, artifact, + email_suggestion, notification_log, application, job_posting, cv_section, profile, schema_migrations CASCADE """ diff --git a/apps/api/app/imap_watch.py b/apps/api/app/imap_watch.py new file mode 100644 index 0000000..28f7090 --- /dev/null +++ b/apps/api/app/imap_watch.py @@ -0,0 +1,468 @@ +"""IMAP email watch: polls UNSEEN messages and classifies them. + +Uses stdlib imaplib (SSL). Enabled only when EMAIL_WATCH_ENABLED=true. +Config: IMAP_HOST, IMAP_PORT, IMAP_USER, IMAP_PASS. + +Flow: +1. Connect via IMAP SSL. +2. Fetch UNSEEN messages since last poll. +3. For each message, match sender domain + subject/body keywords to + open applications (status in sent/interviewing). +4. Classify via LLM task 'email_classify' -> {classification, state_proposal, reason}. +5. Insert email_suggestion rows (skip noise and dedupe by from+subject+day). +""" + +from __future__ import annotations + +import email +import email.utils +import hashlib +import imaplib +import logging +import os +from datetime import datetime, timezone +from typing import Any, Sequence + +from app.db import execute, fetch_all, fetch_one +from app import llm + +logger = logging.getLogger(__name__) + +VALID_CLASSIFICATIONS = frozenset({ + "interview_invite", + "rejection", + "question", + "noise", +}) + +# Keywords for cheap pre-matching before LLM classify +INTERVIEW_KEYWORDS = ("interview", "invite", "meeting", "schedule", "call") +REJECTION_KEYWORDS = ("regret", "unfortunately", "not moving", "rejection", "position has been filled") +QUESTION_KEYWORDS = ("question", "clarif", "additional", "could you", "please provide") +NOISE_KEYWORDS = ("newsletter", "unsubscribe", "promotion", "advert", "offer") + + +def is_email_watch_enabled() -> bool: + """Check if email watch is enabled.""" + return os.environ.get("EMAIL_WATCH_ENABLED", "false").lower() in ( + "true", + "1", + "yes", + ) + + +def _extract_sender_domain(from_addr: str) -> str: + """Extract the domain from an email From header.""" + parsed = email.utils.parseaddr(from_addr) + addr = parsed[1] or from_addr + parts = addr.split("@") + if len(parts) >= 2: + return parts[-1].lower().strip() + return "" + + +def _build_snippet(body: str, max_len: int = 300) -> str: + """Truncate body to a snippet.""" + body = body.replace("\r", " ").replace("\n", " ").strip() + if len(body) > max_len: + return body[:max_len] + "..." + return body + + +def _parse_email_message(raw_bytes: bytes) -> dict[str, str]: + """Parse raw email bytes into a dict with from, subject, body.""" + msg = email.message_from_bytes(raw_bytes) + from_addr = msg.get("From", "") + subject = msg.get("Subject", "") + date_str = msg.get("Date", "") + + # Extract body (prefer plain text) + body = "" + if msg.is_multipart(): + for part in msg.walk(): + ct = part.get_content_type() + if ct == "text/plain": + payload = part.get_payload(decode=True) + if payload: + body = payload.decode("utf-8", errors="replace") + break + if not body: + for part in msg.walk(): + if part.get_content_type().startswith("text/"): + payload = part.get_payload(decode=True) + if payload: + body = payload.decode("utf-8", errors="replace") + break + else: + payload = msg.get_payload(decode=True) + if payload: + body = payload.decode("utf-8", errors="replace") + + return { + "from": from_addr, + "subject": subject, + "body": body, + "date": date_str, + } + + +def _parse_date(date_str: str) -> datetime: + """Parse RFC 2822 date string to UTC datetime. Falls back to now().""" + if date_str: + try: + parsed = email.utils.parsedate_to_datetime(date_str) + if parsed is not None: + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + except (ValueError, TypeError): + pass + return datetime.now(timezone.utc) + + +def match_application( + from_addr: str, + subject: str, + body: str, + applications: Sequence[dict[str, Any]], +) -> dict[str, Any] | None: + """Match an email to an open application. + + Match by: + 1. Company name in subject/body + 2. Sender domain in posting URL + + Only matches applications with state in ('sent', 'interviewing'). + """ + sender_domain = _extract_sender_domain(from_addr) + text_lower = (subject + " " + body).lower() + + for app_row in applications: + state = app_row.get("state", "") + if state not in ("sent", "interviewing"): + continue + + company = (app_row.get("company") or "").lower() + title = (app_row.get("title") or "").lower() + posting_url = app_row.get("url") or "" + + # Match 1: company name in subject/body + if company and len(company) > 2 and company in text_lower: + return app_row + + # Match 2: sender domain in posting URL + if sender_domain and sender_domain in (posting_url or "").lower(): + return app_row + + # Match 3: title keywords in subject (looser) + if title and len(title) > 3: + title_words = [w for w in title.split() if len(w) > 3] + matches = sum(1 for w in title_words if w in text_lower) + if matches >= 2 and len(title_words) >= 2: + return app_row + + return None + + +def classify_email(subject: str, body: str) -> dict[str, Any]: + """Classify an email via LLM task 'email_classify'. + + Returns {classification, state_proposal, reason}. + Falls back to inline keyword heuristics if LLM fails. + """ + text_lower = (subject + " " + body).lower() + + result = llm.run_task( + "email_classify", + f"Subject: {subject}\nBody: {body[:1000]}", + ) + + classification = result.get("classification", "noise") + state_proposal = result.get("state_proposal") + reason = result.get("reason", "") + + # Validate + if classification not in VALID_CLASSIFICATIONS: + classification = "noise" + + return { + "classification": classification, + "state_proposal": state_proposal, + "reason": reason, + } + + +def is_duplicate(from_addr: str, subject: str, received_at: datetime) -> bool: + """Check if a similar email_suggestion already exists (same from+subject+day).""" + day_start = received_at.replace(hour=0, minute=0, second=0, microsecond=0) + row = fetch_one( + """ + SELECT id FROM email_suggestion + WHERE mailbox_from = %s AND subject = %s + AND received_at >= %s AND received_at < %s + interval '1 day' + LIMIT 1 + """, + (from_addr, subject, day_start, day_start), + ) + return row is not None + + +def create_email_suggestion( + application_id: str | None, + mailbox_from: str, + subject: str, + snippet: str, + classification: str, + state_proposal: str | None, + received_at: datetime, +) -> dict[str, Any]: + """Insert an email_suggestion row.""" + row = execute( + """ + INSERT INTO email_suggestion + (application_id, mailbox_from, subject, snippet, classification, state_proposal, status, received_at) + VALUES + (%s, %s, %s, %s, %s, %s, 'pending', %s) + RETURNING * + """, + (application_id, mailbox_from, subject, snippet, classification, state_proposal, received_at), + ) + if row is None: + raise RuntimeError("insert email_suggestion failed") + return _normalize_suggestion(row) + + +def _normalize_suggestion(row: dict[str, Any]) -> dict[str, Any]: + return { + "id": str(row["id"]), + "application_id": str(row["application_id"]) if row.get("application_id") else None, + "mailbox_from": row["mailbox_from"], + "subject": row["subject"], + "snippet": row["snippet"], + "classification": row["classification"], + "state_proposal": row.get("state_proposal"), + "status": row["status"], + "received_at": row["received_at"].isoformat() if row.get("received_at") else None, + "created_at": row["created_at"].isoformat() if row.get("created_at") else None, + } + + +def list_pending_suggestions() -> list[dict[str, Any]]: + """Return all pending email_suggestions, newest first.""" + rows = fetch_all( + """ + SELECT * FROM email_suggestion + WHERE status = 'pending' + ORDER BY received_at DESC + """ + ) + return [_normalize_suggestion(r) for r in rows] + + +def get_suggestion(suggestion_id: str) -> dict[str, Any] | None: + row = fetch_one("SELECT * FROM email_suggestion WHERE id = %s", (suggestion_id,)) + if row is None: + return None + return _normalize_suggestion(row) + + +def update_suggestion_status( + suggestion_id: str, + status: str, +) -> dict[str, Any] | None: + row = execute( + "UPDATE email_suggestion SET status = %s WHERE id = %s RETURNING *", + (status, suggestion_id), + ) + if row is None: + return None + return _normalize_suggestion(row) + + +def get_open_applications() -> list[dict[str, Any]]: + """Return applications in sent/interviewing state for matching.""" + rows = fetch_all( + """ + SELECT a.*, j.company, j.title, j.url + FROM application a + JOIN job_posting j ON a.job_posting_id = j.id + WHERE a.state IN ('sent', 'interviewing') + """ + ) + result: list[dict[str, Any]] = [] + for row in rows: + result.append({ + "id": str(row["id"]), + "state": row["state"], + "company": row.get("company", ""), + "title": row.get("title", ""), + "url": row.get("url", ""), + }) + return result + + +# --- IMAP poll --- + +class FakeImap: + """Test double for imaplib IMAP4_SSL. No network. + + Usage: + fake = FakeImap(messages=[(uid1, raw1), (uid2, raw2)]) + poll_inbox(fake) # uses fake instead of real connection + """ + + def __init__(self, messages: list[tuple[bytes, bytes]] | None = None) -> None: + # messages: list of (uid, raw_email_bytes) + self._messages = messages or [] + self._seen_uids: set[bytes] = set() + self.selected = False + + def select(self, mailbox: str = "INBOX") -> tuple[str, list[bytes]]: + self.selected = True + count = len(self._messages) + return ("OK", [str(count).encode()]) + + def search(self, charset: str | None, *criteria: str) -> tuple[str, list[bytes]]: + # Return uids of messages matching criteria (we keep it simple) + uids = [uid for uid, _ in self._messages] + return ("OK", [b" ".join(uids)]) + + def fetch(self, uid: bytes, parts: str) -> tuple[str, list[tuple[bytes, bytes]]]: + for msg_uid, raw in self._messages: + if msg_uid == uid: + return ("OK", [(uid, raw)]) + return ("OK", []) + + def store(self, uid: bytes, flags: str, flag_set: str) -> tuple[str, list[bytes]]: + self._seen_uids.add(uid) + return ("OK", [uid]) + + def close(self) -> tuple[str, list[bytes]]: + self.selected = False + return ("OK", [b""]) + + def logout(self) -> tuple[str, list[bytes]]: + return ("OK", [b"BYE"]) + + +def _connect_imap() -> Any: + """Connect to IMAP server using env config.""" + host = os.environ.get("IMAP_HOST", "") + port = int(os.environ.get("IMAP_PORT", "993")) + user = os.environ.get("IMAP_USER", "") + password = os.environ.get("IMAP_PASS", "") + + conn = imaplib.IMAP4_SSL(host, port) + conn.login(user, password) + return conn + + +def poll_inbox(conn: Any = None) -> list[dict[str, Any]]: + """Poll the inbox for UNSEEN messages, classify, and create suggestions. + + If conn is provided (e.g. FakeImap for tests), uses it instead of connecting. + Returns list of created suggestions. + + No network is used when conn is a FakeImap. + """ + created: list[dict[str, Any]] = [] + own_conn = False + + if conn is None: + conn = _connect_imap() + own_conn = True + + try: + conn.select("INBOX") + status, data = conn.search(None, "UNSEEN") + if status != "OK": + logger.warning("imap_watch: search failed: %s", status) + return created + + uids = [] + if data and data[0]: + uids = data[0].split() + + if not uids: + return created + + # Get open applications for matching + applications = get_open_applications() + + for uid in uids: + status, fetch_data = conn.fetch(uid, "(RFC822)") + if status != "OK" or not fetch_data: + continue + + raw_bytes = b"" + for item in fetch_data: + if isinstance(item, tuple) and len(item) >= 2: + raw_bytes = item[1] + break + + if not raw_bytes: + continue + + parsed = _parse_email_message(raw_bytes) + from_addr = parsed["from"] + subject = parsed["subject"] + body = parsed["body"] + received_at = _parse_date(parsed["date"]) + snippet = _build_snippet(body) + + # Dedupe: same from+subject+day + if is_duplicate(from_addr, subject, received_at): + logger.debug("imap_watch: dedupe skip: %s / %s", from_addr, subject) + continue + + # Match to application + app_row = match_application(from_addr, subject, body, applications) + + # Classify + classification_result = classify_email(subject, body) + classification = classification_result["classification"] + state_proposal = classification_result["state_proposal"] + + # Skip pure noise (don't create suggestion rows) + if classification == "noise": + continue + + suggestion = create_email_suggestion( + application_id=app_row["id"] if app_row else None, + mailbox_from=from_addr, + subject=subject, + snippet=snippet, + classification=classification, + state_proposal=state_proposal, + received_at=received_at, + ) + created.append(suggestion) + + # Send notification for interview invites + if classification == "interview_invite" and app_row: + from app.notify import send_notification + send_notification( + "email_suggestion", + f"Interview invite from {app_row.get('company', 'unknown')}", + { + "suggestion_id": suggestion["id"], + "application_id": app_row["id"], + "classification": classification, + }, + ) + + # Mark as seen + try: + conn.store(uid, "+FLAGS", "\\Seen") + except Exception: + logger.debug("imap_watch: store failed for uid %s", uid) + + finally: + if own_conn: + try: + conn.close() + conn.logout() + except Exception: + pass + + return created \ No newline at end of file diff --git a/apps/api/app/llm.py b/apps/api/app/llm.py index c9c1515..7b9bc84 100644 --- a/apps/api/app/llm.py +++ b/apps/api/app/llm.py @@ -123,6 +123,11 @@ MOCK_OUTPUTS: dict[str, dict[str, Any]] = { "posting requirements. Be specific." ), }, + "email_classify": { + "classification": "interview_invite", + "state_proposal": "interviewing", + "reason": "The email mentions an interview invitation.", + }, } diff --git a/apps/api/app/main.py b/apps/api/app/main.py index b5857ee..b3ae3df 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -37,11 +37,13 @@ from app.schemas import ( CvSectionOut, CvSectionUpdate, DigestItem, + EmailSuggestionOut, ErrorOut, InterviewPrepResponse, JobPostingCreate, JobPostingOut, NudgeItem, + NotificationLogOut, OutboxOut, OutboxSendRequest, PostingsFetchRequest, @@ -932,4 +934,109 @@ def seed_demo() -> Any: "postings": postings_count, "applications": apps_count, "sections": sections_count, - } \ No newline at end of file + } + + +# --- v1.1: Email Suggestions --- + +@app.get("/api/suggestions", response_model=list[EmailSuggestionOut]) +def get_suggestions() -> Any: + """Return all pending email suggestions, newest first.""" + from app.imap_watch import list_pending_suggestions + return list_pending_suggestions() + + +@app.post("/api/suggestions/{suggestion_id}/accept") +def accept_suggestion(suggestion_id: str) -> Any: + """Accept a suggestion: apply state_proposal via guarded transition. + + If the suggestion has an application_id and a state_proposal, apply the + state transition through the normal guard path. Updates last_activity_at. + Marks the suggestion as 'accepted'. + """ + from app.imap_watch import get_suggestion, update_suggestion_status + + suggestion = get_suggestion(suggestion_id) + if suggestion is None: + raise HTTPException(status_code=404, detail="Suggestion not found") + + if suggestion["status"] != "pending": + raise HTTPException( + status_code=409, + detail={"code": "not_pending", "message": "Suggestion is not pending"}, + ) + + app_id = suggestion.get("application_id") + state_proposal = suggestion.get("state_proposal") + + if app_id and state_proposal: + # Apply guarded transition + app_row = repo_app.get_application(app_id) + if app_row is None: + raise HTTPException(status_code=404, detail="Linked application not found") + + from_state = app_row["state"] + to_state = state_proposal + + from app.statemachine import TransitionContext, check_transition, InvalidTransition + + has_score = app_row.get("score") is not None + has_confirmed_approval = False + artifact_hash_match = False + + ctx = TransitionContext( + application_id=app_id, + from_state=from_state, + to_state=to_state, + has_score=has_score, + has_confirmed_approval=has_confirmed_approval, + artifact_hash_match=artifact_hash_match, + ) + + try: + check_transition(ctx) + except InvalidTransition as exc: + raise HTTPException( + status_code=409, + detail={"code": "invalid_transition", "message": str(exc)}, + ) + + updated = repo_app.update_application_state(app_id, to_state) + if updated is None: + raise HTTPException(status_code=500, detail="State update failed") + + # Mark suggestion as accepted + result = update_suggestion_status(suggestion_id, "accepted") + if result is None: + raise HTTPException(status_code=500, detail="Failed to update suggestion") + return result + + +@app.post("/api/suggestions/{suggestion_id}/dismiss") +def dismiss_suggestion(suggestion_id: str) -> Any: + """Dismiss a suggestion (mark as dismissed).""" + from app.imap_watch import get_suggestion, update_suggestion_status + + suggestion = get_suggestion(suggestion_id) + if suggestion is None: + raise HTTPException(status_code=404, detail="Suggestion not found") + + if suggestion["status"] != "pending": + raise HTTPException( + status_code=409, + detail={"code": "not_pending", "message": "Suggestion is not pending"}, + ) + + result = update_suggestion_status(suggestion_id, "dismissed") + if result is None: + raise HTTPException(status_code=500, detail="Failed to update suggestion") + return result + + +# --- v1.1: Notification Log --- + +@app.get("/api/notifications/log", response_model=list[NotificationLogOut]) +def get_notification_log() -> Any: + """Return last 50 notification log entries.""" + from app.notify import list_notification_log + return list_notification_log(limit=50) \ No newline at end of file diff --git a/apps/api/app/notify.py b/apps/api/app/notify.py new file mode 100644 index 0000000..48c312a --- /dev/null +++ b/apps/api/app/notify.py @@ -0,0 +1,183 @@ +"""Notification channels: LogChannel + WebhookChannel. + +Protocol-based: NotificationChannel defines send(kind, text, data) -> bool. +LogChannel writes to notification_log (delivered=true always). +WebhookChannel POSTs to NOTIFY_WEBHOOK_URL (2xx=delivered, else error row). + +send_notification(kind, text, data) is the public entry point used by +the scheduler and email watch modules. +""" + +from __future__ import annotations + +import json +import logging +import os +from typing import Any, Protocol + +from app.db import execute, fetch_all + +logger = logging.getLogger(__name__) + + +class NotificationChannel(Protocol): + """Protocol for notification delivery channels.""" + + def send(self, kind: str, text: str, data: dict[str, Any]) -> bool: + """Send a notification. Returns True if delivered, False otherwise.""" + ... + + +def _normalize_log_row(row: dict[str, Any]) -> dict[str, Any]: + payload = row.get("payload") + if payload is not None and not isinstance(payload, dict): + payload = json.loads(payload) if isinstance(payload, str) else payload + return { + "id": str(row["id"]), + "channel": row["channel"], + "kind": row["kind"], + "payload": payload or {}, + "delivered": row["delivered"], + "error": row.get("error"), + "created_at": row["created_at"].isoformat() if row.get("created_at") else None, + } + + +class LogChannel: + """Writes notification entries to notification_log with delivered=true.""" + + channel_name: str = "log" + + def send(self, kind: str, text: str, data: dict[str, Any]) -> bool: + row = execute( + """ + INSERT INTO notification_log (channel, kind, payload, delivered, error) + VALUES (%s, %s, %s, true, NULL) + RETURNING * + """, + (self.channel_name, kind, json.dumps({"text": text, **data})), + ) + if row is None: + logger.error("LogChannel: insert notification_log failed") + return False + logger.info("LogChannel: delivered kind=%s", kind) + return True + + +class WebhookChannel: + """POSTs notification payload to NOTIFY_WEBHOOK_URL. + + On 2xx response: writes notification_log with delivered=true. + On non-2xx or exception: writes notification_log with delivered=false and error. + """ + + channel_name: str = "webhook" + + def __init__(self, url: str | None = None) -> None: + self.url = url or os.environ.get("NOTIFY_WEBHOOK_URL", "") + + def send(self, kind: str, text: str, data: dict[str, Any]) -> bool: + payload: dict[str, Any] = {"kind": kind, "text": text, "data": data} + error: str | None = None + delivered = False + + if not self.url: + error = "NOTIFY_WEBHOOK_URL not configured" + logger.warning("WebhookChannel: %s", error) + else: + try: + import httpx + + resp = httpx.post(self.url, json=payload, timeout=10) + if 200 <= resp.status_code < 300: + delivered = True + else: + error = f"HTTP {resp.status_code}: {resp.text[:200]}" + logger.warning("WebhookChannel: %s", error) + except Exception as exc: + error = str(exc) + logger.warning("WebhookChannel: exception: %s", error) + + row = execute( + """ + INSERT INTO notification_log (channel, kind, payload, delivered, error) + VALUES (%s, %s, %s, %s, %s) + RETURNING * + """, + ( + self.channel_name, + kind, + json.dumps(payload), + delivered, + error, + ), + ) + if row is None: + logger.error("WebhookChannel: insert notification_log failed") + return delivered + + +# --- Channel registry --- + +_channels: list[NotificationChannel] | None = None + + +def get_channels() -> list[NotificationChannel]: + """Return the list of active notification channels. + + LogChannel is always included. + WebhookChannel is included when NOTIFY_WEBHOOK_URL is set. + """ + global _channels + if _channels is not None: + return _channels + + channels: list[NotificationChannel] = [LogChannel()] + webhook_url = os.environ.get("NOTIFY_WEBHOOK_URL", "").strip() + if webhook_url: + channels.append(WebhookChannel()) + _channels = channels + return channels + + +def set_channels(channels: list[NotificationChannel] | None) -> None: + """Override channel list (for testing).""" + global _channels + _channels = channels + + +def reset_channels() -> None: + """Reset to default (for testing).""" + global _channels + _channels = None + + +def send_notification(kind: str, text: str, data: dict[str, Any] | None = None) -> None: + """Send a notification via all active channels. + + kind: e.g. 'daily_digest', 'email_suggestion' + text: human-readable notification text + data: structured payload dict + """ + chs = get_channels() + payload = data or {} + for ch in chs: + try: + ch.send(kind, text, payload) + except Exception: + logger.exception("send_notification: channel %s failed", type(ch).__name__) + + +# --- Repository helpers --- + +def list_notification_log(limit: int = 50) -> list[dict[str, Any]]: + """Return recent notification_log rows, newest first.""" + rows = fetch_all( + """ + SELECT * FROM notification_log + ORDER BY created_at DESC + LIMIT %s + """, + (limit,), + ) + return [_normalize_log_row(r) for r in rows] \ No newline at end of file diff --git a/apps/api/app/scheduler.py b/apps/api/app/scheduler.py index 45de8bc..74748ce 100644 --- a/apps/api/app/scheduler.py +++ b/apps/api/app/scheduler.py @@ -1,7 +1,10 @@ -"""APScheduler integration: daily fetch + batch score job. +"""APScheduler integration: daily fetch + batch score + imap poll + digest. Starts during app lifespan when SCHEDULER_ENABLED=true (default false). -Runs a daily job at 07:00 that fetches postings and batch-scores pending applications. +Jobs: + - daily_fetch_score at 07:00: fetch postings + batch score + - daily_digest at 07:30: send daily digest notification + - imap_poll every 15 min: poll inbox for new emails (gated by EMAIL_WATCH_ENABLED) """ from __future__ import annotations @@ -23,6 +26,15 @@ def is_scheduler_enabled() -> bool: ) +def is_email_watch_enabled() -> bool: + """Check if email watch is enabled via env.""" + return os.environ.get("EMAIL_WATCH_ENABLED", "false").lower() in ( + "true", + "1", + "yes", + ) + + async def _daily_fetch_and_score() -> None: """Daily job: fetch postings and batch-score pending applications.""" logger.info("Scheduler: running daily fetch + batch score") @@ -56,6 +68,67 @@ async def _daily_fetch_and_score() -> None: logger.exception("Scheduler: daily job failed") +async def _daily_digest() -> None: + """Daily digest: build /today payload text and send notification.""" + logger.info("Scheduler: running daily digest") + try: + from app.notify import send_notification + from app.db import repo_app + + digest_apps = repo_app.get_digest(limit=20) + nudge_apps = repo_app.get_nudge_applications() + pending = repo_app.count_pending_approvals() + + lines: list[str] = [] + lines.append("Daily Digest") + lines.append(f"Scored applications: {len(digest_apps)}") + if digest_apps: + lines.append("") + lines.append("Top opportunities:") + for item in digest_apps[:5]: + score = item.get("score") + score_str = f" (score: {int(score)})" if score else "" + lines.append( + f" - {item.get('title', '?')} at {item.get('company', '?')}{score_str}" + ) + + if nudge_apps: + lines.append("") + lines.append(f"Follow-up nudges: {len(nudge_apps)}") + for n in nudge_apps[:5]: + lines.append( + f" - {n.get('title', '?')} at {n.get('company', '?')}" + f" ({n.get('days_since_sent', 0)} days since sent)" + ) + + if pending: + lines.append("") + lines.append(f"Pending approvals: {pending}") + + text = "\n".join(lines) + send_notification("daily_digest", text, { + "digest_count": len(digest_apps), + "nudge_count": len(nudge_apps), + "pending_approvals": pending, + }) + logger.info("Scheduler: daily digest sent") + except Exception: + logger.exception("Scheduler: daily digest failed") + + +async def _imap_poll() -> None: + """IMAP poll job: check for new emails and create suggestions.""" + if not is_email_watch_enabled(): + return + logger.info("Scheduler: running imap poll") + try: + from app.imap_watch import poll_inbox + created = poll_inbox() + logger.info("Scheduler: imap poll created %s suggestions", len(created)) + except Exception: + logger.exception("Scheduler: imap poll failed") + + def start_scheduler() -> None: """Start the APScheduler if enabled.""" global _scheduler @@ -66,6 +139,7 @@ def start_scheduler() -> None: try: from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger + from apscheduler.triggers.interval import IntervalTrigger except ImportError: logger.warning( "APScheduler not installed; scheduler will not start." @@ -79,8 +153,20 @@ def start_scheduler() -> None: id="daily_fetch_score", replace_existing=True, ) + _scheduler.add_job( + _daily_digest, + CronTrigger(hour=7, minute=30), + id="daily_digest", + replace_existing=True, + ) + _scheduler.add_job( + _imap_poll, + IntervalTrigger(minutes=15), + id="imap_poll", + replace_existing=True, + ) _scheduler.start() - logger.info("Scheduler started: daily fetch+score at 07:00") + logger.info("Scheduler started: daily fetch+score at 07:00, digest at 07:30, imap poll every 15 min") def stop_scheduler() -> None: diff --git a/apps/api/app/schemas.py b/apps/api/app/schemas.py index 47b746d..30acad0 100644 --- a/apps/api/app/schemas.py +++ b/apps/api/app/schemas.py @@ -319,4 +319,31 @@ class SeedDemoResponse(BaseModel): profile: str postings: int applications: int - sections: int \ No newline at end of file + sections: int + + +# --- v1.1: Email Suggestions --- + +class EmailSuggestionOut(BaseModel): + id: str + application_id: str | None = None + mailbox_from: str + subject: str + snippet: str + classification: str + state_proposal: str | None = None + status: str + received_at: str | None = None + created_at: str | None = None + + +# --- v1.1: Notification Log --- + +class NotificationLogOut(BaseModel): + id: str + channel: str + kind: str + payload: dict[str, Any] = {} + delivered: bool + error: str | None = None + created_at: str | None = None \ No newline at end of file diff --git a/apps/api/migrations/003_email_notify.sql b/apps/api/migrations/003_email_notify.sql new file mode 100644 index 0000000..817fc63 --- /dev/null +++ b/apps/api/migrations/003_email_notify.sql @@ -0,0 +1,24 @@ +-- 003_email_notify.sql -- email suggestions + notification log + +CREATE TABLE IF NOT EXISTS email_suggestion ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + application_id uuid REFERENCES application(id) ON DELETE SET NULL, + mailbox_from text NOT NULL, + subject text NOT NULL, + snippet text NOT NULL, + classification text NOT NULL CHECK (classification IN ('interview_invite','rejection','question','noise')), + state_proposal text, + status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','accepted','dismissed')), + received_at timestamptz NOT NULL, + created_at timestamptz DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS notification_log ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + channel text NOT NULL, + kind text NOT NULL, + payload jsonb NOT NULL, + delivered boolean NOT NULL, + error text, + created_at timestamptz DEFAULT now() +); \ No newline at end of file diff --git a/apps/api/tests/conftest.py b/apps/api/tests/conftest.py index 2235f0a..303e767 100644 --- a/apps/api/tests/conftest.py +++ b/apps/api/tests/conftest.py @@ -38,7 +38,7 @@ def _truncate_tables(): with psycopg.connect(DATABASE_URL) as conn: conn.execute( """ - TRUNCATE TABLE task_run, outbox, approval, artifact, + TRUNCATE TABLE notification_log, email_suggestion, task_run, outbox, approval, artifact, application, job_posting, cv_section, profile RESTART IDENTITY CASCADE """