nothiro/cereal-applicant-api

1"""2Response sync service — Email classification, domain matching, confirmation watching.3 4Key rules from PRD Addendum v1.2:5- RESPONSE_SIGNALS = {interview, offer, rejection, follow_up} (NOT confirmation)6- Use employer_domain stored at index time, not guess_domain()7- classify_signal() takes both snippet AND subject8"""9 10import logging11from datetime import datetime, timedelta12from urllib.parse import urlparse13 14import httpx15 16logger = logging.getLogger("cereal.services.response_sync")17 18 19RESPONSE_SIGNALS = {"interview", "offer", "rejection", "follow_up"}20 21# Keyword-based classification (used as fallback / fast path)22INTERVIEW_KEYWORDS = [23    "interview", "phone screen", "technical assessment", "coding challenge",24    "meet the team", "next round", "schedule a call", "availability",25    "onsite", "virtual interview", "panel interview",26]27 28OFFER_KEYWORDS = [29    "offer letter", "congratulations", "pleased to offer", "job offer",30    "compensation package", "extend an offer", "welcome aboard",31]32 33REJECTION_KEYWORDS = [34    "unfortunately", "not moving forward", "other candidates",35    "decided not to proceed", "position has been filled", "regret to inform",36    "will not be moving forward", "after careful consideration",37]38 39FOLLOW_UP_KEYWORDS = [40    "following up", "checking in", "status update", "additional information",41    "next steps", "wanted to reach out", "update on your application",42]43 44CONFIRMATION_KEYWORDS = [45    "received your application", "thank you for applying",46    "application has been received", "successfully submitted",47    "we have received", "application confirmed",48]49 50 51def classify_signal(snippet: str, subject: str) -> str:52    """53    Classify an email into a signal type using keyword matching.54    Both snippet AND subject matter — check both.55    Returns: confirmation | interview | offer | rejection | follow_up56    """57    combined = f"{subject} {snippet}".lower()58 59    # Check in priority order (offer > interview > rejection > follow_up > confirmation)60    for keyword in OFFER_KEYWORDS:61        if keyword in combined:62            return "offer"63 64    for keyword in INTERVIEW_KEYWORDS:65        if keyword in combined:66            return "interview"67 68    for keyword in REJECTION_KEYWORDS:69        if keyword in combined:70            return "rejection"71 72    for keyword in FOLLOW_UP_KEYWORDS:73        if keyword in combined:74            return "follow_up"75 76    for keyword in CONFIRMATION_KEYWORDS:77        if keyword in combined:78            return "confirmation"79 80    return "confirmation"  # Default fallback81 82 83def is_response(signal_type: str) -> bool:84    """Check if a signal_type counts as a response (not confirmation)."""85    return signal_type in RESPONSE_SIGNALS86 87 88# --- ATS Domain Extraction (from PRD Addendum v1.2 Section 5) ---89 90ATS_DOMAINS = {91    "greenhouse.io", "lever.co", "ashbyhq.com", "workday.com",92    "taleo.net", "icims.com", "jobvite.com", "smartrecruiters.com",93    "breezy.hr", "bamboohr.com",94}95 96 97def extract_employer_domain(apply_url: str) -> str:98    """99    Extract root domain from apply URL.100    For ATS-hosted URLs (greenhouse.io, lever.co etc.), extract the company subdomain.101    For direct company URLs, extract the root domain.102    """103    try:104        parsed = urlparse(apply_url)105        host = parsed.netloc.lower().lstrip("www.")106        parts = host.split(".")107        root = ".".join(parts[-2:])108 109        if root in ATS_DOMAINS and len(parts) >= 3:110            # e.g. 'notion.greenhouse.io' → 'notion'111            return parts[-3]112 113        return root  # e.g. 'stripe.com', 'linear.app'114    except Exception:115        return ""116 117 118# --- App-Open Sync ---119 120async def sync_responses_for_user(user_id: str) -> dict:121    """122    App-open sync: check email for updates on recent applications.123    Targets only applications from the last 72 hours.124    """125    from services.elasticsearch import es_client126 127    client = es_client()128    since = (datetime.utcnow() - timedelta(hours=72)).isoformat()129 130    # Get recent applications that are still in 'applied' status131    result = await client.search(132        index="applications",133        body={134            "query": {135                "bool": {136                    "must": [137                        {"term": {"user_id": user_id}},138                        {"term": {"status": "applied"}},139                        {"range": {"applied_at": {"gte": since}}},140                    ]141                }142            },143            "size": 50,144        },145    )146 147    applications = result["hits"]["hits"]148    checked = 0149    updated = 0150 151    for app_hit in applications:152        app = app_hit["_source"]153        app_id = app_hit["_id"]154        employer_domain = app.get("employer_domain", "")155 156        if not employer_domain:157            checked += 1158            continue159 160        # Check email_context for messages from this employer domain161        email_result = await client.search(162            index="email_context",163            body={164                "query": {165                    "bool": {166                        "must": [167                            {"term": {"user_id": user_id}},168                            {"term": {"from_domain": employer_domain}},169                            {"range": {"date": {"gte": app.get("applied_at", since)}}},170                        ]171                    }172                },173                "sort": [{"date": {"order": "desc"}}],174                "size": 1,175            },176        )177 178        checked += 1179 180        if email_result["hits"]["hits"]:181            email = email_result["hits"]["hits"][0]["_source"]182            signal = classify_signal(183                email.get("snippet", ""),184                email.get("subject", ""),185            )186 187            # Update application status188            update_fields = {"status": signal}189            if signal != "confirmation":190                update_fields["confirmation_email_snippet"] = email.get("snippet", "")191 192            if signal in RESPONSE_SIGNALS:193                update_fields["confirmed_at"] = datetime.utcnow().isoformat()194 195            await client.update(196                index="applications",197                id=app_id,198                body={"doc": update_fields},199            )200            updated += 1201            logger.info(f"Updated application {app_id} status to {signal}")202 203    return {"checked": checked, "updated": updated}204 205 206# --- Background Confirmation Watch (called by APScheduler) ---207 208async def run_confirmation_watch():209    """210    Background job: poll all recent 'applied' applications across all users.211    Runs every 30 minutes via APScheduler.212    """213    from services.elasticsearch import es_client214 215    client = es_client()216    since = (datetime.utcnow() - timedelta(hours=72)).isoformat()217 218    # Get all recent 'applied' applications219    result = await client.search(220        index="applications",221        body={222            "query": {223                "bool": {224                    "must": [225                        {"term": {"status": "applied"}},226                        {"range": {"applied_at": {"gte": since}}},227                    ]228                }229            },230            "size": 200,231        },232    )233 234    # Group by user_id and sync each235    user_apps: dict[str, list] = {}236    for hit in result["hits"]["hits"]:237        uid = hit["_source"]["user_id"]238        if uid not in user_apps:239            user_apps[uid] = []240        user_apps[uid].append(hit)241 242    total_updated = 0243    for uid in user_apps:244        try:245            result = await sync_responses_for_user(uid)246            total_updated += result["updated"]247        except Exception as e:248            logger.error(f"Confirmation watch failed for user {uid}: {e}")249 250    logger.info(f"Confirmation watch complete: {total_updated} applications updated across {len(user_apps)} users")251