from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.utils.html import escape
from django.utils import timezone
import httpx
import logging
from .models import Notification
logger = logging.getLogger(__name__)
def send_email_notification(
*,
to_emails,
subject,
heading,
intro,
details=None,
cta_label=None,
cta_url=None,
secondary_cta_label=None,
secondary_cta_url=None,
outro="Thank you.",
fail_silently=None,
):
"""Send a clean, responsive HTML (+plain-text) transactional email.
Reusable across workflow notifications (JD approval request, approval,
rejection, etc.) — pass structured content and it renders a consistent,
professional template.
Args:
to_emails: a single email string or a list of recipient addresses.
subject: email subject line.
heading: greeting/heading shown at the top of the body (e.g. "Hello Jane,").
intro: one or two sentences introducing the message.
details: optional list of (label, value) tuples rendered as a table.
cta_label / cta_url: optional call-to-action button.
secondary_cta_label / secondary_cta_url: optional second, outlined
button rendered beside the primary one (e.g. a "View Candidate"
link alongside "View Job Description").
outro: closing line (default "Thank you.").
fail_silently: when None, auto-detects the console backend.
Never raises — logs any failure and returns True/False so callers can fire
it as a non-critical side effect without guarding every call site.
"""
recipients = [to_emails] if isinstance(to_emails, str) else list(to_emails or [])
recipients = [e for e in recipients if e]
if not recipients:
logger.info("send_email_notification skipped: no recipients for '%s'.", subject)
return False
if fail_silently is None:
fail_silently = "console" in (settings.EMAIL_BACKEND or "")
details = details or []
# ---- Plain-text part (always included for accessibility / non-HTML clients).
text_lines = [heading, "", intro, ""]
for label, value in details:
text_lines.append(f"{label}: {value}")
if cta_url:
text_lines += ["", f"{cta_label or 'Open'}: {cta_url}"]
if secondary_cta_url:
text_lines += ["", f"{secondary_cta_label or 'Open'}: {secondary_cta_url}"]
text_lines += ["", outro]
text_body = "\n".join(text_lines)
# ---- HTML part: table-based layout with inline styles for email-client
# compatibility; max-width + fluid width keeps it responsive on mobile.
rows_html = "".join(
f"""
| {escape(str(label))} |
{escape(str(value))} |
"""
for label, value in details
)
details_block = (
f''
if rows_html
else ""
)
# Buttons sit in one row of a presentation table so Outlook keeps them
# side by side; the spacer cell provides the gap without relying on margins.
primary_button = (
f"""
{escape(cta_label or 'View')}
| """
if cta_url
else ""
)
secondary_button = (
f"""
{escape(secondary_cta_label or 'View')}
| """
if secondary_cta_url
else ""
)
spacer = ' | ' if primary_button and secondary_button else ""
cta_block = (
f"""
{primary_button}{spacer}{secondary_button}
"""
if primary_button or secondary_button
else ""
)
html_body = f"""\
|
TA-ATS
|
|
{escape(heading)}
{escape(intro)}
{details_block}
{cta_block}
{escape(outro)}
|
|
This is an automated message from TA-ATS. Please do not reply.
|
|
"""
try:
msg = EmailMultiAlternatives(
subject=subject,
body=text_body,
from_email=settings.DEFAULT_FROM_EMAIL,
to=recipients,
)
msg.attach_alternative(html_body, "text/html")
msg.send(fail_silently=fail_silently)
logger.info("Email '%s' sent to %d recipient(s).", subject, len(recipients))
return True
except Exception as e: # noqa: BLE001 — email must never break the caller's workflow
logger.exception("Failed to send email '%s' to %s: %s", subject, recipients, e)
return False
def send_notification(user, title, message, type_code, metadata=None):
"""Save a database Notification and broadcast it via the Gateway's WebSocket."""
notification = Notification.objects.create(
user=user,
title=title,
message=message,
type=type_code,
metadata=metadata or {},
)
# Relay notification to FastAPI Gateway for live WebSocket dispatch
try:
gateway_url = getattr(settings, "GATEWAY_URL", "http://localhost:8000")
broadcast_endpoint = f"{gateway_url}/api/v1/gateway/broadcast"
headers = {
"Authorization": f"Bearer {settings.INTERNAL_API_SECRET}"
}
payload = {
"event": type_code,
"notification": {
"id": notification.id,
"user_id": user.id,
"user_email": user.email,
"title": title,
"message": message,
"type": type_code,
"is_read": False,
"metadata": metadata or {},
"created_at": notification.created_at.isoformat(),
}
}
httpx.post(broadcast_endpoint, json=payload, headers=headers, timeout=1.0)
except Exception as e:
logger.error(f"Failed to broadcast live notification: {e}")
return notification
def broadcast_event(event_type, payload=None):
"""Broadcast an arbitrary event to the Gateway WebSocket."""
try:
gateway_url = getattr(settings, "GATEWAY_URL", "http://localhost:8000")
broadcast_endpoint = f"{gateway_url}/api/v1/gateway/broadcast"
headers = {
"Authorization": f"Bearer {settings.INTERNAL_API_SECRET}"
}
httpx.post(
broadcast_endpoint,
json={"event": event_type, "data": payload or {}},
headers=headers,
timeout=1.0
)
except Exception as e:
logger.error(f"Failed to broadcast event {event_type}: {e}")
def send_jd_assignment_email(*, recruiter, jd, assigned_by=None, assigned_at=None):
"""Send an individual email notification to a recruiter when assigned a JD.
Modular and reusable across recruiter assignment workflow events.
Never raises — logs any failure per recruiter.
"""
to_email = getattr(recruiter, "email", "") or ""
if not to_email:
logger.warning("send_jd_assignment_email skipped: no email for recruiter ID %s", getattr(recruiter, "id", None))
return False
recruiter_name = (getattr(recruiter, "full_name", "") or "").strip() or to_email
pm = assigned_by or getattr(jd, "created_by", None)
pm_name = (getattr(pm, "full_name", "") or "").strip() or getattr(pm, "email", "") or "Project Manager"
pm_email = getattr(pm, "email", "") or "—"
project = jd.client.name if getattr(jd, "client", None) else ""
department = (getattr(jd, "department", "") or "").strip()
assigned_time = assigned_at or timezone.now()
assigned_on = timezone.localtime(assigned_time).strftime("%d %b %Y, %I:%M %p")
status_str = "Assigned to Recruiter"
# The internal JD ID is deliberately kept out of this email — not useful
# to the recruiter and the email may be forwarded outside the org.
details = [
("Project Manager", pm_name),
("Email", pm_email),
("JD Title", jd.title),
]
if project:
details.append(("Project", project))
if department:
details.append(("Department", department))
details.extend([
("Assigned On", assigned_on),
("Status", status_str),
])
frontend_url = getattr(settings, "FRONTEND_URL", "http://localhost:3000")
view_url = f"{frontend_url}/jobs/{jd.id}" if getattr(jd, "id", None) else f"{frontend_url}/jobs"
try:
return send_email_notification(
to_emails=to_email,
subject="New Job Description Assigned to You",
heading=f"Hello {recruiter_name},",
intro="You have been assigned a new Job Description by a Project Manager.",
details=details,
cta_label="View JD",
cta_url=view_url,
outro="Please log in to the system to review the Job Description and start the recruitment process.",
)
except Exception as e:
logger.exception("Failed to send JD assignment email to %s: %s", to_email, e)
return False
def _candidate_display_name(candidate):
"""Best available human name for a candidate, falling back to their email."""
name = " ".join(
p for p in [
(getattr(candidate, "first_name", "") or "").strip(),
(getattr(candidate, "last_name", "") or "").strip(),
] if p
).strip()
return name or (getattr(candidate, "email", "") or "").strip() or "Candidate"
def notify_recruiters_of_application(application, applied_at=None):
"""Email every recruiter assigned to a JD that a candidate has applied.
Called from the candidate-initiated apply flows (careers portal apply and
register-and-apply). Recruiter-initiated additions to a pipeline (bulk
import, "add candidate to job") deliberately do NOT trigger this — the
recruiter already knows.
Never raises: a failed email must not roll back or fail the application the
candidate just submitted. Returns the number of emails sent.
"""
try:
job = getattr(application, "job", None)
candidate = getattr(application, "candidate", None)
if not job or not candidate:
logger.warning(
"notify_recruiters_of_application skipped: application %s has no job/candidate.",
getattr(application, "id", None),
)
return 0
recruiters = list(job.assigned_recruiters.all())
recipients = [r for r in recruiters if (getattr(r, "email", "") or "").strip()]
if not recipients:
# Common and not an error: JDs can be published before assignment.
logger.info(
"No assigned recruiter with an email for JD %s — application %s email skipped.",
job.id, application.id,
)
return 0
candidate_name = _candidate_display_name(candidate)
candidate_email = (getattr(candidate, "email", "") or "").strip() or "—"
applied_time = applied_at or getattr(application, "created_at", None) or timezone.now()
applied_on = timezone.localtime(applied_time).strftime("%d %b %Y, %I:%M %p")
frontend_url = (getattr(settings, "FRONTEND_URL", "") or "http://localhost:3000").rstrip("/")
jd_url = f"{frontend_url}/jobs/{job.id}"
candidate_url = f"{frontend_url}/candidates/{candidate.id}"
# Job ID is deliberately NOT shown: the internal JD reference isn't
# useful to the recruiter and the email may be forwarded outside the org.
details = [
("Candidate Name", candidate_name),
("Candidate Email", candidate_email),
("Applied On", applied_on),
("Job Title", job.title),
]
if getattr(job, "client", None):
details.append(("Client", job.client.name))
if (getattr(job, "location", "") or "").strip():
details.append(("Location", job.location))
sent = 0
# Send individually so each recruiter's greeting is personalised and one
# bad address can't suppress delivery to the others.
for recruiter in recipients:
recruiter_name = (getattr(recruiter, "full_name", "") or "").strip() or recruiter.email
ok = send_email_notification(
to_emails=recruiter.email,
subject=f"New Candidate Application for {job.title}",
heading=f"Hello {recruiter_name},",
intro="A new candidate has applied for one of your assigned job openings.",
details=details,
cta_label="View Job Description",
cta_url=jd_url,
secondary_cta_label="View Candidate",
secondary_cta_url=candidate_url,
outro="Please review the application at your earliest convenience.",
)
sent += int(bool(ok))
# In-app + live notification alongside the email, matching how the
# rest of the app surfaces workflow events. Guarded separately so a
# DB/WebSocket problem can't stop the remaining recruiters' emails.
try:
send_notification(
recruiter,
title=f"New application for {job.title}",
message=f"{candidate_name} applied for {job.title}.",
type_code="CANDIDATE_APPLIED",
metadata={
"job_id": job.id,
"application_id": application.id,
"candidate_id": candidate.id,
},
)
except Exception: # noqa: BLE001
logger.exception(
"In-app application notification failed for recruiter %s", recruiter.email
)
return sent
except Exception as e: # noqa: BLE001 — must never break the apply flow
logger.exception(
"notify_recruiters_of_application failed for application %s: %s",
getattr(application, "id", None), e,
)
return 0
def notify_candidate_of_application(application, applied_at=None):
"""Email the candidate confirming their job application was received.
Called from the candidate-initiated apply flow (careers portal apply),
once, right after the application row is first created. A repeated
apply attempt on an existing application does NOT re-trigger this — the
caller only invokes it on `created is True`, mirroring
`notify_recruiters_of_application`.
Handles the edge cases a candidate-facing confirmation must survive
without ever failing the apply request itself:
* missing job/candidate link on the application row,
* candidate with no registered email (nothing to send to),
* job with no client/company assigned (omitted from the email rather
than showing a blank "at" clause),
* job missing its public_id (falls back to the generic careers URL
instead of building a broken link),
* SMTP/network failures (caught here AND inside
`send_email_notification`, which never raises).
Never raises. Returns True if the email was sent, False otherwise.
"""
try:
job = getattr(application, "job", None)
candidate = getattr(application, "candidate", None)
if not job or not candidate:
logger.warning(
"notify_candidate_of_application skipped: application %s has no job/candidate.",
getattr(application, "id", None),
)
return False
candidate_email = (getattr(candidate, "email", "") or "").strip()
if not candidate_email:
logger.warning(
"notify_candidate_of_application skipped: candidate %s has no registered email (application %s).",
getattr(candidate, "id", None), getattr(application, "id", None),
)
return False
candidate_name = _candidate_display_name(candidate)
job_title = (getattr(job, "title", "") or "").strip() or "the role you applied for"
company_name = (job.client.name if getattr(job, "client", None) else "") or ""
company_name = company_name.strip()
applied_time = applied_at or getattr(application, "created_at", None) or timezone.now()
applied_on = timezone.localtime(applied_time).strftime("%d %b %Y, %I:%M %p")
# The public careers job-detail page resolves by the unguessable
# `public_id` UUID (see PublicJobDetailView.get_object) — plain
# integer ids are deliberately rejected there to stop job enumeration,
# so the link MUST use public_id, not job.id.
frontend_url = (getattr(settings, "FRONTEND_URL", "") or "http://localhost:3000").rstrip("/")
job_public_id = getattr(job, "public_id", None)
job_url = f"{frontend_url}/careers/jobs/{job_public_id}" if job_public_id else frontend_url
at_company = f" at {company_name}" if company_name else ""
details = [("Position", job_title)]
if company_name:
details.append(("Company", company_name))
details.append(("Applied On", applied_on))
ok = send_email_notification(
to_emails=candidate_email,
subject=f"Application Received — {job_title}{at_company}",
heading=f"Hello {candidate_name},",
intro=(
f"Thank you for applying for the {job_title} role{at_company}. "
"This email confirms that we have successfully received your application."
),
details=details,
cta_label="View Job Posting",
cta_url=job_url,
outro=(
"Our recruitment team will carefully review your application and will "
"reach out to you directly if you are shortlisted for the next steps. "
"We appreciate your interest and thank you for taking the time to apply."
),
)
if not ok:
logger.warning(
"notify_candidate_of_application: send failed for application %s (candidate %s).",
getattr(application, "id", None), getattr(candidate, "id", None),
)
return ok
except Exception as e: # noqa: BLE001 — must never break the apply flow
logger.exception(
"notify_candidate_of_application failed for application %s: %s",
getattr(application, "id", None), e,
)
return False