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

mapped Candidate CV with job_Post_id

Backend_CODEBASE
ahmed.mujtaba 2026-08-06 18:53:18 +05:00
parent cc92dc744f
commit dcf0bf5d50
13 changed files with 266 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
backend/agent/decorators.py
View File

@ -36,7 +36,7 @@ def normalize_job_posts(job_posts) -> list[dict]:
return normalized
def parse_match_response(data, allowed_ids) -> tuple[list[str], str]:
def parse_match_response(data, allowed_ids) -> tuple[list[str], str, str]:
"""Filter model JSON ids to the allowed job-post set."""
if not isinstance(data, dict):
raise RuntimeError(f"model did not return a JSON object: {data!r}")
@ -59,7 +59,13 @@ def parse_match_response(data, allowed_ids) -> tuple[list[str], str]:
seen.add(job_id)
suggested.append(job_id)
summary = data.get("summary")
if not isinstance(summary, str):
summary = ""
reasoning = data.get("reasoning")
if isinstance(reasoning, list):
reasoning = "\n".join(str(item) for item in reasoning)
if not isinstance(reasoning, str):
reasoning = ""
return suggested, reasoning.strip()
return suggested, summary.strip(), reasoning.strip()

1
backend/agent/models.py
View File

@ -15,6 +15,7 @@ class AgentState(TypedDict, total=False):
resume_text: str
job_posts: list[dict]
suggested_job_post_ids: list[str]
summary: str
reasoning: str
error: str
status: Literal["pending", "ready", "matched", "skipped", "failed"]

23
backend/agent/prompt.py
View File

@ -11,20 +11,23 @@ import json
def prompt():
return """You are an HR-ATS recruiting assistant.
Given a candidate email subject, resume text, and a list of active job posts,
identify which job posts the candidate is most likely applying for.
You are given a candidate email subject, CV/resume text extracted from an
attachment, and a list of active job posts (id, title, description, requirements).
Return JSON only with this shape:
{
"suggested_job_post_ids": ["uuid-string", ...],
"reasoning": "brief explanation"
}
Identify which job posts the candidate is most likely applying for.
Rules:
- suggested_job_post_ids must only contain ids from the provided job_posts list.
- Return an empty list when nothing matches confidently.
- Only suggest job_post_id values that appear in the provided job_posts list.
- A candidate may match zero, one, or multiple posts.
- Prefer title/subject alignment, then skills and experience in the resume.
- Base matches on skills, role title, experience, and the subject line — not guesses.
- If confidence is low, return an empty list rather than forcing a match.
Respond with JSON only:
{
"suggested_job_post_ids": ["uuid", "..."],
"summary": "one short sentence for the recruiter",
"reasoning": "brief bullet-style explanation per suggested match"
}
"""

1
backend/agent/serializers.py
View File

