diff --git a/backend/.env.example b/backend/.env.example index 438ab71..e47fb09 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -56,9 +56,11 @@ MAX_RESUME_CHARS=60000 REDIS_URL=redis://localhost:6379/0 TASKIQ_QUEUE_NAME=inbox +TASKIQ_CV_QUEUE_NAME=cv_upload TASKIQ_MAX_RETRIES=3 TASKIQ_RETRY_DELAY=5 TASKIQ_MAX_DELAY=120 TASKIQ_DLQ_STREAM=taskiq:dlq TASKIQ_IDLE_TIMEOUT_MS=600000 +MANUAL_UPLOAD_TO_ADDRESS=manual-cv-upload@hr-ats.local APP_VERSION=dev diff --git a/backend/README.md b/backend/README.md index 290a607..6f3f6cd 100644 --- a/backend/README.md +++ b/backend/README.md @@ -47,9 +47,13 @@ flowchart TB API -->|enqueue| REDIS[("Redis Streams")] REDIS --> W["Taskiq worker\ninbox.tasks + inbox.sync_tasks"] + REDIS --> WCV["Taskiq CV worker\ninbox.cv_tasks"] SCHED["Taskiq scheduler\ncron"] --> REDIS + SCHEDCV["Taskiq CV scheduler\nretries"] --> REDIS W --> PG + WCV --> PG W --> AGENT["LangGraph agent\nagent/"] + WCV --> AGENT AGENT --> OAI["OpenAI"] API -->|GET /emails, /sync/read-status| MAILAPI["Email API (MS Graph proxy)"] @@ -305,7 +309,7 @@ Base URL: `http://localhost:8000`. Interactive docs at `/docs`. | GET | `/jobs/alias` | public | Accepted platform shorthands (`fb`, `ig`, `li`, `x`, …) | | POST | `/job/post-job` | `job_board.create` | Render the ad, create the Buffer post, persist the result | | GET | `/job/buffer/channels` | `job_board.view` | Connected Buffer channels across all organizations | -| POST | `/candidate/cv_upload` | `candidates.create` | Upload a PDF, get extracted text back | +| POST | `/candidate/cv_upload` | `candidates.create` | Upload a PDF; extract email, persist like an emailed CV, enqueue matching on the CV stream | | POST | `/candidate/inbox-match?inbox_message_id=` | `candidates.edit` | Queue a forced re-match for a stored message | | GET | `/candidate/fetch?user_id=` | `candidates.view` | Candidate profile via the `inbox` join | @@ -367,7 +371,8 @@ Taskiq over **Redis Streams**, with a result backend and a Redis-backed schedule | Task | Trigger | What it does | |---|---|---| -| `inbox.match_message` | enqueued by `/email/fetch`, `/inbox/{id}/match`, `/candidate/inbox-match` | Extract résumé text → run the agent → write match results | +| `inbox.match_message` | enqueued by `/email/fetch`, `/inbox/{id}/match`, `/candidate/inbox-match` onto the `inbox` stream | Extract résumé text → run the agent → write match results | +| `inbox.match_message` (CV broker) | enqueued by `/candidate/cv_upload` onto the `cv_upload` stream | Same matcher as above; isolated so uploads never sit behind `/email/fetch` backlog | | `inbox.sync_read_status` | cron, `EMAIL_SYNC_CRON` (default every minute) | Pull read-status deltas from the Email API and apply them | | `ping` | manual | Framework smoke test | @@ -512,6 +517,7 @@ own keys with `os.getenv`. |---|---| | `REDIS_URL` | `redis://localhost:6379/0` | | `TASKIQ_QUEUE_NAME` | `inbox` | +| `TASKIQ_CV_QUEUE_NAME` | `cv_upload` | | `TASKIQ_CONSUMER_GROUP` | `taskiq` | | `TASKIQ_MAX_RETRIES` | `3` | | `TASKIQ_RETRY_DELAY` | `5` | @@ -519,6 +525,7 @@ own keys with `os.getenv`. | `TASKIQ_IDLE_TIMEOUT_MS` | `600000` | | `TASKIQ_DLQ_STREAM` | `taskiq:dlq` | | `TASKIQ_WORKER_NAME` | falls back to `HOSTNAME` | +| `MANUAL_UPLOAD_TO_ADDRESS` | `manual-cv-upload@hr-ats.local` — To address stamped on synthetic inbox rows so source resolves to `Manual CV Upload` | | `APP_VERSION` | `dev` | --- @@ -556,12 +563,24 @@ taskiq worker taskiq_management.broker_setup:broker \ inbox.tasks inbox.sync_tasks taskiq_management.tasks ``` +**CV-upload worker** (isolated stream for manual uploads): + +```bash +taskiq worker taskiq_management.cv_broker_setup:cv_broker inbox.cv_tasks +``` + **Scheduler** (cron ticks for `inbox.sync_read_status`): ```bash taskiq scheduler taskiq_management.broker_setup:scheduler inbox.sync_tasks ``` +**CV-upload scheduler** (retries for the CV stream): + +```bash +taskiq scheduler taskiq_management.cv_broker_setup:cv_scheduler inbox.cv_tasks +``` + Docs: --- @@ -594,13 +613,14 @@ and a module called `alembic.py` would shadow the installed package. ## Docker -The repo-root `docker-compose.yml` runs Redis plus the two Taskiq processes; the API itself is -expected to run on the host (the compose file points the containers at -`host.docker.internal` for the database). +The repo-root `docker-compose.yml` runs Redis plus the four Taskiq processes (inbox +worker/scheduler and CV-upload worker/scheduler); the API itself is expected to run on the +host (the compose file points the containers at `host.docker.internal` for the database). ```bash docker compose up -d # from the repo root docker compose logs -f taskiq-worker +docker compose logs -f taskiq-cv-worker ``` `backend/Dockerfile` builds a `python:3.12-slim` image whose default command is the Taskiq diff --git a/backend/inbox/cv_tasks.py b/backend/inbox/cv_tasks.py new file mode 100644 index 0000000..8051c90 --- /dev/null +++ b/backend/inbox/cv_tasks.py @@ -0,0 +1,17 @@ +"""CV-upload Taskiq tasks — same matcher as inbox.tasks, own broker/stream.""" + +from __future__ import annotations + +from inbox.tasks import match_inbox_message +from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY +from taskiq_management.cv_broker_setup import cv_broker + + +@cv_broker.task( + task_name="inbox.match_message", + retry_on_error=True, + max_retries=MAX_RETRIES, + delay=RETRY_DELAY, +) +async def match_uploaded_cv(record_id:str,force:bool=False) -> dict: + return await match_inbox_message(record_id,force) diff --git a/backend/job/app.py b/backend/job/app.py index 586f70a..2efa0d9 100644 --- a/backend/job/app.py +++ b/backend/job/app.py @@ -15,7 +15,6 @@ from job.job_post.serializers import serialize_job_post import logging from job.job_post.plugins import PlatformAlias from fastapi import UploadFile, File, Form -from pydantic import BaseModel from dotenv import load_dotenv from datetime import datetime, time, timezone from pydantic import BaseModel @@ -98,10 +97,65 @@ async def get_job_alias(): except Exception as e: raise HTTPException(status_code=500,detail=str(e)) +@router.post("/candidate/create/candidate") +async def create_manual_candidate( + file: UploadFile = File(...), + candidate_email: str | None = Form(None), + candidate_name: str | None = Form(None), + candidate_phone: str | None = Form(None), + job_post_id: str | None = Form(None), + current_company: str | None = Form(None), + platform: str | None = Form(None), + experience: str | None = Form(None), + status: str | None = Form(None), + referral_by: str | None = Form(None), + current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_CREATE)), + session: AsyncSession = Depends(get_session), +): + saved_path=None + try: + file_content = await file.read() + logger.info(f"Received file: {file.filename} ({len(file_content)} bytes)") + reader=FileRead(session=session,filename=file.filename,file=file_content) + # Parse first: an unreadable PDF is a 400, and doing it before the write + # keeps a file that can never back a row off the disk entirely. + parsed=await reader.injest_manual_upload() + saved=await reader.save_manual_upload() + saved_path=saved.get("file_path") + service=CandidateView(session=session) + data=await service.create_candidate( + candidate_email=candidate_email, + candidate_name=candidate_name, + candidate_phone=candidate_phone, + job_post_id=job_post_id, + current_company=current_company, + platform=platform, + experience=experience, + status=status, + referral_by=referral_by, + file_name=saved.get("file_name"), + file_path=saved_path, + full_text=parsed.get("text") or "", + current_user=current_user.get("id"), + ) + return JSONResponse(content={"data":data,"status_code":200}) + except HTTPException: + # create_candidate rejects a blank email with a 422 AFTER the file has + # landed, so without this every such attempt would leave an orphan PDF. + FileRead.discard_upload(saved_path) + raise + except Exception as e: + FileRead.discard_upload(saved_path) + raise HTTPException(status_code=500,detail=str(e)) + @router.post("/candidate/cv_upload") async def cv_upload( file: UploadFile = File(...), + candidate_email: str | None = Form(None), + candidate_name: str | None = Form(None), + candidate_phone: str | None = Form(None), + job_post_id: str | None = Form(None), current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_CREATE)), session: AsyncSession = Depends(get_session), ): @@ -109,8 +163,9 @@ async def cv_upload( file_content = await file.read() logger.info(f"Received file: {file.filename} ({len(file_content)} bytes)") service=FileRead(session=session,filename=file.filename,file=file_content) - data=await service.read_file() - + data=await service.ingest_upload( + candidate_email=candidate_email,candidate_name=candidate_name, + ) return JSONResponse(content={"data":data,"status_code":200}) except HTTPException: raise diff --git a/backend/job/candidate/models.py b/backend/job/candidate/models.py index d758e3a..e360eba 100644 --- a/backend/job/candidate/models.py +++ b/backend/job/candidate/models.py @@ -10,6 +10,7 @@ from sqlmodel import Field, Relationship, SQLModel, select if TYPE_CHECKING: from inbox.models import Inbox from users.models import Users + from job.job_post.models import JobPosts def _now() -> datetime: @@ -21,6 +22,95 @@ def _now() -> datetime: # `datetime` to TIMESTAMP WITHOUT TIME ZONE, and asyncpg refuses to bind an aware # value to one — "can't subtract offset-naive and offset-aware datetimes" — which # turns every insert here into a 500. Same pairing as job/job_post/models.py. +class Manual_UPLOAD_CANDIDATE(SQLModel, table=True): + __tablename__ = "manual_upload_candidate" + id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) + candidate_email: str = Field(default="") + candidate_name: str = Field(default="") + candidate_phone: str = Field(default="") + job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id") + full_text: str = Field(default="") + current_company: str = Field(default="") + user_id: uuid.UUID | None = Field(default=None, foreign_key="users.id") + platform: str = Field(default="") + created_by: uuid.UUID | None = Field(default=None, foreign_key="users.id") + experience: str = Field(default="") + status: str = Field(default="") + # Free text, not a users FK: a referrer is often someone outside the system + # (a client, a former colleague), and recruiters type whatever the candidate + # told them. "" rather than NULL keeps it consistent with the columns above. + # + # server_default is load-bearing and NOT decoration, unlike the columns above + # — they arrived with the CREATE TABLE, this one arrives as an ALTER. The + # startup autogenerate would emit `ADD COLUMN referral_by VARCHAR NOT NULL`, + # which Postgres rejects outright on a table that already holds rows. The + # DEFAULT backfills them. Pass the bare "" — SQLAlchemy quotes a plain string + # into DEFAULT '', whereas "''" would render DEFAULT '''''' instead. + referral_by: str = Field(default="", sa_column_kwargs={"server_default": ""}) + # The CV as uploaded: file_name is the recruiter-facing original, file_path + # the absolute location under inbox/decoded_attachments. They differ on + # purpose — the stored basename is uniquified so two candidates uploading + # "resume.pdf" cannot overwrite one another (see FileRead.save_manual_upload). + # Same ALTER-on-a-populated-table reasoning as referral_by above, so both + # carry a server default. + file_name: str = Field(default="", sa_column_kwargs={"server_default": ""}) + file_path: str = Field(default="", sa_column_kwargs={"server_default": ""}) + created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) + updated_at: datetime = Field(default_factory=_now, 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 create_manual_upload_candidate(cls, session: AsyncSession, fields: dict): + import os + + from role.models import EnumRoles, Roles + from users.models import Users + from users.plugins import hash_password + + email=(fields.get("candidate_email") or "").strip().lower() + name=(fields.get("candidate_name") or "").strip() or email + default_pw=os.getenv("DEFAULT_CANDIDATE_PASSWORD","Utopia!@#") + + user=await Users.get_user_by_email(session,email) + if not user: + role=await Roles.get_role_by_name(session,EnumRoles.CANDIDATE.value) + user=await Users.insert_user(session,{ + "name":name, + "email":email, + "role_id":role.id if role else 8, + "password":hash_password(default_pw), + "is_active":True, + "is_deleted":False, + }) + + row=cls( + candidate_email=email, + candidate_name=name, + candidate_phone=(fields.get("candidate_phone") or "").strip(), + job_post_id=cls._as_uuid(fields.get("job_post_id")), + full_text=fields.get("full_text") or "", + current_company=(fields.get("current_company") or "").strip(), + user_id=user.id, + platform=(fields.get("platform") or "").strip(), + created_by=cls._as_uuid(fields.get("created_by")), + experience=(fields.get("experience") or "").strip(), + status=(fields.get("status") or "").strip(), + referral_by=(fields.get("referral_by") or "").strip(), + file_name=(fields.get("file_name") or "").strip(), + file_path=(fields.get("file_path") or "").strip(), + ) + session.add(row) + await session.commit() + await session.refresh(row) + return row class Candidates(SQLModel, table=True): diff --git a/backend/job/candidate/plugins.py b/backend/job/candidate/plugins.py index fc06911..2b68fac 100644 --- a/backend/job/candidate/plugins.py +++ b/backend/job/candidate/plugins.py @@ -150,3 +150,111 @@ def documents_from_message(file_name: str | None, file_path: str | None) -> list out.append({"name": path.rsplit("/", 1)[-1].rsplit("\\", 1)[-1], "path": path}) return out + +# Same system prefixes inbox/models._is_linkable_sender rejects — anything we +# accept here must remain linkable when insert_email creates the users row. +_SKIP_SENDER_PREFIXES = ( + "noreply", "no-reply", "donotreply", "do-not-reply", + "mailer-daemon", "postmaster", "bounce", +) +_ROLE_LOCAL_PARTS = frozenset({ + "info", "hr", "careers", "jobs", "admin", "support", "contact", "sales", + "recruitment", "office", "team", "hello", "enquiry", "inquiry", "recruit", + "talent", "hiring", "apply", "applications", "webmaster", "helpdesk", +}) +_EMAIL_RE = re.compile( + r"(?i)\b([a-z0-9][a-z0-9._%+\-]{0,63})@([a-z0-9](?:[a-z0-9\-]{0,61}[a-z0-9])?" + r"(?:\.[a-z0-9](?:[a-z0-9\-]{0,61}[a-z0-9])?)+)\b" +) +_PHONE_RE = re.compile(r"(?:\+?\d[\d\s\-().]{7,}\d)") +_LABEL_RE = re.compile(r"(?i)\b(?:e[\-\s]?mail|mail[\s\-]?id|contact)\b") +_REF_HEADING_RE = re.compile(r"(?i)^\s*(?:references?|referees?)\b") +_REF_MENTION_RE = re.compile( + r"(?i)\b(?:reference|referee|manager|supervisor|contact\s+person)\b" +) +_HEADER_LINE_COUNT = 12 +_MIN_ACCEPT_SCORE = 3 + + +def _presumed_name_tokens(lines: list[str]) -> list[str]: + """First non-empty line with 2+ alpha tokens and no digits/@ — CV name header.""" + for line in lines: + stripped = line.strip() + if not stripped: + continue + if any(ch.isdigit() for ch in stripped) or "@" in stripped: + continue + tokens = [_letters_only(t) for t in re.split(r"\s+", stripped) if _letters_only(t)] + if len(tokens) >= 2: + return tokens + return [] + + +def _email_local_ok(local: str) -> bool: + lowered = (local or "").lower() + if lowered in _ROLE_LOCAL_PARTS: + return False + return not lowered.startswith(_SKIP_SENDER_PREFIXES) + + +def extract_candidate_email(text: str) -> tuple[str | None, list[str]]: + """Pick the candidate's own email from CV text, or None when ambiguous/absent. + + Returns ``(best, all_plausible)``. Ambiguity is intentional — a wrong guess + would create a user under a stranger's address and mail them a confirm link. + """ + if not text or not text.strip(): + return None, [] + + lines = text.splitlines() + name_tokens = _presumed_name_tokens(lines) + in_references = False + scored: list[tuple[int, int, str]] = [] # (score, first_line_idx, email) + seen: dict[str, int] = {} # lower email -> index in scored + + for idx, line in enumerate(lines): + if _REF_HEADING_RE.search(line): + in_references = True + for match in _EMAIL_RE.finditer(line): + local, domain = match.group(1), match.group(2) + if not _email_local_ok(local): + continue + email = f"{local}@{domain}".lower() + score = 0 + if idx < _HEADER_LINE_COUNT: + score += 3 + local_letters = _letters_only(local) + if local_letters and any( + tok and (tok in local_letters or local_letters in tok) + for tok in name_tokens + ): + score += 3 + if _LABEL_RE.search(line) or _PHONE_RE.search(line): + score += 1 + if in_references: + score -= 5 + if _REF_MENTION_RE.search(line): + score -= 3 + + if email in seen: + prev_i = seen[email] + prev_score, _, _ = scored[prev_i] + if score > prev_score: + scored[prev_i] = (score, idx, email) + continue + seen[email] = len(scored) + scored.append((score, idx, email)) + + if not scored: + return None, [] + + scored.sort(key=lambda t: (-t[0], t[1])) + plausible = [email for _, _, email in scored] + top_score, _, top_email = scored[0] + runner_up = scored[1][0] if len(scored) > 1 else None + if top_score < _MIN_ACCEPT_SCORE: + return None, plausible + if runner_up is not None and top_score <= runner_up: + return None, plausible + return top_email, plausible + diff --git a/backend/job/candidate/serializers.py b/backend/job/candidate/serializers.py index 4f995e5..da0651a 100644 --- a/backend/job/candidate/serializers.py +++ b/backend/job/candidate/serializers.py @@ -33,6 +33,28 @@ def serialize_candidate(row) -> dict: } +def serialize_manual_upload_candidate(row) -> Dict[str,Any]: + return { + "id":str(row.id) if row.id else None, + "candidate_email":row.candidate_email, + "candidate_name":row.candidate_name, + "candidate_phone":row.candidate_phone, + "job_post_id":str(row.job_post_id) if row.job_post_id else None, + "full_text":row.full_text, + "current_company":row.current_company, + "user_id":str(row.user_id) if row.user_id else None, + "platform":row.platform, + "created_by":str(row.created_by) if row.created_by else None, + "experience":row.experience, + "status":row.status, + "referral_by":row.referral_by, + "file_name":row.file_name, + "file_path":row.file_path, + "created_at":row.created_at.isoformat() if row.created_at else None, + "updated_at":row.updated_at.isoformat() if row.updated_at else None, + } + + def serialize_candidate_profile( link:Inbox|List[Inbox]|Dict[str,Any]|List[Dict[str,Any]], *, diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index c1a4846..7d36fb5 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -1,6 +1,8 @@ from sqlalchemy.ext.asyncio import AsyncSession -import asyncio,dataclasses,hashlib,os,logging,io,uuid +import asyncio,base64,dataclasses,hashlib,io,logging,os,uuid from datetime import datetime,timezone +from pathlib import Path +from dotenv import load_dotenv from fastapi import HTTPException from pypdf import PdfReader from sqlalchemy import select @@ -22,8 +24,17 @@ from job.candidate.plugins import ( from job.candidate.serializers import serialize_candidate,serialize_candidate_profile from job.job_post.models import JobPosts from job.job_post.serializers import serialize_job_post -from job.candidate.models import Notes +from job.candidate.serializers import serialize_manual_upload_candidate +from job.candidate.models import Notes,Manual_UPLOAD_CANDIDATE from job.notes.serializers import serialize_note +from job.candidate.plugins import extract_candidate_email + +load_dotenv() +logger=logging.getLogger("job.candidate.views") +CV_QUEUE_NAME=os.getenv("TASKIQ_CV_QUEUE_NAME","cv_upload") +MANUAL_UPLOAD_TO_ADDRESS=os.getenv( + "MANUAL_UPLOAD_TO_ADDRESS","manual-cv-upload@hr-ats.local" +) class FileRead: def __init__(self,session:AsyncSession,filename=None,file=None): @@ -46,6 +57,152 @@ class FileRead: raise except Exception as e: raise HTTPException(400, str(e)) + async def injest_manual_upload(self): + try: + parsed=await self.read_file() + text=(parsed.get("text") or "").strip() + if not text: + raise HTTPException(status_code=400,detail="No usable text could be extracted from the PDF") + return parsed + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=400,detail=str(e)) + + async def save_manual_upload(self): + """Write the uploaded CV under inbox/decoded_attachments. + + Returns ``{"file_name", "file_path"}``: the recruiter-facing original + name, and the absolute path actually written. + + Those two differ deliberately. decode_attachment writes ``Path(name).name`` + with plain ``write_bytes`` — no collision handling — so two candidates + uploading "resume.pdf" would silently clobber each other and the first + row's file_path would then serve the second candidate's CV. Prefixing the + stored basename with a uuid makes every upload its own file, while + file_name keeps what the recruiter recognises. resolve_attachment_path + handles the result either way: the stored absolute path wins, and its + basename-under-attachments fallback still finds the prefixed name. + """ + from inbox.file_decoder import AttachmentDecodeError,decode_attachment + + # Separators normalized before taking the basename: a Windows client can + # send "C:\Users\x\cv.pdf", whose Path(...).name on Linux is the whole + # string. Same reasoning as inbox.plugins.resolve_attachment_path. + original=Path((self.filename or "resume.pdf").replace("\\","/")).name or "resume.pdf" + stored=f"{uuid.uuid4().hex}-{original}" + try: + paths=await decode_attachment([{ + "name":stored, + "contentBytes":base64.b64encode(self.file).decode("ascii"), + }]) + except AttachmentDecodeError as e: + raise HTTPException(status_code=400,detail=str(e)) + if not paths: + # decode_attachment skips rather than raises on an unsupported + # extension, so an empty list is the only signal that nothing landed. + raise HTTPException(status_code=400,detail="attachment could not be saved") + return {"file_name":original,"file_path":paths[0]} + + @staticmethod + def discard_upload(file_path): + """Best-effort removal of a saved CV whose row never got created. + + Called on the failure path so a rejected request (a missing email, a DB + error) does not leave an orphan PDF behind. Failure to delete is logged + and swallowed — it must never mask the error that got us here. + """ + if not file_path: + return + try: + Path(file_path).unlink(missing_ok=True) + except OSError as e: + logger.warning("could not remove orphaned upload %s: %s",file_path,e) + + async def ingest_upload(self,candidate_email=None,candidate_name=None): + """Persist a recruiter-uploaded CV with full email-ingestion parity.""" + from inbox.file_decoder import AttachmentDecodeError,decode_attachment + from inbox.cv_tasks import match_uploaded_cv + from inbox.views import Email + + parsed=await self.read_file() + text=parsed.get("text") or "" + detected,emails_found=extract_candidate_email(text) + supplied=(candidate_email or "").strip().lower() or None + email=supplied or detected + email_source="recruiter" if supplied else ("cv" if detected else None) + + if not email: + raise HTTPException( + status_code=422, + detail={ + "error_code":"CANDIDATE_EMAIL_REQUIRED", + "filename":parsed.get("filename"), + "num_pages":parsed.get("num_pages"), + "emails_found":emails_found, + "text":text, + }, + ) + + filename=self.filename or "resume.pdf" + try: + paths=await decode_attachment([{ + "name":filename, + "contentBytes":base64.b64encode(self.file).decode("ascii"), + }]) + except AttachmentDecodeError as e: + raise HTTPException(status_code=400,detail=str(e)) + if not paths: + raise HTTPException(status_code=400,detail="attachment could not be saved") + + now=datetime.now(timezone.utc).isoformat() + email_data={ + "id":f"manual-cv:{uuid.uuid4()}", + "subject":f"Manual CV upload — {filename}", + "body":{"content":"","contentType":"text"}, + "hasAttachments":True, + "attachments":[{"name":filename}], + "from":{"emailAddress":{"address":email,"name":(candidate_name or "").strip()}}, + "toRecipients":[{"emailAddress":{"address":MANUAL_UPLOAD_TO_ADDRESS}}], + "ccRecipients":[], + "bccRecipients":[], + "replyTo":[], + "isRead":False, + "sentDateTime":now, + "receivedDateTime":now, + } + row,new_user_email=await Inbox_Messages.insert_email( + self.session,email_data,file_path=paths, + ) + + created_at=datetime.now(timezone.utc).isoformat() + task=await match_uploaded_cv.kicker().with_labels( + created_at=created_at, + correlation_id=str(row.id), + queue=CV_QUEUE_NAME, + ).kiq(str(row.id),force=False) + + account_setup=None + if new_user_email: + try: + account_setup=await Email(session=self.session).send_account_setup( + [new_user_email] + ) + except Exception as e: + logger.warning("account setup mail failed for %s: %s",new_user_email,e) + account_setup=[{"email":new_user_email,"sent":False}] + + return { + "queued":True, + "inbox_message_id":str(row.id), + "task_id":task.task_id, + "filename":parsed.get("filename"), + "num_pages":parsed.get("num_pages"), + "candidate_email":email, + "email_source":email_source, + "account_setup":account_setup, + "text":text, + } async def match_inbox_cv(self,inbox_message_id): from inbox.plugins import resolve_attachment_path @@ -310,6 +467,36 @@ class CandidateView: scores.setdefault(row.inbox_message_id,[]).append(row) return scores + async def create_candidate(self,candidate_email=None,candidate_name=None,candidate_phone=None,job_post_id=None,current_company=None,platform=None,experience=None,status=None,referral_by=None,file_name=None,file_path=None,full_text=None,current_user=None): + try: + email=(candidate_email or "").strip().lower() + if not email: + raise HTTPException(status_code=422,detail="candidate_email is required") + if not current_user: + raise HTTPException(status_code=400,detail="created_by is required") + data={ + "candidate_email":email, + "candidate_name":(candidate_name or "").strip(), + "candidate_phone":(candidate_phone or "").strip(), + "job_post_id":job_post_id, + "current_company":(current_company or "").strip(), + "platform":(platform or "").strip(), + "experience":(experience or "").strip(), + "status":(status or "").strip(), + "referral_by":(referral_by or "").strip(), + "file_name":(file_name or "").strip(), + "file_path":(file_path or "").strip(), + "full_text":full_text or "", + "created_by":current_user, + + } + row=await Manual_UPLOAD_CANDIDATE.create_manual_upload_candidate(session=self.session,fields=data) + return serialize_manual_upload_candidate(row) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + async def get_candidate(self,user_id=None,limit=10,offset=0,search=None): try: detail=bool(user_id) diff --git a/backend/main.py b/backend/main.py index 87ec33d..af4b7b1 100644 --- a/backend/main.py +++ b/backend/main.py @@ -19,11 +19,13 @@ logger=logging.getLogger("main") async def lifespan(app): async with db_lifespan(app): broker_ready=False + cv_broker_ready=False llm_ready=False agent_ready=False close_llm=None close_agent=None broker=None + cv_broker=None try: from taskiq_management.broker_setup import broker as _broker broker=_broker @@ -31,6 +33,13 @@ async def lifespan(app): broker_ready=True except Exception as exc: logger.warning("taskiq broker startup skipped: %s",exc) + try: + from taskiq_management.cv_broker_setup import cv_broker as _cv_broker + cv_broker=_cv_broker + await cv_broker.startup() + cv_broker_ready=True + except Exception as exc: + logger.warning("taskiq cv broker startup skipped: %s",exc) try: from llm_setup import init_llm,close_llm as _close_llm from agent.agent_setup import init_agent,close_agent as _close_agent @@ -49,6 +58,8 @@ async def lifespan(app): await close_agent() if llm_ready and close_llm is not None: await close_llm() + if cv_broker_ready and cv_broker is not None: + await cv_broker.shutdown() if broker_ready and broker is not None: await broker.shutdown() diff --git a/backend/users/models.py b/backend/users/models.py index d84b1b6..d79f1f3 100644 --- a/backend/users/models.py +++ b/backend/users/models.py @@ -55,6 +55,15 @@ class Users(SQLModel, table=True): is_active: bool = Field(default=False) is_deleted: bool = Field(default=False) + @classmethod + async def get_user_id(cls, session: AsyncSession, user_id: str): + uid = cls._as_uuid(user_id) + if uid is None: + return None + statement = select(cls).where(cls.id == uid) + result = await session.execute(statement) + return result.scalars().first() + @classmethod def _search_filter(cls, search: str): pattern = f"%{search}%" diff --git a/docker-compose.yml b/docker-compose.yml index 7f99556..e29ec44 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -72,5 +72,66 @@ services: condition: service_healthy restart: unless-stopped + taskiq-cv-worker: + build: + context: ./backend + container_name: hrms-taskiq-cv-worker + working_dir: /app + command: + [ + "taskiq", + "worker", + "taskiq_management.cv_broker_setup:cv_broker", + "inbox.cv_tasks", + "--workers", + "1", + ] + env_file: + - ./backend/.env + environment: + PYTHONPATH: /app + REDIS_URL: redis://redis:6379/0 + TASKIQ_CV_QUEUE_NAME: cv_upload + TASKIQ_WORKER_NAME: cv-worker-01 + DB_HOST: host.docker.internal + EMAIL_URL: http://host.docker.internal:5000 + BACKEND_URL: http://host.docker.internal:8000 + extra_hosts: + - "host.docker.internal:host-gateway" + volumes: + - ./backend/inbox/decoded_attachments:/app/inbox/decoded_attachments + depends_on: + redis: + condition: service_healthy + restart: unless-stopped + + taskiq-cv-scheduler: + build: + context: ./backend + container_name: hrms-taskiq-cv-scheduler + working_dir: /app + command: + [ + "taskiq", + "scheduler", + "taskiq_management.cv_broker_setup:cv_scheduler", + "inbox.cv_tasks", + ] + env_file: + - ./backend/.env + environment: + PYTHONPATH: /app + REDIS_URL: redis://redis:6379/0 + TASKIQ_CV_QUEUE_NAME: cv_upload + DB_HOST: host.docker.internal + EMAIL_URL: http://host.docker.internal:5000 + BACKEND_URL: http://host.docker.internal:8000 + extra_hosts: + - "host.docker.internal:host-gateway" + depends_on: + redis: + condition: service_healthy + restart: unless-stopped + volumes: redis-data: diff --git a/frontend/src/api/candidates.js b/frontend/src/api/candidates.js index 60fcae9..26183ab 100644 --- a/frontend/src/api/candidates.js +++ b/frontend/src/api/candidates.js @@ -135,6 +135,54 @@ export function update(userId, payload) { return request('/candidate/update', { method: 'PATCH', params: { user_id: userId }, body: payload }) } +/** + * Manual candidate creation — POST /candidate/create/candidate (backend/job/app.py:98). + * + * Multipart, and the CV is REQUIRED, not an extra: the route declares + * `file: UploadFile = File(...)`, so a request without one is a 422, and + * injest_manual_upload then rejects the upload with 400 when pypdf extracts no + * text. The extracted text IS the record — it is what later scoring reads — so + * a scanned or image-only PDF fails here rather than storing an empty row. + * PDF only: read_file goes straight to PdfReader, so DOC/DOCX 400s. + * + * Every other field is an optional Form value, with one exception — + * candidate_email, which create_candidate rejects when blank (422). It is also + * the identity key: an unknown address creates the `users` row (role CANDIDATE, + * default password from DEFAULT_CANDIDATE_PASSWORD), a known one reuses it. + * That user write is why the route sits behind candidates.create. + * + * job_post_id must be a real job_posts UUID. Anything unparseable is coerced to + * NULL rather than raising (Manual_UPLOAD_CANDIDATE._as_uuid), so a seed id like + * "JOB-101" would silently drop the link — the picker must offer live posts from + * /job/fetch, never the seed catalogue. + * + * `platform`, `status` and `referral_by` are free-text columns, not enums; the + * UI's Source and Stage vocabularies go in verbatim, and a referrer is whatever + * the recruiter typed — often someone with no account here. + */ +export function createManual({ + file, name, email, phone, jobPostId, company, source, experience, stage, referralBy, +}) { + const form = new FormData() + form.append('file', file) + // Blank optional fields are omitted rather than sent as "": Form(None) then + // leaves them None, and the model's own defaults apply. + const put = (key, value) => { + const text = value == null ? '' : String(value).trim() + if (text) form.append(key, text) + } + put('candidate_email', email) + put('candidate_name', name) + put('candidate_phone', phone) + put('job_post_id', jobPostId) + put('current_company', company) + put('platform', source) + put('experience', experience) + put('status', stage) + put('referral_by', referralBy) + return request('/candidate/create/candidate', { method: 'POST', body: form }) +} + /* ------------------------------------------------------------------ Child records of a profile. diff --git a/frontend/src/lib/apiClient.js b/frontend/src/lib/apiClient.js index 02d1505..63dfc8a 100644 --- a/frontend/src/lib/apiClient.js +++ b/frontend/src/lib/apiClient.js @@ -49,19 +49,23 @@ export async function request( } } + // Multipart uploads pass a FormData body. Content-Type is deliberately NOT + // set for those: the browser has to write it itself so the generated boundary + // token ends up in the header, and a hand-set value strips it and the server + // reads an unparseable body. FormData is replayable, so the 401 retry below + // can re-send the same object. + const multipart = typeof FormData !== 'undefined' && body instanceof FormData + const send = async () => { const headers = { Accept: 'application/json' } - // FormData bodies (file uploads) set their own multipart boundary — adding a - // Content-Type here would break the request, and they must not be stringified. - const isForm = typeof FormData !== 'undefined' && body instanceof FormData - if (body != null && !isForm) headers['Content-Type'] = 'application/json' + if (body != null && !multipart) headers['Content-Type'] = 'application/json' const bearer = token ?? (auth ? getAccessToken() : null) if (bearer) headers.Authorization = `Bearer ${bearer}` return fetch(buildUrl(path, params), { method, headers, signal, - body: body == null ? undefined : isForm ? body : JSON.stringify(body), + body: body == null ? undefined : multipart ? body : JSON.stringify(body), }) } diff --git a/frontend/src/screens/Candidates.jsx b/frontend/src/screens/Candidates.jsx index 76883e3..2e18ea3 100644 --- a/frontend/src/screens/Candidates.jsx +++ b/frontend/src/screens/Candidates.jsx @@ -8,20 +8,22 @@ candidates happens through CV Import (real scoring), not a manual form. ============================================================ */ -import { useCallback, useEffect, useMemo, useState } from 'react' +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import { useLocation, useNavigate } from 'react-router-dom' -import { useQuery, useQueryClient } from '@tanstack/react-query' +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' import Modal from '../ui/Modal' import { Pagination, useDataTable } from '../ui/DataTable' -import { Avatar, Badge, EmptyState, Icon, ScoreChip } from '../ui/primitives' +import { Avatar, Badge, EmptyState, FieldError, Icon, ScoreChip } from '../ui/primitives' import { useToast } from '../ui/Toast' import CandidateProfile from './ScoredCandidateProfile' import { qk } from '../lib/queryKeys' import { friendlyAuthError } from '../lib/errors' import * as candidatesApi from '../api/candidates' +import * as jobPostsApi from '../api/jobPosts' +import { useFormState } from '../components/AuthLayout' import { persist } from '../data/seedQueries' -import { atsRecommendationClass, avatarColor, initials as initialsOf } from '../data/seed' +import { atsRecommendationClass, avatarColor, initials as initialsOf, sources, stages } from '../data/seed' const ATS_BANDS = ['85+', '70-84', '<70'] const SOURCE_LABEL = { upload: 'Upload', inbox: 'Inbox' } @@ -44,6 +46,34 @@ function recommendationOf(c) { return c.aiScore >= 82 ? 'Strong Match' : c.aiScore >= 65 ? 'Potential Match' : 'Weak Match' } +/* Client-side guard only — the route has no size cap of its own, so this just + stops an obviously wrong file from being read into memory and posted. */ +const MAX_CV_MB = 10 + +/** Live job posts offered by the Add Candidate picker. */ +const JOB_POST_LIMIT = 100 + +/* Referral By must name a colleague, so it is constrained to a company address: + a referral from outside the company is not a referral, and a bare name + ("Sarah") cannot be resolved to a person later. + + This is the ONLY place the rule lives. `referral_by` is a free-text column and + the route does not check it, so anything posted outside this form is stored + as-is — the constraint is a data-entry guard, not an invariant. */ +const REFERRAL_DOMAIN = 'utopiabrands.com' +const REFERRAL_RE = new RegExp( + `^[a-z0-9][a-z0-9._%+-]*@${REFERRAL_DOMAIN.replace(/\./g, '\\.')}$`, + 'i', +) + +/** + * The one reading of the Referral By box: surrounding whitespace is stripped, so + * a field holding only spaces is absent rather than invalid, and the address is + * lower-cased so " Ada@UtopiaBrands.com " and "ada@utopiabrands.com" are stored + * as one referrer rather than two. + */ +const referralValue = (raw) => (raw || '').trim().toLowerCase() + export default function Candidates() { const { toast } = useToast() const qc = useQueryClient() @@ -70,6 +100,7 @@ export default function Candidates() { const [sortMode, setSortMode] = useState('relevance') const [profileFor, setProfileFor] = useState(null) const [atsFor, setAtsFor] = useState(null) + const [adding, setAdding] = useState(false) const jobTitleOf = useCallback( (c) => jobsById[c.jobId]?.title ?? '—', @@ -101,9 +132,12 @@ export default function Candidates() { // Deep links from Talent Pool, global search, dashboard… useEffect(() => { const st = location.state - if (!st?.openCandidate) return - const c = candidates.find((x) => x.id === st.openCandidate) - if (c) openProfile(c) + if (!st) return + if (st.openAdd) setAdding(true) + if (st.openCandidate) { + const c = candidates.find((x) => x.id === st.openCandidate) + if (c) openProfile(c) + } }, [location.state, candidates, openProfile]) const skillOptions = useMemo(() => { @@ -190,9 +224,12 @@ export default function Candidates() { - + @@ -371,6 +408,17 @@ export default function Candidates() { onAtsMatch={(c) => { setProfileFor(null); openAts(c) }} /> )} + + {adding && ( + setAdding(false)} + onSave={() => { + setAdding(false) + toast('Candidate added to pipeline', 'success') + }} + onInvalid={() => toast('Please fix the highlighted fields', 'error')} + /> + )} ) } @@ -463,3 +511,242 @@ export function AtsMatch({ candidate: c, jobTitle, onClose, onProfile }) { ) } + +/** + * Add Candidate — the only writer on this screen that reaches the server. + * + * POST /candidate/create/candidate persists the row and, for an unseen email, + * the `users` record behind it. The CV is not optional there: the route requires + * the file and refuses it when no text can be extracted, so the dropzone below + * the fields is part of the contract rather than a convenience. + * + * Applied Job lists LIVE job posts (/job/fetch), not the seed catalogue, because + * job_post_id is a job_posts FK and a seed id would be coerced to NULL without + * an error — the link would look saved and simply not exist. + * + * Manual rows do not pass through `inbox`, so /candidate/fetch may not surface + * them immediately; the save still invalidates the candidates query so the + * live-backed screens refetch and pick the row up once an application links it. + */ +function AddCandidate({ onClose, onSave, onInvalid }) { + const { toast } = useToast() + const qc = useQueryClient() + const fileInput = useRef(null) + const [cv, setCv] = useState(null) + const [dragging, setDragging] = useState(false) + + const postsQuery = useQuery({ + queryKey: qk.jobPosts.list({ top: JOB_POST_LIMIT }), + queryFn: async () => { + const res = await jobPostsApi.list({ top: JOB_POST_LIMIT }) + return Array.isArray(res?.data) ? res.data : [] + }, + }) + const posts = postsQuery.data ?? [] + + const form = useFormState({ + name: '', email: '', phone: '', job: '', + experience: '3', company: '', source: sources[0], stage: stages[0], + referral: '', + }) + + // Defaulting by derivation rather than in an effect: the picker resolves after + // first paint, and useFormState's setters are new every render, so seeding the + // field from an effect would either loop or need a ref to guard it. + const jobPostId = form.values.job || (posts[0] ? String(posts[0].id) : '') + + const create = useMutation({ + mutationFn: (vars) => candidatesApi.createManual(vars), + onError: (err) => toast(friendlyAuthError(err, 'Could not add the candidate.'), 'error'), + onSuccess: () => { + // The new user_id lands in /candidate/fetch's join the moment an + // application exists for them, so let the live-backed screens refetch. + qc.invalidateQueries({ queryKey: qk.candidates.all() }) + onSave() + }, + }) + + function pickFile(next) { + if (!next) return + setCv(next) + form.setErrors((prev) => { + if (!prev.cv) return prev + const rest = { ...prev } + delete rest.cv + return rest + }) + } + + function submit() { + if (create.isPending) return + const v = form.values + const errors = {} + if (!v.name.trim()) errors.name = 'Required' + if (!/^\S+@\S+\.\S+$/.test(v.email)) errors.email = 'Valid email required' + // Only enforceable when the picker actually has something to pick — the + // column is nullable server-side. + if (posts.length && !jobPostId) errors.job = 'Required' + if (!cv) errors.cv = 'Attach the candidate’s CV' + else if (!/\.pdf$/i.test(cv.name)) errors.cv = 'Only PDF resumes can be parsed' + else if (cv.size > MAX_CV_MB * 1024 * 1024) errors.cv = `Keep the file under ${MAX_CV_MB} MB` + // Optional: trimmed first, so a field holding only spaces is genuinely empty + // and passes rather than failing the pattern. Anything left must be a + // company address — the pattern rejects interior spaces on its own. + const referral = referralValue(v.referral) + if (referral && !REFERRAL_RE.test(referral)) { + errors.referral = `Must be a @${REFERRAL_DOMAIN} address` + } + form.setErrors(errors) + if (Object.keys(errors).length) { + onInvalid() + return + } + create.mutate({ + file: cv, + name: v.name, + email: v.email, + phone: v.phone, + jobPostId, + company: v.company, + source: v.source, + experience: v.experience, + stage: v.stage, + referralBy: referral, + }) + } + + const field = (n) => ({ value: form.values[n], onChange: (e) => form.setField(n, e.target.value) }) + + return ( + + + + + } + > +
{ e.preventDefault(); submit() }}> +
+
+ + + {form.errors.name} +
+
+ + + {form.errors.email} +
+
+
+ + + {form.errors.job} +
+
+
+
+ + +
+
+ + +
+ {/* Optional, and deliberately not gated on Source === 'Referral': + a referrer is worth recording whenever there is one, and referrals + routinely arrive tagged as LinkedIn or Company Site. */} +
+ + form.setField('referral', referralValue(e.target.value))} + className={form.errors.referral ? 'err' : ''} + placeholder={`name@${REFERRAL_DOMAIN}`} + /> + {form.errors.referral} +
+
+ + {/* .req is scoped to `.form-field label .req`, so tint it here. */} +
+ CV / Resume * +
+ { pickFile(e.target.files?.[0]); e.target.value = '' }} + /> +
{ if (!create.isPending) fileInput.current?.click() }} + onKeyDown={(e) => { + if (e.key === 'Enter' || e.key === ' ') { e.preventDefault(); fileInput.current?.click() } + }} + onDragOver={(e) => { e.preventDefault(); setDragging(true) }} + onDragLeave={() => setDragging(false)} + onDrop={(e) => { + e.preventDefault() + setDragging(false) + pickFile(e.dataTransfer.files?.[0]) + }} + > +
+ +
+

Drop the CV here or click to browse

+

+ PDF only · text-based resumes · up to {MAX_CV_MB} MB +

+
+ {cv && ( +
+ +
+
{cv.name}
+
{Math.max(1, Math.round(cv.size / 1024))} KB
+
+ +
+ )} + {form.errors.cv} +
+ + ) +}