utopia-ai
/
HR-ATS-Portal
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

LINK_X_USER_INBOX #10

Merged
ahmed.mujtaba merged 4 commits from LINK_X_USER_INBOX into main 2026-08-12 08:56:44 +00:00
Conversation 0 Commits 4 Files Changed 14
14 changed files with 895 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/.env.example
View File

@ -46,9 +46,11 @@ OPENAI_PROJECT=
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

30
backend/README.md
View File

@ -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: <http://localhost:8000/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

17
backend/inbox/cv_tasks.py Normal file
View File

@ -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)

62
backend/job/app.py
View File

@ -12,7 +12,7 @@ from users.permissions import PermissionTag, require_permission
from job.job_post.views import JobPost,JobPostCreate
import logging
from job.job_post.plugins import PlatformAlias
from fastapi import UploadFile, File
from fastapi import UploadFile, File, Form
from dotenv import load_dotenv
from datetime import datetime, time, timezone
from pydantic import BaseModel
@ -95,10 +95,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),
):
@ -106,8 +161,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

90
backend/job/candidate/models.py
View File

@ -9,6 +9,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:
@ -20,6 +21,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 Interviews(SQLModel, table=True):

108
backend/job/candidate/plugins.py
View File

@ -71,3 +71,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

22
backend/job/candidate/serializers.py
View File

@ -7,6 +7,28 @@ from job.activity.serializers import serialize_activity
from job.feedback.serializers import serialize_feedback
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]],
*,

193
backend/job/candidate/views.py
View File

@ -1,6 +1,8 @@
from sqlalchemy.ext.asyncio import AsyncSession
import os,logging,io
import base64,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
@ -8,11 +10,18 @@ from sqlalchemy.orm import selectinload
from sqlmodel import true
from job.job_post.models import JobPosts
from job.job_post.serializers import serialize_job_post
from job.candidate.serializers import serialize_candidate_profile
from job.candidate.models import Notes
from job.candidate.serializers import serialize_candidate_profile,serialize_manual_upload_candidate
from job.candidate.models import Notes,Manual_UPLOAD_CANDIDATE
from job.notes.serializers import serialize_note
from inbox.models import Inbox_Messages,Inbox
from job.candidate.plugins import normalize_spaced_text
from job.candidate.plugins import extract_candidate_email,normalize_spaced_text
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):
@ -35,6 +44,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
@ -78,6 +233,36 @@ class CandidateView:
def __init__(self,session:AsyncSession):
self.session=session
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)

11
backend/main.py
View File

@ -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()

9
backend/users/models.py
View File

@ -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}%"

61
docker-compose.yml
View File

@ -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:

48
frontend/src/api/candidates.js
View File

@ -48,6 +48,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.

11
frontend/src/lib/apiClient.js
View File

@ -49,16 +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' }
if (body != null) 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 ? JSON.stringify(body) : undefined,
body: body == null ? undefined : multipart ? body : JSON.stringify(body),
})
}

270
frontend/src/screens/Candidates.jsx
View File