@ -10,6 +10,7 @@ def serialize_agent_result(state: dict) -> dict:
"""Plain dict for services/serializers — no ORM objects."""
return {
"suggested_job_post_ids": state.get("suggested_job_post_ids") or [],
"summary": state.get("summary") or "",
"reasoning": state.get("reasoning") or "",
"status": state.get("status") or "failed",
"error": state.get("error") or "",

7
backend/agent/views.py
View File

@ -30,6 +30,7 @@ async def prepare_context(state: AgentState) -> dict:
"status": "skipped",
"error": "resume_text is empty",
"suggested_job_post_ids": [],
"summary": "",
"reasoning": "",
}
if not job_posts:
@ -37,6 +38,7 @@ async def prepare_context(state: AgentState) -> dict:
"status": "skipped",
"error": "no active job posts to match against",
"suggested_job_post_ids": [],
"summary": "",
"reasoning": "",
}
@ -60,10 +62,11 @@ async def match_jobs(state: AgentState) -> dict:
try:
data = await llm_call(prompt(), user_prompt(state), json_mode=True)
allowed_ids = {item["id"] for item in state.get("job_posts") or []}
suggested, reasoning = parse_match_response(data, allowed_ids)
suggested, summary, reasoning = parse_match_response(data, allowed_ids)
return {
"status": "matched",
"suggested_job_post_ids": suggested,
"summary": summary,
"reasoning": reasoning,
}
except Exception as exc:
@ -72,6 +75,7 @@ async def match_jobs(state: AgentState) -> dict:
"status": "failed",
"error": str(exc),
"suggested_job_post_ids": [],
"summary": "",
"reasoning": "",
}
@ -80,6 +84,7 @@ async def finalize(state: AgentState) -> dict:
"""Normalize terminal state for callers."""
return {
"suggested_job_post_ids": state.get("suggested_job_post_ids") or [],
"summary": state.get("summary") or "",
"reasoning": state.get("reasoning") or "",
"status": state.get("status") or "failed",
"error": state.get("error") or "",

31
backend/inbox/app.py
View File

@ -1,16 +1,23 @@
from fastapi import APIRouter,Depends, Query
from fastapi import APIRouter,BackgroundTasks,Depends, Query
from fastapi.responses import JSONResponse
from fastapi import HTTPException
from db_setup import get_session
from sqlalchemy.ext.asyncio import AsyncSession
from inbox.views import Email
from users.permissions import PermissionTag, require_permission
from dotenv import load_dotenv
load_dotenv()
router = APIRouter()
@router.get("/email/fetch")
async def fetch_email(top:int=Query(100),skip:int=Query(0,ge=0),token=Query(...),session: AsyncSession = Depends(get_session)):
async def fetch_email(
background_tasks: BackgroundTasks,
top:int=Query(100),
skip:int=Query(0,ge=0),
token=Query(...),
session: AsyncSession = Depends(get_session),
):
try:
if not token:
raise HTTPException(status_code=401,detail="Unauthorized")
@ -23,6 +30,8 @@ async def fetch_email(top:int=Query(100),skip:int=Query(0,ge=0),token=Query(...)
service_per_email=await service.get_email_by_id(message_id)
items_lst.append({"message_id":message_id,"email_contents":service_per_email})
if service.pending_match_ids:
background_tasks.add_task(Email.run_inbox_matching, list(service.pending_match_ids))
return JSONResponse(content={"data":items_lst,"status_code":200})
@ -53,3 +62,21 @@ async def fetch_inbox(
raise
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
@router.post("/inbox/{record_id}/match")
async def rematch_inbox(
record_id: str,
background_tasks: BackgroundTasks,
current_user: dict = Depends(require_permission(PermissionTag.INBOX_EDIT)),
session: AsyncSession = Depends(get_session),
):
try:
service=Email(session=session)
queued_id=await service.queue_rematch(record_id)
background_tasks.add_task(Email.run_inbox_matching, [queued_id], force=True)
return JSONResponse(content={"data":{"queued":True},"status_code":200})
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))

45
backend/inbox/models.py
View File

@ -1,8 +1,8 @@
import uuid
from datetime import datetime
from datetime import datetime, timezone
from typing import Any, Optional
from sqlalchemy import Column, func, or_
from sqlalchemy import Column, DateTime, func, or_
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import Field, Relationship, SQLModel, select
@ -62,6 +62,13 @@ class Inbox_Messages(SQLModel, table=True):
message_reply: str | None = Field(default=None)
file_name: str | None = Field(default=None)
file_path: str | None = Field(default=None)
resume_text: str | None = Field(default=None)
suggested_job_post_ids: list[str] | None = Field(default=None, sa_column=Column(JSONB))
match_summary: str | None = Field(default=None)
match_reasoning: str | None = Field(default=None)
match_status: str | None = Field(default=None)
match_error: str | None = Field(default=None)
matched_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True))
inbox: list[Inbox] = Relationship(back_populates="messages")
@ -73,12 +80,36 @@ class Inbox_Messages(SQLModel, table=True):
if isinstance(body, str):
return body
return email_data.get("bodyPreview") or ""
@classmethod
async def candidate_x_inbox(cls,session:AsyncSession,candidate_id:Any):
try:
pass
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
async def set_match_result(
cls,
session: AsyncSession,
record_id,
*,
resume_text=None,
suggested_job_post_ids=None,
summary="",
reasoning="",
status="",
error="",
):
"""Persist agent output onto one inbox row; returns the row or None."""
row = await cls.get_inbox_message_by_id(session, record_id)
if not row:
return None
if resume_text is not None:
row.resume_text = resume_text
row.suggested_job_post_ids = suggested_job_post_ids
row.match_summary = summary or None
row.match_reasoning = reasoning or None
row.match_status = status or None
row.match_error = error or None
row.matched_at = datetime.now(timezone.utc)
session.add(row)
await session.commit()
await session.refresh(row)
return row
@classmethod
def _fields_from_email(cls, email_data: dict, file_path: list[str] | None = None) -> dict:

