"""APScheduler integration: daily fetch + batch score + imap poll + digest. Starts during app lifespan when SCHEDULER_ENABLED=true (default false). 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 import logging import os logger = logging.getLogger(__name__) _scheduler = None def is_scheduler_enabled() -> bool: """Check if the scheduler is enabled via env.""" return os.environ.get("SCHEDULER_ENABLED", "false").lower() in ( "true", "1", "yes", ) 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") try: # Import here to avoid circular imports from app.main import _fetch_and_create_postings, _batch_score_internal # Fetch default query fetch_result = _fetch_and_create_postings( query="developer", region="Skane lan", ) logger.info( "Scheduler: fetched %s new, %s dupes", fetch_result.get("new", 0), fetch_result.get("dupes", 0), ) # Batch score all discovered applications from app.db import repo_app apps = repo_app.list_applications() discovered_ids = [ a["id"] for a in apps if a["state"] == "discovered" ] if discovered_ids: results = _batch_score_internal(discovered_ids) logger.info( "Scheduler: batch-scored %s applications", len(results) ) except Exception: 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 if not is_scheduler_enabled(): logger.info("Scheduler disabled (SCHEDULER_ENABLED != true)") return 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." ) return _scheduler = AsyncIOScheduler() _scheduler.add_job( _daily_fetch_and_score, CronTrigger(hour=7, minute=0), 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, digest at 07:30, imap poll every 15 min") def stop_scheduler() -> None: """Stop the scheduler if running.""" global _scheduler if _scheduler is not None: _scheduler.shutdown(wait=False) _scheduler = None logger.info("Scheduler stopped")