HR-ATS-Portal/backend/job/job_post/views.py

599 lines
25 KiB
Python

from datetime import date, time
import logging
import os
import uuid
from pathlib import Path
from uuid import UUID
import httpx
from dotenv import load_dotenv
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel, model_validator
from inbox.models import Inbox_Messages
from job.assignment.views import Assignment
from job.job_post.enums import RequisitionStatus
from job.job_post.models import JobPostImages,JobPostStatusHistory,JobPosts,SocialPlatform
from role.models import EnumRoles
from users.models import Users
from job.job_post.plugins import (
BufferError,
create_buffer_post,
list_buffer_channels,
local_status,
normalize_platform,
parse_buffer_datetime,
render_job_post,
resolve_channel,
)
from job.job_post.serializers import serialize_job_post, serialize_job_row, serialize_job_stats, serialize_status_history
load_dotenv()
logger=logging.getLogger("job.job_post")
# Cover images live in the job_post_images table (bytea), NOT on disk:
# production containers have ephemeral filesystems, so a file-backed image
# would vanish on every redeploy. One row per post; re-upload replaces it.
ALLOWED_IMAGE_TYPES={"image/png","image/jpeg","image/webp","image/gif"}
IMAGE_TYPE_BY_EXT={"png":"image/png","jpg":"image/jpeg","jpeg":"image/jpeg","webp":"image/webp","gif":"image/gif"}
MAX_JOB_IMAGE_BYTES=5*1024*1024
def _payload_recruiter_ids(payload):
"""Prefer current_recruiter_ids; fall back to current_recruiter_id. None = omitted."""
has_list="current_recruiter_ids" in payload and payload.get("current_recruiter_ids") is not None
has_one="current_recruiter_id" in payload
if has_list:
raw=payload.get("current_recruiter_ids") or []
if not isinstance(raw,(list,tuple)):
raw=[raw]
ids=list(raw)
if not ids and has_one and payload.get("current_recruiter_id") not in (None,""):
ids=[payload.get("current_recruiter_id")]
return ids
if has_one:
raw=payload.get("current_recruiter_id")
return [] if raw in (None,"") else [raw]
return None
def _recruiter_fields(users):
ids=[str(u.id) for u in users]
return {
"current_recruiter_ids": ids,
"current_recruiter_id": users[0].id if users else None,
}
def _job_image_key(job_post_id) -> uuid.UUID:
try:
return uuid.UUID(str(job_post_id))
except ValueError as e:
raise HTTPException(status_code=422,detail="job_post_id must be a UUID") from e
class JobPostCreate(BaseModel):
title: str
experience_min: int | None = None
experience_max: int | None = None
requirements: list[str] = []
optional_skills: list[str] = []
salary: str = "Anonymous"
location: str | None = None
employment_type: str | None = None
department: str | None = None
vacancies: int = 1
description: str | None = None
platform: str | None = None
channel_id: str | None = None
mode: str = "addToQueue"
scheduler_time: time | None = time(0, 0, 0)
scheduler_date: date | None = None
due_at: str | None = None
hiring_manager_id: UUID | None = None
current_recruiter_id: UUID | None = None
current_recruiter_ids: list[UUID] | None = None
requisition_id: UUID | None = None
@model_validator(mode="after")
def validate_mode_and_due_at(self):
allowed = {"addToQueue", "shareNow", "customScheduled"}
if self.mode not in allowed:
raise ValueError(f"mode must be one of {sorted(allowed)}")
if self.mode == "customScheduled" and not self.due_at and not self.scheduler_date:
raise ValueError("due_at or scheduler_date is required when mode is customScheduled")
return self
class JobPost:
def __init__(self,session:AsyncSession):
self.session=session
self.buffer_api=os.getenv("BUFFER_API")
self.channel_id=os.getenv("BUFFER_CHANNEL_ID")
async def _resolve_recruiters(self,assignment,raw_ids):
"""Validate each id is an active recruiter. Dedup, preserve order."""
users=[]
seen=set()
for raw in raw_ids or []:
if raw in (None,""):
continue
rec=await assignment.require_role(raw,EnumRoles.RECRUITER,"current_recruiter_ids")
key=str(rec.id)
if key in seen:
continue
seen.add(key)
users.append(rec)
return users
async def _names_for(self,row):
ids=JobPosts.recruiter_ids_of(row)
extra=[]
if getattr(row,"hiring_manager_id",None):
extra.append(row.hiring_manager_id)
return await Users.names_by_ids(self.session,ids+extra)
async def _serialize_post(self,row):
return serialize_job_post(row,names=await self._names_for(row))
async def _resolve_target(self,payload,aliases=None):
"""Pick the Buffer channel to post to, and the service it belongs to.
Precedence: an explicit channel_id, then the requested platform, then the
configured default channel. Returns (channel_id, service) where service is
None if we did not have to look the channel up.
"""
if payload.get("channel_id"):
return payload["channel_id"],None
if payload.get("platform"):
channel=await resolve_channel(payload["platform"],aliases=aliases)
return channel["id"],channel.get("service")
if self.channel_id:
return self.channel_id,None
raise HTTPException(
status_code=400,
detail="Provide channel_id or platform, or configure BUFFER_CHANNEL_ID",
)
async def post_job(self,payload,current_user):
aliases=await SocialPlatform.alias_map(self.session)
# No channel_id and no platform means an internal-only requisition: save the
# row for the board/dashboard and never touch Buffer. The env-default channel
# fallback only applies when the caller explicitly asked to publish.
publish=bool(payload.get("channel_id") or payload.get("platform"))
if publish:
try:
channel_id,service=await self._resolve_target(payload,aliases)
except (httpx.HTTPError,BufferError,RuntimeError) as e:
raise HTTPException(status_code=502,detail=f"Failed to resolve Buffer channel: {e}") from e
else:
channel_id,service="",None
text=render_job_post(payload)
fields={
"title":payload.get("title"),
"employment_type":payload.get("employment_type"),
"location":payload.get("location"),
"experience_min":payload.get("experience_min"),
"experience_max":payload.get("experience_max"),
"requirements":list(payload.get("requirements") or []),
"optional_skills":list(payload.get("optional_skills") or []),
"salary":payload.get("salary") or "Anonymous",
# department is NOT NULL with a server_default of "" — pass "", never None.
"department":payload.get("department") or "",
"vacancies":payload.get("vacancies") or 1,
"description":payload.get("description"),
"post_text":text,
"channel_id":channel_id,
"status":"draft",
"created_by":current_user["id"],
}
# Only set platform when it is actually known: passing None would override the
# column default and break the NOT NULL constraint. Buffer's channelService
# replaces this with the authoritative value once the post is created.
known_platform=service or normalize_platform(payload.get("platform"),aliases)
if known_platform:
fields["platform"]=known_platform
elif not publish:
# Column default is "linkedin"; an unpublished requisition must not
# masquerade as a LinkedIn post.
fields["platform"]="internal"
assignment=Assignment(self.session)
hm=None
if payload.get("hiring_manager_id"):
hm=await assignment.require_role(
payload.get("hiring_manager_id"),EnumRoles.HIRING_MANAGER,"hiring_manager_id",
)
fields["hiring_manager_id"]=hm.id
rec_users=[]
raw_ids=_payload_recruiter_ids(payload)
if raw_ids:
rec_users=await self._resolve_recruiters(assignment,raw_ids)
fields.update(_recruiter_fields(rec_users))
if payload.get("requisition_id"):
from candidate_forms.models import Requisition
req=await Requisition.get_form_by_id(
self.session, record_id=str(payload["requisition_id"]),
)
if not req:
raise HTTPException(status_code=404, detail="Requisition not found")
held=await JobPosts.get_by_requisition_id(self.session, req.id)
if held:
raise HTTPException(
status_code=409,
detail="This requisition is already linked to a job post",
)
fields["requisition_id"]=req.id
try:
row=await JobPosts.insert_job_post(self.session,fields)
except IntegrityError as e:
orig=str(getattr(e,"orig",e)).lower()
if "requisition" in orig:
raise HTTPException(
status_code=409,detail="This requisition is already linked to a job post",
) from e
raise
assigned_by=current_user.get("id") if isinstance(current_user,dict) else None
if hm:
await assignment.record_job_owner(row.id,hm.id,"hiring_manager",assigned_by)
if rec_users:
await assignment.record_job_recruiters(row.id,[u.id for u in rec_users],assigned_by)
try:
from notifications.views import notify_job_created
await notify_job_created(self.session,row,actor_id=assigned_by)
except Exception as exc:
logger.warning("notification insert skipped: %s",exc)
# A new opening is the moment the CV Bank is worth reading. Ranking it
# here is what turns the bank from a pile someone has to remember into
# something that offers itself up. Fire-and-forget: the job is already
# created, and a queue that is down must not fail the request.
await self._rank_cv_bank(row.id)
if not publish:
return await self._serialize_post(row)
try:
post=await create_buffer_post(
text,
channel_id,
mode=payload.get("mode") or "addToQueue",
due_at=payload.get("due_at"),
)
except (httpx.HTTPError,BufferError,RuntimeError) as e:
await JobPosts.mark_failed(self.session,str(row.id),str(e))
raise HTTPException(status_code=502,detail=f"Failed to publish job post to Buffer: {e}") from e
saved=await JobPosts.mark_buffer_result(
self.session,
str(row.id),
buffer_post_id=post["id"],
status=local_status(post.get("status")),
external_link=post.get("externalLink"),
sent_at=parse_buffer_datetime(post.get("sentAt")),
platform=post.get("channelService"),
)
return serialize_job_post(saved,names=await self._names_for(saved))
async def _rank_cv_bank(self,job_post_id):
"""Queue the tier-1 rank of every banked CV against a brand-new job.
Best effort by design: this is a convenience signal, not part of
creating the job post. Redis being unavailable must not turn a
successful job creation into a 500.
"""
try:
from datetime import datetime as _dt,timezone as _tz
from job.candidate.bank_tasks import rank_bank_for_job
await rank_bank_for_job.kicker().with_labels(
created_at=_dt.now(_tz.utc).isoformat(),
correlation_id=str(job_post_id),
queue="inbox",
).kiq(str(job_post_id))
except Exception as exc:
logger.warning("cv-bank rank not queued for job %s: %s",job_post_id,exc)
async def list_channels(self):
try:
return await list_buffer_channels()
except (httpx.HTTPError,BufferError,RuntimeError) as e:
raise HTTPException(status_code=502,detail="Failed to list Buffer channels") from e
async def _restrict_ids_for_requisition_scope(self,current_user):
"""None = unscoped. Empty list = no jobs. Else owned job-post ids."""
from users.permissions import scopes_to_own_requisitions
if not scopes_to_own_requisitions(current_user):
return None
return await JobPosts.ids_for_manager(self.session,current_user.get("id") if current_user else None)
async def fetch_job_posts(self,search=None,top=None,skip=0,ids=None,active_only=True,current_user=None):
restrict=await self._restrict_ids_for_requisition_scope(current_user)
if restrict is not None:
owned={str(i) for i in restrict}
if ids:
ids=[i for i in ids if str(i) in owned]
if not ids:
return [],0
restrict=None
elif not restrict:
return [],0
rows,total=await JobPosts.fetch_job_posts(
self.session,
search=search,
top=top,
skip=skip,
ids=ids,
active_only=active_only,
restrict_ids=restrict,
)
names=await Users.names_by_ids(
self.session,
[uid for r in rows for uid in JobPosts.recruiter_ids_of(r)],
)
return [serialize_job_post(r,names=names) for r in rows],total
async def fetch_job_stats(self,job_post_id=None,search=None,ids=None,top=None,skip=0,active_only=False):
uid=None
if job_post_id not in (None,""):
uid=JobPosts._as_uuid(job_post_id)
if uid is None:
raise HTTPException(status_code=422,detail="job_post_id must be a UUID")
rows,total=await JobPosts.fetch_job_stats(
self.session,
job_post_id=uid,
search=search,
ids=ids,
top=top,
skip=skip,
active_only=active_only,
)
names=await Users.names_by_ids(
self.session,
[uid for r in rows for uid in JobPosts.recruiter_ids_of(r)],
)
data=[serialize_job_stats(r,names=names) for r in rows]
if uid is not None:
if not data:
raise HTTPException(status_code=404,detail="Job post not found")
return data[0],1
return data,total
async def fetch_departments(self,active_only=False):
return await JobPosts.list_departments(self.session,active_only=active_only)
async def fetch_requisition_statuses(self):
return RequisitionStatus.as_list()
async def fetch_status_history(self,job_post_id):
job=await JobPosts.get_job_post_by_id(self.session,job_post_id)
if not job or job.is_deleted:
raise HTTPException(status_code=404,detail="Job post not found")
rows=await JobPostStatusHistory.fetch_by_job(self.session,job_post_id)
names=await Users.names_by_ids(self.session,[r.changed_by for r in rows])
return [
serialize_status_history(r,changed_by_name=names.get(str(r.changed_by)))
for r in rows
]
async def fetch_jobs(self,search=None,department=None,requisition_status=None,
employment_type=None,hiring_manager_id=None,top=None,skip=0,active_only=True,
current_user=None):
restrict=await self._restrict_ids_for_requisition_scope(current_user)
if restrict is not None and not restrict:
return [],0
hm_uid=None
if hiring_manager_id:
hm_uid=JobPosts._as_uuid(hiring_manager_id)
if hm_uid is None:
raise HTTPException(status_code=422,detail="hiring_manager_id must be a UUID")
rows,total=await JobPosts.fetch_job_posts(
self.session,search=search,top=top,skip=skip,active_only=active_only,
department=department,requisition_status=requisition_status,
employment_type=employment_type,hiring_manager_id=hm_uid,
restrict_ids=restrict,
)
names=await Users.names_by_ids(
self.session,
[uid for r in rows for uid in JobPosts.recruiter_ids_of(r)]+[r.hiring_manager_id for r in rows],
)
counts=await Inbox_Messages.counts_by_job_post_ids(self.session,[r.id for r in rows])
return [
serialize_job_row(
r,
names=names,
hiring_manager_name=names.get(str(r.hiring_manager_id)),
applicant_count=counts.get(str(r.id),0),
)
for r in rows
],total
async def _job_row(self,row):
names=await Users.names_by_ids(
self.session,
JobPosts.recruiter_ids_of(row)+[row.hiring_manager_id],
)
return serialize_job_row(
row,
names=names,
hiring_manager_name=names.get(str(row.hiring_manager_id)),
)
async def update_job(self,job_post_id,payload,current_user):
if not current_user:
raise HTTPException(status_code=401,detail="Not authenticated")
existing=await JobPosts.get_job_post_by_id(self.session,job_post_id)
if not existing or existing.is_deleted:
raise HTTPException(status_code=404,detail="Job post not found")
allowed=("title","department","location","employment_type","vacancies",
"salary","experience_min","experience_max","description")
fields={k:payload[k] for k in allowed if k in payload}
if "salary" not in fields and ("salary_min" in payload or "salary_max" in payload):
low=payload.get("salary_min")
high=payload.get("salary_max")
if low is not None and high is not None:
fields["salary"]=f"{low} - {high}"
elif low is not None:
fields["salary"]=str(low)
elif high is not None:
fields["salary"]=str(high)
if "department" in fields and fields["department"] is None:
fields["department"]=""
assignment=Assignment(self.session)
assigned_by=current_user.get("id") if isinstance(current_user,dict) else None
hm_changed=False
rec_changed=False
if "hiring_manager_id" in payload:
raw=payload.get("hiring_manager_id")
if raw is None or raw=="":
fields["hiring_manager_id"]=None
hm_changed=existing.hiring_manager_id is not None
else:
hm=await assignment.require_role(raw,EnumRoles.HIRING_MANAGER,"hiring_manager_id")
fields["hiring_manager_id"]=hm.id
hm_changed=str(existing.hiring_manager_id)!=str(hm.id)
if "requisition_id" in payload:
raw=payload.get("requisition_id")
if raw is None or raw=="":
fields["requisition_id"]=None
else:
from candidate_forms.models import Requisition
req=await Requisition.get_form_by_id(self.session,record_id=str(raw))
if not req:
raise HTTPException(status_code=404,detail="Requisition not found")
held=await JobPosts.get_by_requisition_id(self.session, req.id)
if held and str(held.id)!=str(existing.id):
raise HTTPException(
status_code=409,
detail="This requisition is already linked to a job post",
)
fields["requisition_id"]=req.id
rec_users=None
raw_ids=_payload_recruiter_ids(payload)
if raw_ids is not None:
rec_users=await self._resolve_recruiters(assignment,raw_ids)
fields.update(_recruiter_fields(rec_users))
rec_changed=JobPosts.recruiter_ids_of(existing)!=[str(u.id) for u in rec_users]
if not fields:
raise HTTPException(status_code=400,detail="No fields to update")
try:
row=await JobPosts.update_job_post(self.session,job_post_id,fields)
except IntegrityError as e:
orig=str(getattr(e,"orig",e)).lower()
if "requisition" in orig:
raise HTTPException(
status_code=409,detail="This requisition is already linked to a job post",
) from e
raise
if not row:
raise HTTPException(status_code=404,detail="Job post not found")
if hm_changed:
await assignment.record_job_owner(
job_post_id,fields["hiring_manager_id"],"hiring_manager",assigned_by,
)
if rec_changed:
await assignment.record_job_recruiters(
job_post_id,fields.get("current_recruiter_ids") or [],assigned_by,
)
if hm_changed or rec_changed:
try:
from notifications.views import notify_job_assignment
labels=[]
if hm_changed:
labels.append("hiring manager")
if rec_changed:
labels.append("recruiter")
await notify_job_assignment(
self.session,row,
role_label=" and ".join(labels),
actor_id=assigned_by,
previous_ids=[existing.hiring_manager_id,*JobPosts.recruiter_ids_of(existing)],
)
except Exception as exc:
logger.warning("notification insert skipped: %s",exc)
return await self._job_row(row)
async def delete_job(self,job_post_id,current_user):
if not current_user:
raise HTTPException(status_code=401,detail="Not authenticated")
row=await JobPosts.soft_delete_job_post(self.session,job_post_id)
if not row:
raise HTTPException(status_code=404,detail="Job post not found")
return {"id":str(row.id),"deleted":True}
async def save_job_image(self,job_post_id,filename,content_type,content,current_user):
if not current_user:
raise HTTPException(status_code=401,detail="Not authenticated")
key=_job_image_key(job_post_id)
media=(content_type or "").lower()
if media not in ALLOWED_IMAGE_TYPES:
# Fall back to the filename extension; browsers occasionally send
# application/octet-stream for perfectly valid images.
suffix=Path((filename or "").replace("\\","/")).suffix.lstrip(".").lower()
media=IMAGE_TYPE_BY_EXT.get(suffix)
if not media:
raise HTTPException(status_code=415,detail="Image must be PNG, JPG, WEBP or GIF")
if not content:
raise HTTPException(status_code=400,detail="Empty image upload")
if len(content)>MAX_JOB_IMAGE_BYTES:
raise HTTPException(status_code=413,detail="Image must be under 5 MB")
rows,total=await JobPosts.fetch_job_posts(self.session,ids=[str(key)],active_only=False)
if not total:
raise HTTPException(status_code=404,detail="Job post not found")
raw_user=(current_user or {}).get("id")
uploaded_by=uuid.UUID(str(raw_user)) if raw_user else None
await JobPostImages.upsert(
self.session,key,
content_type=media,
file_name=Path((filename or "").replace("\\","/")).name or None,
data=content,
uploaded_by=uploaded_by,
)
return {"job_post_id":str(key),"has_image":True}
async def get_job_image(self,job_post_id):
key=_job_image_key(job_post_id)
row=await JobPostImages.get(self.session,key)
if not row:
raise HTTPException(status_code=404,detail="No image for this job post")
return row.data,row.content_type
async def set_job_status(self,job_post_id,payload,current_user):
if not current_user:
raise HTTPException(status_code=401,detail="Not authenticated")
status=(payload.get("requisition_status") or "").strip()
parsed=RequisitionStatus.parse(status)
if parsed is None:
raise HTTPException(
status_code=422,
detail=f"requisition_status must be one of {', '.join(RequisitionStatus.values())}",
)
status=parsed.value
actor=current_user.get("id") if isinstance(current_user,dict) else None
previous=None
existing=await JobPosts.get_job_post_by_id(self.session,job_post_id)
if existing:
previous=existing.requisition_status
row=await JobPosts.set_requisition_status(
self.session,job_post_id,status,changed_by=actor,
)
if not row:
raise HTTPException(status_code=404,detail="Job post not found")
if previous!=status:
try:
from notifications.views import notify_job_status
await notify_job_status(
self.session,row,
from_status=previous,to_status=status,actor_id=actor,
)
except Exception as exc:
logger.warning("notification insert skipped: %s",exc)
return await self._job_row(row)