From 79098aefd544f9e4c0173c9e9f07bca781dbd159 Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Tue, 8 Sep 2026 15:25:43 +0500 Subject: [PATCH 1/2] RESCAN IMPLEMENTED --- app/models/scoring.py | 3 +- app/prompts/ats.py | 4 + backend/g_sheet/enums.py | 1 + backend/g_sheet/models.py | 27 ++ backend/g_sheet/scoring.py | 23 +- backend/inbox/app.py | 40 +++ backend/inbox/models.py | 227 +++++++++++++++++ backend/inbox/serializers.py | 28 +++ backend/inbox/tasks.py | 37 ++- backend/inbox/views.py | 238 +++++++++++++++++- backend/job/candidate/models.py | 14 ++ backend/job/candidate/plugins.py | 1 + backend/job/candidate/serializers.py | 6 + backend/job/candidate/views.py | 76 +++++- backend/job/job_post/models.py | 9 + .../manual/032_inbox_rescan_runs.sql | 26 ++ .../manual/033_professional_summary.sql | 30 +++ backend/tests/test_g_sheet_scoring.py | 2 + backend/tests/test_professional_summary.py | 83 ++++++ backend/users/models.py | 16 ++ frontend/src/api/inbox.js | 13 + frontend/src/lib/queryKeys.js | 1 + frontend/src/screens/CandidateProfile.jsx | 20 +- frontend/src/screens/Inbox.jsx | 181 ++++++++++--- frontend/src/styles/styles.css | 7 + tests/integration/test_api.py | 1 + tests/unit/test_models.py | 12 +- tests/unit/test_prompts.py | 6 + 28 files changed, 1072 insertions(+), 60 deletions(-) create mode 100644 backend/migrations/manual/032_inbox_rescan_runs.sql create mode 100644 backend/migrations/manual/033_professional_summary.sql create mode 100644 backend/tests/test_professional_summary.py diff --git a/app/models/scoring.py b/app/models/scoring.py index d05265e..18e37ef 100644 --- a/app/models/scoring.py +++ b/app/models/scoring.py @@ -55,13 +55,14 @@ class ATSScore(StrictModel): matched_keywords: list[str] = Field(default_factory=list, max_length=30) missing_keywords: list[str] = Field(default_factory=list, max_length=30) summary_critique: str = Field(min_length=1, max_length=500) + professional_summary: str | None = Field(default=None, max_length=500) @field_validator("matched_keywords", "missing_keywords", mode="before") @classmethod def _normalize(cls, value: Any) -> Any: return _normalize_keywords(value) - @field_validator("candidate_name", "job_title", "current_company", mode="before") + @field_validator("candidate_name", "job_title", "current_company", "professional_summary", mode="before") @classmethod def _blank_profile_text_to_none(cls, value: Any) -> Any: if isinstance(value, str): diff --git a/app/prompts/ats.py b/app/prompts/ats.py index 7ed2143..1b7f3a9 100644 --- a/app/prompts/ats.py +++ b/app/prompts/ats.py @@ -52,6 +52,10 @@ entries; null if neither is stated. (for example "6 years of experience"), use that stated number; otherwise compute \ whole years only from dates or durations explicitly stated in the resume; null \ whenever neither is available. +- professional_summary: one or two sentences naming the candidate's tech-stack \ +speciality and functional department from the resume alone. Ignore the job \ +description. This is not summary_critique. Null if the resume does not evidence \ +either a stack or a department. Return concise, evidence-based fields matching the supplied JSON schema. \ matched_keywords must contain only skills that appear in the resume, written with the \ diff --git a/backend/g_sheet/enums.py b/backend/g_sheet/enums.py index 55e3f46..5ee5910 100644 --- a/backend/g_sheet/enums.py +++ b/backend/g_sheet/enums.py @@ -207,6 +207,7 @@ class FormDataColumn(str, Enum): AREA_OF_RESIDENCE = "area_of_residence" RESIDING_CITY = "residing_city" CITY = "city" + PROFESSIONAL_SUMMARY = "professional_summary" RESIDING_COUNTRY = "residing_country" COMMUNICATION_SKILLS = "communication_skills" PREFERRED_TIMINGS = "preferred_timings" diff --git a/backend/g_sheet/models.py b/backend/g_sheet/models.py index 8018093..a06c006 100644 --- a/backend/g_sheet/models.py +++ b/backend/g_sheet/models.py @@ -86,6 +86,7 @@ class FormData(SQLModel, table=True): residing_city: str | None = Field(default=None) residing_country: str | None = Field(default=None) city: str | None = Field(default=None) + professional_summary: str | None = Field(default=None) reapplied: list[str] = Field(default_factory=list, sa_type=JSONB, sa_column_kwargs={"server_default": "[]"}) communication_skills: int | None = Field(default=None) preferred_timings: str | None = Field(default=None) @@ -229,6 +230,17 @@ class FormData(SQLModel, table=True): result = await session.execute(select(cls).where(cls.id == rid)) return result.scalars().first() + @classmethod + async def set_professional_summary(cls, session: AsyncSession, record_id, summary): + row = await cls.get_form_data_by_id(session, record_id) + if not row: + return None + row.professional_summary = (summary or "").strip() or None + session.add(row) + await session.commit() + await session.refresh(row) + return row + @classmethod async def get_with_job(cls, session: AsyncSession, record_id, job_post_id): """Form row + one job it may be scored against (suggested or assigned).""" @@ -390,6 +402,21 @@ class FormData(SQLModel, table=True): result = await session.execute(statement) return result.scalars().all() + @classmethod + async def list_on_hold_scan_rows(cls, session: AsyncSession, sheet=None): + """On-Hold Sheet Forms: id + email. Entire catalogue, optional sheet tab.""" + statement = select(cls.id, cls.candidate_email) + for clause in cls._filters(sheet=sheet, no_suggestions=True): + statement = statement.where(clause) + result = await session.execute(statement) + rows = [] + for record_id, email in result.all(): + rows.append({ + "id": record_id, + "email": (email or "").strip().lower() or None, + }) + return rows + @classmethod async def list_by_emails(cls, session: AsyncSession, emails): """Sheet applicants for these addresses. Promoted rows are omitted — diff --git a/backend/g_sheet/scoring.py b/backend/g_sheet/scoring.py index 593aa49..e7d49fd 100644 --- a/backend/g_sheet/scoring.py +++ b/backend/g_sheet/scoring.py @@ -17,7 +17,7 @@ from app.services.pdf import ExtractedResume from app.services.scoring import score_batch from db_setup import session_scope from g_sheet.models import FormData -from inbox.models import AtsResults +from inbox.models import AtsResults, InboxRescanRun from job.candidate.plugins import build_job_description, get_scorer, get_scoring_settings from job.job_post.models import JobPosts @@ -46,6 +46,7 @@ def serialize_form_ats(row) -> dict: "job_post_id": str(row.job_post_id) if row.job_post_id else None, "overall_score": row.overall_score, "band": row.band or None, + "professional_summary": row.professional_summary or None, "computed_at": row.computed_at.isoformat() if row.computed_at else None, } @@ -96,7 +97,7 @@ async def enqueue_form_row_scores(form_row) -> None: await enqueue_form_scores(form_row.id,job_ids) -async def score_form_against_job(form_data_id: str, job_id: str) -> dict: +async def score_form_against_job(form_data_id: str, job_id: str, rescan_run_id=None) -> dict: """Score one Sheet Forms CV against one job. Idempotent per (form, job).""" try: uuid.UUID(str(form_data_id)) @@ -143,6 +144,14 @@ async def score_form_against_job(form_data_id: str, job_id: str) -> dict: logger.warning("form ats failed form_data=%s job=%s code=%s", form_data_id, job_id, error) return {"status": "failed", "error_code": error} + summary = (result.professional_summary or "").strip() or None + run_id = None + if rescan_run_id not in (None, ""): + try: + run_id = uuid.UUID(str(rescan_run_id)) + except (TypeError, ValueError): + run_id = None + async with session_scope() as session: existing = await AtsResults.get_for_form_job(session, form_pk, job_pk) if existing is not None: @@ -150,6 +159,7 @@ async def score_form_against_job(form_data_id: str, job_id: str) -> dict: job = await JobPosts.get_job_post_by_id(session, job_pk) if job is None or job.is_deleted: return {"status": "skipped", "reason": "job_gone"} + await FormData.set_professional_summary(session, form_pk, summary) await AtsResults.insert_result(session, { "inbox_id": None, "user_id": None, @@ -159,8 +169,17 @@ async def score_form_against_job(form_data_id: str, job_id: str) -> dict: "overall_score": float(result.match_score), "band": _band(result.match_score), "model_name": settings.openai_model, + "professional_summary": summary, + "rescan_run_id": run_id, "is_current": True, }) + if run_id: + await InboxRescanRun.append_summary(session, run_id, { + "kind": "form", + "record_id": str(form_pk), + "job_post_id": str(job_pk), + "professional_summary": summary, + }) return { "status": "scored", "overall_score": result.match_score, diff --git a/backend/inbox/app.py b/backend/inbox/app.py index 9b253da..029cd06 100644 --- a/backend/inbox/app.py +++ b/backend/inbox/app.py @@ -74,6 +74,11 @@ class DuplicateBody(BaseModel): is_duplicate: bool +class OnHoldRescanBody(BaseModel): + channel: str = "all" + sheet: str | None = None + + class ReadBody(BaseModel): read: bool = True @@ -513,3 +518,38 @@ async def reply_email( raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) + + +@router.post("/inbox/rescan-on-hold") +async def start_on_hold_rescan( + payload: OnHoldRescanBody, + current_user: dict = Depends(require_permission(PermissionTag.INBOX_EDIT)), + session: AsyncSession = Depends(get_session), +): + """Score every On-Hold CV against all job posts. Poll GET until completed.""" + try: + service=Email(session=session) + data=await service.start_on_hold_rescan( + channel=payload.channel,sheet=payload.sheet,current_user=current_user, + ) + return JSONResponse(content={"data":data,"total":1,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + + +@router.get("/inbox/rescan-on-hold") +async def fetch_on_hold_rescan( + run_id: str | None = Query(None), + current_user: dict = Depends(require_permission(PermissionTag.INBOX_VIEW)), + session: AsyncSession = Depends(get_session), +): + try: + service=Email(session=session) + data=await service.get_on_hold_rescan(run_id=run_id) + return JSONResponse(content={"data":data,"total":1,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) diff --git a/backend/inbox/models.py b/backend/inbox/models.py index a1e20a2..5d01fe8 100644 --- a/backend/inbox/models.py +++ b/backend/inbox/models.py @@ -682,6 +682,8 @@ class Inbox_Messages(SQLModel, table=True): source_channel_id: int | None = Field(default=None, foreign_key="source_channels.id") processing_state: str = Field(default="unread", sa_column_kwargs={"server_default": "unread"}) city: str | None = Field(default=None) + # Job-agnostic tech-stack + department, overwritten on every completed ATS run. + professional_summary: str | None = Field(default=None) # Ingestion time — list screens order by this (newest first). server_default # backfills existing rows on the ALTER so NOT NULL is safe on a populated table. created_at: datetime = Field( @@ -1137,6 +1139,23 @@ class Inbox_Messages(SQLModel, table=True): result = await session.execute(statement) return result.scalars().all() + @classmethod + async def list_on_hold_scan_rows(cls, session: AsyncSession): + """On-Hold email applications: id + sender + CV path. No TOAST columns.""" + statement = cls._apply_filters( + select(cls.id, cls.message_from, cls.file_path), + no_suggestions=True, + ) + result = await session.execute(statement) + rows = [] + for record_id, email, file_path in result.all(): + rows.append({ + "id": record_id, + "email": (email or "").strip().lower() or None, + "file_path": (file_path or "").strip() or None, + }) + return rows + @classmethod async def get_inbox_message_by_id(cls, session: AsyncSession, record_id: str): try: @@ -1158,6 +1177,18 @@ class Inbox_Messages(SQLModel, table=True): await session.refresh(row) return row + @classmethod + async def set_professional_summary(cls, session: AsyncSession, record_id, summary): + row = await cls.get_inbox_message_by_id(session, record_id) + if not row: + return None + text=(summary or "").strip() or None + row.professional_summary = text + session.add(row) + await session.commit() + await session.refresh(row) + return row + @classmethod async def set_assigned_job_post(cls, session: AsyncSession, record_id, job_post_id): """Set or clear assigned_job_post_id; returns the row or None if missing.""" @@ -1813,6 +1844,9 @@ class AtsResults(SQLModel, table=True): job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id") # Sheet Forms scores: no inbox/user. Identity is (form_data_id, job_post_id). form_data_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="form_data.id") + # On-Hold ReScan run that produced this row. NULL for assign-job / upload scores. + rescan_run_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="inbox_rescan_runs.id") + professional_summary: str | None = Field(default=None) overall_score: float = Field(default=0.0) band: str = Field(default="") is_current: bool = Field(default=True) @@ -1961,6 +1995,96 @@ class AtsResults(SQLModel, table=True): grouped.setdefault(row.form_data_id, []).append(row) return grouped + @classmethod + async def job_ids_by_emails(cls, session: AsyncSession, emails) -> dict: + """{email: {job_post_id, ...}} for any prior ATS score of these people.""" + from g_sheet.models import FormData + + lowers = sorted({(e or "").strip().lower() for e in (emails or []) if (e or "").strip()}) + if not lowers: + return {} + grouped: dict[str, set[str]] = {} + + def _add(email, job_id): + key = (email or "").strip().lower() + if not key or job_id is None: + return + grouped.setdefault(key, set()).add(str(job_id)) + + inbox_rows = await session.execute( + select(func.lower(Inbox_Messages.message_from), cls.job_post_id) + .join(Inbox, cls.inbox_id == Inbox.id) + .join(Inbox_Messages, Inbox.message_id == Inbox_Messages.id) + .where(func.lower(Inbox_Messages.message_from).in_(lowers)) + .where(cls.job_post_id.is_not(None)) + ) + for email, job_id in inbox_rows.all(): + _add(email, job_id) + + form_rows = await session.execute( + select(func.lower(FormData.candidate_email), cls.job_post_id) + .join(FormData, cls.form_data_id == FormData.id) + .where(func.lower(FormData.candidate_email).in_(lowers)) + .where(cls.job_post_id.is_not(None)) + ) + for email, job_id in form_rows.all(): + _add(email, job_id) + + user_rows = await session.execute( + select(func.lower(Users.email), cls.job_post_id) + .join(Users, cls.user_id == Users.id) + .where(func.lower(Users.email).in_(lowers)) + .where(cls.job_post_id.is_not(None)) + ) + for email, job_id in user_rows.all(): + _add(email, job_id) + return grouped + + @classmethod + async def job_ids_for_messages(cls, session: AsyncSession, message_ids) -> dict: + """{message_id: {job_post_id, ...}} scored for these inbox_messages rows.""" + keys = [] + for raw in message_ids or []: + uid = cls._as_uuid(raw) + if uid is not None: + keys.append(uid) + if not keys: + return {} + result = await session.execute( + select(Inbox.message_id, cls.job_post_id) + .join(Inbox, cls.inbox_id == Inbox.id) + .where(Inbox.message_id.in_(keys)) + .where(cls.job_post_id.is_not(None)) + ) + grouped: dict[str, set[str]] = {} + for mid, job_id in result.all(): + if mid is None or job_id is None: + continue + grouped.setdefault(str(mid), set()).add(str(job_id)) + return grouped + + @classmethod + async def job_ids_for_forms(cls, session: AsyncSession, form_ids) -> dict: + """{form_data_id: {job_post_id, ...}} scored for these Sheet Forms rows.""" + keys = [] + for raw in form_ids or []: + uid = cls._as_uuid(raw) + if uid is not None: + keys.append(uid) + if not keys: + return {} + result = await session.execute( + select(cls.form_data_id, cls.job_post_id) + .where(cls.form_data_id.in_(keys)) + .where(cls.job_post_id.is_not(None)) + ) + grouped: dict[str, set[str]] = {} + for fid, job_id in result.all(): + if fid is None or job_id is None: + continue + grouped.setdefault(str(fid), set()).add(str(job_id)) + return grouped + @classmethod async def resolve_identity(cls, session: AsyncSession, email, candidate_id): """XOR identity for a score row from the scored candidate's email. @@ -2046,6 +2170,16 @@ class AtsResults(SQLModel, table=True): await session.commit() return row + @classmethod + async def list_by_rescan_run(cls, session: AsyncSession, rescan_run_id): + uid = cls._as_uuid(rescan_run_id) + if uid is None: + return [] + result = await session.execute( + select(cls).where(cls.rescan_run_id == uid).order_by(cls.computed_at.desc()) + ) + return list(result.scalars().all()) + class MailboxSyncRun(SQLModel, table=True): """One Outlook mailbox sync job — survives tab close because work runs in Taskiq. @@ -2125,3 +2259,96 @@ class MailboxSyncRun(SQLModel, table=True): await session.commit() await session.refresh(row) return row + + +class InboxRescanRun(SQLModel, table=True): + """On-Hold catalogue ATS rescan — one row the Inbox ReScan button polls.""" + + __tablename__ = "inbox_rescan_runs" + + id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) + status: str = Field(default="queued", index=True) # queued|running|scoring|completed|failed + channel: str = Field(default="all") + sheet: str | None = Field(default=None) + task_id: str | None = Field(default=None) + created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id") + job_count: int = Field(default=0) + candidate_count: int = Field(default=0) + skipped_candidates: int = Field(default=0) + skipped_pairs: int = Field(default=0) + pair_count: int = Field(default=0) + done_count: int = Field(default=0) + entries: list | None = Field(default=None, sa_column=Column(JSONB)) + summaries: list = Field(default_factory=list, sa_type=JSONB, sa_column_kwargs={"server_default": "[]"}) + error: str | None = Field(default=None) + created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) + started_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) + finished_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) + + @staticmethod + def _as_uuid(record_id) -> uuid.UUID | None: + if record_id in (None, ""): + return None + try: + return uuid.UUID(str(record_id)) + except ValueError: + return None + + @classmethod + async def get_by_id(cls, session: AsyncSession, record_id): + uid = cls._as_uuid(record_id) + if uid is None: + return None + result = await session.execute(select(cls).where(cls.id == uid)) + return result.scalars().first() + + @classmethod + async def get_active(cls, session: AsyncSession): + result = await session.execute( + select(cls) + .where(cls.status.in_(("queued", "running", "scoring"))) + .order_by(cls.created_at.desc()) + ) + return result.scalars().first() + + @classmethod + async def get_latest(cls, session: AsyncSession): + result = await session.execute( + select(cls).order_by(cls.created_at.desc()).limit(1) + ) + return result.scalars().first() + + @classmethod + async def insert_run(cls, session: AsyncSession, fields: dict, *, commit: bool = True): + row = cls(**fields) + session.add(row) + if commit: + await session.commit() + await session.refresh(row) + return row + + @classmethod + async def update_run(cls, session: AsyncSession, record_id, fields: dict, *, commit: bool = True): + row = await cls.get_by_id(session, record_id) + if not row: + return None + for key, value in fields.items(): + setattr(row, key, value) + session.add(row) + if commit: + await session.commit() + await session.refresh(row) + return row + + @classmethod + async def append_summary(cls, session: AsyncSession, record_id, entry): + row = await cls.get_by_id(session, record_id) + if not row: + return None + items = list(row.summaries or []) + items.append(entry) + row.summaries = items + session.add(row) + await session.commit() + await session.refresh(row) + return row diff --git a/backend/inbox/serializers.py b/backend/inbox/serializers.py index 843a9d3..c2c63d8 100644 --- a/backend/inbox/serializers.py +++ b/backend/inbox/serializers.py @@ -84,6 +84,7 @@ def serialize_message(message: Inbox_Messages, *, linkedin_url=None) -> dict: "resume_text": message.resume_text, "ats_score": message.ats_score, "ats_band": message.ats_band or None, + "professional_summary": message.professional_summary or None, } @@ -93,6 +94,7 @@ def serialize_ats_result(row) -> dict: "job_post_id": str(row.job_post_id) if row.job_post_id else None, "overall_score": row.overall_score, "band": row.band or None, + "professional_summary": row.professional_summary or None, "computed_at": row.computed_at.isoformat() if row.computed_at else None, } @@ -151,6 +153,7 @@ def serialize_application(message: Inbox_Messages, *, linkedin_url=None, light: "matched_at": message.matched_at.isoformat() if message.matched_at else None, "ats_score": message.ats_score, "ats_band": message.ats_band or None, + "professional_summary": message.professional_summary or None, "phone": message.candidate_phone_number, "experience": message.experience or "", "current_employment": message.current_employment or "", @@ -212,3 +215,28 @@ def serialize_mailbox_sync_run(row) -> dict: "started_at": row.started_at.isoformat() if row.started_at else None, "finished_at": row.finished_at.isoformat() if row.finished_at else None, } + + +def serialize_inbox_rescan_run(row) -> dict: + """Progress for the On-Hold catalogue ATS rescan, plus per-pair summaries.""" + pair_count = int(row.pair_count or 0) + done_count = int(row.done_count or 0) + return { + "id": str(row.id), + "status": row.status, + "channel": row.channel, + "sheet": row.sheet, + "task_id": row.task_id, + "created_by": str(row.created_by) if row.created_by else None, + "job_count": int(row.job_count or 0), + "candidate_count": int(row.candidate_count or 0), + "skipped_candidates": int(row.skipped_candidates or 0), + "skipped_pairs": int(row.skipped_pairs or 0), + "pair_count": pair_count, + "done_count": done_count, + "summaries": list(row.summaries or []), + "error": row.error, + "created_at": row.created_at.isoformat() if row.created_at else None, + "started_at": row.started_at.isoformat() if row.started_at else None, + "finished_at": row.finished_at.isoformat() if row.finished_at else None, + } diff --git a/backend/inbox/tasks.py b/backend/inbox/tasks.py index b7cf2bc..5c4b42a 100644 --- a/backend/inbox/tasks.py +++ b/backend/inbox/tasks.py @@ -22,7 +22,7 @@ logger=logging.getLogger("inbox.tasks") _DONE=frozenset({"matched","skipped","no_text","failed","dlq"}) -async def score_message_against_job(record_id:str,job_id:str) -> dict: +async def score_message_against_job(record_id:str,job_id:str,rescan_run_id=None) -> dict: """ATS-score one inbox CV against one job post — the no-upload path. The decoded attachment already on disk is the CV; the job post in the @@ -57,7 +57,7 @@ async def score_message_against_job(record_id:str,job_id:str) -> dict: try: # Attribute the rows to the job's owner — there is no request user # in a background task. - results=await service.score_inbox(job_id,[record_id],{"id":str(job.created_by)}) + results=await service.score_inbox(job_id,[record_id],{"id":str(job.created_by)},rescan_run_id=rescan_run_id) except HTTPException as exc: # 400/404 from score_inbox are permanent (no attachment, bad ids); # retrying cannot fix them. @@ -138,6 +138,39 @@ async def score_inbox_message(record_id:str,job_id:str) -> dict: return await score_message_against_job(record_id,job_id) +@broker.task( + task_name="inbox.rescan_on_hold", + retry_on_error=True, + max_retries=MAX_RETRIES, + delay=RETRY_DELAY, +) +async def rescan_on_hold_run(run_id:str,cursor:int=0) -> dict: + """Score On-Hold CVs against every job post, in short idempotent chunks.""" + from g_sheet.scoring import score_form_against_job + from inbox.views import Email + + async with session_scope() as session: + prepared=await Email(session=session).prepare_on_hold_rescan_chunk(run_id,cursor) + status=prepared.get("status") + if status in ("completed","failed","missing"): + return prepared + for item in prepared.get("batch") or []: + kind=item.get("kind") + record_id=item.get("record_id") + job_id=item.get("job_id") + try: + if kind=="form": + await score_form_against_job(record_id,job_id,run_id) + else: + await score_message_against_job(record_id,job_id,run_id) + except Exception as exc: + logger.warning("on-hold rescan failed for %s vs %s: %s",record_id,job_id,exc) + async with session_scope() as session: + return await Email(session=session).finish_on_hold_rescan_chunk( + run_id,prepared.get("next_cursor") or 0,bool(prepared.get("more")), + ) + + @broker.task( task_name="inbox.match_message", retry_on_error=True, diff --git a/backend/inbox/views.py b/backend/inbox/views.py index 0c0933c..834bdba 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -4,9 +4,9 @@ import uuid import httpx,os from fastapi import HTTPException from inbox.enums import Candidate_application_Status -from inbox.models import Inbox_Messages,Inbox_Message_Triage,MailboxSyncRun,Inbox,SourceChannels,AtsResults +from inbox.models import Inbox_Messages,Inbox_Message_Triage,MailboxSyncRun,InboxRescanRun,Inbox,SourceChannels,AtsResults from inbox.file_decoder import extract_pdf_attachments -from inbox.serializers import serialize_application, serialize_message, serialize_triage, serialize_mailbox_sync_run, serialize_ats_result +from inbox.serializers import serialize_application, serialize_message, serialize_triage, serialize_mailbox_sync_run, serialize_inbox_rescan_run, serialize_ats_result from inbox.plugins import ( EMAIL_API_TOKEN, attach_email_pdfs_to_s3, @@ -426,6 +426,240 @@ class Email: raise HTTPException(status_code=404,detail="No sync runs yet") return serialize_mailbox_sync_run(row) + async def start_on_hold_rescan(self,channel="all",sheet=None,current_user=None): + """Queue an On-Hold catalogue ATS scan against every job_posts row. + + Returns an existing queued/running/scoring run instead of stacking another. + """ + channel=(channel or "all").strip().lower() + if channel not in ("all","email","forms"): + raise HTTPException(status_code=422,detail="channel must be all, email, or forms") + sheet=(sheet or "").strip() or None + if channel!="forms": + sheet=None + + active=await InboxRescanRun.get_active(self.session) + if active: + return serialize_inbox_rescan_run(active) + + created_by=None + if isinstance(current_user,dict) and current_user.get("id"): + created_by=InboxRescanRun._as_uuid(current_user.get("id")) + row=await InboxRescanRun.insert_run(self.session,{ + "status":"queued", + "channel":channel, + "sheet":sheet, + "created_by":created_by, + }) + from inbox.tasks import rescan_on_hold_run + try: + task=await rescan_on_hold_run.kicker().with_labels( + created_at=datetime.now(timezone.utc).isoformat(), + correlation_id=str(row.id), + queue="inbox", + ).kiq(str(row.id),0) + except Exception as exc: + await InboxRescanRun.update_run(self.session,row.id,{ + "status":"failed", + "error":str(exc), + "finished_at":datetime.now(timezone.utc), + }) + raise HTTPException(status_code=503,detail="Could not queue On-Hold rescan") from exc + row=await InboxRescanRun.update_run(self.session,row.id,{"task_id":task.task_id}) + return serialize_inbox_rescan_run(row) + + async def get_on_hold_rescan(self,run_id=None): + if run_id: + row=await InboxRescanRun.get_by_id(self.session,run_id) + if not row: + raise HTTPException(status_code=404,detail="Rescan run not found") + return serialize_inbox_rescan_run(row) + row=await InboxRescanRun.get_active(self.session) + if row: + return serialize_inbox_rescan_run(row) + row=await InboxRescanRun.get_latest(self.session) + if not row: + return None + return serialize_inbox_rescan_run(row) + + async def plan_on_hold_pairs(self,channel,sheet=None): + """Build (candidate, job) pairs that have never been ATS-scored. + + Skip a candidate who already has a score against an active job. + Skip a pair that already exists on ats_results for that person/row. + """ + from g_sheet.models import FormData + from job.job_post.models import JobPosts + + job_ids=[str(jid) for jid in await JobPosts.list_ids(self.session)] + active_ids={str(jid) for jid in await JobPosts.list_ids(self.session,active_only=True)} + inbox_rows=[] + form_rows=[] + if channel in ("all","email"): + inbox_rows=await Inbox_Messages.list_on_hold_scan_rows(self.session) + if channel in ("all","forms"): + form_rows=await FormData.list_on_hold_scan_rows(self.session,sheet=sheet) + + emails=[] + for row in inbox_rows: + if row.get("email"): + emails.append(row["email"]) + for row in form_rows: + if row.get("email"): + emails.append(row["email"]) + scored_by_email=await AtsResults.job_ids_by_emails(self.session,emails) + scored_by_message=await AtsResults.job_ids_for_messages( + self.session,[row["id"] for row in inbox_rows], + ) + scored_by_form=await AtsResults.job_ids_for_forms( + self.session,[row["id"] for row in form_rows], + ) + skipped_active=set() + for email,jobs in scored_by_email.items(): + if jobs & active_ids: + skipped_active.add(email) + known_by_email={key:set(jobs) for key,jobs in scored_by_email.items()} + + pairs=[] + skipped_candidates=0 + skipped_pairs=0 + + def _already(email,row_jobs): + jobs=set(row_jobs or ()) + if email: + jobs |= known_by_email.get(email) or set() + return jobs + + def _mark(email,job_id): + if email: + known_by_email.setdefault(email,set()).add(job_id) + + def _skip_for_active(email,row_jobs): + if email and email in skipped_active: + return True + if (row_jobs or set()) & active_ids: + return True + return False + + for row in inbox_rows: + email=row.get("email") + mid=str(row["id"]) + row_jobs=scored_by_message.get(mid) or set() + if not row.get("file_path"): + skipped_candidates += 1 + continue + if _skip_for_active(email,row_jobs): + skipped_candidates += 1 + continue + known=_already(email,row_jobs) + for job_id in job_ids: + if job_id in known: + skipped_pairs += 1 + continue + pairs.append({"kind":"inbox","record_id":mid,"job_id":job_id}) + known.add(job_id) + _mark(email,job_id) + + for row in form_rows: + email=row.get("email") + fid=str(row["id"]) + row_jobs=scored_by_form.get(fid) or set() + if _skip_for_active(email,row_jobs): + skipped_candidates += 1 + continue + known=_already(email,row_jobs) + for job_id in job_ids: + if job_id in known: + skipped_pairs += 1 + continue + pairs.append({"kind":"form","record_id":fid,"job_id":job_id}) + known.add(job_id) + _mark(email,job_id) + + return { + "job_count":len(job_ids), + "candidate_count":len(inbox_rows)+len(form_rows), + "skipped_candidates":skipped_candidates, + "skipped_pairs":skipped_pairs, + "pairs":pairs, + } + + async def prepare_on_hold_rescan_chunk(self,run_id,cursor=0): + """Plan pairs if needed and return the next batch. Scoring happens outside.""" + row=await InboxRescanRun.get_by_id(self.session,run_id) + if row is None: + return {"status":"missing"} + if row.status in ("failed","completed"): + return {"status":row.status} + + now=datetime.now(timezone.utc) + entries=list(row.entries or []) + if not entries and int(cursor or 0)==0: + await InboxRescanRun.update_run(self.session,run_id,{ + "status":"running", + "started_at":row.started_at or now, + }) + plan=await self.plan_on_hold_pairs(row.channel,row.sheet) + entries=plan["pairs"] + await InboxRescanRun.update_run(self.session,run_id,{ + "status":"scoring" if entries else "completed", + "job_count":plan["job_count"], + "candidate_count":plan["candidate_count"], + "skipped_candidates":plan["skipped_candidates"], + "skipped_pairs":plan["skipped_pairs"], + "pair_count":len(entries), + "done_count":0, + "entries":entries, + "finished_at":None if entries else now, + }) + if not entries: + return {"status":"completed","pair_count":0,"batch":[]} + + if not entries: + await InboxRescanRun.update_run(self.session,run_id,{ + "status":"failed", + "error":"Rescan has no stored pairs", + "finished_at":datetime.now(timezone.utc), + }) + return {"status":"failed","batch":[]} + + chunk=4 + start=int(cursor or 0) + batch=entries[start:start+chunk] + return { + "status":"scoring", + "batch":batch, + "pair_count":len(entries), + "next_cursor":min(start+len(batch),len(entries)), + "more":(start+len(batch)) dict: "matched_keywords": list(row.matched_keywords or []), "missing_keywords": list(row.missing_keywords or []), "summary_critique": row.summary_critique, + "professional_summary": row.professional_summary or None, "linkedin_url": row.linkedin_url or None, "status": row.status, "error_code": row.error_code, @@ -208,6 +209,10 @@ def serialize_candidate_profile( "match_status": message.match_status if message else None, "match_error": message.match_error if message else None, "matched_at": message.matched_at.isoformat() if message and message.matched_at else None, + "professional_summary": ( + (message.professional_summary if message else None) + or (user.professional_summary if user else None) + ), "file_path": _first_file_path(message.file_path if message else None), "job_posts": [], } @@ -278,6 +283,7 @@ def serialize_manual_candidate_profile(row, user, job_post) -> Dict[str, Any]: "match_status": None, "match_error": None, "matched_at": None, + "professional_summary": row.professional_summary or (user.professional_summary if user else None), "job_posts": [job_payload] if job_payload else [], "favorite": None, "rating": None, diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index d2ff7dd..0106187 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -9,7 +9,7 @@ from app.core.errors import ATSError,ErrorCode from app.models.scoring import CompletedCandidate from app.services.pdf import extract_resume,sanitize_filename from app.services.scoring import score_batch -from inbox.models import Inbox_Messages,Inbox,Inbox_Message_Triage,AtsResults +from inbox.models import Inbox_Messages,Inbox,Inbox_Message_Triage,InboxRescanRun,AtsResults from job.candidate.models import Candidates from job.candidate.plugins import ( FILE_NOT_FOUND, @@ -632,7 +632,7 @@ class CandidateScoring: sources.append(source) return await self._score_and_persist(job_id,sources,"upload",current_user) - async def score_inbox(self,job_id,message_ids,current_user): + async def score_inbox(self,job_id,message_ids,current_user,rescan_run_id=None): """Score PDF attachments of inbox messages (S3 URLs or legacy local paths).""" from inbox.plugins import load_file_bytes @@ -671,7 +671,7 @@ class CandidateScoring: sources.append(source) if not sources: raise HTTPException(status_code=400,detail="No attachments found for the given message(s)") - return await self._score_and_persist(job_id,sources,"inbox",current_user) + return await self._score_and_persist(job_id,sources,"inbox",current_user,rescan_run_id=rescan_run_id) async def score_bank(self,job_id,record_ids,current_user): """ATS-score CVs already sitting in the bank — tier 2, the paid step. @@ -747,7 +747,7 @@ class CandidateScoring: data=serialize_candidate(row) return await CandidateView(session=self.session).attach_application_history(data) - async def _score_and_persist(self,job_id,sources,source_kind,current_user): + async def _score_and_persist(self,job_id,sources,source_kind,current_user,rescan_run_id=None): job=await JobPosts.get_job_post_by_id(self.session,job_id) if job is None or job.is_deleted: raise HTTPException(status_code=404,detail="Job post not found") @@ -790,7 +790,7 @@ class CandidateScoring: ): await self.session.commit() rows.append(row) - await self._sync_ats_results(source_kind,job,rows,sources,current_user) + await self._sync_ats_results(source_kind,job,rows,sources,current_user,rescan_run_id=rescan_run_id) rows.sort(key=lambda r:(0,-(r.match_score or 0)) if r.status=="completed" else (1,0)) return [serialize_candidate(row) for row in rows] @@ -833,7 +833,8 @@ class CandidateScoring: fields_by_slot[slot]=candidate_failed_fields(source,result.error_code,result.error_message) return fields_by_slot - async def _sync_ats_results(self,source_kind,job,rows,sources,current_user=None): + async def _sync_ats_results(self,source_kind,job,rows,sources,current_user=None,rescan_run_id=None): + run_id=self._as_rescan_run_id(rescan_run_id) if source_kind=="inbox": best={} for source,row in zip(sources,rows): @@ -844,21 +845,50 @@ class CandidateScoring: best[mid]=row for message_id,row in best.items(): try: - await self._sync_inbox_ats(message_id,job,row,current_user=current_user) + await self._sync_inbox_ats(message_id,job,row,current_user=current_user,rescan_run_id=run_id) except Exception: await self.session.rollback() logger.exception("inbox ATS denorm failed for message %s",message_id) return - for row in rows: + for source,row in zip(sources,rows): if row.status!="completed": continue try: - await self._sync_upload_ats(job,row,current_user=current_user) + await self._sync_upload_ats(job,row,source,current_user=current_user,rescan_run_id=run_id) except Exception: await self.session.rollback() logger.exception("upload ATS history failed for candidate %s",row.id) - async def _sync_inbox_ats(self,message_id,job,row,current_user=None): + @staticmethod + def _as_rescan_run_id(value): + if value in (None,""): + return None + try: + return uuid.UUID(str(value)) + except (TypeError,ValueError): + return None + + @staticmethod + def _summary_text(row): + return (getattr(row,"professional_summary",None) or "").strip() or None + + async def _stamp_identity_summary(self,row,summary): + identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id) + if identity.get("user_id"): + await Users.set_professional_summary(self.session,identity["user_id"],summary) + return identity + + async def _append_rescan_summary(self,rescan_run_id,kind,record_id,job,summary): + if not rescan_run_id: + return + await InboxRescanRun.append_summary(self.session,rescan_run_id,{ + "kind":kind, + "record_id":str(record_id), + "job_post_id":str(job.id), + "professional_summary":summary, + }) + + async def _sync_inbox_ats(self,message_id,job,row,current_user=None,rescan_run_id=None): """Land a completed score on inbox_messages / inbox / ats_results. message_id is the scoring call's known inbox_messages PK, not a column @@ -868,6 +898,7 @@ class CandidateScoring: if msg is None: return band=CandidateView._recommendation(row.match_score) or "" + summary=self._summary_text(row) denorm=True assigned=msg.assigned_job_post_id link=await Inbox.get_inbox_by_message_id(self.session,message_id) @@ -876,11 +907,12 @@ class CandidateScoring: denorm=existing is None if denorm: await Inbox_Messages.set_ats_score(self.session,message_id,row.match_score,band) + await Inbox_Messages.set_professional_summary(self.session,message_id,summary) + identity=await self._stamp_identity_summary(row,summary) if link is None: return old=await AtsResults.get_current_for_inbox(self.session,link.id) old_score=old.overall_score if old else None - identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id) await AtsResults.insert_result(self.session,{ "inbox_id":link.id, **identity, @@ -888,8 +920,11 @@ class CandidateScoring: "overall_score":float(row.match_score), "band":band, "model_name":row.model, + "professional_summary":summary, + "rescan_run_id":rescan_run_id, "is_current":True, }) + await self._append_rescan_summary(rescan_run_id,"email",message_id,job,summary) await HistoryRecorder(self.session).record( HistoryEvent.ATS_SCORED.value, current_user=current_user,user_id=link.user_id,inbox_id=link.id, @@ -898,10 +933,18 @@ class CandidateScoring: description=f"{band} against {job.title or 'job'}",commit=True, ) - async def _sync_upload_ats(self,job,row,current_user=None): + async def _sync_upload_ats(self,job,row,source=None,current_user=None,rescan_run_id=None): """History row for an upload-sourced score — inbox_id stays NULL.""" band=CandidateView._recommendation(row.match_score) or "" - identity=await AtsResults.resolve_identity(self.session,row.candidate_email,row.id) + summary=self._summary_text(row) + identity=await self._stamp_identity_summary(row,summary) + mid=(source or {}).get("manual_upload_candidate_id") + if not mid and row.candidate_email: + manual=await Manual_UPLOAD_CANDIDATE.get_by_email_and_job(self.session,row.candidate_email,job.id) + if manual: + mid=manual.id + if mid: + await Manual_UPLOAD_CANDIDATE.set_professional_summary(self.session,mid,summary) old=None if identity.get("user_id"): old=await AtsResults.get_current_for_user(self.session,identity["user_id"],job.id) @@ -915,8 +958,11 @@ class CandidateScoring: "overall_score":float(row.match_score), "band":band, "model_name":row.model, + "professional_summary":summary, + "rescan_run_id":rescan_run_id, "is_current":True, }) + await self._append_rescan_summary(rescan_run_id,"upload",mid or row.id,job,summary) await HistoryRecorder(self.session).record( HistoryEvent.ATS_SCORED.value, current_user=current_user,user_id=identity.get("user_id"), @@ -1453,6 +1499,8 @@ class CandidateView: base["candidate_id"]=str(chosen.candidate_id) if chosen.candidate_id else None if chosen.user_id and not base.get("user_id"): base["user_id"]=str(chosen.user_id) + if chosen.professional_summary and not base.get("professional_summary"): + base["professional_summary"]=chosen.professional_summary scored=None if chosen.candidate_id: scored=await Candidates.get_candidate_by_id(self.session,str(chosen.candidate_id)) @@ -1464,6 +1512,8 @@ class CandidateView: base["matched_keywords"]=list(scored.matched_keywords or []) base["missing_keywords"]=list(scored.missing_keywords or []) base["summary_critique"]=scored.summary_critique + if scored.professional_summary and not base.get("professional_summary"): + base["professional_summary"]=scored.professional_summary else: for record in records: score,band=self._score_from_message(record) diff --git a/backend/job/job_post/models.py b/backend/job/job_post/models.py index fafa934..2708396 100644 --- a/backend/job/job_post/models.py +++ b/backend/job/job_post/models.py @@ -109,6 +109,15 @@ class JobPosts(SQLModel, table=True): ) return result.scalars().all() + @classmethod + async def list_ids(cls, session: AsyncSession, *, active_only: bool = False): + """Non-deleted job_posts.id values. active_only limits to live openings.""" + statement = select(cls.id).where(cls.is_deleted == False) # noqa: E712 + if active_only: + statement = statement.where(cls.is_active == True) # noqa: E712 + result = await session.execute(statement) + return [row[0] for row in result.all()] + @classmethod async def get_by_ids(cls, session: AsyncSession, ids: list[str], *, active_only: bool = True): uids = [] diff --git a/backend/migrations/manual/032_inbox_rescan_runs.sql b/backend/migrations/manual/032_inbox_rescan_runs.sql new file mode 100644 index 0000000..2b66262 --- /dev/null +++ b/backend/migrations/manual/032_inbox_rescan_runs.sql @@ -0,0 +1,26 @@ +-- 032_inbox_rescan_runs.sql +-- On-Hold catalogue ATS rescan: one run row so the Inbox ReScan button can +-- poll progress while workers score (candidate, job_post) pairs. + +CREATE TABLE IF NOT EXISTS app.inbox_rescan_runs ( + id uuid PRIMARY KEY, + status varchar NOT NULL DEFAULT 'queued', + channel varchar NOT NULL DEFAULT 'all', + sheet varchar, + task_id varchar, + created_by uuid REFERENCES app.users (id), + job_count integer NOT NULL DEFAULT 0, + candidate_count integer NOT NULL DEFAULT 0, + skipped_candidates integer NOT NULL DEFAULT 0, + skipped_pairs integer NOT NULL DEFAULT 0, + pair_count integer NOT NULL DEFAULT 0, + done_count integer NOT NULL DEFAULT 0, + entries jsonb, + error text, + created_at timestamptz NOT NULL DEFAULT now(), + started_at timestamptz, + finished_at timestamptz +); + +CREATE INDEX IF NOT EXISTS ix_inbox_rescan_runs_status + ON app.inbox_rescan_runs (status); diff --git a/backend/migrations/manual/033_professional_summary.sql b/backend/migrations/manual/033_professional_summary.sql new file mode 100644 index 0000000..8e3d972 --- /dev/null +++ b/backend/migrations/manual/033_professional_summary.sql @@ -0,0 +1,30 @@ +-- 033_professional_summary.sql +-- Job-agnostic professional summary (tech-stack speciality + department). +-- Latest denorm on each intake row; per-run snapshot on ats_results. +-- ats_results.rescan_run_id is the 1:N link from an On-Hold ReScan run. +-- Applied at startup by alembic_setup.run_manual_sql(). + +ALTER TABLE app.inbox_messages + ADD COLUMN IF NOT EXISTS professional_summary TEXT; + +ALTER TABLE app.form_data + ADD COLUMN IF NOT EXISTS professional_summary TEXT; + +ALTER TABLE app.manual_upload_candidate + ADD COLUMN IF NOT EXISTS professional_summary TEXT; + +ALTER TABLE app.users + ADD COLUMN IF NOT EXISTS professional_summary TEXT; + +ALTER TABLE app.candidates + ADD COLUMN IF NOT EXISTS professional_summary TEXT; + +ALTER TABLE app.ats_results + ADD COLUMN IF NOT EXISTS professional_summary TEXT, + ADD COLUMN IF NOT EXISTS rescan_run_id UUID REFERENCES app.inbox_rescan_runs (id); + +ALTER TABLE app.inbox_rescan_runs + ADD COLUMN IF NOT EXISTS summaries JSONB NOT NULL DEFAULT '[]'::jsonb; + +CREATE INDEX IF NOT EXISTS ix_ats_results_rescan_run_id + ON app.ats_results (rescan_run_id); diff --git a/backend/tests/test_g_sheet_scoring.py b/backend/tests/test_g_sheet_scoring.py index 585a149..ae1fec9 100644 --- a/backend/tests/test_g_sheet_scoring.py +++ b/backend/tests/test_g_sheet_scoring.py @@ -34,12 +34,14 @@ def test_serialize_form_ats_shape(): job_post_id = job_id overall_score = 81.0 band = "Potential Match" + professional_summary = "Python backend in Engineering." computed_at = None assert serialize_form_ats(Row()) == { "job_post_id": str(job_id), "overall_score": 81.0, "band": "Potential Match", + "professional_summary": "Python backend in Engineering.", "computed_at": None, } diff --git a/backend/tests/test_professional_summary.py b/backend/tests/test_professional_summary.py new file mode 100644 index 0000000..a56d243 --- /dev/null +++ b/backend/tests/test_professional_summary.py @@ -0,0 +1,83 @@ +"""Professional summary — persist fields and serializer shape. No DB.""" + +from types import SimpleNamespace +from uuid import uuid4 + +from g_sheet.scoring import serialize_form_ats +from inbox.serializers import serialize_ats_result, serialize_inbox_rescan_run +from job.candidate.plugins import candidate_completed_fields, candidate_failed_fields + + +def test_failed_fields_omit_professional_summary(): + source = {"safe_name": "a.pdf", "file_path": None, "sha256": "x"} + fields = candidate_failed_fields(source, "PDF_TEXT_UNAVAILABLE", "empty") + assert "professional_summary" not in fields + + +def test_completed_fields_include_professional_summary(): + result = SimpleNamespace( + candidate_name="Ada", + job_title="Engineer", + current_company="Acme", + years_experience=6, + match_score=82, + matched_keywords=["Python"], + missing_keywords=["AWS"], + summary_critique="Strong Python, missing cloud.", + professional_summary="Python backend specialist in Engineering.", + ) + source = {"safe_name": "a.pdf", "file_path": None, "sha256": "x"} + fields = candidate_completed_fields(source, result) + assert fields["professional_summary"] == "Python backend specialist in Engineering." + + +def test_serialize_ats_result_includes_summary(): + job_id = uuid4() + row = SimpleNamespace( + job_post_id=job_id, + overall_score=81.0, + band="Potential Match", + professional_summary="Python backend in Engineering.", + computed_at=None, + ) + assert serialize_ats_result(row) == { + "job_post_id": str(job_id), + "overall_score": 81.0, + "band": "Potential Match", + "professional_summary": "Python backend in Engineering.", + "computed_at": None, + } + assert serialize_form_ats(row) == serialize_ats_result(row) + + +def test_serialize_inbox_rescan_run_includes_summaries(): + run_id = uuid4() + entry = { + "kind": "email", + "record_id": str(uuid4()), + "job_post_id": str(uuid4()), + "professional_summary": "Python backend in Engineering.", + } + row = SimpleNamespace( + id=run_id, + status="completed", + channel="all", + sheet=None, + task_id=None, + created_by=None, + job_count=2, + candidate_count=1, + skipped_candidates=0, + skipped_pairs=0, + pair_count=1, + done_count=1, + summaries=[entry], + error=None, + created_at=None, + started_at=None, + finished_at=None, + ) + payload = serialize_inbox_rescan_run(row) + assert payload["summaries"] == [entry] + assert payload["id"] == str(run_id) + assert payload["done_count"] == 1 diff --git a/backend/users/models.py b/backend/users/models.py index be94d9c..a5b1f90 100644 --- a/backend/users/models.py +++ b/backend/users/models.py @@ -62,6 +62,7 @@ class Users(SQLModel, table=True): # mentions LinkedIn; never overwrite a stored value with empty. linkedin_url: str | None = Field(default=None) city: str | None = Field(default=None) + professional_summary: str | None = Field(default=None) # Prior job_post_ids for this email across candidate tables. [] until a # later application finds an already-linked job. reapplied: list[str] = Field(default_factory=list, sa_type=JSONB, sa_column_kwargs={"server_default": "[]"}) @@ -354,5 +355,20 @@ class Users(SQLModel, table=True): await session.commit() return updated + @classmethod + async def set_professional_summary(cls, session: AsyncSession, user_id, summary): + uid = cls._as_uuid(user_id) + if uid is None: + return None + user = await cls.get_user_id(session, uid) + if user is None: + return None + user.professional_summary = (summary or "").strip() or None + user.updated_at = _now() + session.add(user) + await session.commit() + await session.refresh(user) + return user + import job.candidate.models as _candidate_models # noqa: E402, F401 diff --git a/frontend/src/api/inbox.js b/frontend/src/api/inbox.js index ef00b41..ed168d8 100644 --- a/frontend/src/api/inbox.js +++ b/frontend/src/api/inbox.js @@ -81,6 +81,19 @@ export function getMailboxSync(runId) { return request('/email/sync/fetch', { params: { run_id: runId } }) } +/** Score every On-Hold CV against all job posts. Returns a run immediately. */ +export function startOnHoldRescan({ channel = 'all', sheet } = {}) { + return request('/inbox/rescan-on-hold', { + method: 'POST', + body: { channel, sheet }, + }) +} + +/** Poll one On-Hold rescan (or the active/latest run when runId is omitted). */ +export function getOnHoldRescan(runId) { + return request('/inbox/rescan-on-hold', { params: { run_id: runId } }) +} + /** * Legacy synchronous sync — blocks until the page is ingested. Prefer * startMailboxSync + getMailboxSync so closing the tab cannot kill the job. diff --git a/frontend/src/lib/queryKeys.js b/frontend/src/lib/queryKeys.js index ff2e8b6..ac2aab1 100644 --- a/frontend/src/lib/queryKeys.js +++ b/frontend/src/lib/queryKeys.js @@ -29,6 +29,7 @@ export const qk = { // override or a sync needs no extra invalidation. triage: (p = {}) => ['mailbox', 'triage', p], sync: (id) => ['mailbox', 'sync', id], + rescan: (id) => ['mailbox', 'rescan', id], // Sheet form applicants live under the same mailbox prefix so the Inbox // channel toggle can invalidate both email and form caches together. formSheets: () => ['mailbox', 'form-sheets'], diff --git a/frontend/src/screens/CandidateProfile.jsx b/frontend/src/screens/CandidateProfile.jsx index 56dd373..477b10f 100644 --- a/frontend/src/screens/CandidateProfile.jsx +++ b/frontend/src/screens/CandidateProfile.jsx @@ -316,14 +316,19 @@ export default function CandidateProfile({ {/* No score anywhere -> the whole block goes, rather than a ring drawn around a blank. Seed-backed callers still pass a number and are unaffected; only live candidates the engine never scored drop out. */} - {(atsScore ?? live?.ai_score ?? c.aiScore) != null && ( -
- - {/* The band caption replaces the static label only when the caller - actually fetched one — every other screen keeps "AI Match". */} -
{recommendation || 'AI Match'}
+ {(atsScore ?? live?.ai_score ?? c.aiScore) != null || live?.professional_summary ? ( +
+ {(atsScore ?? live?.ai_score ?? c.aiScore) != null && ( +
+ + {/* The band caption replaces the static label only when the caller + actually fetched one — every other screen keeps "AI Match". */} +
{recommendation || 'AI Match'}
+
+ )} + {live?.professional_summary ?
{live.professional_summary}
: null}
- )} + ) : null}
@@ -346,6 +351,7 @@ export default function CandidateProfile({ + diff --git a/frontend/src/screens/Inbox.jsx b/frontend/src/screens/Inbox.jsx index 515f7b9..f773393 100644 --- a/frontend/src/screens/Inbox.jsx +++ b/frontend/src/screens/Inbox.jsx @@ -324,6 +324,23 @@ function emailAtsScore(row) { ?? (fromJobs.length ? Math.max(...fromJobs) : null) } +function AtsHero({ score, band, ringColor, summary }) { + if (score == null && !summary) return null + return ( +
+ {score != null && ( +
+
+
{Math.round(score)}
+
+
{band || 'ATS Score'}
+
+ )} + {summary ?
{summary}
: null} +
+ ) +} + /** Assigned job title for export — list rows may carry the post or only an id + jobPosts. */ function assignedJobTitle(row) { const fromPost = row?.assignedPost?.title @@ -418,6 +435,7 @@ function mapFormRow(row) { : [], atsResults, atsScore, + professionalSummary: (row.professional_summary || '').trim() || '', assignedId, assignedPost: row.assigned_job_post || null, isReapplicant: Boolean(row.is_reapplicant), @@ -565,6 +583,7 @@ async function fetchMessageDetail(recordId) { assignedPost: row.assigned_job_post || null, atsResults: Array.isArray(row.ats_results) ? row.ats_results : [], atsScore: emailAtsScore(row), + professionalSummary: (row.professional_summary || '').trim() || '', isReapplicant: Boolean(row.is_reapplicant), previousApplications: Array.isArray(row.previous_applications) ? row.previous_applications : [], } @@ -623,6 +642,7 @@ async function fetchApplications(params) { assignedPost: row.assigned_job_post || null, atsResults: Array.isArray(row.ats_results) ? row.ats_results : [], isReapplicant: Boolean(row.is_reapplicant), + professionalSummary: (row.professional_summary || '').trim() || '', previousApplications: Array.isArray(row.previous_applications) ? row.previous_applications : [], } }), @@ -1093,7 +1113,7 @@ function InboxFilters({ filters, cities, sources, showLinkFilters, active, open, * Owns its own interval so the minute count stays current without re-rendering * the queue: under "Show all" that parent render is a thousand rows. */ -function QueueFreshness({ at, refreshing, onRefresh }) { +function QueueFreshness({ at, refreshing, onRefresh, rescan }) { const [, tick] = useState(0) useEffect(() => { if (!at) return undefined @@ -1101,24 +1121,44 @@ function QueueFreshness({ at, refreshing, onRefresh }) { return () => clearInterval(t) }, [at]) + const rescanning = Boolean(rescan?.busy) + const progress = rescan?.progress + return (
- {refreshing - ? 'Updating…' - : at - ? `Updated ${agoLabel(at)}` - : 'Loading…'} + {rescanning + ? (progress + ? `ReScan ${progress.done}/${progress.total}…` + : 'ReScan in progress…') + : refreshing + ? 'Updating…' + : at + ? `Updated ${agoLabel(at)}` + : 'Loading…'} - +
+ {rescan?.visible && ( + + )} + +
) } @@ -1736,6 +1776,70 @@ export default function Inbox() { return undefined }, [qc, syncRun.data, toast]) + const onHoldTab = tab === 'On-Hold' + const [rescanRunId, setRescanRunId] = useState(null) + const rescanToastShown = useRef(null) + + const rescan = useMutation({ + mutationFn: () => inboxApi.startOnHoldRescan({ + channel, + sheet: channel === 'forms' ? formSheet : undefined, + }), + onSuccess: (res) => { + const id = res?.data?.id + if (id) setRescanRunId(id) + toast('ReScan started — On-Hold CVs will be scored against every job post', 'info') + }, + onError: (err) => toast(friendlyAuthError(err, 'ReScan failed'), 'error'), + }) + + const rescanRun = useQuery({ + queryKey: qk.mailbox.rescan(rescanRunId || 'latest'), + queryFn: async () => { + const res = await inboxApi.getOnHoldRescan(rescanRunId) + return res?.data ?? null + }, + enabled: onHoldTab, + refetchInterval: (q) => { + const status = q.state.data?.status + return status === 'queued' || status === 'running' || status === 'scoring' ? 1500 : false + }, + }) + + const rescanBusy = ['queued', 'running', 'scoring'].includes(rescanRun.data?.status) + || rescan.isPending + + useEffect(() => { + const run = rescanRun.data + if (!run?.id) return undefined + const active = ['queued', 'running', 'scoring'].includes(run.status) + if (active) { + setRescanRunId((id) => id || run.id) + const timer = setInterval(() => { + qc.invalidateQueries({ queryKey: qk.mailbox.all() }) + }, 8000) + return () => clearInterval(timer) + } + if (rescanRunId !== run.id) return undefined + if (run.status === 'completed' && rescanToastShown.current !== run.id) { + rescanToastShown.current = run.id + const scored = run.done_count ?? 0 + const skipped = (run.skipped_candidates ?? 0) + (run.skipped_pairs ?? 0) + toast( + scored + ? `ReScan finished — ${scored} ATS score${scored === 1 ? '' : 's'} saved${skipped ? `, ${skipped} skipped` : ''}` + : `ReScan finished — nothing new to score${skipped ? ` (${skipped} skipped)` : ''}`, + 'success', + ) + qc.invalidateQueries({ queryKey: qk.mailbox.all() }) + } + if (run.status === 'failed' && rescanToastShown.current !== run.id) { + rescanToastShown.current = run.id + toast(run.error || 'ReScan failed', 'error') + } + return undefined + }, [qc, rescanRun.data, rescanRunId, toast]) + return (
{ setInboxFilters(EMPTY_INBOX_FILTERS); setSkip(0); setSelectedId(null) }} /> - + rescan.mutate(), + progress: rescanRun.data?.pair_count + ? { done: rescanRun.data.done_count || 0, total: rescanRun.data.pair_count } + : null, + } : null} + />
{/* Hairline, not a skeleton: the rows below stay readable and @@ -2247,14 +2364,14 @@ function FormApplicantDetail({
{i.rowNumber}
)} - {selectedAts != null && ( -
-
-
{Math.round(selectedAts.score)}
-
-
{selectedAts.band || 'ATS Score'}
-
- )} + {selectedAts != null || i.professionalSummary ? ( + + ) : null}
@@ -2622,14 +2739,14 @@ function ApplicationDetail({ {loading && Loading details…} - {i.atsScore != null && ( -
-
-
{Math.round(i.atsScore)}
-
-
ATS Score
-
- )} + {i.atsScore != null || i.professionalSummary ? ( + + ) : null} diff --git a/frontend/src/styles/styles.css b/frontend/src/styles/styles.css index b74a459..07ab26c 100644 --- a/frontend/src/styles/styles.css +++ b/frontend/src/styles/styles.css @@ -1173,6 +1173,12 @@ canvas { width: 100%; max-width: 100%; display: block; } font-size: 12px; flex-shrink: 0; } +.inbox-freshness-actions { + display: flex; + align-items: center; + gap: 6px; + flex-shrink: 0; +} /* Sheet Forms link filters. Deliberately NOT .filter-panel: that grid is four columns wide and its breakpoints watch the viewport, but this lives in a minmax(280px, 34%) column, so on a wide screen it would pack four columns into @@ -1319,6 +1325,7 @@ canvas { width: 100%; max-width: 100%; display: block; } .ats-ring .ats-val { position: relative; z-index: 1; text-align: center; } .ats-ring .ats-num { font-family: var(--font-display); font-size: 30px; font-weight: 700; letter-spacing: -0.02em; line-height: 1; } .ats-ring .ats-lbl { font-size: 11px; color: var(--text-3); font-weight: 600; } +.ats-summary { font-size: var(--fs-xs); color: var(--text-2); line-height: 1.4; max-width: 280px; display: -webkit-box; -webkit-line-clamp: 3; -webkit-box-orient: vertical; overflow: hidden; } .skill-pill { display: inline-flex; align-items: center; gap: 5px; font-size: var(--fs-xs); font-weight: 600; padding: 4px 10px; border-radius: var(--radius-sm); } .skill-pill svg { width: 12px; height: 12px; } .skill-matched { background: var(--success-soft); color: var(--success); } diff --git a/tests/integration/test_api.py b/tests/integration/test_api.py index 1d3bf09..46169cb 100644 --- a/tests/integration/test_api.py +++ b/tests/integration/test_api.py @@ -62,6 +62,7 @@ class TestHappyPath: "matched_keywords", "missing_keywords", "summary_critique", + "professional_summary", } assert result["status"] == "completed" diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py index afe8c20..4d53fa0 100644 --- a/tests/unit/test_models.py +++ b/tests/unit/test_models.py @@ -82,12 +82,14 @@ class TestProfileFields: assert result.job_title is None assert result.current_company is None assert result.years_experience is None + assert result.professional_summary is None def test_blank_strings_normalise_to_none(self) -> None: - result = score(candidate_name=" ", job_title="", current_company="\n\t") + result = score(candidate_name=" ", job_title="", current_company="\n\t", professional_summary=" ") assert result.candidate_name is None assert result.job_title is None assert result.current_company is None + assert result.professional_summary is None def test_whitespace_is_collapsed(self) -> None: result = score(candidate_name=" Ada Lovelace ", job_title="Senior\nEngineer") @@ -107,6 +109,14 @@ class TestProfileFields: with pytest.raises(ValidationError): score(candidate_name="x" * 121) + def test_professional_summary_whitespace_is_collapsed(self) -> None: + result = score(professional_summary=" Python backend\n in Engineering. ") + assert result.professional_summary == "Python backend in Engineering." + + def test_over_length_professional_summary_rejected(self) -> None: + with pytest.raises(ValidationError): + score(professional_summary="x" * 501) + class TestCritique: def test_whitespace_is_collapsed(self) -> None: diff --git a/tests/unit/test_prompts.py b/tests/unit/test_prompts.py index 23b8399..10b616e 100644 --- a/tests/unit/test_prompts.py +++ b/tests/unit/test_prompts.py @@ -35,6 +35,12 @@ def test_system_prompt_stabilises_profile_extraction() -> None: assert "most recent employment entry" in SYSTEM_PROMPT +def test_system_prompt_asks_for_job_agnostic_professional_summary() -> None: + assert "professional_summary" in SYSTEM_PROMPT + assert "Ignore the job" in SYSTEM_PROMPT + assert "This is not summary_critique" in SYSTEM_PROMPT + + def test_injection_text_stays_inside_resume_delimiters() -> None: content = build_user_content("Backend engineer", INJECTION) resume_block = content[1]["text"] From 9f6f4532203330de0f1473be15c2f9090c72276f Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Tue, 8 Sep 2026 16:12:04 +0500 Subject: [PATCH 2/2] creds --- backend/credentials/application_default_credentials.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/backend/credentials/application_default_credentials.json b/backend/credentials/application_default_credentials.json index c109842..a6b6581 100644 --- a/backend/credentials/application_default_credentials.json +++ b/backend/credentials/application_default_credentials.json @@ -2,10 +2,10 @@ "type": "authorized_user", "client_id": "679334897177-ufal3rogbg8cgm3qcren6pv20phqd7nl.apps.googleusercontent.com", "client_secret": "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", - "refresh_token": "1//03C8LMkk-9mSyCgYIARAAGAMSNwF-L9IrAcPhShsp3cDeprSlqI-P6lArpmbyzu-PcKiLfQ5gC3H_MEk930IaKfewy3cxP3T0Oo8", + "refresh_token": "1//03Tu7LRVp_zBACgYIARAAGAMSNwF-L9IrIXwqqdEZGxpxULAKbtgCygCih8DHmO-ELciPtMV8VCVNyZCrO9l6veq0WPwT6fbcAC8", "universe_domain": "googleapis.com", "account": "ahmed.mujtaba@utopiabrands.com", - "token": "ya29.a0AdMD6EgKB22VSy--W0qCtRkMOYECCDhvL4c14xNUSvizbooOBC-ctQeyWR_XybUqa6PZvQ0csVqrIDR6e_uQazKuTAxKPLgpgbTmJ96-sHWbj_981xNrcWe6JsxIpQuGLX9GiKSGa8y5t50ZgWbDy0ECUoOQDvzlUq-hgNdLve1ECxDG4twpL3-2ZpGgskHlhuRL4-PQaCgYKARISARASFQHGX2MiOWOjgYvNnX5ZWhBoYqO9Cg0207", - "expiry": "2026-09-02T16:27:05Z", + "token": "ya29.a0AdMD6Eh9Kvd-ACJZT90CDywa396Zsrf84OWg17u8X-AVffmKhB0nuql60ail5cAY8XlkRuySHSZRKSdXQ7W3IM2dticjCoYgeMmVErMi5UAawUQAd6q0CEsCbi7EPnLTgraOXTAO1MRlWaHwU-R179t0GsAQwnlj9SWBC5Zgfu7Ubf8dGyBcuzg9zknfCF2zbmZFZOsaCgYKAV0SARASFQHGX2MiOeWelgu56Tql44hrVoy1iw0206", + "expiry": "2026-09-08T12:10:05Z", "quota_project_id": "hrms-ats-portal" }