From 08068b2b18fb80b7d6dcb3491e7eb82d05a6ba9c Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Mon, 10 Aug 2026 16:19:29 +0500 Subject: [PATCH 1/3] Linking user to candidate to job --- backend/agent/decorators.py | 7 ++- backend/agent/models.py | 1 + backend/agent/prompt.py | 3 +- backend/agent/serializers.py | 1 + backend/agent/views.py | 5 +- backend/inbox/app.py | 9 ++- backend/inbox/models.py | 109 ++++++++++++++++++++++++++++++++- backend/inbox/plugins.py | 15 +++++ backend/inbox/serializers.py | 2 +- backend/inbox/tasks.py | 3 +- backend/inbox/views.py | 34 +++++++--- backend/job/app.py | 21 ++++++- backend/job/candidate/views.py | 11 ++++ 13 files changed, 201 insertions(+), 20 deletions(-) diff --git a/backend/agent/decorators.py b/backend/agent/decorators.py index a5c0079..53bd62c 100644 --- a/backend/agent/decorators.py +++ b/backend/agent/decorators.py @@ -68,4 +68,9 @@ def parse_match_response(data, allowed_ids) -> tuple[list[str], str, str]: reasoning = "\n".join(str(item) for item in reasoning) if not isinstance(reasoning, str): reasoning = "" - return suggested, summary.strip(), reasoning.strip() + + experience = data.get("experience") + if not isinstance(experience, str): + experience = "" + + return suggested, summary.strip(), reasoning.strip(), experience.strip() diff --git a/backend/agent/models.py b/backend/agent/models.py index fa4e5b6..51fb99a 100644 --- a/backend/agent/models.py +++ b/backend/agent/models.py @@ -13,6 +13,7 @@ class AgentState(TypedDict, total=False): subject: str resume_text: str + experience: str job_posts: list[dict] suggested_job_post_ids: list[str] summary: str diff --git a/backend/agent/prompt.py b/backend/agent/prompt.py index 25e979a..0484963 100644 --- a/backend/agent/prompt.py +++ b/backend/agent/prompt.py @@ -26,7 +26,8 @@ Respond with JSON only: { "suggested_job_post_ids": ["uuid", "..."], "summary": "one short sentence for the recruiter", - "reasoning": "brief bullet-style explanation per suggested match" + "reasoning": "brief bullet-style explanation per suggested match", + "experience": "the relevant experience of the candidate in years for the suggested match" } """ diff --git a/backend/agent/serializers.py b/backend/agent/serializers.py index 163027e..347f7e0 100644 --- a/backend/agent/serializers.py +++ b/backend/agent/serializers.py @@ -12,6 +12,7 @@ def serialize_agent_result(state: dict) -> dict: "suggested_job_post_ids": state.get("suggested_job_post_ids") or [], "summary": state.get("summary") or "", "reasoning": state.get("reasoning") or "", + "experience": state.get("experience") or "", "status": state.get("status") or "failed", "error": state.get("error") or "", } diff --git a/backend/agent/views.py b/backend/agent/views.py index 0a6a867..8520014 100644 --- a/backend/agent/views.py +++ b/backend/agent/views.py @@ -62,12 +62,13 @@ 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, summary, reasoning = parse_match_response(data, allowed_ids) + suggested, summary, reasoning, experience = parse_match_response(data, allowed_ids) return { "status": "matched", "suggested_job_post_ids": suggested, "summary": summary, "reasoning": reasoning, + "experience": experience, } except Exception as exc: logger.exception("agent match_jobs failed") @@ -77,6 +78,7 @@ async def match_jobs(state: AgentState) -> dict: "suggested_job_post_ids": [], "summary": "", "reasoning": "", + "experience": "", } @@ -86,6 +88,7 @@ async def finalize(state: AgentState) -> dict: "suggested_job_post_ids": state.get("suggested_job_post_ids") or [], "summary": state.get("summary") or "", "reasoning": state.get("reasoning") or "", + "experience": state.get("experience") or "", "status": state.get("status") or "failed", "error": state.get("error") or "", } diff --git a/backend/inbox/app.py b/backend/inbox/app.py index dce4c6c..6d4321e 100644 --- a/backend/inbox/app.py +++ b/backend/inbox/app.py @@ -15,6 +15,7 @@ router = APIRouter() async def fetch_email( top:int=Query(100), skip:int=Query(0,ge=0), + test_on: bool = Query(True), token: str | None = Query(None), session: AsyncSession = Depends(get_session), ): @@ -27,13 +28,17 @@ async def fetch_email( items_lst=[] for item in value: message_id=item.get("id") - service_per_email=await service.get_email_by_id(message_id) + service_per_email=await service.get_email_by_id(message_id,test_on) items_lst.append({"message_id":message_id,"email_contents":service_per_email}) if service.pending_match_ids: await service.enqueue_matching(list(service.pending_match_ids),force=False) - return JSONResponse(content={"data":items_lst,"status_code":200}) + account_setup=[] + if service.pending_confirmation_emails: + account_setup=await service.send_account_setup(list(service.pending_confirmation_emails)) + + return JSONResponse(content={"data":items_lst,"account_setup":account_setup,"status_code":200}) except HTTPException: raise diff --git a/backend/inbox/models.py b/backend/inbox/models.py index 75b02a6..97f9699 100644 --- a/backend/inbox/models.py +++ b/backend/inbox/models.py @@ -1,14 +1,31 @@ +import logging +import os import uuid from datetime import datetime, timezone from typing import Any, Optional + +from dotenv import load_dotenv from fastapi import HTTPException from inbox.enums import Candidate_application_Status +from role.models import EnumRoles, Roles from sqlalchemy import Column, DateTime, func, or_, update from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel import Field, Relationship, SQLModel, select, true from users.models import Users +from users.plugins import hash_password + +load_dotenv() +logger = logging.getLogger("inbox.models") + +# Placeholder only. The account lands inactive and the candidate is mailed a +# confirmation link; the real password comes from the reset flow afterwards. +DEFAULT_CANDIDATE_PASSWORD = os.getenv("DEFAULT_CANDIDATE_PASSWORD", "Utopia!@#") +CANDIDATE_ROLE_ID_FALLBACK = 8 # mirrors users/views.py:signup_user +SKIP_SENDER_PREFIXES = ("noreply", "no-reply", "donotreply", "do-not-reply", + "mailer-daemon", "postmaster", "bounce") class Inbox(SQLModel, table=True): @@ -65,6 +82,7 @@ class Inbox_Messages(SQLModel, table=True): file_name: str | None = Field(default=None) file_path: str | None = Field(default=None) resume_text: str | None = Field(default=None) + experience: 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) @@ -83,6 +101,15 @@ class Inbox_Messages(SQLModel, table=True): return body return email_data.get("bodyPreview") or "" + # @classmethod + # async def get_candidate_profile(cls,session:AsyncSession,user_id:uuid.UUID|None=None): + # try: + # qryy=select(cls,Users).join(cls,cls.) + # if user_id + + # except Exception as e: + # raise HTTPException(status_code=500,detail=str(e)) + @classmethod async def get_all_applications(cls,session:AsyncSession,message_id:uuid.UUID|int|None=None): try: @@ -101,6 +128,7 @@ class Inbox_Messages(SQLModel, table=True): record_id, *, resume_text=None, + experience=None, suggested_job_post_ids=None, summary="", reasoning="", @@ -118,6 +146,7 @@ class Inbox_Messages(SQLModel, table=True): row.match_reasoning = reasoning or None row.match_status = status or None row.match_error = error or None + row.experience = experience or None row.matched_at = datetime.now(timezone.utc) session.add(row) await session.commit() @@ -156,6 +185,72 @@ class Inbox_Messages(SQLModel, table=True): "full_email_response": email_data, } + @classmethod + def _sender_address(cls, email_data: dict) -> str: + return ( + email_data.get("from", {}) + .get("emailAddress", {}) + .get("address", "") + or "" + ).strip().lower() + + @classmethod + def _sender_display_name(cls, email_data: dict, address: str) -> str: + name = ( + email_data.get("from", {}) + .get("emailAddress", {}) + .get("name") + or "" + ).strip() + if name: + return name + return address.split("@", 1)[0] if address else "candidate" + + @classmethod + def _is_linkable_sender(cls, address: str) -> bool: + if not address or "@" not in address: + return False + local = address.split("@", 1)[0] + return not local.startswith(SKIP_SENDER_PREFIXES) + + @classmethod + async def _link_sender(cls,session:AsyncSession,email_data:dict,email): + address=cls._sender_address(email_data) + if not cls._is_linkable_sender(address): + return None + try: + user=(await session.execute( + select(Users).where(func.lower(Users.email)==address) + )).scalars().first() + + if not user: + role=await Roles.get_role_by_name(session,EnumRoles.CANDIDATE.value) + user=Users( + name=cls._sender_display_name(email_data,address), + email=address, + role_id=role.id if role else CANDIDATE_ROLE_ID_FALLBACK, + password=hash_password(DEFAULT_CANDIDATE_PASSWORD), + ) + session.add(user) + session.add(Inbox(user_id=user.id,message_id=email.id)) + await session.commit() + return address + + link=(await session.execute( + select(Inbox).where(Inbox.message_id==email.id,Inbox.user_id==user.id) + )).scalars().first() + if not link: + session.add(Inbox(user_id=user.id,message_id=email.id)) + await session.commit() + return None + except IntegrityError: + await session.rollback() + return None + except Exception as e: + await session.rollback() + logger.warning("sender link failed for %s: %s",address,e) + return None + @classmethod async def insert_email( cls, @@ -163,9 +258,11 @@ class Inbox_Messages(SQLModel, table=True): email_data: dict, file_path: list[str] | None = None, ): + """Returns (row, new_user_email). new_user_email is set only when this call + created the sender's Users row.""" fields = cls._fields_from_email(email_data, file_path) external_id = fields.get("message_id") - + link_user=None if external_id: existing = ( await session.execute( @@ -178,12 +275,18 @@ class Inbox_Messages(SQLModel, table=True): session.add(existing) await session.commit() await session.refresh(existing) - return existing + + if fields.get("attachment"): + link_user=await cls._link_sender(session, email_data, existing) + return existing, link_user email = cls(**fields) session.add(email) await session.commit() - return email + + if fields.get("attachment"): + link_user=await cls._link_sender(session, email_data, email) + return email, link_user @classmethod def _search_filter(cls, search: str): diff --git a/backend/inbox/plugins.py b/backend/inbox/plugins.py index 45294fe..5176f49 100644 --- a/backend/inbox/plugins.py +++ b/backend/inbox/plugins.py @@ -17,10 +17,25 @@ load_dotenv() EMAIL_URL=os.getenv("EMAIL_URL") EMAIL_API_TOKEN=os.getenv("EMAIL_API_TOKEN") +BACKEND_URL=os.getenv("BACKEND_URL","http://localhost:8000") _ATTACHMENTS_DIR=Path(__file__).resolve().parent/"decoded_attachments" +async def request_email_confirmation(email): + """POST /users/confirm-email/resend on this same service -> status code. + + Goes through the endpoint rather than importing Confirmation so the token row, + resend cooldown and mail send stay on one code path. + """ + async with httpx.AsyncClient(timeout=20.0) as client: + response=await client.post( + f"{BACKEND_URL.rstrip('/')}/users/confirm-email/resend", + json={"email":email}, + ) + return response.status_code + + async def fetch_read_status_delta(folder, since=None, limit=1000, max_pages=10, token=None): """GET /sync/read-status -> the raw round dict.""" if not EMAIL_URL: diff --git a/backend/inbox/serializers.py b/backend/inbox/serializers.py index fcab277..4419fd6 100644 --- a/backend/inbox/serializers.py +++ b/backend/inbox/serializers.py @@ -97,7 +97,7 @@ def serialize_application(message: Inbox_Messages) -> dict: "resume_text": message.resume_text, "ats_score": None, "phone": None, - "experience": None, + "experience": message.experience or "", "recruiter": None, "duplicate": None, } diff --git a/backend/inbox/tasks.py b/backend/inbox/tasks.py index c0d65cb..554b8f6 100644 --- a/backend/inbox/tasks.py +++ b/backend/inbox/tasks.py @@ -80,12 +80,13 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict: if status=="failed": raise RuntimeError(error or "agent returned failed status") - + logger.info("agent result: %s,%s",result,result.get("experience")) async with session_scope() as session: await Inbox_Messages.set_match_result( session, record_id, resume_text=text, + experience=result.get("experience") or "", suggested_job_post_ids=result.get("suggested_job_post_ids") or [], summary=result.get("summary") or "", reasoning=result.get("reasoning") or "", diff --git a/backend/inbox/views.py b/backend/inbox/views.py index 3855cc0..e7d88b1 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -9,6 +9,7 @@ from inbox.plugins import ( EMAIL_API_TOKEN, fetch_message_read_status, load_message_files, + request_email_confirmation, ) from dotenv import load_dotenv load_dotenv() @@ -24,6 +25,7 @@ class Email: self.get_url=os.getenv("EMAIL_URL") self.token=token or EMAIL_API_TOKEN self.pending_match_ids:list[str]=[] + self.pending_confirmation_emails:list[str]=[] # async def get_all_applications(self,app_id=None): # try: @@ -49,7 +51,7 @@ class Email: except Exception as e: raise HTTPException(status_code=500,detail=str(e)) - async def get_email_by_id(self,message_id): + async def get_email_by_id(self,message_id,test_on=True): async with httpx.AsyncClient() as client: try: response=await client.get(f"{self.get_url}/emails/{message_id}", @@ -58,14 +60,14 @@ class Email: if response.status_code==200: 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() + row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) + if row.attachment and row.file_path and row.match_status is None: + self.pending_match_ids.append(str(row.id)) + if test_on: + return data + if new_user_email: + self.pending_confirmation_emails.append(new_user_email) + return data else: raise HTTPException(status_code=response.status_code,detail=response.text) except Exception as e: @@ -128,6 +130,20 @@ class Email: task_ids.append(task.task_id) return task_ids + async def send_account_setup(self,emails): + """Mail a confirmation link per freshly created sender. Best effort: a failed + mail must not fail a fetch whose messages are already stored.""" + results=[] + for email in emails or []: + try: + status=await request_email_confirmation(email) + except Exception as exc: + logger.warning("confirmation request failed for %s: %s",email,exc) + results.append({"email":email,"sent":False}) + continue + results.append({"email":email,"sent":status==200}) + return results + async def count_inbox_messages(self,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED): if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED: return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status) diff --git a/backend/job/app.py b/backend/job/app.py index ce910a2..cdeb38c 100644 --- a/backend/job/app.py +++ b/backend/job/app.py @@ -2,7 +2,7 @@ from fastapi import APIRouter,Depends,Query from fastapi.responses import JSONResponse from fastapi import HTTPException from db_setup import get_session -from job.candidate.views import FileRead +from job.candidate.views import FileRead,CandidateView from sqlalchemy.ext.asyncio import AsyncSession from users.permissions import PermissionTag, require_permission from job.job_post.views import JobPost,JobPostCreate @@ -101,3 +101,22 @@ async def buffer_channels( raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) + +@router.get("/candidate/fetch") +async def fetch_candidate( + user_id:str=Query(None), + current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), + session: AsyncSession = Depends(get_session), +): + try: + service=CandidateView(session=session) + if user_id: + data=await service.get_candidate(user_id=user_id) + else: + data=await service.get_candidate() + return JSONResponse(content={"data":data,"status_code":200}) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) + \ No newline at end of file diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index 5c09149..0307168 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -4,6 +4,7 @@ from datetime import datetime,timezone from fastapi import HTTPException from pypdf import PdfReader from sqlalchemy import select +from sqlmodel import true from inbox.models import Inbox_Messages from job.candidate.plugins import normalize_spaced_text @@ -66,3 +67,13 @@ class FileRead: # try: # get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id) # get_file= + +class CandidateView: + def __init__(self,session:AsyncSession): + self.session=session + + async def get_candidate(self,user_id=None): + try: + call_func=Inbox_Messages.get_candidate_profile(user_id=user_id) + except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) \ No newline at end of file From 4d340fd6d5b2a133fb42a387333fcc077c45a2eb Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Mon, 10 Aug 2026 16:24:47 +0500 Subject: [PATCH 2/3] agent coide simplifgication --- backend/agent/agent_setup.py | 30 +++++------- backend/agent/decorators.py | 72 +++++++++++++-------------- backend/agent/execute_agent.py | 17 +++---- backend/agent/models.py | 24 +++++---- backend/agent/serializers.py | 15 +++--- backend/agent/views.py | 90 ++++++++++++---------------------- backend/inbox/tasks.py | 54 ++++++-------------- backend/inbox/views.py | 13 ++--- backend/job/app.py | 36 +++++++------- backend/job/candidate/views.py | 16 +++--- 10 files changed, 148 insertions(+), 219 deletions(-) diff --git a/backend/agent/agent_setup.py b/backend/agent/agent_setup.py index 5fe000a..2bd9bba 100644 --- a/backend/agent/agent_setup.py +++ b/backend/agent/agent_setup.py @@ -13,45 +13,39 @@ from __future__ import annotations import logging -from langgraph.graph import END, START, StateGraph +from langgraph.graph import END,START,StateGraph from agent.models import AgentState -from agent.views import finalize, match_jobs, prepare_context, route_after_prepare +from agent.views import match_jobs,prepare_context,route_after_prepare -logger = logging.getLogger("agent") +logger=logging.getLogger("agent") -_graph = None +_graph=None def build_graph(): - """Construct and compile the HR-ATS candidate matching graph.""" - graph = StateGraph(AgentState) - graph.add_node("prepare", prepare_context) - graph.add_node("match_jobs", match_jobs) - graph.add_node("finalize", finalize) - graph.add_edge(START, "prepare") - graph.add_conditional_edges("prepare", route_after_prepare) - graph.add_edge("match_jobs", "finalize") - graph.add_edge("finalize", END) + graph=StateGraph(AgentState) + graph.add_node("prepare",prepare_context) + graph.add_node("match_jobs",match_jobs) + graph.add_edge(START,"prepare") + graph.add_conditional_edges("prepare",route_after_prepare) + graph.add_edge("match_jobs",END) return graph.compile() def get_graph(): - """Return the cached compiled graph, building it on first use.""" global _graph if _graph is None: - _graph = build_graph() + _graph=build_graph() logger.info("langgraph compiled") return _graph async def init_agent(): - """Warm the compiled graph. LLM init stays on llm_setup.init_llm().""" get_graph() async def close_agent(): - """Drop the cached graph.""" global _graph - _graph = None + _graph=None logger.info("agent graph closed") diff --git a/backend/agent/decorators.py b/backend/agent/decorators.py index 53bd62c..5cdbbeb 100644 --- a/backend/agent/decorators.py +++ b/backend/agent/decorators.py @@ -11,45 +11,41 @@ import uuid def normalize_job_posts(job_posts) -> list[dict]: - """Keep only dict items with an id field; stringify ids for the LLM.""" if not job_posts: return [] - normalized: list[dict] = [] + normalized=[] for item in job_posts: - if not isinstance(item, dict): + if not isinstance(item,dict): continue - job_id = item.get("id") + job_id=item.get("id") if job_id is None: continue - normalized.append( - { - "id": str(job_id), - "title": item.get("title") or "", - "description": item.get("description") or "", - "post_text": item.get("post_text") or "", - "requirements": item.get("requirements") or [], - "optional_skills": item.get("optional_skills") or [], - "location": item.get("location") or "", - "employment_type": item.get("employment_type") or "", - } - ) + normalized.append({ + "id":str(job_id), + "title":item.get("title") or "", + "description":item.get("description") or "", + "post_text":item.get("post_text") or "", + "requirements":item.get("requirements") or [], + "optional_skills":item.get("optional_skills") or [], + "location":item.get("location") or "", + "employment_type":item.get("employment_type") or "", + }) return normalized -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): +def parse_match_response(data,allowed_ids) -> tuple[list[str],str,str,str]: + if not isinstance(data,dict): raise RuntimeError(f"model did not return a JSON object: {data!r}") - allowed = set(allowed_ids or []) - raw_ids = data.get("suggested_job_post_ids") or [] - if not isinstance(raw_ids, list): - raw_ids = [] + allowed=set(allowed_ids or []) + raw_ids=data.get("suggested_job_post_ids") or [] + if not isinstance(raw_ids,list): + raw_ids=[] - suggested: list[str] = [] - seen: set[str] = set() + suggested=[] + seen=set() for raw_id in raw_ids: - job_id = str(raw_id).strip() + job_id=str(raw_id).strip() if not job_id or job_id not in allowed or job_id in seen: continue try: @@ -59,18 +55,18 @@ def parse_match_response(data, allowed_ids) -> tuple[list[str], str, str]: seen.add(job_id) suggested.append(job_id) - summary = data.get("summary") - if not isinstance(summary, str): - summary = "" + 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 = "" + reasoning=data.get("reasoning") + if isinstance(reasoning,list): + reasoning="\n".join(str(item) for item in reasoning) + if not isinstance(reasoning,str): + reasoning="" - experience = data.get("experience") - if not isinstance(experience, str): - experience = "" + experience=data.get("experience") + if not isinstance(experience,str): + experience="" - return suggested, summary.strip(), reasoning.strip(), experience.strip() + return suggested,summary.strip(),reasoning.strip(),experience.strip() diff --git a/backend/agent/execute_agent.py b/backend/agent/execute_agent.py index e7ef8d0..f7e0c03 100644 --- a/backend/agent/execute_agent.py +++ b/backend/agent/execute_agent.py @@ -9,14 +9,11 @@ from agent.agent_setup import get_graph from agent.serializers import serialize_agent_result -async def run_agent(*, subject="", resume_text="", job_posts=None) -> dict: - """Run the default graph and return a serialized result dict.""" - final_state = await get_graph().ainvoke( - { - "subject": subject or "", - "resume_text": resume_text or "", - "job_posts": job_posts or [], - "status": "pending", - } - ) +async def run_agent(*,subject="",resume_text="",job_posts=None) -> dict: + final_state=await get_graph().ainvoke({ + "subject":subject or "", + "resume_text":resume_text or "", + "job_posts":job_posts or [], + "status":"pending", + }) return serialize_agent_result(final_state) diff --git a/backend/agent/models.py b/backend/agent/models.py index 51fb99a..01d15de 100644 --- a/backend/agent/models.py +++ b/backend/agent/models.py @@ -5,18 +5,16 @@ Pure module: no FastAPI imports and no HTTPException. from __future__ import annotations -from typing import Literal, TypedDict +from typing import Literal,TypedDict -class AgentState(TypedDict, total=False): - """Shared state passed between graph nodes.""" - - subject: str - resume_text: str - experience: str - job_posts: list[dict] - suggested_job_post_ids: list[str] - summary: str - reasoning: str - error: str - status: Literal["pending", "ready", "matched", "skipped", "failed"] +class AgentState(TypedDict,total=False): + subject:str + resume_text:str + experience:str + job_posts:list[dict] + suggested_job_post_ids:list[str] + summary:str + reasoning:str + error:str + status:Literal["pending","ready","matched","skipped","failed"] diff --git a/backend/agent/serializers.py b/backend/agent/serializers.py index 347f7e0..833aac7 100644 --- a/backend/agent/serializers.py +++ b/backend/agent/serializers.py @@ -6,13 +6,12 @@ Pure module: no FastAPI imports and no HTTPException. from __future__ import annotations -def serialize_agent_result(state: dict) -> dict: - """Plain dict for services/serializers — no ORM objects.""" +def serialize_agent_result(state:dict) -> dict: return { - "suggested_job_post_ids": state.get("suggested_job_post_ids") or [], - "summary": state.get("summary") or "", - "reasoning": state.get("reasoning") or "", - "experience": state.get("experience") or "", - "status": state.get("status") or "failed", - "error": state.get("error") or "", + "suggested_job_post_ids":state.get("suggested_job_post_ids") or [], + "summary":state.get("summary") or "", + "reasoning":state.get("reasoning") or "", + "experience":state.get("experience") or "", + "status":state.get("status") or "failed", + "error":state.get("error") or "", } diff --git a/backend/agent/views.py b/backend/agent/views.py index 8520014..a128e70 100644 --- a/backend/agent/views.py +++ b/backend/agent/views.py @@ -11,84 +11,58 @@ from typing import Literal from langgraph.graph import END -from agent.decorators import normalize_job_posts, parse_match_response +from agent.decorators import normalize_job_posts,parse_match_response from agent.models import AgentState -from agent.prompt import prompt, user_prompt +from agent.prompt import prompt,user_prompt from llm_setup import llm_call -logger = logging.getLogger("agent") +logger=logging.getLogger("agent") -async def prepare_context(state: AgentState) -> dict: - """Validate inputs and decide whether matching should run.""" - subject = (state.get("subject") or "").strip() - resume_text = (state.get("resume_text") or "").strip() - job_posts = normalize_job_posts(state.get("job_posts")) +async def prepare_context(state:AgentState) -> dict: + subject=(state.get("subject") or "").strip() + resume_text=(state.get("resume_text") or "").strip() + job_posts=normalize_job_posts(state.get("job_posts")) if not resume_text: - return { - "status": "skipped", - "error": "resume_text is empty", - "suggested_job_post_ids": [], - "summary": "", - "reasoning": "", - } + return {"status":"skipped","error":"resume_text is empty","suggested_job_post_ids":[]} if not job_posts: - return { - "status": "skipped", - "error": "no active job posts to match against", - "suggested_job_post_ids": [], - "summary": "", - "reasoning": "", - } + return {"status":"skipped","error":"no active job posts to match against","suggested_job_post_ids":[]} return { - "subject": subject, - "resume_text": resume_text, - "job_posts": job_posts, - "status": "ready", - "error": "", + "subject":subject, + "resume_text":resume_text, + "job_posts":job_posts, + "status":"ready", + "error":"", } -def route_after_prepare(state: AgentState) -> Literal["match_jobs", "__end__"]: - if state.get("status") == "ready": +def route_after_prepare(state:AgentState) -> Literal["match_jobs","__end__"]: + if state.get("status")=="ready": return "match_jobs" return END -async def match_jobs(state: AgentState) -> dict: - """Ask the LLM (via llm_setup.llm_call) to map the candidate to job posts.""" +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, summary, reasoning, experience = parse_match_response(data, allowed_ids) + data=await llm_call(prompt(),user_prompt(state),json_mode=True) + allowed_ids={item["id"] for item in state.get("job_posts") or []} + suggested,summary,reasoning,experience=parse_match_response(data,allowed_ids) return { - "status": "matched", - "suggested_job_post_ids": suggested, - "summary": summary, - "reasoning": reasoning, - "experience": experience, + "status":"matched", + "suggested_job_post_ids":suggested, + "summary":summary, + "reasoning":reasoning, + "experience":experience, } - except Exception as exc: + except Exception as e: logger.exception("agent match_jobs failed") return { - "status": "failed", - "error": str(exc), - "suggested_job_post_ids": [], - "summary": "", - "reasoning": "", - "experience": "", + "status":"failed", + "error":str(e), + "suggested_job_post_ids":[], + "summary":"", + "reasoning":"", + "experience":"", } - - -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 "", - "experience": state.get("experience") or "", - "status": state.get("status") or "failed", - "error": state.get("error") or "", - } diff --git a/backend/inbox/tasks.py b/backend/inbox/tasks.py index 554b8f6..2d59cc9 100644 --- a/backend/inbox/tasks.py +++ b/backend/inbox/tasks.py @@ -3,19 +3,19 @@ from __future__ import annotations import logging -from datetime import datetime, timezone +from datetime import datetime,timezone +from agent.execute_agent import run_agent from db_setup import session_scope from inbox.models import Inbox_Messages from inbox.plugins import extract_resume_text from job.job_post.models import JobPosts from job.job_post.serializers import serialize_job_post -from taskiq_management.broker_setup import MAX_RETRIES, RETRY_DELAY, broker +from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY,broker from taskiq_management.middleware import PermanentTaskError logger=logging.getLogger("inbox.tasks") - -_DONE_STATUSES=frozenset({"matched","skipped","no_text","failed","dlq"}) +_DONE=frozenset({"matched","skipped","no_text","failed","dlq"}) @broker.task( @@ -27,60 +27,41 @@ _DONE_STATUSES=frozenset({"matched","skipped","no_text","failed","dlq"}) async def match_inbox_message(record_id:str,force:bool=False) -> dict: if not record_id or not str(record_id).strip(): raise PermanentTaskError("record_id is required") - record_id=str(record_id).strip() async with session_scope() as session: row=await Inbox_Messages.get_inbox_message_by_id(session,record_id) if not row: raise PermanentTaskError(f"inbox message {record_id} not found") - - if not force and row.match_status in _DONE_STATUSES: - logger.info("skip %s — already %s",record_id,row.match_status) + if not force and row.match_status in _DONE: return {"status":row.match_status,"skipped":True} - if not row.attachment or not row.file_path: raise PermanentTaskError("message has no attachment to match") + paths=[p.strip() for p in row.file_path.split(",") if p.strip()] + subject=row.message_subject or "" row.match_status="processing" row.match_error=None row.matched_at=datetime.now(timezone.utc) session.add(row) await session.commit() - await session.refresh(row) - paths=[p.strip() for p in (row.file_path or "").split(",") if p.strip()] - subject=row.message_subject or "" + posts=await JobPosts.get_active_job_posts(session) + job_posts=[serialize_job_post(p) for p in posts] text,extract_err=await extract_resume_text(paths) if not text: async with session_scope() as session: await Inbox_Messages.set_match_result( - session, - record_id, - status="no_text", - error=extract_err or "no text extracted", + session,record_id,status="no_text",error=extract_err or "no text extracted", ) return {"status":"no_text","error":extract_err} - from agent.execute_agent import run_agent - - async with session_scope() as session: - posts=await JobPosts.get_active_job_posts(session) - job_posts=[serialize_job_post(p) for p in posts] - - try: - result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts) - except Exception as exc: - logger.exception("agent failed for %s",record_id) - raise RuntimeError(f"agent matching failed: {exc}") from exc - + result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts) status=result.get("status") or "failed" - error=result.get("error") or "" - if status=="failed": - raise RuntimeError(error or "agent returned failed status") - logger.info("agent result: %s,%s",result,result.get("experience")) + raise RuntimeError(result.get("error") or "agent returned failed status") + async with session_scope() as session: await Inbox_Messages.set_match_result( session, @@ -91,11 +72,6 @@ async def match_inbox_message(record_id:str,force:bool=False) -> dict: summary=result.get("summary") or "", reasoning=result.get("reasoning") or "", status=status, - error=error, + error=result.get("error") or "", ) - - logger.info("matched inbox %s status=%s ids=%s",record_id,status,result.get("suggested_job_post_ids")) - return { - "status":status, - "suggested_job_post_ids":result.get("suggested_job_post_ids") or [], - } + return {"status":status,"suggested_job_post_ids":result.get("suggested_job_post_ids") or []} diff --git a/backend/inbox/views.py b/backend/inbox/views.py index e7d88b1..8c6092d 100644 --- a/backend/inbox/views.py +++ b/backend/inbox/views.py @@ -51,7 +51,7 @@ class Email: except Exception as e: raise HTTPException(status_code=500,detail=str(e)) - async def get_email_by_id(self,message_id,test_on=True): + async def get_email_by_id(self,message_id): async with httpx.AsyncClient() as client: try: response=await client.get(f"{self.get_url}/emails/{message_id}", @@ -63,8 +63,6 @@ class Email: row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file) if row.attachment and row.file_path and row.match_status is None: self.pending_match_ids.append(str(row.id)) - if test_on: - return data if new_user_email: self.pending_confirmation_emails.append(new_user_email) return data @@ -131,17 +129,14 @@ class Email: return task_ids async def send_account_setup(self,emails): - """Mail a confirmation link per freshly created sender. Best effort: a failed - mail must not fail a fetch whose messages are already stored.""" results=[] for email in emails or []: try: status=await request_email_confirmation(email) - except Exception as exc: - logger.warning("confirmation request failed for %s: %s",email,exc) + results.append({"email":email,"sent":status==200}) + except Exception as e: + logger.warning("confirmation request failed for %s: %s",email,e) results.append({"email":email,"sent":False}) - continue - results.append({"email":email,"sent":status==200}) return results async def count_inbox_messages(self,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED): diff --git a/backend/job/app.py b/backend/job/app.py index cdeb38c..682bb75 100644 --- a/backend/job/app.py +++ b/backend/job/app.py @@ -2,7 +2,7 @@ from fastapi import APIRouter,Depends,Query from fastapi.responses import JSONResponse from fastapi import HTTPException from db_setup import get_session -from job.candidate.views import FileRead,CandidateView +from job.candidate.views import FileRead from sqlalchemy.ext.asyncio import AsyncSession from users.permissions import PermissionTag, require_permission from job.job_post.views import JobPost,JobPostCreate @@ -102,21 +102,21 @@ async def buffer_channels( except Exception as e: raise HTTPException(status_code=500,detail=str(e)) -@router.get("/candidate/fetch") -async def fetch_candidate( - user_id:str=Query(None), - current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), - session: AsyncSession = Depends(get_session), -): - try: - service=CandidateView(session=session) - if user_id: - data=await service.get_candidate(user_id=user_id) - else: - data=await service.get_candidate() - return JSONResponse(content={"data":data,"status_code":200}) - except HTTPException: - raise - except Exception as e: - raise HTTPException(status_code=500,detail=str(e)) +# @router.get("/candidate/fetch") +# async def fetch_candidate( +# user_id:str=Query(None), +# current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)), +# session: AsyncSession = Depends(get_session), +# ): +# try: +# service=CandidateView(session=session) +# if user_id: +# data=await service.get_candidate(user_id=user_id) +# else: +# data=await service.get_candidate() +# return JSONResponse(content={"data":data,"status_code":200}) +# except HTTPException: +# raise +# except Exception as e: +# raise HTTPException(status_code=500,detail=str(e)) \ No newline at end of file diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index 0307168..6b10aa3 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -68,12 +68,12 @@ class FileRead: # get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id) # get_file= -class CandidateView: - def __init__(self,session:AsyncSession): - self.session=session +# class CandidateView: +# def __init__(self,session:AsyncSession): +# self.session=session - async def get_candidate(self,user_id=None): - try: - call_func=Inbox_Messages.get_candidate_profile(user_id=user_id) - except Exception as e: - raise HTTPException(status_code=500,detail=str(e)) \ No newline at end of file +# async def get_candidate(self,user_id=None): +# try: +# call_func=Inbox_Messages.get_candidate_profile(user_id=user_id) +# except Exception as e: +# raise HTTPException(status_code=500,detail=str(e)) \ No newline at end of file From 1d8b16760639cedcffc63c16f0bc1f71de4e0d46 Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Mon, 10 Aug 2026 17:23:10 +0500 Subject: [PATCH 3/3] readme file --- backend/README.md | 676 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 676 insertions(+) create mode 100644 backend/README.md diff --git a/backend/README.md b/backend/README.md new file mode 100644 index 0000000..290a607 --- /dev/null +++ b/backend/README.md @@ -0,0 +1,676 @@ +# HR-ATS-Portal — Backend + +FastAPI service behind **TalentFlow**, an HR / Applicant-Tracking portal. It pulls candidate +applications out of a mailbox, decodes CV attachments, runs an LLM agent to match each CV +against live job posts, publishes job ads to social channels through Buffer, and serves the +whole thing to the frontend behind JWT auth and a database-driven RBAC model. + +Everything in this document describes `backend/` only. + +--- + +## Table of contents + +- [Architecture](#architecture) +- [Tech stack](#tech-stack) +- [Directory layout](#directory-layout) +- [House style — what lives in which file](#house-style--what-lives-in-which-file) +- [Domains](#domains) +- [Data model](#data-model) +- [API reference](#api-reference) +- [Authentication and RBAC](#authentication-and-rbac) +- [Background jobs](#background-jobs) +- [The matching agent](#the-matching-agent) +- [External integrations](#external-integrations) +- [Configuration](#configuration) +- [Running locally](#running-locally) +- [Database migrations](#database-migrations) +- [Docker](#docker) +- [Response conventions](#response-conventions) +- [Known gaps and gotchas](#known-gaps-and-gotchas) + +--- + +## Architecture + +```mermaid +flowchart TB + FE["Frontend (Vite, :5173)"] -->|JWT Bearer| API + + subgraph API["FastAPI — main.py :8000"] + R1["users / role / forget_password / notifications"] + R2["inbox"] + R3["job (job_post + candidate)"] + end + + API --> PG[("PostgreSQL — schema app")] + API -->|enqueue| REDIS[("Redis Streams")] + + REDIS --> W["Taskiq worker\ninbox.tasks + inbox.sync_tasks"] + SCHED["Taskiq scheduler\ncron"] --> REDIS + W --> PG + W --> AGENT["LangGraph agent\nagent/"] + AGENT --> OAI["OpenAI"] + + API -->|GET /emails, /sync/read-status| MAILAPI["Email API (MS Graph proxy)"] + W --> MAILAPI + API -->|multipart send| TEAMS["Teams Mail API"] + API -->|GraphQL| BUF["Buffer"] +``` + +**The application flow, end to end:** + +1. `GET /email/fetch` pulls messages from the external Email API, decodes PDF/DOC/DOCX + attachments to `inbox/decoded_attachments/`, and upserts them into `inbox_messages`. +2. When a message carries an attachment, the sender is linked to a `users` row — created with + the `candidate` role if new — through the `inbox` join table. +3. Any message with a stored attachment and no match result yet is enqueued onto Redis as an + `inbox.match_message` task. +4. The worker extracts the résumé text, hands it plus the active job posts to the LangGraph + agent, and writes `suggested_job_post_ids`, `match_summary`, `match_reasoning` and + `experience` back onto the row. +5. New candidate accounts land inactive and are mailed a confirmation link; the link is what + flips `is_active`. +6. A cron task sweeps Outlook read-status deltas back onto `inbox_messages.message_read`. +7. Recruiters read all of it through `/inbox/all-applications` and publish new roles with + `POST /job/post-job`, which renders the ad copy and pushes it to Buffer. + +--- + +## Tech stack + +| Concern | Choice | +|---|---| +| Web framework | FastAPI 0.136 + Uvicorn | +| ORM / models | SQLModel 0.0.38 on SQLAlchemy 2.0 (async, `asyncpg`) | +| Database | PostgreSQL, application objects in the `app` schema | +| Migrations | Alembic 1.18, driven by `alembic_setup.py` | +| Auth | PyJWT (HS256) access / refresh / reset tokens, `bcrypt` hashing | +| Task queue | Taskiq on Redis Streams, with a smart-retry + dead-letter middleware | +| LLM | OpenAI async client, orchestrated by LangGraph | +| PDF extraction | `pypdf` | +| HTTP client | `httpx` | +| Python | 3.12 (see `Dockerfile`) | + +--- + +## Directory layout + +``` +backend/ +├── main.py # FastAPI app, lifespan, CORS, router mounting +├── db_setup.py # Settings, async engine, sessions, init_db/lifespan +├── alembic_setup.py # Alembic scaffolding, autogenerate, migrate-on-boot +├── llm_setup.py # AsyncOpenAI client + llm_call helper +├── requirements.txt +├── Dockerfile # image for the Taskiq worker / scheduler +├── alembic.ini # generated by alembic_setup.py, not hand-written +├── migrations/ # generated env.py + versions/ +├── LLM_CONTEXT_PROMPT.md # house-style prompt to paste into an LLM before editing +│ +├── users/ # accounts, login, signup, RBAC enforcement +├── role/ # roles, permission bundles, permission tags +├── forget_password/ # reset-code request → verify → new password +├── notifications/ # email-confirmation tokens and mail +├── inbox/ # mailbox sync, attachments, applications +├── job/ +│ ├── app.py # routes for both sub-domains +│ ├── job_post/ # job ads + Buffer publishing +│ └── candidate/ # CV reading, candidate profile +├── agent/ # LangGraph CV → job-post matching agent +└── taskiq_management/ # broker, scheduler, DLQ middleware, smoke task +``` + +There are **no `__init__.py` files**. The service is run from `backend/`, so imports are +top-level (`from users.app import router`, `from db_setup import get_session`). + +--- + +## House style — what lives in which file + +Every domain package follows the same six-file shape. This is enforced by convention, and +`LLM_CONTEXT_PROMPT.md` is the canonical statement of it. + +| File | Owns | Must not do | +|---|---|---| +| `app.py` | Routes, inline request models, `JSONResponse`, dependency injection | Business rules, SQL | +| `views.py` | Business checks, calls models, raises `HTTPException` | Build login token envelopes | +| `models.py` | SQLModel table + `@classmethod async def` accessors | Import FastAPI, raise `HTTPException` | +| `serializers.py` | Hand-built `dict` builders (`serialize_*`) | Touch the DB or `Depends` | +| `plugins.py` | Pure helpers — hashing, JWT, HTTP calls to third parties | Import FastAPI | +| `permissions.py` | Bearer schemes and `Depends` aliases (auth domains only) | Hold route handlers | + +Additional rules that matter when you edit this code: + +- Request bodies are Pydantic models declared **inline in `app.py`**, never in `serializers.py`. +- There are **no Pydantic response models** — responses are hand-built dicts. +- Route paths are verb-in-path (`/users/create`, `/users/fetch`), not REST-resource-only, and + there is no `/api/v1` prefix. +- Non-DB config is module-level `load_dotenv()` + `os.getenv(...)`. Only database settings go + through `db_setup.Settings`. + +--- + +## Domains + +### `users/` +Signup, login, refresh, CRUD, role assignment, and the RBAC machinery every other domain +depends on. `users/permissions.py` defines the full `PermissionTag` vocabulary (13 modules × +8 actions = 104 tags) and the `require_permission(...)` dependency. A startup assertion +(`_assert_vocabulary_complete`) fails loudly if the tag list ever drifts from +`PermissionModule × PermissionAction`. + +Signup always assigns the `candidate` role, creates the account **inactive**, and sends a +confirmation email. Login rejects unconfirmed accounts. + +### `role/` +Three-level permission model: `permission_tags` (atomic `module.action` rows) → +`permissions` (named bundles holding a JSONB array of tag ids) → `roles` (holding a JSONB +array of bundle ids). `Roles.resolve_tags()` walks that chain and returns a flat tuple of tag +names; dangling or inactive ids simply contribute nothing rather than erroring. + +Eight system roles are seeded: `system_administrator`, `hr_administrator`, `recruiter`, +`hiring_manager`, `department_head`, `interviewer`, `ceo`, `candidate`. + +### `inbox/` +The heart of the ingestion pipeline. + +- `views.py::Email` talks to the external Email API, decodes attachments, upserts messages, + and enqueues matching work. +- `models.py` holds `Inbox_Messages` (the mail rows plus all agent output columns), + `Inbox_Alerts`, and `Inbox` — the join table linking a message to the candidate `Users` row + it came from. `_link_sender` creates the candidate account on first contact, skipping + `noreply@`-style senders. +- `file_decoder.py` turns Graph `contentBytes` into real PDF / DOCX / DOC files, validating + magic bytes for each format and stripping path traversal from filenames. +- `plugins.py` resolves attachment paths (handling Windows paths written by the host API but + read from a Linux worker), extracts résumé text, and calls the read-status sync endpoints. +- `tasks.py` / `sync_tasks.py` are the two Taskiq tasks. + +### `job/` +Two sub-domains behind one router: + +- **`job_post/`** — renders LinkedIn-shaped ad copy from a structured payload, resolves the + Buffer channel (by explicit id, by platform alias, or by the configured default), creates + the post over Buffer's GraphQL API, and records the mapped status. A queued post is recorded + as `scheduled`, not `published`; only Buffer reporting `sent` promotes it. +- **`candidate/`** — `FileRead` extracts text from an uploaded PDF (`pypdf`), and + `match_inbox_cv` force-requeues an existing inbox message for matching. `CandidateView` + reads the candidate profile through the `inbox` join. + +### `notifications/` and `forget_password/` +Two parallel token flows, deliberately kept separate so each owns its own mail copy and env +reads: + +- **Confirmation** — a 32-byte url-safe secret, bcrypt-hashed in + `email_confirmation_tokens`; the link carries `.` because a bcrypt hash + cannot be looked up. Replays (mail scanners, back button) are handled idempotently. +- **Password reset** — a short code mailed to the user, bcrypt-hashed in + `password_reset_codes`, with a resend cooldown and a max-attempts cap. Verifying the code + mints a `type=reset` JWT carrying the code row id (`crid`), which is the only thing that + authorises the new-password call. + +### `agent/` +LangGraph state machine — see [The matching agent](#the-matching-agent). + +### `taskiq_management/` +Broker, scheduler, DLQ middleware, and a `ping` smoke task. + +--- + +## Data model + +All tables live in the `app` schema (`DB_DEFAULT_SCHEMA`), with a shared naming convention for +indexes, constraints and foreign keys. `SQLModel.metadata` is pointed at `Base.metadata` in +`db_setup.py` so SQLModel and DeclarativeBase share one registry and Alembic sees everything. + +| Table | Key columns | Notes | +|---|---|---| +| `users` | `id` (uuid PK), `email` (unique), `role_id` → `roles.id`, `password`, `is_active`, `is_deleted` | Soft delete. `role` is `selectin`-loaded; lazy loads would raise `MissingGreenlet` under asyncio | +| `roles` | `id`, `role_name` (unique), `permissions` (JSONB int[]), `is_system` | | +| `permissions` | `id`, `name` (unique), `permission_tags` (JSONB int[]) | Named bundles | +| `permission_tags` | `id`, `tag_name` (unique), `module`, `action` | Unique on (`module`, `action`) | +| `inbox_messages` | `id` (uuid), `message_id` (upstream id, unique), `full_email_response` (JSONB), subject/body/from/to/cc/bcc, `message_read`, `attachment`, `file_name`, `file_path`, `application_status`, `resume_text`, `experience`, `suggested_job_post_ids` (JSONB), `match_summary`, `match_reasoning`, `match_status`, `match_error`, `matched_at` | One row per mail; agent output lands here | +| `inbox` | `id`, `user_id` → `users.id`, `message_id` → `inbox_messages.id`, `alert_id` | Join table linking a candidate to a message | +| `inbox_alerts` | `id`, `alert_sender_name`, `alert_sender_email`, `is_read` | | +| `job_posts` | `id` (uuid), `title`, `platform`, `channel_id`, `post_text`, `requirements`/`optional_skills` (JSON), `status`, `buffer_post_id`, `buffer_external_link`, `buffer_sent_at`, `buffer_error`, `created_by` → `users.id` | | +| `password_reset_codes` | `id`, `email`, `code_hash`, `expires_at`, `attempts`, `is_used`, `verified_at` | | +| `email_confirmation_tokens` | `id`, `user_id`, `email`, `token_hash`, `expires_at`, `is_used`, `confirmed_at` | | + +`application_status` is a `str` enum: `PROCESS`, `PENDING`, `APPROVED`, `REJECTED`, `ONHOLD`, +`CLOSED`. + +`match_status` is free-form text written by the worker: `processing`, `matched`, `skipped`, +`no_text`, `failed`, `dlq`. + +--- + +## API reference + +Base URL: `http://localhost:8000`. Interactive docs at `/docs`. + +### Auth — `users/app.py`, `notifications/app.py`, `forget_password/app.py` + +| Method | Path | Guard | Purpose | +|---|---|---|---| +| POST | `/users/signup` | public | Create a candidate account (inactive) and mail a confirmation link | +| POST | `/users/login` | public | Email/username + password → token envelope | +| POST | `/users/refresh` | public | Refresh token → new token pair | +| GET | `/users/me` | any authenticated user | Current user with resolved permissions | +| POST | `/users/confirm-email` | public | Consume a confirmation token, activate the account | +| POST | `/users/confirm-email/resend` | public | Re-issue a confirmation link (cooldown enforced) | +| POST | `/users/forget-password` | public | Mail a reset code | +| POST | `/users/forget-password/verify-code` | public | Verify the code → `type=reset` JWT | +| POST | `/users/forget-password/new-password` | reset JWT | Set the new password | + +### Users — `users/app.py` + +| Method | Path | Required tag | +|---|---|---| +| GET | `/users/fetch` | `rbac_users.view` | +| POST | `/users/create` | `rbac_users.create` | +| PUT | `/users/update?record_id=` | `rbac_users.edit` | +| PUT | `/users/assign-role?record_id=` | `rbac_users.edit` (+ `rbac_users.manage` inside the service) | +| PUT | `/users/remove-role?record_id=` | `rbac_users.edit` (+ `rbac_users.manage`) | +| DELETE | `/users/delete?record_id=` | `rbac_users.delete` | + +### Roles and permissions — `role/app.py` + +| Method | Path | Required tag | +|---|---|---| +| GET | `/roles/fetch` | `rbac_users.view` | +| POST | `/roles/create` | `rbac_users.create` | +| PUT | `/roles/update?record_id=` | `rbac_users.edit` | +| DELETE | `/roles/delete?record_id=` | `rbac_users.delete` | +| GET | `/permissions/fetch` | `rbac_users.view` | +| POST | `/permissions/create` | `rbac_users.manage` | +| PUT | `/permissions/update?record_id=` | `rbac_users.manage` | +| GET | `/permission-tags/fetch` | `rbac_users.view` | + +### Inbox — `inbox/app.py` + +| Method | Path | Guard | Purpose | +|---|---|---|---| +| GET | `/email/fetch` | upstream token only | Pull from the Email API, decode attachments, upsert, enqueue matching. `test_on=true` (default) returns raw payloads and skips the account-setup mails | +| GET | `/inbox/fetch` | none | Stored messages, with attachments inlined as base64 | +| GET | `/inbox/all-applications` | `inbox.view` | The Applications tab. Filters: `application_status`, `isread`, `search`, `record_id`, `top`, `skip` | +| POST | `/inbox/{record_id}/match` | `inbox.edit` | Force a re-match of one message | +| POST | `/inbox/{record_id}/read` | `inbox.edit` | Mark read locally | +| GET | `/inbox/{record_id}/read-status` | `inbox.edit` | Re-pull read status from upstream for one message | + +### Jobs and candidates — `job/app.py` + +| Method | Path | Required tag | Purpose | +|---|---|---|---| +| 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/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 | + +`POST /job/post-job` takes `mode` ∈ `addToQueue` | `shareNow` | `customScheduled`; the +`customScheduled` mode requires `scheduler_date` (and optionally `scheduler_time`), which the +route combines into a UTC `due_at`. + +--- + +## Authentication and RBAC + +**Tokens.** PyJWT, HS256, with a `type` claim that `decode_token(..., expected_type=...)` +rejects on mismatch. Three types: + +| Type | Lifetime (default) | Extra claims | +|---|---|---| +| `access` | 30 min | `email`, `role_id` | +| `refresh` | 7 days | — | +| `reset` | 10 min | `crid` (reset-code row id) | + +`iat` / `exp` are always timezone-aware UTC. Every token carries a `jti`. + +**Login response** puts the OAuth2 fields at the root so Swagger's Authorize button can read +them: + +```json +{ + "access_token": "...", + "refresh_token": "...", + "token_type": "bearer", + "expires_in": 1800, + "data": { "id": "...", "email": "...", "role_id": 3, "role_name": "recruiter", "...": "..." }, + "status_code": 200 +} +``` + +**Passwords** use `bcrypt` directly rather than `passlib` — passlib 1.7.4 reads +`bcrypt.__about__.__version__`, which bcrypt dropped in 4.1, and the failed probe makes it +reject every password as over 72 bytes. Input is truncated to 72 bytes on a character +boundary before hashing. + +**Authorization.** `get_current_user` decodes the access token, loads the user by `sub`, +rejects missing / deleted / inactive accounts, resolves the role's tags, and returns a +serialized user dict with a `permissions` list. `require_permission(*tags, require_all=True)` +is the dependency that guards routes; a user with no role assigned gets a 403 before any tag +check runs. + +Role assignment is additionally guarded in `users/views.py`: you cannot grant a role holding +permissions you do not yourself hold. + +--- + +## Background jobs + +Taskiq over **Redis Streams**, with a result backend and a Redis-backed schedule source. + +**Middleware order matters** — `DeadLetterMiddleware` sits before `SmartRetryMiddleware` so a +`PermanentTaskError` can set `retry_on_error=False` before retry logic runs. + +| 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.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 | + +**Retries.** `SmartRetryMiddleware` with jitter and exponential delay, `TASKIQ_MAX_RETRIES` +attempts, capped at `TASKIQ_MAX_DELAY`. Raising `PermanentTaskError` (missing record, no +attachment, blank `record_id`) skips retries entirely. + +**Dead letter queue.** Exhausted or permanently-failed tasks are written to the +`taskiq:dlq` Redis stream as a JSON payload. For `inbox.match_message` specifically, the +middleware also stamps `match_status="dlq"` on the row so the failure is visible in the UI +rather than only in Redis. + +**The read-status latch.** `apply_read_status` only ever applies `false → true`. Nothing +pushes local reads back to Outlook, so upstream keeps reporting `isRead=false`; without the +latch the every-minute sweep would un-read a mail the user just opened. The trade-off is that +un-reading a mail in Outlook no longer propagates here. `sync_read_status` holds a Redis lock +(`inbox:sync_read_status:lock`, 300s TTL) so overlapping cron ticks cannot double-run, and +pages at most 10 rounds per tick. + +--- + +## The matching agent + +A small LangGraph `StateGraph` over an `AgentState` TypedDict. + +``` +START → prepare → (conditional) → match_jobs → END + └→ END (when resume_text or job_posts is empty) +``` + +- **`prepare_context`** trims the subject and résumé text and normalizes the job posts down to + the fields the model needs. Empty résumé text or no active posts short-circuits to + `status="skipped"`. +- **`match_jobs`** calls `llm_call(prompt(), user_prompt(state), json_mode=True)` and parses + the response. + +The parser (`agent/decorators.py::parse_match_response`) is deliberately strict: a suggested +id must be present in the posts that were actually sent, must be a valid UUID, and duplicates +are dropped. The model cannot invent a job post. `reasoning` returned as a list is joined into +a string; non-string fields fall back to `""`. + +The graph is compiled once per process (`init_agent()` from the FastAPI lifespan, or lazily on +first use inside the worker) and the OpenAI client is created lazily and closed on shutdown. + +Startup is **fault-tolerant**: if the broker or the LLM/agent fails to initialize, `main.py` +logs a warning and the API still serves. Only the database is a hard requirement. + +--- + +## External integrations + +| Service | Used by | Contract | +|---|---|---| +| **Email API** (a Microsoft Graph proxy) | `inbox/` | `GET {EMAIL_URL}/emails`, `GET {EMAIL_URL}/emails/{id}`, `GET {EMAIL_URL}/sync/read-status`, `GET {EMAIL_URL}/sync/read-status/message/{id}` — Bearer `EMAIL_API_TOKEN` | +| **Teams Mail API** | `notifications/`, `forget_password/` | multipart POST to `TEAMS_MAIL_API_URL`; success is HTTP **202**, anything else raises | +| **Buffer** | `job/job_post/` | GraphQL against `BUFFER_API_URL` — `createPost` mutation, `account { organizations }` and `channels` queries | +| **OpenAI** | `agent/`, `llm_setup.py` | Chat Completions with `response_format: json_object` | + +Attachments are written to `backend/inbox/decoded_attachments/`. In Docker this directory is +bind-mounted into the worker so both processes see the same files. + +--- + +## Configuration + +Copy `.env.example` to `.env` and fill it in. `.env` is git-ignored; `.env.example` is not. +`db_setup.Settings` reads `backend/.env` or the repo-root `.env`; every other module reads its +own keys with `os.getenv`. + +### Database + +| Variable | Default | Notes | +|---|---|---| +| `DB_USERNAME`, `DB_PASSWORD`, `DB_HOST`, `DB_PORT`, `DB_NAME` | — | **Required.** `DB_PORT` must be an integer | +| `DATABASE_URL` | — | Full DSN; wins over the parts above | +| `DB_SSLMODE` | — | e.g. `require` on Azure; translated to asyncpg's `ssl` | +| `DB_SCHEMAS` | `app` | Comma-separated; created on startup | +| `DB_DEFAULT_SCHEMA` | `app` | Schema for models that declare none | +| `DB_ECHO` | `false` | SQL logging | +| `DB_POOL_SIZE` / `DB_MAX_OVERFLOW` / `DB_POOL_RECYCLE` | `5` / `10` / `1800` | | +| `DB_CONNECT_RETRIES` | `10` | Startup wait-for-Postgres | +| `DB_AUTO_MIGRATE` | `true` | Run `upgrade head` on startup | +| `DB_AUTOGENERATE` | `true` | Write a revision when models drift | +| `APP_NAME` | `hr-ats-portal` | Postgres `application_name` | + +### Auth + +| Variable | Default | +|---|---| +| `JWT_SECRET_KEY` | — (**required**) | +| `JWT_ALGORITHM` | `HS256` | +| `JWT_ACCESS_TOKEN_EXPIRE_MINUTES` | `30` | +| `JWT_REFRESH_TOKEN_EXPIRE_DAYS` | `7` | +| `JWT_RESET_TOKEN_EXPIRE_MINUTES` | `10` | + +### Email ingestion + +| Variable | Default | +|---|---| +| `EMAIL_URL`, `EMAIL_API_TOKEN` | — | +| `EMAIL_SYNC_FOLDER` | `inbox` | +| `EMAIL_SYNC_SINCE` | — | +| `EMAIL_SYNC_CRON` | `* * * * *` | +| `BACKEND_URL` | `http://localhost:8000` (used for the internal confirmation-resend call) | +| `DEFAULT_CANDIDATE_PASSWORD` | `Utopia!@#` — placeholder only; the account is inactive until confirmed | + +### Mail out + +| Variable | Default | +|---|---| +| `TEAMS_MAIL_API_URL`, `TEAMS_API_TOKEN` | — | +| `FRONTEND_URL` | `http://localhost:5173` | +| `CONFIRM_EMAIL_PATH` | `/auth/confirm-email` | +| `CONFIRM_TOKEN_TTL_SECONDS` | `86400` | +| `CONFIRM_TOKEN_RESEND_SECONDS` | `60` | +| `RESET_CODE_TTL_SECONDS` | `60` | +| `RESET_CODE_RESEND_SECONDS` | `30` | +| `RESET_CODE_MAX_ATTEMPTS` | `5` | + +### Buffer + +| Variable | Default | +|---|---| +| `BUFFER_API` | — (access token) | +| `BUFFER_API_URL` | `https://api.buffer.com` | +| `BUFFER_CHANNEL_ID` | — (fallback channel) | + +### OpenAI + +| Variable | Default | +|---|---| +| `OPENAI_API_KEY` | — | +| `OPENAI_MODEL` | `gpt-5.4-mini` | +| `OPENAI_TEMPERATURE` | `0` — leave blank to omit the parameter for models that reject it | +| `OPENAI_MAX_OUTPUT_TOKENS` | `32768` (`.env.example` ships `4096`) | +| `OPENAI_TIMEOUT` / `OPENAI_MAX_RETRIES` / `OPENAI_CONNECT_RETRIES` | `60` / `3` / `3` | +| `OPENAI_BASE_URL`, `OPENAI_ORGANIZATION`, `OPENAI_PROJECT` | — (set only for Azure or a gateway) | + +### Taskiq / Redis + +| Variable | Default | +|---|---| +| `REDIS_URL` | `redis://localhost:6379/0` | +| `TASKIQ_QUEUE_NAME` | `inbox` | +| `TASKIQ_CONSUMER_GROUP` | `taskiq` | +| `TASKIQ_MAX_RETRIES` | `3` | +| `TASKIQ_RETRY_DELAY` | `5` | +| `TASKIQ_MAX_DELAY` | `120` | +| `TASKIQ_IDLE_TIMEOUT_MS` | `600000` | +| `TASKIQ_DLQ_STREAM` | `taskiq:dlq` | +| `TASKIQ_WORKER_NAME` | falls back to `HOSTNAME` | +| `APP_VERSION` | `dev` | + +--- + +## Running locally + +**Prerequisites:** Python 3.12, PostgreSQL, Redis. + +```bash +cd backend + +python -m venv .venv +source .venv/bin/activate # Windows: .venv\Scripts\activate +pip install -r requirements.txt + +cp .env.example .env # then fill it in +``` + +All commands must be run from `backend/` — the import paths depend on it. + +**API:** + +```bash +uvicorn main:app --reload --port 8000 +``` + +Startup connects to Postgres (retrying with backoff), creates the configured schemas, runs +migrations to head, then starts the broker, the OpenAI client and the agent graph. Broker and +LLM failures are logged and skipped; the API still comes up. + +**Worker** (needs Redis): + +```bash +taskiq worker taskiq_management.broker_setup:broker \ + inbox.tasks inbox.sync_tasks taskiq_management.tasks +``` + +**Scheduler** (cron ticks for `inbox.sync_read_status`): + +```bash +taskiq scheduler taskiq_management.broker_setup:scheduler inbox.sync_tasks +``` + +Docs: + +--- + +## Database migrations + +`alembic_setup.py` wraps Alembic so the plain `alembic` CLI and the app's own +migrate-on-startup share one configuration. It scaffolds `alembic.ini`, `migrations/env.py` +and `script.py.mako` on first use and never overwrites them. Model modules are discovered +automatically — every `/models.py` under `backend/` is imported before the metadata is +diffed. + +```bash +python alembic_setup.py migrate # upgrade to head, then autogenerate any drift +python alembic_setup.py revision -m "add x" # write a revision if the models have drifted +python alembic_setup.py upgrade -r head +python alembic_setup.py downgrade -r -1 +python alembic_setup.py current +python alembic_setup.py head +``` + +Migrations run under a Postgres advisory lock, so several workers booting at once cannot +migrate concurrently. Empty revisions are suppressed. Alembic's own `alembic_version` table is +excluded from autogenerate, as is anything outside the configured schemas. + +The module is named `alembic_setup` rather than `alembic` because `backend/` is on `sys.path` +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). + +```bash +docker compose up -d # from the repo root +docker compose logs -f taskiq-worker +``` + +`backend/Dockerfile` builds a `python:3.12-slim` image whose default command is the Taskiq +worker. `backend/inbox/decoded_attachments` is bind-mounted so the worker can read the +attachments the API wrote. + +--- + +## Response conventions + +Every handler wraps its body in the same try/except: + +```python +try: + service=Email(session=session) + data=await service.some_method(...) + return JSONResponse(content={"data":data,"status_code":200}) +except HTTPException: + raise +except Exception as e: + raise HTTPException(status_code=500,detail=str(e)) +``` + +| Shape | Response | +|---|---| +| List | `{"data": [...], "total": , "status_code": 200}` | +| Single record | `{"data": {...}, "total": 1, "status_code": 200}` | +| Login / refresh | OAuth2 fields at the root, user under `data` | +| Error | FastAPI's `{"detail": "..."}` with the real status code | + +Serializers always `str()` UUIDs, `.isoformat()` datetimes, and never emit `password`. + +--- + +## Known gaps and gotchas + +- **`DB_PORT` must be set.** `db_setup.Settings` evaluates `int(os.getenv("DB_PORT"))` at class + definition time, so a missing value raises `TypeError` on import rather than a friendly + config error. +- **CORS is fully open** (`allow_origins=["*"]` with credentials). Fine for development, needs + tightening before production. +- **`/email/fetch` and `/inbox/fetch` carry no permission guard.** `/email/fetch` authenticates + only against the upstream Email API token. +- **`.doc` / `.docx` résumés are decoded and stored but not parsed.** `extract_resume_text` + handles PDFs only and reports `no PDF attachment to extract` for the rest. +- **`serialize_application` returns `null` for `ats_score`, `phone`, `recruiter` and + `duplicate`** — `inbox_messages` has no columns for them yet, and `processing` is derived + from `message_read` alone, so it is only ever `"Read"` or `"Unread"`. +- **`Inbox.get_candidate_profile` filters on `cls.user.role_id`**, which is a relationship + attribute rather than a joined column; the candidate-profile query needs a join before it + behaves as intended. +- **Attachment paths may be Windows absolutes** written by the host API but read by a Linux + worker. `resolve_attachment_path` normalizes separators and falls back to the basename under + the mounted `decoded_attachments` directory. +- **Read status is a one-way latch** — see [Background jobs](#background-jobs). +- **There is no test suite** in `backend/` at present. + +--- + +## Editing this codebase + +Before changing anything here, read [`LLM_CONTEXT_PROMPT.md`](LLM_CONTEXT_PROMPT.md). It states +the house style in full and is the reference used to keep new code indistinguishable from +`users/` and `inbox/`. The short version: mirror the neighbouring file, keep the layer duties +intact, add no new layers, and do not reformat code you did not otherwise need to touch. + +Adding an endpoint, in order: + +1. Model accessor in `models.py` (if it touches the DB). +2. Service method in `views.py`. +3. `serialize_*` in `serializers.py` if the shape is new. +4. Route in `app.py` with the standard try/except and `JSONResponse`. +5. `CurrentUser` or `Depends(require_permission(...))` if the route is protected.