"""CV Bank Taskiq tasks — profile backfill and job-opening rank. Worker: taskiq worker taskiq_management.broker_setup:broker job.candidate.bank_tasks Two jobs live here, both about the bank being useful rather than merely stored: cvbank.backfill_profiles one-off, for CVs banked before extraction existed cvbank.rank_for_job fired when a job opens, so the bank is offered up instead of waiting to be remembered """ from __future__ import annotations import logging import os from db_setup import session_scope from taskiq_management.broker_setup import MAX_RETRIES, RETRY_DELAY, broker from taskiq_management.middleware import PermanentTaskError logger = logging.getLogger("cvbank.tasks") # One agent call per CV, so a backfill of a large bank is paced across runs # rather than fired as one unbounded burst. BACKFILL_BATCH = 25 # 03:00 daily. The sweep only flags, so the exact hour does not matter; off-peak # just keeps it away from the scoring workload. RETENTION_SWEEP_CRON = os.getenv("CV_BANK_RETENTION_CRON", "0 3 * * *") @broker.task( task_name="cvbank.backfill_profiles", retry_on_error=True, max_retries=MAX_RETRIES, delay=RETRY_DELAY, ) async def backfill_bank_profiles(limit: int = BACKFILL_BATCH) -> dict: """Extract skills/title/company/years for CVs banked before migration 029. Re-runnable: rows are selected by "has no extraction yet", so a finished bank returns scanned=0 and the task becomes a no-op. Returns `remaining` so a caller can decide whether to enqueue another batch. """ from job.candidate.models import Manual_UPLOAD_CANDIDATE from job.candidate.views import extract_bank_profile_from_cv updated = 0 failed = 0 async with session_scope() as session: rows = await Manual_UPLOAD_CANDIDATE.list_bank_needing_profile( session, limit=max(1, int(limit or BACKFILL_BATCH)), ) for row in rows: # extract_bank_profile_from_cv never raises, but a bad row must not # cost the whole batch either. try: profile = await extract_bank_profile_from_cv(row.full_text) except Exception: logger.exception("bank profile backfill failed id=%s", row.id) failed += 1 continue if not any(profile.get(k) for k in ("skills", "current_position", "current_company")): continue await Manual_UPLOAD_CANDIDATE.set_bank_profile(session, row.id, profile) updated += 1 remaining = len( await Manual_UPLOAD_CANDIDATE.list_bank_needing_profile(session, limit=1) ) return {"scanned": len(rows), "updated": updated, "failed": failed, "remaining": remaining} @broker.task( task_name="cvbank.rank_for_job", retry_on_error=True, max_retries=MAX_RETRIES, delay=RETRY_DELAY, ) async def rank_bank_for_job(job_post_id: str) -> dict: """Score every banked CV against a newly opened job — tier 1, free. Deterministic keyword overlap only. No LLM call, so this runs over the whole bank on every job opening without a bill; the paid ATS score happens later and only for the handful a recruiter shortlists. """ from job.candidate.models import CvBankMatches, Manual_UPLOAD_CANDIDATE from job.job_post.models import JobPosts from matching.ranking import rank_bank_row if not job_post_id or not str(job_post_id).strip(): raise PermanentTaskError("job_post_id is required") job_post_id = str(job_post_id).strip() async with session_scope() as session: job = await JobPosts.get_job_post_by_id(session, job_post_id) if job is None or job.is_deleted: raise PermanentTaskError("job post missing or deleted") job_fields = { "title": job.title, "requirements": job.requirements, "optional_skills": job.optional_skills, } rows = await Manual_UPLOAD_CANDIDATE.list_bank_for_ranking(session) scores = [(row.id, rank_bank_row(job_fields, row)) for row in rows] await CvBankMatches.replace_for_job(session, job.id, scores) threshold = _suggest_threshold() strong = [s for _, s in scores if s >= threshold] if strong: await _notify_owner(job_post_id, len(strong)) return {"ranked": len(scores), "above_threshold": len(strong)} @broker.task( task_name="cvbank.sweep_expired", schedule=[{"cron": RETENTION_SWEEP_CRON}], ) async def sweep_expired_bank_cvs() -> dict: """Flag banked CVs past their retention window — nightly. Flags, never deletes. These are resumes a person sent us: dropping them on a timer with no record would be worse than holding them, and a wrongly configured window would silently destroy the whole bank. A human decides, the sweep only makes the decision unavoidable. Expired rows are already excluded from ranking (list_bank_for_ranking), so nothing is being surfaced to recruiters in the meantime. """ from job.candidate.models import Manual_UPLOAD_CANDIDATE async with session_scope() as session: rows = await Manual_UPLOAD_CANDIDATE.list_bank_expired(session) for row in rows: logger.info( "cv-bank retention expired id=%s banked_at=%s expired_at=%s", row.id, row.created_at.isoformat() if row.created_at else None, row.bank_expires_at.isoformat() if row.bank_expires_at else None, ) if rows: await _notify_retention_review(len(rows)) return {"expired": len(rows)} async def _notify_retention_review(count: int) -> None: """Tell whoever banked the CVs that the window has run out. Best effort — the log line above is the durable record. """ import uuid as _uuid try: from notifications.models import Notifications from users.models import Users recipient = os.getenv("CV_BANK_RETENTION_NOTIFY_EMAIL", "").strip().lower() if not recipient: return async with session_scope() as session: user = await Users.get_user_by_email(session, recipient) if user is None: return await Notifications.insert_notification(session, { "user_id": _uuid.UUID(str(user.id)), "kind": "system", "title": "CV Bank retention review", "body": ( f"{count} stored CV{'s' if count != 1 else ''} passed the retention " "window and need to be kept with a reason or deleted." ), "link_path": "/cvbank", }) except Exception: logger.exception("cv-bank retention notification failed") def _suggest_threshold() -> int: return int(os.getenv("CV_BANK_SUGGEST_THRESHOLD", "55")) async def _notify_owner(job_post_id: str, count: int) -> None: """Tell the job's recruiter the bank already holds plausible candidates. This is the whole point of ranking on job creation: without it the bank only gets searched by someone who remembers it exists. Best effort — a missing notification must never fail the ranking that has already been persisted. """ import uuid as _uuid try: from job.job_post.models import JobPosts from notifications.models import Notifications async with session_scope() as session: job = await JobPosts.get_job_post_by_id(session, job_post_id) if job is None: return raw = getattr(job, "current_recruiter_id", None) or getattr(job, "created_by", None) if not raw: return await Notifications.insert_notification(session, { "user_id": _uuid.UUID(str(raw)), "kind": "application", "title": "CVs in the bank match this job", "body": ( f"{count} stored CV{'s' if count != 1 else ''} look relevant to " f"{job.title}. Open the CV Bank to review them." ), "link_path": f"/cvbank?job={job_post_id}", "job_post_id": job.id, }) except Exception: logger.exception("cv-bank suggestion notification failed job=%s", job_post_id)