525 lines
18 KiB
Python
525 lines
18 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
import httpx
|
|
import logging
|
|
import uuid
|
|
|
|
from notifications.models import EmailConfirmationTokens,Notifications
|
|
from role.models import EnumRoles
|
|
from notifications.plugins import (
|
|
CONFIRM_TOKEN_RESEND_SECONDS,
|
|
CONFIRM_TOKEN_TTL_SECONDS,
|
|
build_confirmation_link,
|
|
compose_token,
|
|
confirmation_expiry,
|
|
generate_token_secret,
|
|
hash_token,
|
|
now_utc,
|
|
render_confirmation_email,
|
|
send_confirmation_mail,
|
|
split_token,
|
|
verify_token,
|
|
)
|
|
from notifications.serializers import (
|
|
serialize_confirmation_request,
|
|
serialize_confirmation_result,
|
|
serialize_notification,
|
|
)
|
|
from users.models import Users
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Candidate-history event_type → in-app kind, title, and profile tab.
|
|
_HISTORY_KIND = {
|
|
"stage.changed": "application",
|
|
"candidate.created": "application",
|
|
"candidate.imported": "application",
|
|
"interview.created": "interview",
|
|
"interview.updated": "interview",
|
|
"calendar.created": "interview",
|
|
"calendar.rescheduled": "interview",
|
|
"calendar.cancelled": "interview",
|
|
"ats.scored": "assessment",
|
|
"form.created": "approval",
|
|
"form.updated": "approval",
|
|
"note.created": "message",
|
|
"note.updated": "message",
|
|
"feedback.created": "message",
|
|
"feedback.updated": "message",
|
|
}
|
|
_HISTORY_TITLE = {
|
|
"stage.changed": "Stage changed",
|
|
"note.created": "Note added",
|
|
"note.updated": "Note updated",
|
|
"feedback.created": "Feedback added",
|
|
"feedback.updated": "Feedback updated",
|
|
"interview.created": "Interview scheduled",
|
|
"interview.updated": "Interview updated",
|
|
"calendar.created": "Calendar event created",
|
|
"calendar.rescheduled": "Interview rescheduled",
|
|
"calendar.cancelled": "Interview cancelled",
|
|
"favorite.changed": "Favorite updated",
|
|
"rating.changed": "Rating updated",
|
|
"candidate.created": "Candidate added",
|
|
"candidate.imported": "Candidate imported",
|
|
"document.uploaded": "Document uploaded",
|
|
"ats.scored": "ATS score ready",
|
|
"form.created": "Form submitted",
|
|
"form.updated": "Form updated",
|
|
}
|
|
_HISTORY_TAB = {
|
|
"stage.changed": "History",
|
|
"interview.created": "Interview",
|
|
"interview.updated": "Interview",
|
|
"calendar.created": "Interview",
|
|
"calendar.rescheduled": "Interview",
|
|
"calendar.cancelled": "Interview",
|
|
"note.created": "Notes",
|
|
"note.updated": "Notes",
|
|
"feedback.created": "Activity",
|
|
"feedback.updated": "Activity",
|
|
"form.created": "Forms",
|
|
"form.updated": "Forms",
|
|
"ats.scored": "Resume",
|
|
"document.uploaded": "History",
|
|
"candidate.created": "History",
|
|
"candidate.imported": "History",
|
|
"favorite.changed": "History",
|
|
"rating.changed": "History",
|
|
}
|
|
|
|
|
|
class Confirmation:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def send_confirmation(self,user):
|
|
"""Issue a fresh token for an already committed Users row and mail the link."""
|
|
await EmailConfirmationTokens.invalidate_tokens_for_user(self.session,str(user.id))
|
|
|
|
secret=generate_token_secret()
|
|
expires_at=confirmation_expiry()
|
|
row=await EmailConfirmationTokens.insert_token(self.session,{
|
|
"user_id":user.id,
|
|
"email":user.email,
|
|
"token_hash":hash_token(secret),
|
|
"expires_at":expires_at,
|
|
})
|
|
|
|
link=build_confirmation_link(compose_token(row.id,secret))
|
|
subject,html=render_confirmation_email(link,CONFIRM_TOKEN_TTL_SECONDS)
|
|
try:
|
|
await send_confirmation_mail(user.email,subject,html)
|
|
except (httpx.HTTPError,RuntimeError) as e:
|
|
await EmailConfirmationTokens.mark_used(self.session,str(row.id))
|
|
raise HTTPException(status_code=502,detail="Failed to send confirmation email") from e
|
|
|
|
return serialize_confirmation_request(user.email,expires_at)
|
|
|
|
async def confirm(self,token):
|
|
record_id,secret=split_token(token)
|
|
if not record_id:
|
|
raise HTTPException(status_code=400,detail="Invalid confirmation link")
|
|
|
|
row=await EmailConfirmationTokens.get_token_by_id(self.session,record_id)
|
|
if not row or not verify_token(secret,row.token_hash):
|
|
raise HTTPException(status_code=400,detail="Invalid confirmation link")
|
|
|
|
user=await Users.get_user_by_id(self.session,str(row.user_id))
|
|
if not user or user.is_deleted:
|
|
raise HTTPException(status_code=404,detail="User not found")
|
|
|
|
# Mail clients, link scanners and the back button all replay this link.
|
|
if row.is_used:
|
|
if row.confirmed_at and user.is_active:
|
|
return serialize_confirmation_result(user,already_confirmed=True)
|
|
raise HTTPException(status_code=400,detail="This confirmation link is no longer valid. Request a new one.")
|
|
|
|
if row.expires_at<=now_utc():
|
|
raise HTTPException(status_code=400,detail="Confirmation link has expired. Request a new one.")
|
|
|
|
if user.is_active:
|
|
await EmailConfirmationTokens.mark_confirmed(self.session,str(row.id))
|
|
return serialize_confirmation_result(user,already_confirmed=True)
|
|
|
|
updated=await Users.update_user(self.session,str(user.id),{"is_active":True})
|
|
await EmailConfirmationTokens.mark_confirmed(self.session,str(row.id))
|
|
return serialize_confirmation_result(updated)
|
|
|
|
async def resend(self,email):
|
|
user=await Users.get_user_by_email(self.session,email)
|
|
if not user or user.is_deleted:
|
|
raise HTTPException(status_code=404,detail="No account found for this email")
|
|
if user.is_active:
|
|
raise HTTPException(status_code=400,detail="This account is already confirmed")
|
|
|
|
active=await EmailConfirmationTokens.get_active_token_by_user(self.session,str(user.id))
|
|
if active:
|
|
age=(now_utc()-active.created_at).total_seconds()
|
|
if age<CONFIRM_TOKEN_RESEND_SECONDS and active.expires_at>now_utc():
|
|
raise HTTPException(status_code=429,detail="Please wait before requesting another confirmation email")
|
|
|
|
return await self.send_confirmation(user)
|
|
|
|
|
|
def _as_uuid(value):
|
|
if value in (None,""):
|
|
return None
|
|
try:
|
|
return uuid.UUID(str(value))
|
|
except (TypeError,ValueError):
|
|
return None
|
|
|
|
|
|
def _user_id(current_user):
|
|
if not current_user or not current_user.get("id"):
|
|
raise HTTPException(status_code=401,detail="Not authenticated")
|
|
uid=_as_uuid(current_user["id"])
|
|
if uid is None:
|
|
raise HTTPException(status_code=401,detail="Invalid user id")
|
|
return uid
|
|
|
|
|
|
def _humanize_event(event_type):
|
|
text = str(event_type or "").replace(".", " ").replace("_", " ").strip()
|
|
return text[:1].upper() + text[1:] if text else "Update"
|
|
|
|
|
|
def _candidate_link(user_id, tab="History"):
|
|
path = f"/candidate/{user_id}"
|
|
if tab:
|
|
return f"{path}?tab={tab}"
|
|
return path
|
|
|
|
|
|
def _job_link(job_post_id, tab=None):
|
|
path = f"/jobs?job={job_post_id}"
|
|
if tab:
|
|
return f"{path}&tab={tab}"
|
|
return path
|
|
|
|
|
|
async def system_admin_ids(session):
|
|
return await Users.ids_by_role_names(session, [EnumRoles.SYSTEM_ADMINISTRATOR.value])
|
|
|
|
|
|
async def job_recruiter_ids(session, job):
|
|
"""Recruiters currently linked to the job post.
|
|
|
|
Uses the live pointer (current_recruiter_id) and open job_assignments
|
|
rows with assignment_role=primary_recruiter.
|
|
"""
|
|
ids = set()
|
|
if job is None:
|
|
return ids
|
|
uid = _as_uuid(getattr(job, "current_recruiter_id", None))
|
|
if uid is not None:
|
|
ids.add(uid)
|
|
from job.assignment.models import JobAssignments
|
|
rows = await JobAssignments.fetch_by_job(
|
|
session, job.id, current_only=True, assignment_role="primary_recruiter",
|
|
)
|
|
for row in rows:
|
|
rid = _as_uuid(row.user_id)
|
|
if rid is not None:
|
|
ids.add(rid)
|
|
return ids
|
|
|
|
|
|
async def job_stakeholder_ids(session, jobs, *, extra_ids=None, include_admins=True):
|
|
"""Recruiter, hiring manager, created_by, plus system admins.
|
|
|
|
`jobs` may be one row or an iterable. Extra ids cover people who just
|
|
left an assignment so they still see the history entry.
|
|
"""
|
|
rows = jobs if isinstance(jobs, (list, tuple, set)) else [jobs]
|
|
ids = set()
|
|
for job in rows:
|
|
if job is None:
|
|
continue
|
|
ids.update(await job_recruiter_ids(session, job))
|
|
for raw in (job.hiring_manager_id, job.created_by):
|
|
uid = _as_uuid(raw)
|
|
if uid is not None:
|
|
ids.add(uid)
|
|
for raw in extra_ids or []:
|
|
uid = _as_uuid(raw)
|
|
if uid is not None:
|
|
ids.add(uid)
|
|
if include_admins:
|
|
ids.update(await system_admin_ids(session))
|
|
return ids
|
|
|
|
|
|
async def notify_users(
|
|
session,
|
|
user_ids,
|
|
*,
|
|
kind,
|
|
title,
|
|
body=None,
|
|
link_path=None,
|
|
inbox_id=None,
|
|
job_post_id=None,
|
|
exclude_ids=None,
|
|
commit=True,
|
|
):
|
|
"""Fan-out one in-app row per recipient. Failures never raise."""
|
|
try:
|
|
exclude = {_as_uuid(x) for x in (exclude_ids or [])}
|
|
exclude.discard(None)
|
|
seen = set()
|
|
payloads = []
|
|
for raw in user_ids or []:
|
|
uid = _as_uuid(raw)
|
|
if uid is None or uid in exclude or uid in seen:
|
|
continue
|
|
seen.add(uid)
|
|
payloads.append({
|
|
"user_id": uid,
|
|
"kind": kind,
|
|
"title": title,
|
|
"body": body,
|
|
"link_path": link_path,
|
|
"inbox_id": inbox_id,
|
|
"job_post_id": job_post_id,
|
|
})
|
|
if not payloads:
|
|
return []
|
|
return await Notifications.insert_many(session, payloads, commit=commit)
|
|
except Exception as exc:
|
|
logger.warning("notification insert skipped: %s", exc)
|
|
return []
|
|
|
|
|
|
async def notify_job_stakeholders(
|
|
session,
|
|
job,
|
|
*,
|
|
kind,
|
|
title,
|
|
body=None,
|
|
link_path=None,
|
|
extra_ids=None,
|
|
exclude_ids=None,
|
|
commit=True,
|
|
):
|
|
if job is None:
|
|
return []
|
|
recipients = await job_stakeholder_ids(session, job, extra_ids=extra_ids)
|
|
return await notify_users(
|
|
session,
|
|
recipients,
|
|
kind=kind,
|
|
title=title,
|
|
body=body,
|
|
link_path=link_path or _job_link(job.id),
|
|
job_post_id=job.id,
|
|
exclude_ids=exclude_ids,
|
|
commit=commit,
|
|
)
|
|
|
|
|
|
async def notify_job_created(session, job, *, actor_id=None):
|
|
"""New requisition: recruiter, hiring manager, created_by, system admins.
|
|
|
|
created_by is included even when they are the actor — that is who asked.
|
|
"""
|
|
if job is None:
|
|
return []
|
|
title = (job.title or "A job post").strip() or "A job post"
|
|
return await notify_job_stakeholders(
|
|
session,
|
|
job,
|
|
kind="approval",
|
|
title="New job post",
|
|
body=f"{title} was created",
|
|
link_path=_job_link(job.id),
|
|
exclude_ids=None,
|
|
commit=True,
|
|
)
|
|
|
|
|
|
async def notify_job_status(session, job, *, from_status, to_status, actor_id=None):
|
|
"""Status history: assigned recruiter, created_by, and every system admin."""
|
|
if job is None:
|
|
return []
|
|
title = (job.title or "A job post").strip() or "A job post"
|
|
if from_status:
|
|
body = f"{title} moved from {from_status} to {to_status}"
|
|
heading = "Requisition closed" if to_status == "closed" else "Requisition updated"
|
|
else:
|
|
body = f"{title} is now {to_status}"
|
|
heading = "Requisition updated"
|
|
recipients = await job_recruiter_ids(session, job)
|
|
created = _as_uuid(job.created_by)
|
|
if created is not None:
|
|
recipients.add(created)
|
|
recipients.update(await system_admin_ids(session))
|
|
return await notify_users(
|
|
session,
|
|
recipients,
|
|
kind="approval",
|
|
title=heading,
|
|
body=body,
|
|
link_path=_job_link(job.id, tab="history"),
|
|
job_post_id=job.id,
|
|
exclude_ids=None,
|
|
commit=True,
|
|
)
|
|
|
|
|
|
async def notify_job_assignment(session, job, *, role_label, actor_id=None, previous_ids=None):
|
|
if job is None:
|
|
return []
|
|
title = (job.title or "A job post").strip() or "A job post"
|
|
return await notify_job_stakeholders(
|
|
session,
|
|
job,
|
|
kind="approval",
|
|
title="Job assignment updated",
|
|
body=f"{title}: {role_label} changed",
|
|
link_path=_job_link(job.id, tab="history"),
|
|
extra_ids=previous_ids,
|
|
exclude_ids=[actor_id] if actor_id else None,
|
|
commit=True,
|
|
)
|
|
|
|
|
|
async def _job_ids_for_candidate_event(session, *, user_id, inbox_id, manual_upload_candidate_id):
|
|
"""Resolve job posts for a history row without importing candidate views."""
|
|
from inbox.models import Inbox
|
|
from job.candidate.models import Manual_UPLOAD_CANDIDATE
|
|
|
|
ids = set()
|
|
if inbox_id is not None:
|
|
link = await Inbox.get_inbox_with_message(session, inbox_id)
|
|
msg = getattr(link, "messages", None) if link is not None else None
|
|
jid = getattr(msg, "assigned_job_post_id", None) if msg is not None else None
|
|
if jid:
|
|
ids.add(jid)
|
|
if manual_upload_candidate_id is not None:
|
|
manual = await Manual_UPLOAD_CANDIDATE.get_by_id(session, manual_upload_candidate_id)
|
|
if manual and manual.job_post_id:
|
|
ids.add(manual.job_post_id)
|
|
scoped = inbox_id is not None or manual_upload_candidate_id is not None
|
|
if ids or user_id is None or scoped:
|
|
return ids
|
|
rows = await Inbox.get_candidate_profile(session=session, user_id=user_id, limit=1000, offset=0)
|
|
records = rows if isinstance(rows, list) else ([rows] if rows else [])
|
|
for rec in records:
|
|
msg = getattr(rec, "messages", None)
|
|
jid = getattr(msg, "assigned_job_post_id", None) if msg is not None else None
|
|
if jid:
|
|
ids.add(jid)
|
|
manual = await Manual_UPLOAD_CANDIDATE.get_by_user_id(session, user_id)
|
|
if manual and manual.job_post_id:
|
|
ids.add(manual.job_post_id)
|
|
return ids
|
|
|
|
|
|
async def notify_candidate_history(
|
|
session,
|
|
row,
|
|
*,
|
|
inbox_id=None,
|
|
manual_upload_candidate_id=None,
|
|
commit=True,
|
|
):
|
|
"""One in-app row per job stakeholder for a candidate_history write."""
|
|
if row is None:
|
|
return []
|
|
try:
|
|
from job.job_post.models import JobPosts
|
|
|
|
event_type = row.event_type
|
|
names = await Users.names_by_ids(session, [row.user_id])
|
|
candidate_name = names.get(str(row.user_id)) or "A candidate"
|
|
heading = _HISTORY_TITLE.get(event_type) or _humanize_event(event_type)
|
|
kind = _HISTORY_KIND.get(event_type, "system")
|
|
tab = _HISTORY_TAB.get(event_type, "History")
|
|
if row.description:
|
|
detail = row.description
|
|
elif row.from_value and row.to_value:
|
|
detail = f"{row.from_value} → {row.to_value}"
|
|
elif row.to_value:
|
|
detail = str(row.to_value)
|
|
else:
|
|
detail = heading
|
|
body = f"{candidate_name}: {detail}"
|
|
link_path = _candidate_link(row.user_id, tab=tab)
|
|
|
|
job_ids = await _job_ids_for_candidate_event(
|
|
session,
|
|
user_id=row.user_id,
|
|
inbox_id=inbox_id if inbox_id is not None else row.inbox_id,
|
|
manual_upload_candidate_id=(
|
|
manual_upload_candidate_id
|
|
if manual_upload_candidate_id is not None
|
|
else row.manual_upload_candidate_id
|
|
),
|
|
)
|
|
jobs = []
|
|
if job_ids:
|
|
jobs = await JobPosts.get_by_ids(session, list(job_ids), active_only=False)
|
|
if jobs:
|
|
recipients = await job_stakeholder_ids(session, jobs, include_admins=True)
|
|
else:
|
|
recipients = set(await system_admin_ids(session))
|
|
|
|
exclude = [row.user_id]
|
|
if row.actor_id:
|
|
exclude.append(row.actor_id)
|
|
job_post_id = jobs[0].id if jobs else None
|
|
inbox_value = row.inbox_id if row.inbox_id is not None else inbox_id
|
|
return await notify_users(
|
|
session,
|
|
recipients,
|
|
kind=kind,
|
|
title=heading,
|
|
body=body,
|
|
link_path=link_path,
|
|
inbox_id=inbox_value,
|
|
job_post_id=job_post_id,
|
|
exclude_ids=exclude,
|
|
commit=commit,
|
|
)
|
|
except Exception as exc:
|
|
logger.warning("notification insert skipped: %s", exc)
|
|
return []
|
|
|
|
|
|
class Notification:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def get_notifications(self,current_user,unread_only=False,top=None,skip=0):
|
|
rows,total,unread=await Notifications.fetch_notifications(
|
|
self.session,
|
|
user_id=_user_id(current_user),
|
|
unread_only=bool(unread_only),
|
|
top=top,
|
|
skip=skip or 0,
|
|
)
|
|
return [serialize_notification(r) for r in rows],total,unread
|
|
|
|
async def mark_read(self,record_id,current_user):
|
|
row=await Notifications.mark_read(
|
|
self.session,record_id,user_id=_user_id(current_user)
|
|
)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Notification not found")
|
|
return serialize_notification(row)
|
|
|
|
async def mark_all_read(self,current_user):
|
|
count=await Notifications.mark_all_read(self.session,_user_id(current_user))
|
|
return {"updated": count}
|
|
|
|
async def delete_notification(self,record_id,current_user):
|
|
row=await Notifications.soft_delete_notification(
|
|
self.session,record_id,user_id=_user_id(current_user)
|
|
)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Notification not found")
|
|
return {"id": str(row.id),"deleted": True}
|