30
backend/inbox/plugins.py
View File

@ -6,6 +6,7 @@ import base64
from pathlib import Path
from inbox.models import Inbox_Messages
from job.candidate.views import FileRead
def load_message_files(message: Inbox_Messages) -> list[dict]:
@ -30,3 +31,32 @@ def load_message_files(message: Inbox_Messages) -> list[dict]:
}
)
return files
async def extract_resume_text(file_paths: list[str]) -> tuple[str, str]:
"""Extract text from the PDFs among file_paths. Returns (combined_text, error)."""
pdf_paths = [
Path(p.strip())
for p in (file_paths or [])
if p and p.strip() and Path(p.strip()).suffix.lower() == ".pdf"
]
existing = [p for p in pdf_paths if p.is_file()]
if not existing:
return "", "no PDF attachment to extract (.doc/.docx not supported)"
texts: list[str] = []
errors: list[str] = []
for path in existing:
try:
raw = path.read_bytes()
result = await FileRead(session=None, filename=path.name, file=raw).read_file()
text = (result.get("text") or "").strip()
if text:
texts.append(text)
except Exception as exc:
errors.append(f"{path.name}: {exc}")
if not texts:
return "", "; ".join(errors) if errors else "no text extracted from PDF"
return "\n\n---\n\n".join(texts), ""

6
backend/inbox/serializers.py
View File

@ -40,4 +40,10 @@ def serialize_message(message: Inbox_Messages) -> dict:
"message_sent_time": message.message_sent_time,
"message_reply": message.message_reply,
"file_path": message.file_path,
"suggested_job_post_ids": list(message.suggested_job_post_ids or []),
"match_summary": message.match_summary,
"match_reasoning": message.match_reasoning,
"match_status": message.match_status,
"match_error": message.match_error,
"matched_at": message.matched_at.isoformat() if message.matched_at else None,
}

93
backend/inbox/views.py
View File

@ -1,19 +1,96 @@
import logging
import httpx,os
from fastapi import HTTPException
from inbox.models import Inbox_Messages
from inbox.file_decoder import decode_attachment, AttachmentDecodeError
from inbox.serializers import serialize_message
from inbox.plugins import load_message_files
from inbox.plugins import load_message_files, extract_resume_text
from dotenv import load_dotenv
load_dotenv()
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
from db_setup import session_scope
from job.job_post.models import JobPosts
from job.job_post.serializers import serialize_job_post
logger = logging.getLogger("inbox.match")
class Email:
def __init__(self,session:AsyncSession,token=None):
self.session=session
self.get_url=os.getenv("EMAIL_URL")
self.token=token
self.pending_match_ids: list[str] = []
@staticmethod
async def run_inbox_matching(inbox_ids: list[str], *, force: bool = False) -> None:
"""Background worker: extract CV text, run the matching agent, persist results.
Uses its own session_scope per message - never the request session.
"""
if not inbox_ids:
return
# Local import so inbox routes still load when langgraph/openai are missing.
from agent.execute_agent import run_agent
cached_posts = None
for record_id in inbox_ids:
try:
async with session_scope() as session:
row = await Inbox_Messages.get_inbox_message_by_id(session, record_id)
if not row:
continue
if not force and row.match_status is not None:
continue
paths = [p.strip() for p in (row.file_path or "").split(",") if p.strip()]
text, extract_err = await extract_resume_text(paths)
if not text:
await Inbox_Messages.set_match_result(
session,
record_id,
status="no_text",
error=extract_err or "no text extracted",
)
continue
if cached_posts is None:
posts = await JobPosts.get_active_job_posts(session)
cached_posts = [serialize_job_post(p) for p in posts]
result = await run_agent(
subject=row.message_subject or "",
resume_text=text,
job_posts=cached_posts,
)
await Inbox_Messages.set_match_result(
session,
record_id,
resume_text=text,
suggested_job_post_ids=result.get("suggested_job_post_ids") or [],
summary=result.get("summary") or "",
reasoning=result.get("reasoning") or "",
status=result.get("status") or "failed",
error=result.get("error") or "",
)
logger.info(
"matched inbox %s status=%s ids=%s",
record_id,
result.get("status"),
result.get("suggested_job_post_ids"),
)
except Exception as exc:
logger.exception("inbox matching failed for %s", record_id)
try:
async with session_scope() as session:
await Inbox_Messages.set_match_result(
session, record_id, status="failed", error=str(exc)
)
except Exception:
logger.exception("failed to persist match failure for %s", record_id)
async def service_email(self,top,skip):
async with httpx.AsyncClient() as client:
@ -39,6 +116,12 @@ class Email:
data=response.json()
re_create_file=await decode_attachment(data.get("attachments"))
insert_func=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file)
if (
insert_func.attachment
and insert_func.file_path
and insert_func.match_status is None
):
self.pending_match_ids.append(str(insert_func.id))
return response.json()
else:
raise HTTPException(status_code=response.status_code,detail=response.text)
@ -66,5 +149,13 @@ class Email:
item["files"]=files
return item
async def queue_rematch(self,record_id):
message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
if not message:
raise HTTPException(status_code=404,detail="Message not found")
if not message.attachment or not message.file_path:
raise HTTPException(status_code=400,detail="Message has no attachment to match")
return str(message.id)
async def count_inbox_messages(self,search=None):
return await Inbox_Messages.count_inbox_messages(self.session,search)