@ -8,9 +8,9 @@
selection column needs to render against a Set this component owns.
============================================================ */
import { useCallback, useEffect, useMemo, useState } from 'react'
import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
import { useLocation } 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'
@ -19,6 +19,9 @@ import { useToast } from '../ui/Toast'
import { useFormState } from '../components/AuthLayout'
import CandidateProfile from './CandidateProfile'
import { qk } from '../lib/queryKeys'
import { friendlyAuthError } from '../lib/errors'
import * as candidatesApi from '../api/candidates'
import * as jobPostsApi from '../api/jobPosts'
import { persist, seedQuery, useSeedMutation } from '../data/seedQueries'
import {
atsRecommendationClass, avatarColor, departments, educationLevels, getJob,
@ -32,6 +35,34 @@ const INTERVIEW_STATES = ['Not Scheduled', 'Scheduled', 'Completed']
const NOTICE = ['Immediate', '2 weeks', '1 month', '2 months', '3 months']
const AVAILABILITY = ['Immediate', '2 weeks', '1 month', 'Passive']
/* 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()
const EMPTY_FILTERS = {
job: '', skill: '', dept: '', location: '', exp: '', edu: '', recruiter: '',
manager: '', source: '', ats: '', stage: '', interview: '', notice: '', availability: '',
@ -575,41 +606,141 @@ function BulkAssign({ count, recruiters, onClose, onSave }) {
)
}
/**
* 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.
*
* The row handed to onSave is still seed-shaped. Nothing on this screen reads
* /candidate/fetch — manual rows do not pass through `inbox`, so they surface
* neither here nor in Talent Pool — and dropping the candidate the recruiter
* just created out of the table would read as a failed save. The fabricated
* scoring fields are the pre-existing seed shape, unchanged; only the identity
* fields now carry what was actually posted.
*/
function AddCandidate({ jobs, count, onClose, onSave, onInvalid }) {
const open = jobs.filter((j) => j.status === 'Open')
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: open[0]?.title ?? '',
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: (res) => {
// 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(buildRow(res?.data))
},
})
function buildRow(saved) {
const v = form.values
const post = posts.find((p) => String(p.id) === jobPostId)
const title = post?.title || v.job || jobs[0]?.title || 'Unassigned'
// Department, location, recruiter and skills are presentation-only columns
// the endpoint does not return — borrow them from the seed job of the same
// title so the row renders like every other one.
const job = jobs.find((j) => j.title === title) || jobs[0] || {}
const skills = job.skills ?? []
const score = int(55, 95)
return {
id: `CAN-${5001 + count}`,
userId: saved?.user_id ?? null,
manualUploadId: saved?.id ?? null,
name: v.name, initials: initialsOf(v.name), color: avatarColor(v.name),
email: v.email, phone: v.phone || '+1 (555) 000-0000',
jobId: job.id, jobTitle: title, department: job.department,
experience: Number(v.experience) || 1, currentCompany: v.company || '—',
currentTitle: title, location: job.location,
stage: v.stage, status: v.stage, aiScore: score, source: v.source,
referredBy: referralValue(v.referral) || null,
recruiter: job.recruiter, recruiterId: job.recruiterId,
applied: new Date(TODAY), education: "Bachelor's Degree",
skills: skills.slice(0, 4), rating: '4.0', salary: 120000,
matchedSkills: skills.slice(0, 3), missingSkills: skills.slice(3),
recommendation: score >= 82 ? 'Strong Match' : score >= 65 ? 'Potential Match' : 'Weak Match',
subScores: { skills: score, experience: 80, education: 80, keywords: score, location: 100, salary: 90 },
noticePeriod: '1 month', availability: '2 weeks', certifications: [],
favorite: false, interviewStatus: 'Not Scheduled',
}
}
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
}
const job = jobs.find((j) => j.title === v.job) || jobs[0]
const score = int(55, 95)
onSave({
id: `CAN-${5001 + count}`,
name: v.name, initials: initialsOf(v.name), color: avatarColor(v.name),
email: v.email, phone: v.phone || '+1 (555) 000-0000',
jobId: job.id, jobTitle: job.title, department: job.department,
experience: Number(v.experience) || 1, currentCompany: v.company || '—',
currentTitle: job.title, location: job.location,
stage: v.stage, status: v.stage, aiScore: score, source: v.source,
recruiter: job.recruiter, recruiterId: job.recruiterId,
applied: new Date(TODAY), education: "Bachelor's Degree",
skills: job.skills.slice(0, 4), rating: '4.0', salary: 120000,
matchedSkills: job.skills.slice(0, 3), missingSkills: job.skills.slice(3),
recommendation: score >= 82 ? 'Strong Match' : score >= 65 ? 'Potential Match' : 'Weak Match',
subScores: { skills: score, experience: 80, education: 80, keywords: score, location: 100, salary: 90 },
noticePeriod: '1 month', availability: '2 weeks', certifications: [],
favorite: false, interviewStatus: 'Not Scheduled',
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,
})
}
@ -622,8 +753,10 @@ function AddCandidate({ jobs, count, onClose, onSave, onInvalid }) {
onClose={onClose}
footer={
<>
<button className="btn btn-secondary" onClick={onClose}>Cancel</button>
<button className="btn btn-primary" onClick={submit}><Icon name="check" /> Add Candidate</button>
<button className="btn btn-secondary" onClick={onClose} disabled={create.isPending}>Cancel</button>
<button className="btn btn-primary" onClick={submit} disabled={create.isPending}>
<Icon name="check" /> {create.isPending ? 'Adding…' : 'Add Candidate'}
</button>
</>
}
>
@ -642,7 +775,21 @@ function AddCandidate({ jobs, count, onClose, onSave, onInvalid }) {
<div className="form-field"><label>Phone</label><input {...field('phone')} placeholder="+1 (555) 000-0000" /></div>
<div className="form-field">
<label>Applied Job <span className="req">*</span></label>
<select {...field('job')}>{open.map((j) => <option key={j.id}>{j.title}</option>)}</select>
<select
value={jobPostId}
onChange={(e) => form.setField('job', e.target.value)}
className={form.errors.job ? 'err' : ''}
disabled={postsQuery.isPending || !posts.length}
>
{postsQuery.isPending && <option value="">Loading job posts…</option>}
{!postsQuery.isPending && !posts.length && (
<option value="">{postsQuery.isError ? 'Could not load job posts' : 'No active job posts'}</option>
)}
{posts.map((p) => (
<option key={p.id} value={String(p.id)}>{p.title}</option>
))}
</select>
<FieldError>{form.errors.job}</FieldError>
</div>
<div className="form-field"><label>Experience (years)</label><input type="number" {...field('experience')} /></div>
<div className="form-field"><label>Current Company</label><input {...field('company')} placeholder="Acme Inc." /></div>
@ -654,7 +801,80 @@ function AddCandidate({ jobs, count, onClose, onSave, onInvalid }) {
<label>Stage</label>
<select {...field('stage')}>{stages.map((s) => <option key={s}>{s}</option>)}</select>
</div>
{/* 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. */}
<div className="form-field">
<label>Referral By</label>
<input
type="email"
{...field('referral')}
// Normalising on blur means the value the recruiter sees is the
// value that gets posted — otherwise a pasted address with a
// trailing space would submit clean while still looking untidy.
onBlur={(e) => form.setField('referral', referralValue(e.target.value))}
className={form.errors.referral ? 'err' : ''}
placeholder={`name@${REFERRAL_DOMAIN}`}
/>
<FieldError>{form.errors.referral}</FieldError>
</div>
</div>
{/* .req is scoped to `.form-field label .req`, so tint it here. */}
<div className="form-section-title">
CV / Resume <span className="req" style={{ color: 'var(--danger)' }}>*</span>
</div>
<input
ref={fileInput}
type="file"
accept="application/pdf,.pdf"
hidden
onChange={(e) => { pickFile(e.target.files?.[0]); e.target.value = '' }}
/>
<div
className={`dropzone${dragging ? ' drag' : ''}`}
style={{ padding: '22px 18px', cursor: create.isPending ? 'default' : 'pointer' }}
role="button"
tabIndex={0}
onClick={() => { 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])
}}
>
<div className="dz-icn" style={{ width: 44, height: 44, borderRadius: 13, marginBottom: 10 }}>
<Icon name="upload" />
</div>
<h3 style={{ fontSize: 15 }}>Drop the CV here or click to browse</h3>
<p className="text-muted text-sm">
PDF only · text-based resumes · up to {MAX_CV_MB} MB
</p>
</div>
{cv && (
<div className="upload-row">
<span className="attach-icn" style={{ width: 34, height: 34 }}><Icon name="file" /></span>
<div style={{ flex: 1, minWidth: 0 }}>
<div className="fw-600 text-sm">{cv.name}</div>
<div className="cell-sub">{Math.max(1, Math.round(cv.size / 1024))} KB</div>
</div>
<button
type="button"
className="act-btn"
aria-label="Remove file"
disabled={create.isPending}
onClick={() => setCv(null)}
>
<Icon name="trash" />
</button>
</div>
)}
<FieldError>{form.errors.cv}</FieldError>
</form>
</Modal>
)