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): if payload.get("current_recruiter_ids"): raw=payload.get("current_recruiter_ids") return raw 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)