7
backend/job/job_post/models.py
View File

@ -62,6 +62,13 @@ class JobPosts(SQLModel, table=True):
result = await session.execute(select(cls).where(cls.id == uid))
return result.scalars().first()
@classmethod
async def get_active_job_posts(cls, session: AsyncSession):
result = await session.execute(
select(cls).where(cls.is_active == True, cls.is_deleted == False) # noqa: E712
)
return result.scalars().all()
@classmethod
async def insert_job_post(cls, session: AsyncSession, fields: dict):
row = cls(**fields)

2
backend/llm_setup.py
View File

@ -66,7 +66,7 @@ async def llm_call(system, user, *, model=None, temperature=None, json_mode=Fals
With json_mode the prompt must mention JSON somewhere or the API rejects the call.
"""
kwargs = {
"model": OPENAI_MODEL,
"model": model or OPENAI_MODEL,
"max_completion_tokens": OPENAI_MAX_OUTPUT_TOKENS,
"messages": [
{"role": "system", "content": system},

36
backend/main.py
View File

@ -1,10 +1,11 @@
import logging
from contextlib import asynccontextmanager
import fastapi
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from fastapi import FastAPI,APIRouter
from db_setup import lifespan
from db_setup import lifespan as db_lifespan
from inbox.app import router as inbox_router
from users.app import router as users_router
from role.app import router as role_router
@ -13,6 +14,37 @@ from job.app import router as candidate_router
from notifications.app import router as confirmation_router
# Without this the db/migration logs have no handler and are swallowed under uvicorn.
logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(name)s: %(message)s")
logger = logging.getLogger("main")
@asynccontextmanager
async def lifespan(app):
"""Compose DB lifespan with optional LLM/agent warm-up (degrades on failure)."""
async with db_lifespan(app):
llm_ready = False
agent_ready = False
close_llm = None
close_agent = None
try:
from llm_setup import init_llm, close_llm as _close_llm
from agent.agent_setup import init_agent, close_agent as _close_agent
close_llm = _close_llm
close_agent = _close_agent
await init_llm()
llm_ready = True
await init_agent()
agent_ready = True
except Exception as exc:
logger.warning("llm/agent startup skipped: %s", exc)
try:
yield
finally:
if agent_ready and close_agent is not None:
await close_agent()
if llm_ready and close_llm is not None:
await close_llm()
# lifespan connects to Postgres and brings migrations up to head on startup,
# and disposes of the connection pool on shutdown.
@ -30,4 +62,4 @@ app.include_router(users_router)
app.include_router(role_router)
app.include_router(forget_password_router)
app.include_router(confirmation_router)
app.include_router(candidate_router)
app.include_router(candidate_router)