670 lines
29 KiB
Python
670 lines
29 KiB
Python
from typing import Any
|
|
|
|
|
|
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 AtsResults,Inbox_Messages
|
|
from job.assignment.views import Assignment
|
|
from job.candidate.models import Candidates
|
|
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,
|
|
job_people_of,
|
|
create_buffer_post,
|
|
list_buffer_channels,
|
|
local_status,
|
|
normalize_platform,
|
|
optional_skill_hits,
|
|
parse_buffer_datetime,
|
|
render_job_post,
|
|
resolve_channel,
|
|
suggested_source,
|
|
suggested_summary,
|
|
)
|
|
from job.job_post.serializers import serialize_job_post, serialize_job_row, serialize_job_stats, serialize_status_history, serialize_suggested_candidate
|
|
|
|
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):
|
|
"""None = the client did not send recruiters, so leave them alone.
|
|
[] = the client sent an empty list, so remove every recruiter.
|
|
|
|
Checks for the key rather than truthiness: an empty list is falsy, and
|
|
reading it as "not sent" made the last recruiter impossible to remove.
|
|
"""
|
|
if "current_recruiter_ids" not in payload:
|
|
return None
|
|
return payload.get("current_recruiter_ids") or []
|
|
|
|
def _recruiter_fields(users):
|
|
ids=[str(u.id) for u in users]
|
|
return {
|
|
"current_recruiter_ids": ids,
|
|
}
|
|
|
|
|
|
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
|
|
department_id: UUID | 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_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 "",
|
|
"department_id":None,
|
|
"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.
|
|
if payload.get("department_id"):
|
|
department=await self._require_department(payload.get("department_id"))
|
|
fields["department_id"]=department.id
|
|
fields["department"]=department.name
|
|
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,search=None,top=None,limit=None):
|
|
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,search,top,limit)
|
|
names=await Users.names_by_ids(self.session,[r.changed_by for r in rows],search,top,limit)
|
|
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,
|
|
)
|
|
people=await Users.job_people(
|
|
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,
|
|
people=job_people_of(r,people),
|
|
applicant_count=counts.get(str(r.id),0),
|
|
)
|
|
for r in rows
|
|
],total
|
|
|
|
async def fetch_job_profile(self,job_post_id,current_user=None,search=None,top=None,limit=None,skip=None):
|
|
"""Job profile page: the requisition row, its suggested candidates and the
|
|
Suggested / Top Match header stats — one round trip.
|
|
|
|
Suggested = newest ats_results row per person for this job. Profile fields
|
|
and keywords come from that score's candidates row; a user- or form-identity
|
|
score has none, so it borrows the newest completed row for the same email."""
|
|
uid=JobPosts._as_uuid(job_post_id)
|
|
|
|
if uid is None:
|
|
raise HTTPException(status_code=422,detail="job_post_id must be a UUID")
|
|
|
|
# it's a system admin job profile page, so we don't need to restrict the ids
|
|
restrict=await self._restrict_ids_for_requisition_scope(current_user)
|
|
|
|
if restrict is not None and str(uid) not in {str(i) for i in restrict}:
|
|
raise HTTPException(status_code=404,detail="Job post not found")
|
|
|
|
job=await JobPosts.get_job_post_by_id(self.session,uid)
|
|
if not job or job.is_deleted:
|
|
raise HTTPException(status_code=404,detail="Job post not found")
|
|
|
|
people=await Users.job_people(
|
|
self.session,
|
|
JobPosts.recruiter_ids_of(job),
|
|
[job.hiring_manager_id],
|
|
)
|
|
counts=await Inbox_Messages.counts_by_job_post_ids(self.session,[job.id])
|
|
job_payload=serialize_job_row(
|
|
job,
|
|
people=people,
|
|
applicant_count=counts.get(str(job.id),0),
|
|
)
|
|
|
|
rows=await AtsResults.latest_per_candidate_for_job(self.session,job.id,search,top,limit,skip)
|
|
total=await AtsResults.count_suggested_for_job(self.session,job.id,search)
|
|
emails=[user_email or form_email for row,_,user_email,_,form_email,candidate in rows if candidate is None]
|
|
by_email=await Candidates.latest_completed_for_job_by_emails(self.session,job.id,emails)
|
|
candidates=[]
|
|
for row,user_name,user_email,form_name,form_email,candidate in rows:
|
|
email=user_email or form_email
|
|
scored=candidate or by_email.get((email or "").strip().lower())
|
|
candidates.append(serialize_suggested_candidate(
|
|
row,
|
|
name=user_name or form_name,
|
|
email=email,
|
|
candidate=scored,
|
|
source=suggested_source(row.inbox_id,row.form_data_id,scored.source if scored else None),
|
|
optional_matched=optional_skill_hits(job.optional_skills,scored.matched_keywords if scored else []),
|
|
))
|
|
return {"job":job_payload,**suggested_summary(candidates),"total":total,"candidates":candidates}
|
|
|
|
async def _job_row(self,row):
|
|
people=await Users.job_people(
|
|
self.session,
|
|
JobPosts.recruiter_ids_of(row),
|
|
[row.hiring_manager_id],
|
|
)
|
|
return serialize_job_row(row,people=job_people_of(row,people))
|
|
|
|
async def _require_department(self,department_id):
|
|
from department.models import Department
|
|
department=await Department.get_by_id(self.session,department_id)
|
|
if not department:
|
|
raise HTTPException(status_code=404,detail="Department not found")
|
|
return department
|
|
|
|
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"]=""
|
|
if "department_id" in payload:
|
|
if payload.get("department_id"):
|
|
department=await self._require_department(payload.get("department_id"))
|
|
fields["department_id"]=department.id
|
|
fields["department_ref"]=department
|
|
fields["department"]=department.name
|
|
else:
|
|
fields["department_id"]=None
|
|
fields["department_ref"]=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)
|