From 1399cf3fec947fc53414b69b8b053990cf4ddf05 Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Thu, 3 Sep 2026 15:15:16 +0500 Subject: [PATCH 1/3] Cv_extractor of g_sheet --- .../application_default_credentials.json | 4 +- backend/g_sheet/decorators.py | 331 ++++++++++++++++++ backend/g_sheet/enums.py | 1 + backend/g_sheet/models.py | 42 +++ backend/g_sheet/plugins.py | 126 +++++++ backend/g_sheet/serializers.py | 4 + backend/g_sheet/tasks.py | 19 +- backend/g_sheet/views.py | 9 +- backend/tests/test_g_sheet_plugins.py | 29 ++ frontend/src/screens/Inbox.jsx | 20 +- 10 files changed, 563 insertions(+), 22 deletions(-) create mode 100644 backend/g_sheet/decorators.py diff --git a/backend/credentials/application_default_credentials.json b/backend/credentials/application_default_credentials.json index 4de621d..c109842 100644 --- a/backend/credentials/application_default_credentials.json +++ b/backend/credentials/application_default_credentials.json @@ -5,7 +5,7 @@ "refresh_token": "1//03C8LMkk-9mSyCgYIARAAGAMSNwF-L9IrAcPhShsp3cDeprSlqI-P6lArpmbyzu-PcKiLfQ5gC3H_MEk930IaKfewy3cxP3T0Oo8", "universe_domain": "googleapis.com", "account": "ahmed.mujtaba@utopiabrands.com", - "token": "ya29.a0AdMD6EgILeNb9UszC7bQJbAcqX709J5ky3eM8MEuQayGwhDStmfnR5t7o192x-FPdt53Q29rYL69zqYrgofUqpwxoI_sPjBsb0wrLqYDo6zwJMTx4P5svM4jZJd9nrXzUEyp3uI81e9DQ3z6lIDKtm6aUTPWQ3fm33i2hxJM7i-svhi3OwjnLtpOUHDyw--v8rQLPHdfaCgYKATkSARASFQHGX2MiXDnutGLllH9DnNCBUKWi4Q0207", - "expiry": "2026-09-02T08:29:45Z", + "token": "ya29.a0AdMD6EgKB22VSy--W0qCtRkMOYECCDhvL4c14xNUSvizbooOBC-ctQeyWR_XybUqa6PZvQ0csVqrIDR6e_uQazKuTAxKPLgpgbTmJ96-sHWbj_981xNrcWe6JsxIpQuGLX9GiKSGa8y5t50ZgWbDy0ECUoOQDvzlUq-hgNdLve1ECxDG4twpL3-2ZpGgskHlhuRL4-PQaCgYKARISARASFQHGX2MiOWOjgYvNnX5ZWhBoYqO9Cg0207", + "expiry": "2026-09-02T16:27:05Z", "quota_project_id": "hrms-ats-portal" } diff --git a/backend/g_sheet/decorators.py b/backend/g_sheet/decorators.py new file mode 100644 index 0000000..9a6d863 --- /dev/null +++ b/backend/g_sheet/decorators.py @@ -0,0 +1,331 @@ +"""Drive CV extract wrapper for sheet ingest. + +Hang `@extract_drive_cvs` on `SheetImport.import_sheet` only (the worker). +HTTP enqueue routes must not run this — FormData rows do not exist yet. + +Worker job pattern (same as a Taskiq message): create a temp dir for the run, +stream each Drive CV to a file, extract, write extracted_data, delete that file. +A finally block removes the job dir so a successful run leaves no CVs on disk. +One row at a time — a plain sequential loop, no extra locks. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import shutil +import uuid +from datetime import datetime,timezone +from functools import wraps +from inspect import signature +from pathlib import Path + +from dotenv import load_dotenv + +from g_sheet.models import FormData +from g_sheet.plugins import ( + SheetsApiError, + drive_file_id, + download_drive_file, + ensure_fresh, + load_credentials, +) + +load_dotenv(Path(__file__).resolve().parent.parent/".env") + +logger=logging.getLogger("g_sheet.decorators") + +_MAX_RESUME_CHARS=int(os.getenv("MAX_RESUME_CHARS","60000")) +_MAX_PDF_SIZE_MB=int(os.getenv("MAX_PDF_SIZE_MB","10")) +_TEMP_ROOT=Path(__file__).resolve().parent/"tmp"/"cv_extract" + + +def build_extracted_data( + *, + status, + resume_link, + file_id=None, + filename=None, + mime_type=None, + text=None, + page_count=None, + truncated=None, + error_code=None, + error_message=None, +): + """Stable JSON blob stored on form_data.extracted_data.""" + return { + "status":status, + "resume_link":resume_link or "", + "file_id":file_id, + "filename":filename, + "mime_type":mime_type, + "text":text, + "page_count":page_count, + "truncated":truncated, + "char_count":len(text) if isinstance(text,str) else None, + "error_code":error_code, + "error_message":error_message, + "extracted_at":datetime.now(timezone.utc).isoformat(), + } + + +def _drive_error_code(status_code): + if status_code in (401,403): + return "DRIVE_FORBIDDEN" + if status_code==404: + return "DRIVE_FILE_NOT_FOUND" + if status_code==413: + return "PAYLOAD_TOO_LARGE" + if status_code in (400,415): + return "UNSUPPORTED_FILE_TYPE" + return "DRIVE_DOWNLOAD_FAILED" + + +def _prepare_drive_credentials(service): + """Load/refresh the Google session. Never raises — None means skip extract.""" + try: + if service is None: + return load_credentials() + creds=getattr(service,"credentials",None) + path=getattr(service,"credentials_path",None) + scopes=getattr(service,"scopes",None) + if creds is not None: + return ensure_fresh(creds,path) + return load_credentials(path,scopes) + except Exception: + logger.warning("Google Drive session unavailable; skipping CV extract") + return None + + +def _is_sheet_service(obj): + return obj is not None and hasattr(obj,"session") and hasattr(obj,"spreadsheet_id") + + +def _tab_from(result,args,kwargs): + if isinstance(result,dict) and result.get("tab"): + return result.get("tab") + if kwargs.get("tab"): + return kwargs.get("tab") + if args: + return args[0] + return None + + +def _should_ingest(result): + if not isinstance(result,dict): + return False + if result.get("error"): + return False + if result.get("status") in ("queued","running","failed"): + return False + if result.get("rows_read",1)==0 and result.get("inserted",1)==0: + return False + return True + + +def make_job_temp_dir(root=None): + """Temp dir for one extract job. Caller must remove_job_temp_dir in finally.""" + base=Path(root) if root else _TEMP_ROOT + base.mkdir(parents=True,exist_ok=True) + job_dir=base/uuid.uuid4().hex + job_dir.mkdir() + return job_dir + + +def remove_job_temp_dir(job_dir): + """Delete leftover CVs and the job dir. No-op if missing.""" + if not job_dir: + return + path=Path(job_dir) + if not path.exists(): + return + shutil.rmtree(path,ignore_errors=True) + + +def _unlink(path): + if path is None: + return + try: + Path(path).unlink(missing_ok=True) + except OSError as e: + logger.warning("could not delete temp CV %s: %s",path,e) + + +def _extract_pdf(data,filename,max_chars): + from app.services.pdf import extract_resume,sanitize_filename + from job.candidate.plugins import normalize_spaced_text + + resume=extract_resume(data,sanitize_filename(filename),max_chars) + return { + "text":normalize_spaced_text(resume.text), + "page_count":resume.page_count, + "truncated":resume.truncated, + } + + +def _download_and_extract(credentials,link,max_chars,max_bytes,dest_dir): + """Stream one Drive file into dest_dir, extract, then delete that file.""" + file_id=drive_file_id(link) + if not file_id: + return build_extracted_data( + status="skipped", + resume_link=link, + error_code="NOT_DRIVE_URL", + error_message="Resume link is not a Google Drive file URL", + ) + dest=None + try: + downloaded=download_drive_file( + credentials,link,max_bytes=max_bytes,dest_dir=dest_dir, + ) + dest=downloaded.get("path") + if dest is None: + return build_extracted_data( + status="failed", + resume_link=link, + file_id=downloaded.get("file_id") or file_id, + error_code="DRIVE_DOWNLOAD_FAILED", + error_message="Drive download did not write a file", + ) + data=Path(dest).read_bytes() + try: + parsed=_extract_pdf(data,downloaded.get("filename") or "resume.pdf",max_chars) + finally: + data=b"" + return build_extracted_data( + status="completed", + resume_link=link, + file_id=downloaded.get("file_id") or file_id, + filename=downloaded.get("filename"), + mime_type=downloaded.get("mime_type"), + text=parsed["text"], + page_count=parsed["page_count"], + truncated=parsed["truncated"], + ) + except SheetsApiError as e: + return build_extracted_data( + status="failed", + resume_link=link, + file_id=file_id, + error_code=_drive_error_code(e.status_code), + error_message=(e.message or "")[:300], + ) + except Exception as e: + from app.core.errors import ATSError + if isinstance(e,ATSError): + return build_extracted_data( + status="failed", + resume_link=link, + file_id=file_id, + error_code=e.error_code, + error_message=e.public_message, + ) + logger.exception("drive download/extract failed for file_id=%s",file_id) + return build_extracted_data( + status="failed", + resume_link=link, + file_id=file_id, + error_code="DRIVE_DOWNLOAD_FAILED", + error_message="Drive download failed", + ) + finally: + _unlink(dest) + + +async def extract_one_resume(credentials,resume_link,dest_dir,max_chars=None,max_bytes=None): + """Download one Drive URL into dest_dir and return extracted_data JSON.""" + if max_chars is None: + max_chars=_MAX_RESUME_CHARS + if max_bytes is None: + max_bytes=_MAX_PDF_SIZE_MB*1024*1024 + link=(resume_link or "").strip() + return await asyncio.to_thread( + _download_and_extract,credentials,link,max_chars,max_bytes,dest_dir, + ) + + +async def ingest_form_resume_links(session,sheet,credentials,temp_root=None): + """One Drive file per resume_link: download → extract → DB → delete file. + + The job temp dir is created at start and removed in finally so a finished + run leaves no CVs on disk (Taskiq worker cleanup). + """ + rows=await FormData.fetch_resume_links(session,sheet) + completed=0 + failed=0 + max_chars=_MAX_RESUME_CHARS + max_bytes=_MAX_PDF_SIZE_MB*1024*1024 + job_dir=make_job_temp_dir(temp_root) + logger.info( + "drive CV extract starting tab=%s resumes=%s temp=%s", + sheet,len(rows),job_dir, + ) + try: + for record_id,resume_link in rows: + payload=await extract_one_resume( + credentials,resume_link,job_dir,max_chars,max_bytes, + ) + try: + await FormData.set_extracted_data(session,record_id,payload) + except Exception: + logger.exception("could not persist extracted_data for %s",record_id) + failed+=1 + try: + await session.rollback() + except Exception: + logger.exception("rollback after extracted_data persist failed") + continue + status=payload.get("status") + if status=="completed": + completed+=1 + elif status=="failed": + failed+=1 + logger.info( + "drive CV extract finished tab=%s extracted=%s failed=%s", + sheet,completed,failed, + ) + return {"extracted":completed,"extract_failed":failed} + finally: + remove_job_temp_dir(job_dir) + + +def extract_drive_cvs(func): + """Hang on SheetImport.import_sheet (worker ingest), not on HTTP enqueue. + + After rows are inserted: one Drive download + extract per resume_link, + written to form_data.extracted_data at the end of each row. + Credentials load only if ingest will actually run. + """ + + @wraps(func) + async def wrapper(*args,**kwargs): + result=await func(*args,**kwargs) + service=args[0] if args and _is_sheet_service(args[0]) else None + session=getattr(service,"session",None) if service is not None else None + rest=args[1:] if service is not None else args + tab=_tab_from(result,rest,kwargs) + if session is None or not tab or not _should_ingest(result): + logger.info( + "drive CV extract skipped tab=%s session=%s ingest=%s", + tab,session is not None, + _should_ingest(result) if isinstance(result,dict) else False, + ) + return result + credentials=await asyncio.to_thread(_prepare_drive_credentials,service) + if credentials is None: + logger.warning("Google Drive session unavailable; skipping CV extract") + return result + try: + stats=await ingest_form_resume_links(session,tab,credentials) + except Exception: + logger.exception("drive CV extract after import of %s failed",tab) + stats={"extracted":0,"extract_failed":0} + if isinstance(result,dict): + result["extracted"]=stats.get("extracted",0) + result["extract_failed"]=stats.get("extract_failed",0) + return result + + wrapper.__signature__=signature(func) + return wrapper diff --git a/backend/g_sheet/enums.py b/backend/g_sheet/enums.py index 6d97be6..962cefb 100644 --- a/backend/g_sheet/enums.py +++ b/backend/g_sheet/enums.py @@ -189,6 +189,7 @@ class FormDataColumn(str, Enum): CANDIDATE_EMAIL = "candidate_email" PROFILE_LINK = "profile_link" RESUME_LINK = "resume_link" + EXTRACTED_DATA = "extracted_data" AREA_OF_EXPERTISE = "area_of_expertise" REQUISITION_NUMBER = "requisition_number" POSITION_APPLIED_FOR = "position_applied_for" diff --git a/backend/g_sheet/models.py b/backend/g_sheet/models.py index 69a1f97..d1d1c2b 100644 --- a/backend/g_sheet/models.py +++ b/backend/g_sheet/models.py @@ -51,6 +51,8 @@ class FormData(SQLModel, table=True): candidate_email: str | None = Field(default=None, index=True) profile_link: str | None = Field(default=None) resume_link: str | None = Field(default=None) + # Drive CV extract JSON written by @extract_drive_cvs after sheet ingest. + extracted_data: dict | None = Field(default=None, sa_column=Column(JSONB)) area_of_expertise: str | None = Field(default=None) requisition_number: str | None = Field(default=None, index=True) position_applied_for: str | None = Field(default=None) @@ -190,6 +192,38 @@ class FormData(SQLModel, table=True): await session.refresh(row) return row + @classmethod + async def set_extracted_data( + cls, session: AsyncSession, record_id, extracted_data, *, commit: bool = True, + ): + row = await cls.get_form_data_by_id(session, record_id) + if not row: + return None + row.extracted_data = extracted_data + row.updated_at = _now() + session.add(row) + if commit: + await session.commit() + await session.refresh(row) + return row + + @classmethod + async def fetch_resume_links(cls, session: AsyncSession, sheet: str): + """(id, resume_link) for one tab. Blank links are dropped.""" + statement = ( + select(cls.id, cls.resume_link) + .where(cls.sheet == sheet) + .where(cls.resume_link.is_not(None)) + .order_by(cls.row_number) + ) + result = await session.execute(statement) + rows = [] + for record_id, link in result.all(): + text = (link or "").strip() + if text: + rows.append((record_id, text)) + return rows + @classmethod async def fetch_form_data( cls, session: AsyncSession, *, sheet=None, search=None, @@ -423,6 +457,14 @@ class SheetImportRun(SQLModel, table=True): ) return result.scalars().first() + @classmethod + async def delete_failed(cls, session: AsyncSession, *, commit: bool = True): + """Drop failed import rows so a new job is not blocked by them.""" + result = await session.execute(delete(cls).where(cls.status == "failed")) + if commit: + await session.commit() + return result.rowcount + @classmethod async def insert_run(cls, session: AsyncSession, fields: dict, *, commit: bool = True): row = cls(**fields) diff --git a/backend/g_sheet/plugins.py b/backend/g_sheet/plugins.py index d0cee32..3f449cb 100644 --- a/backend/g_sheet/plugins.py +++ b/backend/g_sheet/plugins.py @@ -11,6 +11,7 @@ the session can be copied to Linux prod. Re-auth lives in g_sheet/store_session. from __future__ import annotations +import io import json import logging import os @@ -25,6 +26,7 @@ from google.auth import default as google_auth_default from google.auth.transport.requests import Request from googleapiclient.discovery import build from googleapiclient.errors import HttpError +from googleapiclient.http import MediaIoBaseDownload from g_sheet.enums import ( ALIAS_TO_FIELD, @@ -206,6 +208,130 @@ def build_sheets_client(credentials): raise SheetsApiError(f"Could not build the Sheets client: {e}") +def build_drive_client(credentials): + """Drive v3 client. Same ADC session as Sheets; cache_discovery=False under threads.""" + try: + return build("drive","v3",credentials=credentials,cache_discovery=False) + except Exception as e: + raise SheetsApiError(f"Could not build the Drive client: {e}") + + +_DRIVE_FILE_ID_PATTERNS=( + re.compile(r"/file/d/([a-zA-Z0-9_-]+)"), + re.compile(r"/document/d/([a-zA-Z0-9_-]+)"), + re.compile(r"[?&]id=([a-zA-Z0-9_-]+)"), + re.compile(r"/d/([a-zA-Z0-9_-]+)"), +) +_GOOGLE_APPS_SHORTCUT="application/vnd.google-apps.shortcut" +_GOOGLE_APPS_DOCUMENT="application/vnd.google-apps.document" +_GOOGLE_APPS_PREFIX="application/vnd.google-apps." + + +def drive_file_id(url): + """Extract a Drive/Docs file id from a Google URL, or None if it is not one.""" + raw=(url or "").strip() + if not raw: + return None + lowered=raw.lower() + if "drive.google.com" not in lowered and "docs.google.com" not in lowered: + return None + if "/folders/" in lowered: + return None + for pattern in _DRIVE_FILE_ID_PATTERNS: + match=pattern.search(raw) + if match: + return match.group(1) + return None + + +def _drive_file_meta(drive,file_id): + request=drive.files().get( + fileId=file_id, + fields="id,name,mimeType,size,shortcutDetails", + supportsAllDrives=True, + ) + return execute(request,"drive file metadata") + + +def _download_media(request,dest_path=None): + """Stream a Drive media request. dest_path set → write that file; else return bytes.""" + try: + if dest_path is None: + buf=io.BytesIO() + downloader=MediaIoBaseDownload(buf,request) + done=False + while not done: + _,done=downloader.next_chunk() + return buf.getvalue() + path=Path(dest_path) + path.parent.mkdir(parents=True,exist_ok=True) + with path.open("wb") as fh: + downloader=MediaIoBaseDownload(fh,request) + done=False + while not done: + _,done=downloader.next_chunk() + return path + except HttpError as e: + status=_status_of(e) + raise SheetsApiError(f"drive download failed: {_reason_of(e)}",status or 502) + + +def _cv_dest_path(dest_dir,file_id,filename): + dest_dir=Path(dest_dir).resolve() + suffix=Path(filename or "resume.pdf").suffix.lower() or ".pdf" + if suffix not in (".pdf",".doc",".docx"): + suffix=".pdf" + safe_id=re.sub(r"[^a-zA-Z0-9_-]","",file_id or "") or "file" + dest=(dest_dir/f"{safe_id}{suffix}").resolve() + if dest.parent!=dest_dir: + raise SheetsApiError("invalid download path",400) + return dest + + +def download_drive_file(credentials,url,*,max_bytes=None,dest_dir=None): + """Download one Drive file via the existing Google session. + + When dest_dir is set the file is streamed to disk and `path` is returned + (`data` is None). Otherwise `data` holds the bytes (tests / callers without a + work dir). + """ + file_id=drive_file_id(url) + if not file_id: + raise SheetsApiError("not a Google Drive file URL",400) + ensure_fresh(credentials) + drive=build_drive_client(credentials) + meta=_drive_file_meta(drive,file_id) + if (meta.get("mimeType") or "")==_GOOGLE_APPS_SHORTCUT: + target=(meta.get("shortcutDetails") or {}).get("targetId") + if not target: + raise SheetsApiError("Drive shortcut has no target",400) + file_id=target + meta=_drive_file_meta(drive,file_id) + mime=meta.get("mimeType") or "" + name=meta.get("name") or "resume.pdf" + size=meta.get("size") + if max_bytes is not None and size is not None: + try: + if int(size)>max_bytes: + raise SheetsApiError("The file exceeds the size limit.",413) + except (TypeError,ValueError): + pass + if mime==_GOOGLE_APPS_DOCUMENT: + request=drive.files().export_media(fileId=file_id,mimeType="application/pdf") + if not name.lower().endswith(".pdf"): + name=f"{name}.pdf" + elif mime.startswith(_GOOGLE_APPS_PREFIX): + raise SheetsApiError("unsupported Google file type",415) + else: + request=drive.files().get_media(fileId=file_id,supportsAllDrives=True) + if dest_dir is None: + data=_download_media(request) + return {"file_id":file_id,"filename":name,"mime_type":mime,"data":data,"path":None} + dest=_cv_dest_path(dest_dir,file_id,name) + _download_media(request,dest) + return {"file_id":file_id,"filename":name,"mime_type":mime,"data":None,"path":dest} + + def _status_of(error): status=getattr(getattr(error,"resp",None),"status",None) if status is None: diff --git a/backend/g_sheet/serializers.py b/backend/g_sheet/serializers.py index affd4e2..3bd8f22 100644 --- a/backend/g_sheet/serializers.py +++ b/backend/g_sheet/serializers.py @@ -128,6 +128,8 @@ def serialize_import(report: dict) -> dict: "ages_parsed": report.get("ages_parsed", 0), "salaries_parsed": report.get("salaries_parsed", 0), "unmapped_headers": report.get("unmapped_headers") or [], + "extracted": report.get("extracted", 0), + "extract_failed": report.get("extract_failed", 0), "error": report.get("error"), } @@ -142,6 +144,8 @@ def serialize_import_all(reports: list[dict]) -> dict: "failed": len(failed), "inserted": sum(r.get("inserted", 0) for r in ok), "deleted": sum(r.get("deleted", 0) for r in ok), + "extracted": sum(r.get("extracted", 0) for r in ok), + "extract_failed": sum(r.get("extract_failed", 0) for r in reports), "reports": [serialize_import(r) for r in reports], } diff --git a/backend/g_sheet/tasks.py b/backend/g_sheet/tasks.py index d47010b..64b2a6b 100644 --- a/backend/g_sheet/tasks.py +++ b/backend/g_sheet/tasks.py @@ -49,13 +49,30 @@ async def import_sheets(run_id:str) -> dict: try: acquired=await client.set(_LOCK_KEY,run_id,nx=True,ex=_LOCK_TTL) if not acquired: - return await _fail(run_id,"another sheet import is already running") + holder=await client.get(_LOCK_KEY) + # Crash/restart redelivers the same run_id while the TTL lock is + # still set. Failing that as "another import" strands the lock + # until expiry and every later click also bounces. + if holder==run_id: + await client.expire(_LOCK_KEY,_LOCK_TTL) + logger.warning("sheet import %s reclaimed its own stale lock",run_id) + else: + logger.warning( + "sheet import %s skipped: lock held by %s",run_id,holder, + ) + return await _fail(run_id,"another sheet import is already running") try: async with session_scope() as session: row=await SheetImportRun.get_by_id(session,run_id) if not row: raise PermanentTaskError(f"import run {run_id} not found") + if row.status=="failed": + await SheetImportRun.delete_failed(session) + return {"status":"failed","error":row.error} + if row.status=="completed": + return {"status":"completed","report":row.report} + await SheetImportRun.delete_failed(session) await SheetImportRun.update_run(session,run_id,{ "status":"running", "started_at":datetime.now(timezone.utc), diff --git a/backend/g_sheet/views.py b/backend/g_sheet/views.py index 88df88e..36e422d 100644 --- a/backend/g_sheet/views.py +++ b/backend/g_sheet/views.py @@ -21,6 +21,7 @@ from datetime import datetime,timezone from fastapi import HTTPException +from g_sheet.decorators import extract_drive_cvs from g_sheet.plugins import ( SCOPES, SPREADSHEET_ID, @@ -232,6 +233,7 @@ class SheetHealth(SheetRead): class SheetImport(SheetRead): """Google Sheet → FormData import + import-run tracking.""" + @extract_drive_cvs async def import_sheet(self,tab): """Read one tab from Google Sheets and replace its FormData rows.""" session=self._require_session() @@ -279,15 +281,18 @@ class SheetImport(SheetRead): return serialize_import_all(reports) async def start_import(self,current_user=None,tab=None): - """Enqueue a sheet import on the shared Taskiq worker; return the run row. + """Enqueue a sheet import. - If a queued/running import already exists, return it instead of stacking another. + At the start of every new job: queued/running → keep that job; + failed → delete those rows and start this one; completed → start this one. """ session=self._require_session() active=await SheetImportRun.get_active(session) if active: return serialize_import_run(active) + await SheetImportRun.delete_failed(session) + created_by=None if isinstance(current_user,dict) and current_user.get("id"): created_by=SheetImportRun._as_uuid(current_user.get("id")) diff --git a/backend/tests/test_g_sheet_plugins.py b/backend/tests/test_g_sheet_plugins.py index 7dee269..8c331fa 100644 --- a/backend/tests/test_g_sheet_plugins.py +++ b/backend/tests/test_g_sheet_plugins.py @@ -151,3 +151,32 @@ def test_rows_to_indexed_records_keeps_true_sheet_row_across_blank(): def test_form_data_fields_match_model(): assert set(FORM_DATA_FIELDS) == set(FormData.model_fields) + + +def test_drive_file_id_from_form_open_url(): + url = "https://drive.google.com/open?id=1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4" + assert plugins.drive_file_id(url) == "1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4" + + +def test_drive_file_id_from_file_view_url(): + url = "https://drive.google.com/file/d/abc123XYZ/view?usp=sharing" + assert plugins.drive_file_id(url) == "abc123XYZ" + + +def test_drive_file_id_from_docs_url(): + url = "https://docs.google.com/document/d/docFileId99/edit" + assert plugins.drive_file_id(url) == "docFileId99" + + +def test_drive_file_id_rejects_folder_and_non_drive(): + assert plugins.drive_file_id("https://drive.google.com/drive/folders/abc") is None + assert plugins.drive_file_id("https://example.com/cv.pdf") is None + assert plugins.drive_file_id("") is None + assert plugins.drive_file_id(None) is None + + +def test_cv_dest_path_stays_inside_dir(tmp_path): + dest = plugins._cv_dest_path(tmp_path, "1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4", "Resume.PDF") + assert dest.parent == tmp_path.resolve() + assert dest.name.endswith(".pdf") + assert dest.name.startswith("1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4") diff --git a/frontend/src/screens/Inbox.jsx b/frontend/src/screens/Inbox.jsx index 0c8e999..c4ddafc 100644 --- a/frontend/src/screens/Inbox.jsx +++ b/frontend/src/screens/Inbox.jsx @@ -154,19 +154,8 @@ function sourceFrom(messageTo) { return { source: raw.split(',')[0].trim(), sourceMeta: null } } -/** - * Form `source_of_application` is a free-text label (LinkedIn, Indeed, …), not a - * To-address. Reuse the email source palette when the spelling matches; otherwise - * tag the row as a Sheet Forms entry so the chip still paints. - */ -function formSourceFrom(raw) { - const label = (raw || '').trim() - if (!label) return { source: 'Google Forms', sourceMeta: SHEET_SOURCE_META } - const flat = label.toLowerCase().replace(/[^a-z]/g, '') - const hit = inboxSources.find((s) => flat.includes(s.toLowerCase().replace(/[^a-z]/g, ''))) - if (hit) return { source: hit, sourceMeta: sourceMeta[hit] } - return { source: label, sourceMeta: SHEET_SOURCE_META } -} +/** Sheet Forms always chip as Google Sheet — not the form's "where did you hear" answer. */ +const FORM_LIST_SOURCE = { source: 'Google Sheet', sourceMeta: SHEET_SOURCE_META } /** entry_date is midnight UTC; entry_time is a separate "HH:MM" string from the sheet. */ function formReceivedAt(entryDate, entryTime) { @@ -195,7 +184,7 @@ function mapFormRow(row) { email: row.candidate_email || '', phone: row.candidate_number || '', position: row.position_applied_for || '—', - ...formSourceFrom(row.source_of_application), + ...FORM_LIST_SOURCE, received: formReceivedAt(row.entry_date, row.entry_time), screenedBy: row.screened_by || '', hrComments: row.hr_comments || '', @@ -1188,9 +1177,6 @@ export default function Inbox() { {i.atsScore != null && (
)} - {isForms && i.noticePeriod && ( -
{i.noticePeriod}
- )}
{i.processing} From 11335171170a6a15a79c6aec65f98edfe27cbd4a Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Thu, 3 Sep 2026 16:03:42 +0500 Subject: [PATCH 2/3] . --- backend/g_sheet/decorators.py | 5 +- backend/g_sheet/enums.py | 2 + backend/g_sheet/models.py | 81 +++++++++++- backend/g_sheet/plugins.py | 1 + backend/g_sheet/scoring.py | 169 ++++++++++++++++++++++++++ backend/g_sheet/serializers.py | 2 + backend/g_sheet/views.py | 98 ++++++++++++--- backend/inbox/models.py | 60 ++++++++- backend/inbox/tasks.py | 21 ++++ backend/job/job_post/models.py | 46 +++++++ backend/tests/test_g_sheet_plugins.py | 62 ++++++++++ frontend/src/screens/Inbox.jsx | 48 +++++++- frontend/src/ui/SuggestedRoles.jsx | 3 +- 13 files changed, 569 insertions(+), 29 deletions(-) create mode 100644 backend/g_sheet/scoring.py diff --git a/backend/g_sheet/decorators.py b/backend/g_sheet/decorators.py index 9a6d863..35e307a 100644 --- a/backend/g_sheet/decorators.py +++ b/backend/g_sheet/decorators.py @@ -268,7 +268,7 @@ async def ingest_form_resume_links(session,sheet,credentials,temp_root=None): credentials,resume_link,job_dir,max_chars,max_bytes, ) try: - await FormData.set_extracted_data(session,record_id,payload) + saved=await FormData.set_extracted_data(session,record_id,payload) except Exception: logger.exception("could not persist extracted_data for %s",record_id) failed+=1 @@ -280,6 +280,9 @@ async def ingest_form_resume_links(session,sheet,credentials,temp_root=None): status=payload.get("status") if status=="completed": completed+=1 + if saved is not None: + from g_sheet.scoring import enqueue_form_row_scores + await enqueue_form_row_scores(saved) elif status=="failed": failed+=1 logger.info( diff --git a/backend/g_sheet/enums.py b/backend/g_sheet/enums.py index 962cefb..6769e68 100644 --- a/backend/g_sheet/enums.py +++ b/backend/g_sheet/enums.py @@ -171,6 +171,8 @@ class FormDataColumn(str, Enum): ID = "id" SHEET = "sheet" JOB_POST_ID = "job_post_id" + ASSIGNED_JOB_POST_ID = "assigned_job_post_id" + SUGGESTED_JOB_POST_IDS = "suggested_job_post_ids" MANUAL_UPLOAD_CANDIDATE_ID = "manual_upload_candidate_id" ROW_NUMBER = "row_number" SERIAL_NO = "serial_no" diff --git a/backend/g_sheet/models.py b/backend/g_sheet/models.py index d1d1c2b..9f19807 100644 --- a/backend/g_sheet/models.py +++ b/backend/g_sheet/models.py @@ -28,9 +28,15 @@ class FormData(SQLModel, table=True): id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) sheet: str = Field(nullable=False, index=True) - # Optional link to a job post. DB FK only — no ORM Relationship (avoids + # Recruiter-assigned job. DB FK only — no ORM Relationship (avoids # pulling job_posts into the sheet worker metadata graph). job_post_id: uuid.UUID | None = Field(default=None, index=True) + assigned_job_post_id: uuid.UUID | None = Field(default=None, index=True) + # ILIKE title matches from Position Applied For. One form row → many jobs. + # Suggested, not assigned. ATS scores each id separately. + suggested_job_post_ids: list[str] | None = Field( + default=None, sa_column=Column(JSONB), + ) # Set when this form row is promoted into the hiring pipeline (Users + # manual_upload_candidate). Idempotency key for assign / shortlist. manual_upload_candidate_id: uuid.UUID | None = Field(default=None, index=True) @@ -133,19 +139,65 @@ class FormData(SQLModel, table=True): result = await session.execute(select(cls).where(cls.id == rid)) return result.scalars().first() + @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).""" + from job.job_post.models import JobPosts + + form = await cls.get_form_data_by_id(session, record_id) + if form is None: + return None, None + try: + jid = uuid.UUID(str(job_post_id)) + except (TypeError, ValueError): + return None, None + if jid not in set(cls.score_job_ids(form)): + return None, None + job = await JobPosts.get_job_post_by_id(session, jid) + if job is None or job.is_deleted: + return None, None + return form, job + + @staticmethod + def score_job_ids(row) -> list[uuid.UUID]: + """Jobs ATS may score: assigned only, else every suggested id.""" + if row is None: + return [] + getter = row.get if isinstance(row, dict) else lambda key, default=None: getattr(row, key, default) + assigned = getter("assigned_job_post_id") or getter("job_post_id") + if assigned not in (None, ""): + try: + return [uuid.UUID(str(assigned))] + except (TypeError, ValueError): + return [] + out: list[uuid.UUID] = [] + seen: set[uuid.UUID] = set() + for raw in getter("suggested_job_post_ids") or []: + try: + uid = uuid.UUID(str(raw)) + except (TypeError, ValueError): + continue + if uid not in seen: + seen.add(uid) + out.append(uid) + return out + @classmethod async def set_job_post(cls, session: AsyncSession, record_id, job_post_id): - """Set or clear job_post_id; returns the row or None if missing.""" + """Set or clear the recruiter assignment; returns the row or None if missing.""" row = await cls.get_form_data_by_id(session, record_id) if not row: return None if job_post_id is None: row.job_post_id = None + row.assigned_job_post_id = None else: try: - row.job_post_id = uuid.UUID(str(job_post_id)) + uid = uuid.UUID(str(job_post_id)) except (TypeError, ValueError): return None + row.job_post_id = uid + row.assigned_job_post_id = uid row.updated_at = _now() session.add(row) await session.commit() @@ -330,6 +382,29 @@ class FormData(SQLModel, table=True): await session.commit() return {"deleted": deleted, "inserted": inserted} + @classmethod + async def stamp_suggested_job_posts( + cls, session: AsyncSession, records: list[dict], + ) -> list[dict]: + """Set suggested_job_post_ids from ILIKE title match on position_applied_for. + + One applied-for title can match many job_posts. Blank or no match → []. + Recruiter assignment (job_post_id / assigned_job_post_id) stays unset. + """ + from job.job_post.models import JobPosts + + found = await JobPosts.ids_for_titles_ilike( + session, + [r.get("position_applied_for") for r in records], + ) + for record in records: + applied = (record.get("position_applied_for") or "").strip() + hits = found.get(applied) or [] if applied else [] + record["suggested_job_post_ids"] = [str(uid) for uid in hits] + record["job_post_id"] = None + record["assigned_job_post_id"] = None + return records + @staticmethod def _cell(data: dict, key: str): """Sheet cell → stripped str, or None if missing/blank.""" diff --git a/backend/g_sheet/plugins.py b/backend/g_sheet/plugins.py index 3f449cb..3bce00e 100644 --- a/backend/g_sheet/plugins.py +++ b/backend/g_sheet/plugins.py @@ -457,6 +457,7 @@ _CURRENCY_STRIP_RE=re.compile(r"(?:rs\.?|pkr|inr|usd|\$|€|£)",re.I) # Every typed column key the mapper must emit (uniform dicts for bulk insert). _FORM_DATA_COLUMN_KEYS=tuple(field.value for field in FormDataField)+( "age_raw","current_salary_value","expected_salary_value","job_post_id", + "assigned_job_post_id","suggested_job_post_ids", ) diff --git a/backend/g_sheet/scoring.py b/backend/g_sheet/scoring.py new file mode 100644 index 0000000..593aa49 --- /dev/null +++ b/backend/g_sheet/scoring.py @@ -0,0 +1,169 @@ +"""ATS-score a form_data CV against every linked job post. + +A form row can match many jobs (suggested_job_post_ids). Recruiter assignment +is assigned_job_post_id. If assigned is set, ATS scores only that job; else +every suggested id. Resume text comes from extracted_data. Each +(form_data_id, job_post_id) pair lands as its own ats_results row. +""" + +from __future__ import annotations + +import logging +import uuid +from datetime import datetime, timezone + +from app.models.scoring import CompletedCandidate +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 job.candidate.plugins import build_job_description, get_scorer, get_scoring_settings +from job.job_post.models import JobPosts + +logger = logging.getLogger("g_sheet.scoring") + + +def resume_text_from_extracted(payload) -> str | None: + """Usable CV text from form_data.extracted_data, or None.""" + if not isinstance(payload, dict): + return None + if payload.get("status") != "completed": + return None + text = (payload.get("text") or "").strip() + return text or None + + +def _band(score) -> str: + if score is None: + return "" + return "Strong Match" if score >= 82 else "Potential Match" if score >= 65 else "Weak Match" + + +def serialize_form_ats(row) -> dict: + """One current ats_results row for a Sheet Forms applicant.""" + return { + "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, + "computed_at": row.computed_at.isoformat() if row.computed_at else None, + } + + +async def enqueue_form_score(form_data_id, job_post_id) -> None: + """Queue ATS for one form row against one job. Broker-down only logs.""" + if not form_data_id or not job_post_id: + return + await enqueue_form_scores(form_data_id, [job_post_id]) + + +async def enqueue_form_scores(form_data_id, job_post_ids) -> None: + """Queue ATS for one form row against each job. Broker-down only logs.""" + if not form_data_id: + return + from inbox.tasks import score_form_data + + seen: set[str] = set() + for raw in job_post_ids or []: + job_id = str(raw or "").strip() + if not job_id or job_id in seen: + continue + seen.add(job_id) + try: + await score_form_data.kicker().with_labels( + created_at=datetime.now(timezone.utc).isoformat(), + correlation_id=str(form_data_id), + queue="inbox", + ).kiq(str(form_data_id), job_id) + except Exception as exc: + logger.warning( + "could not queue form ats score for %s vs %s: %s", + form_data_id, job_id, exc, + ) + + +async def enqueue_form_row_scores(form_row) -> None: + """Queue ATS: assigned job only, else every suggested job.""" + if form_row is None: + return + assigned=getattr(form_row,"assigned_job_post_id",None) or getattr(form_row,"job_post_id",None) + if assigned: + await enqueue_form_score(form_row.id,assigned) + return + job_ids=FormData.score_job_ids(form_row) + if not job_ids: + return + await enqueue_form_scores(form_row.id,job_ids) + + +async def score_form_against_job(form_data_id: str, job_id: str) -> dict: + """Score one Sheet Forms CV against one job. Idempotent per (form, job).""" + try: + uuid.UUID(str(form_data_id)) + uuid.UUID(str(job_id)) + except (TypeError, ValueError): + return {"status": "skipped", "reason": "invalid_ids"} + + async with session_scope() as session: + existing = await AtsResults.get_for_form_job(session, form_data_id, job_id) + if existing is not None: + return {"status": "already_scored"} + form_row, job = await FormData.get_with_job(session, form_data_id, job_id) + if form_row is None or job is None: + return {"status": "skipped", "reason": "no_join"} + text = resume_text_from_extracted(form_row.extracted_data) + if not text: + return {"status": "skipped", "reason": "no_extract"} + filename = (form_row.extracted_data or {}).get("filename") or "resume.pdf" + page_count = int((form_row.extracted_data or {}).get("page_count") or 1) + truncated = bool((form_row.extracted_data or {}).get("truncated")) + settings = get_scoring_settings() + jd = build_job_description(job) + if len(jd) > settings.max_jd_chars: + return {"status": "skipped", "reason": "jd_too_large"} + form_pk = form_row.id + job_pk = job.id + + resume = ExtractedResume( + filename=str(filename), + candidate_id=str(form_pk), + text=text, + page_count=page_count, + truncated=truncated, + ) + scored = await score_batch( + [resume], + job_description=jd, + scorer=get_scorer(), + concurrency=1, + ) + result = scored[0] if scored else None + if not isinstance(result, CompletedCandidate): + error = getattr(result, "error_code", None) if result is not None else "MODEL_UNAVAILABLE" + logger.warning("form ats failed form_data=%s job=%s code=%s", form_data_id, job_id, error) + return {"status": "failed", "error_code": error} + + async with session_scope() as session: + existing = await AtsResults.get_for_form_job(session, form_pk, job_pk) + if existing is not None: + return {"status": "already_scored"} + 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 AtsResults.insert_result(session, { + "inbox_id": None, + "user_id": None, + "candidate_id": None, + "form_data_id": form_pk, + "job_post_id": job_pk, + "overall_score": float(result.match_score), + "band": _band(result.match_score), + "model_name": settings.openai_model, + "is_current": True, + }) + return { + "status": "scored", + "overall_score": result.match_score, + "band": _band(result.match_score), + "job_post_id": str(job_pk), + } diff --git a/backend/g_sheet/serializers.py b/backend/g_sheet/serializers.py index 3bd8f22..780dfe7 100644 --- a/backend/g_sheet/serializers.py +++ b/backend/g_sheet/serializers.py @@ -111,6 +111,8 @@ def serialize_form_data(row) -> dict: out[key] = _iso(value) elif isinstance(value, uuid.UUID): out[key] = str(value) + elif key == "suggested_job_post_ids": + out[key] = [str(v) for v in (value or []) if v not in (None, "")] else: out[key] = value return out diff --git a/backend/g_sheet/views.py b/backend/g_sheet/views.py index 36e422d..3f60fb8 100644 --- a/backend/g_sheet/views.py +++ b/backend/g_sheet/views.py @@ -17,6 +17,7 @@ Hierarchy: import asyncio import logging import threading +import uuid from datetime import datetime,timezone from fastapi import HTTPException @@ -249,6 +250,7 @@ class SheetImport(SheetRead): FormData.from_sheet_row(tab,row_number,record) for row_number,record in indexed ] + mapped=await FormData.stamp_suggested_job_posts(session,mapped) result=await FormData.replace_sheet(session,tab,mapped) return serialize_import({ "tab":tab, @@ -334,39 +336,97 @@ class SheetFormData(Sheet): """FormData DB mirror — query / delete only (no Google client).""" async def _hydrate_job_posts(self,items): - """Attach matching job_posts (title == position_applied_for) + assigned_job_post. + """Attach suggested job_posts, assigned_job_post, and per-job ATS scores. - No AI suggestions — form applicants already name the role. One query for - titles on the page, one for any assigned ids. + Preferred source is suggested_job_post_ids (ILIKE matches stored on + import). Legacy rows without that list still title-match. ATS is one + current score per (form, job). """ if not items: return items + from g_sheet.scoring import serialize_form_ats + from inbox.models import AtsResults from job.job_post.models import JobPosts from job.job_post.serializers import serialize_job_post session=self._require_session() + + def _job_payload(post): + payload=serialize_job_post(post) + if post.is_deleted or not post.is_active: + payload={**payload,"unavailable":True} + return payload + + suggested_ids=[] + for item in items: + for raw in item.get("suggested_job_post_ids") or []: + if raw: + suggested_ids.append(raw) + assigned_ids=[] + for item in items: + aid=item.get("assigned_job_post_id") or item.get("job_post_id") + if aid: + assigned_ids.append(aid) + wanted=list(dict.fromkeys([*suggested_ids,*assigned_ids])) + by_id={} + if wanted: + for post in await JobPosts.get_by_ids(session,wanted,active_only=False): + by_id[str(post.id)]=_job_payload(post) + titles=[(item.get("position_applied_for") or "").strip() for item in items] titles=[t for t in titles if t] by_title={} - if titles: + needs_title=any(not (item.get("suggested_job_post_ids") or []) for item in items) + if titles and needs_title: for post in await JobPosts.get_by_titles(session,titles): key=(post.title or "").strip().lower() - payload=serialize_job_post(post) - if post.is_deleted or not post.is_active: - payload={**payload,"unavailable":True} - by_title.setdefault(key,[]).append(payload) + by_title.setdefault(key,[]).append(_job_payload(post)) - assigned_ids=[item.get("job_post_id") for item in items if item.get("job_post_id")] - assigned_map={} - if assigned_ids: - for post in await JobPosts.get_by_ids(session,assigned_ids,active_only=False): - assigned_map[str(post.id)]=serialize_job_post(post) + ats_by_form=await AtsResults.get_current_for_forms( + session,[item.get("id") for item in items], + ) for item in items: - key=(item.get("position_applied_for") or "").strip().lower() - item["job_posts"]=list(by_title.get(key) or []) - aid=item.get("job_post_id") - item["assigned_job_post"]=assigned_map.get(str(aid)) if aid else None + suggested=[str(raw) for raw in (item.get("suggested_job_post_ids") or []) if raw] + item["suggested_job_post_ids"]=suggested + if suggested: + posts=[] + for sid in suggested: + payload=by_id.get(sid) + if payload is None: + posts.append({"id":sid,"unavailable":True}) + else: + posts.append(dict(payload)) + item["job_posts"]=posts + else: + key=(item.get("position_applied_for") or "").strip().lower() + item["job_posts"]=[dict(p) for p in (by_title.get(key) or [])] + + aid=item.get("assigned_job_post_id") or item.get("job_post_id") + item["assigned_job_post_id"]=str(aid) if aid else None + item["assigned_job_post"]=by_id.get(str(aid)) if aid else None + + fid=item.get("id") + try: + form_uid=uuid.UUID(str(fid)) if fid else None + except (TypeError,ValueError): + form_uid=None + scores=[serialize_form_ats(row) for row in (ats_by_form.get(form_uid) or [])] + item["ats_results"]=scores + score_by_job={ + str(s["job_post_id"]):s for s in scores if s.get("job_post_id") + } + for post in item["job_posts"]: + hit=score_by_job.get(str(post.get("id"))) + if hit: + post["overall_score"]=hit.get("overall_score") + post["band"]=hit.get("band") + assigned_score=score_by_job.get(str(aid)) if aid else None + if assigned_score and assigned_score.get("overall_score") is not None: + item["ats_score"]=assigned_score.get("overall_score") + else: + nums=[s.get("overall_score") for s in scores if s.get("overall_score") is not None] + item["ats_score"]=max(nums) if nums else None return items async def get_form_data( @@ -394,7 +454,7 @@ class SheetFormData(Sheet): return items[0] async def assign_job_post(self,record_id,job_post_id): - """Set or clear form_data.job_post_id (same contract as inbox assign). + """Set or clear form_data.assigned_job_post_id (same contract as inbox assign). Setting a job promotes the row into Users + manual_upload_candidate so Candidates / Talent Pool / Pipeline can see it (platform tag: Form). @@ -410,6 +470,8 @@ class SheetFormData(Sheet): raise HTTPException(status_code=404,detail="Form data not found") if job_post_id is not None: await self._promote_to_application(updated) + from g_sheet.scoring import enqueue_form_score + await enqueue_form_score(updated.id,job_post_id) return await self.get_form_data_by_id(record_id) async def set_processing_state(self,record_id,processing_state,current_user=None): diff --git a/backend/inbox/models.py b/backend/inbox/models.py index 7de274a..637dee8 100644 --- a/backend/inbox/models.py +++ b/backend/inbox/models.py @@ -1472,6 +1472,8 @@ class AtsResults(SQLModel, table=True): candidate_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="candidates.id") user_id: uuid.UUID | None = Field(default=None, index=True, foreign_key="users.id") 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") overall_score: float = Field(default=0.0) band: str = Field(default="") is_current: bool = Field(default=True) @@ -1530,6 +1532,53 @@ class AtsResults(SQLModel, table=True): ) return result.scalars().first() + @classmethod + async def get_for_form_job(cls, session: AsyncSession, form_data_id, job_post_id): + """Any score for this form_data row against this job — current or superseded.""" + fid = cls._as_uuid(form_data_id) + jid = cls._as_uuid(job_post_id) + if fid is None or jid is None: + return None + result = await session.execute( + select(cls) + .where(cls.form_data_id == fid, cls.job_post_id == jid) + .order_by(cls.computed_at.desc()) + ) + return result.scalars().first() + + @classmethod + async def get_current_for_form_job(cls, session: AsyncSession, form_data_id, job_post_id=None): + """Current Sheet Forms score, optionally pinned to one job post.""" + fid = cls._as_uuid(form_data_id) + if fid is None: + return None + qry = select(cls).where(cls.form_data_id == fid, cls.is_current == True) # noqa: E712 + jid = cls._as_uuid(job_post_id) if job_post_id is not None else None + if jid is not None: + qry = qry.where(cls.job_post_id == jid) + result = await session.execute(qry.order_by(cls.computed_at.desc())) + return result.scalars().first() + + @classmethod + async def get_current_for_forms(cls, session: AsyncSession, form_data_ids) -> dict: + """Current Sheet Forms scores grouped by form_data_id (newest first).""" + keys = [] + for raw in form_data_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) + .where(cls.form_data_id.in_(keys), cls.is_current == True) # noqa: E712 + .order_by(cls.computed_at.desc()) + ) + grouped: dict = {} + for row in result.scalars(): + grouped.setdefault(row.form_data_id, []).append(row) + 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. @@ -1579,15 +1628,20 @@ class AtsResults(SQLModel, table=True): Inbox scores chain on inbox_id and repoint inbox.ats_id (candidate_id may be NULL). Upload scores chain on candidate_id, or on (user_id, - job_post_id) when identity resolved to a user. Flush the INSERT first: - with no relationship() edge the unit of work emits the UPDATEs first, - and the FKs reject a pointer to a row not yet inserted.""" + job_post_id) when identity resolved to a user. Sheet Forms scores chain + on (form_data_id, job_post_id) with inbox_id and user_id left NULL. + Flush the INSERT first: with no relationship() edge the unit of work + emits the UPDATEs first, and the FKs reject a pointer to a row not yet + inserted.""" inbox_id = fields.get("inbox_id") candidate_id = fields.get("candidate_id") user_id = fields.get("user_id") job_post_id = fields.get("job_post_id") + form_data_id = fields.get("form_data_id") if inbox_id is not None: prev = await cls.get_current_for_inbox(session, inbox_id) + elif form_data_id is not None: + prev = await cls.get_current_for_form_job(session, form_data_id, job_post_id) elif candidate_id is not None: prev = await cls.get_current_for_candidate(session, candidate_id) elif user_id is not None: diff --git a/backend/inbox/tasks.py b/backend/inbox/tasks.py index aa8e9d7..bb74a22 100644 --- a/backend/inbox/tasks.py +++ b/backend/inbox/tasks.py @@ -62,6 +62,27 @@ async def score_message_against_job(record_id:str,job_id:str) -> dict: return {"status":"scored","results":len(results)} +@broker.task( + task_name="g_sheet.score_form", + retry_on_error=True, + max_retries=MAX_RETRIES, + delay=RETRY_DELAY, +) +async def score_form_data(form_data_id:str,job_id:str) -> dict: + """ATS-score one Sheet Forms CV (extracted_data) against one job post.""" + from g_sheet.scoring import score_form_against_job + + try: + uuid.UUID(str(form_data_id)) + uuid.UUID(str(job_id)) + except (TypeError,ValueError): + raise PermanentTaskError("form_data_id and job_id must be uuids") + result=await score_form_against_job(form_data_id,job_id) + if result.get("status")=="failed": + raise RuntimeError(result.get("error_code") or "form ats failed") + return result + + @broker.task( task_name="inbox.score_message", retry_on_error=True, diff --git a/backend/job/job_post/models.py b/backend/job/job_post/models.py index 3f2ce3b..beb75aa 100644 --- a/backend/job/job_post/models.py +++ b/backend/job/job_post/models.py @@ -148,6 +148,52 @@ class JobPosts(SQLModel, table=True): result = await session.execute(statement) return list(result.scalars().all()) + @staticmethod + def title_ilike_pattern(applied_for: str) -> str | None: + """ILIKE pattern so job_posts.title contains the form's Position Applied For.""" + needle = (applied_for or "").strip() + if not needle: + return None + escaped = needle.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + return f"%{escaped}%" + + @classmethod + async def ids_for_title_ilike(cls, session: AsyncSession, applied_for: str) -> list[uuid.UUID]: + """All non-deleted job_posts.id whose title ILIKE-contains applied_for. + + One form title can match many posts. Order is created_at DESC, id DESC + so the UI/ATS list is stable. Empty if blank or no row. Suggested, + not recruiter-assigned. + """ + pattern = cls.title_ilike_pattern(applied_for) + if not pattern: + return [] + statement = ( + select(cls.id) + .where(cls.is_deleted == False) # noqa: E712 + .where(cls.title.ilike(pattern, escape="\\")) + .order_by(cls.created_at.desc(), cls.id.desc()) + ) + result = await session.execute(statement) + return list(result.scalars().all()) + + @classmethod + async def ids_for_titles_ilike( + cls, session: AsyncSession, titles: list[str], + ) -> dict[str, list[uuid.UUID]]: + """Map stripped Position Applied For → every matching job_posts.id.""" + out: dict[str, list[uuid.UUID]] = {} + seen: set[str] = set() + for raw in titles or []: + key = (raw or "").strip() + if not key or key in seen: + continue + seen.add(key) + found = await cls.ids_for_title_ilike(session, key) + if found: + out[key] = found + return out + @classmethod async def fetch_job_posts( cls, diff --git a/backend/tests/test_g_sheet_plugins.py b/backend/tests/test_g_sheet_plugins.py index 8c331fa..0775b2c 100644 --- a/backend/tests/test_g_sheet_plugins.py +++ b/backend/tests/test_g_sheet_plugins.py @@ -180,3 +180,65 @@ def test_cv_dest_path_stays_inside_dir(tmp_path): assert dest.parent == tmp_path.resolve() assert dest.name.endswith(".pdf") assert dest.name.startswith("1l5HOY5R6KiL6270A_sV56FkdX6EIGfI4") + + +def test_title_ilike_pattern_contains_and_escapes(): + from job.job_post.models import JobPosts + + assert JobPosts.title_ilike_pattern("") is None + assert JobPosts.title_ilike_pattern(" ") is None + assert JobPosts.title_ilike_pattern("Brand Manager") == "%Brand Manager%" + assert JobPosts.title_ilike_pattern("C++ _dev%") == r"%C++ \_dev\%%" + + +async def test_stamp_suggested_job_posts_sets_ids_or_empty(monkeypatch): + import uuid + + job_a = uuid.uuid4() + job_b = uuid.uuid4() + + async def fake_ids(session, titles): + return {"Executive Secretary": [job_a, job_b]} + + monkeypatch.setattr( + "job.job_post.models.JobPosts.ids_for_titles_ilike", + fake_ids, + ) + records = [ + {"position_applied_for": "Executive Secretary"}, + {"position_applied_for": "Unknown Role"}, + {"position_applied_for": None}, + {"position_applied_for": " "}, + ] + out = await FormData.stamp_suggested_job_posts(object(), records) + assert out[0]["suggested_job_post_ids"] == [str(job_a), str(job_b)] + assert out[0]["job_post_id"] is None + assert out[0]["assigned_job_post_id"] is None + assert out[1]["suggested_job_post_ids"] == [] + assert out[1]["job_post_id"] is None + assert out[1]["assigned_job_post_id"] is None + assert out[2]["suggested_job_post_ids"] == [] + assert out[3]["suggested_job_post_ids"] == [] + + +def test_score_job_ids_assigned_wins_else_suggested(): + import uuid + + a = uuid.uuid4() + b = uuid.uuid4() + assigned = uuid.uuid4() + assert FormData.score_job_ids({ + "suggested_job_post_ids": [str(a), str(b)], + "assigned_job_post_id": assigned, + }) == [assigned] + assert FormData.score_job_ids({ + "suggested_job_post_ids": [str(a), str(b)], + "job_post_id": assigned, + }) == [assigned] + assert FormData.score_job_ids({ + "suggested_job_post_ids": [str(a), str(b), str(a)], + "assigned_job_post_id": None, + "job_post_id": None, + }) == [a, b] + assert FormData.score_job_ids({"suggested_job_post_ids": None}) == [] + assert FormData.score_job_ids(None) == [] diff --git a/frontend/src/screens/Inbox.jsx b/frontend/src/screens/Inbox.jsx index c4ddafc..ccb4492 100644 --- a/frontend/src/screens/Inbox.jsx +++ b/frontend/src/screens/Inbox.jsx @@ -169,11 +169,22 @@ function formReceivedAt(entryDate, entryTime) { /** * GET /sheet/form-data/fetch row → the same list/detail shape the email channel * uses for name / avatar / position / source / time, plus form-only profile fields. - * job_posts are title-matched (position_applied_for ↔ job_posts.title), not AI. + * job_posts come from suggested_job_post_ids (ILIKE title matches), not AI. */ function mapFormRow(row) { const name = (row.name || row.candidate_email || 'Unknown').trim() const jobPosts = Array.isArray(row.job_posts) ? row.job_posts : [] + const atsResults = Array.isArray(row.ats_results) ? row.ats_results : [] + const assignedId = row.assigned_job_post_id + ? String(row.assigned_job_post_id) + : (row.job_post_id ? String(row.job_post_id) : null) + const assignedScore = atsResults.find((s) => String(s.job_post_id) === String(assignedId)) + const numericScores = atsResults + .map((s) => Number(s.overall_score)) + .filter((n) => Number.isFinite(n)) + const atsScore = row.ats_score + ?? assignedScore?.overall_score + ?? (numericScores.length ? Math.max(...numericScores) : null) const state = row.processing_state || 'unread' return { kind: 'form', @@ -212,7 +223,12 @@ function mapFormRow(row) { processingState: state, duplicate: Boolean(row.is_duplicate), jobPosts, - assignedId: row.job_post_id ? String(row.job_post_id) : null, + suggestedIds: Array.isArray(row.suggested_job_post_ids) + ? row.suggested_job_post_ids.map(String) + : [], + atsResults, + atsScore, + assignedId, assignedPost: row.assigned_job_post || null, } } @@ -1324,7 +1340,7 @@ function firstResumeKey(item) { /** * Sheet form applicant detail — profile grids + resume/LinkedIn links + - * title-matched job selection (position_applied_for ↔ job_posts.title) + + * ILIKE-matched job selection (one applied-for title → many job posts) + * the same Import / Shortlist / Duplicate / Reject actions as email. */ function FormApplicantDetail({ @@ -1384,6 +1400,23 @@ function FormApplicantDetail({ const alreadyProcessed = i.processing === 'Processed' const shortlistLocked = !alreadyProcessed && !jobChosen const panelBusy = busy || assignMutation.isPending + const selectedAts = useMemo(() => { + const jobId = selection || i.assignedId + if (!jobId) return null + const fromPost = (i.jobPosts || []).find((p) => String(p.id) === String(jobId)) + const fromResults = (i.atsResults || []).find((s) => String(s.job_post_id) === String(jobId)) + const score = fromPost?.overall_score ?? fromResults?.overall_score + if (score == null || !Number.isFinite(Number(score))) return null + const n = Number(score) + return { + score: n, + band: fromPost?.band || fromResults?.band + || (n >= 82 ? 'Strong Match' : n >= 65 ? 'Potential Match' : 'Weak Match'), + } + }, [selection, i.assignedId, i.jobPosts, i.atsResults]) + const selectedRingColor = selectedAts == null + ? undefined + : selectedAts.score >= 82 ? 'var(--success)' : selectedAts.score >= 65 ? 'var(--warning)' : 'var(--danger)' async function handleMove() { if (busy || alreadyProcessed) return @@ -1426,6 +1459,14 @@ function FormApplicantDetail({
{i.rowNumber}
)} + {selectedAts != null && ( +
+
+
{Math.round(selectedAts.score)}
+
+
{selectedAts.band || 'ATS Score'}
+
+ )} {(resumeHref || profileHref) && ( @@ -1563,6 +1604,7 @@ function FormApplicantDetail({
Matching roles
Matched by position applied for: {orDash(i.position)} + {i.jobPosts?.length > 1 ? ` · ${i.jobPosts.length} roles` : ''}
{matchCards.length === 0 && !manualPost ? ( diff --git a/frontend/src/ui/SuggestedRoles.jsx b/frontend/src/ui/SuggestedRoles.jsx index 567c267..eabb3a6 100644 --- a/frontend/src/ui/SuggestedRoles.jsx +++ b/frontend/src/ui/SuggestedRoles.jsx @@ -10,7 +10,7 @@ import { useState } from 'react' import { useQuery } from '@tanstack/react-query' import Modal from './Modal' -import { Badge, EmptyState, Icon } from './primitives' +import { Badge, EmptyState, Icon, ScoreChip } from './primitives' import { friendlyAuthError } from '../lib/errors' import { qk } from '../lib/queryKeys' import * as jobPostsApi from '../api/jobPosts' @@ -74,6 +74,7 @@ export function JobCard({ post, rank, selected, onSelect, resumeText, manual, ba ) : ( {post.status || 'draft'} )} + {post?.overall_score != null && } {selected && } {meta &&
{meta}
} From e22da6ed0bcf7ef6277288ee4fcdd4aa22753db2 Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Thu, 3 Sep 2026 16:13:09 +0500 Subject: [PATCH 3/3] . --- backend/g_sheet/views.py | 4 ++-- frontend/src/screens/Inbox.jsx | 39 +++++++++++++++++++++++++--------- 2 files changed, 31 insertions(+), 12 deletions(-) diff --git a/backend/g_sheet/views.py b/backend/g_sheet/views.py index 3f60fb8..7080edd 100644 --- a/backend/g_sheet/views.py +++ b/backend/g_sheet/views.py @@ -423,10 +423,10 @@ class SheetFormData(Sheet): post["band"]=hit.get("band") assigned_score=score_by_job.get(str(aid)) if aid else None if assigned_score and assigned_score.get("overall_score") is not None: - item["ats_score"]=assigned_score.get("overall_score") + item["ats_score"]=round(float(assigned_score.get("overall_score"))) else: nums=[s.get("overall_score") for s in scores if s.get("overall_score") is not None] - item["ats_score"]=max(nums) if nums else None + item["ats_score"]=round(float(max(nums))) if nums else None return items async def get_form_data( diff --git a/frontend/src/screens/Inbox.jsx b/frontend/src/screens/Inbox.jsx index ccb4492..3557eba 100644 --- a/frontend/src/screens/Inbox.jsx +++ b/frontend/src/screens/Inbox.jsx @@ -166,6 +166,13 @@ function formReceivedAt(entryDate, entryTime) { return d } +/** Same numeric gate as email `ats_score` / pipeline ScoreChip. */ +function asAtsScore(value) { + if (value == null || value === '') return null + const n = Number(value) + return Number.isFinite(n) ? n : null +} + /** * GET /sheet/form-data/fetch row → the same list/detail shape the email channel * uses for name / avatar / position / source / time, plus form-only profile fields. @@ -179,12 +186,12 @@ function mapFormRow(row) { ? String(row.assigned_job_post_id) : (row.job_post_id ? String(row.job_post_id) : null) const assignedScore = atsResults.find((s) => String(s.job_post_id) === String(assignedId)) - const numericScores = atsResults - .map((s) => Number(s.overall_score)) - .filter((n) => Number.isFinite(n)) - const atsScore = row.ats_score - ?? assignedScore?.overall_score - ?? (numericScores.length ? Math.max(...numericScores) : null) + const fromResults = atsResults.map((s) => asAtsScore(s.overall_score)).filter((n) => n != null) + const fromJobs = jobPosts.map((p) => asAtsScore(p.overall_score)).filter((n) => n != null) + const atsScore = asAtsScore(row.ats_score) + ?? asAtsScore(assignedScore?.overall_score) + ?? (fromResults.length ? Math.max(...fromResults) : null) + ?? (fromJobs.length ? Math.max(...fromJobs) : null) const state = row.processing_state || 'unread' return { kind: 'form', @@ -380,7 +387,7 @@ async function fetchApplications(params) { linkedinSlug: row.linkedin_slug || '', linkedinUrl: row.linkedin_url || '', resumeText: row.resume_text || '', - atsScore: row.ats_score, + atsScore: asAtsScore(row.ats_score), phone: row.phone, experience: row.experience, recruiter: row.recruiter, @@ -869,7 +876,19 @@ export default function Inbox() { enabled: Boolean(selectedId), }) - const selectedRow = inbox.find((i) => i.id === selectedId) + const sidebar = useMemo(() => { + if (!isForms) return list + const detail = detailQuery.data + const detailScore = asAtsScore(detail?.atsScore) + if (!detail?.id || detailScore == null) return list + return list.map((row) => ( + String(row.id) === String(detail.id) + ? { ...row, atsScore: asAtsScore(row.atsScore) ?? detailScore } + : row + )) + }, [isForms, list, detailQuery.data]) + + const selectedRow = sidebar.find((i) => i.id === selectedId) const selected = selectedRow || detailQuery.data ? { ...selectedRow, ...(detailQuery.data ?? {}) } : null @@ -1164,7 +1183,7 @@ export default function Inbox() { {isForms ? 'No form responses in this sheet tab.' : 'No applications in this view.'} ) : ( - list.map((i) => ( + sidebar.map((i) => (
{outlookListTime(i.received)}
- {i.atsScore != null && ( + {asAtsScore(i.atsScore) != null && (
)}