146 lines
4.4 KiB
Python
146 lines
4.4 KiB
Python
"""Offer helpers — compensation field list, offer-email template, Teams mail send.
|
|
|
|
Pure module: no FastAPI imports and no HTTPException.
|
|
|
|
`send_offer_mail` intentionally duplicates
|
|
`notifications.plugins.send_confirmation_mail` rather than importing it: that
|
|
helper is domain-named, and each domain owns its own mail copy and env reads.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
|
|
from candidate_forms.plugins import FORM_READY_STATUSES
|
|
|
|
load_dotenv()
|
|
|
|
TEAMS_MAIL_API_URL = os.getenv("TEAMS_MAIL_API_URL")
|
|
TEAMS_API_TOKEN = os.getenv("TEAMS_API_TOKEN")
|
|
MAIL_ACCEPTED_STATUSES = {200, 202}
|
|
|
|
INTERVIEW_PLUS = FORM_READY_STATUSES
|
|
OFFER_SUBJECT = "Offer from UtopiaBrands Recruitement team"
|
|
|
|
|
|
def non_validation_values():
|
|
fields=("base_salary","currency","salary_period","signing_bonus","annual_bonus_pct",
|
|
"equity_units","equity_instrument","start_date","expiry_date",
|
|
"cadre","gross_salary_in_words","subsidized_services","probation_period",
|
|
"notice_period","work_location","work_timings")
|
|
return fields
|
|
|
|
|
|
def email_key(value) -> str:
|
|
return (value or "").strip().lower()
|
|
|
|
|
|
def stage_value(value) -> str:
|
|
if value is None:
|
|
return ""
|
|
return (value.value if hasattr(value,"value") else str(value)).strip().upper()
|
|
|
|
|
|
def is_interview_plus(value) -> bool:
|
|
return stage_value(value) in INTERVIEW_PLUS
|
|
|
|
|
|
def parse_offer_datetime(value):
|
|
if value in (None,""):
|
|
return None
|
|
if isinstance(value,datetime):
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value
|
|
text=str(value).strip()
|
|
if not text:
|
|
return None
|
|
if text.endswith("Z"):
|
|
text=text[:-1]+"+00:00"
|
|
try:
|
|
parsed=datetime.fromisoformat(text)
|
|
except ValueError:
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=timezone.utc)
|
|
return parsed
|
|
|
|
|
|
def _period_label(period) -> str:
|
|
raw=(period or "year").strip().lower()
|
|
if raw in ("annual","year"):
|
|
return "per year"
|
|
if raw=="month":
|
|
return "per month"
|
|
if raw=="hour":
|
|
return "per hour"
|
|
return raw
|
|
|
|
|
|
def _money_line(amount, currency) -> str:
|
|
cur=(currency or "USD").strip() or "USD"
|
|
try:
|
|
n=float(amount)
|
|
pretty=f"{n:,.0f}" if n==int(n) else f"{n:,.2f}"
|
|
except (TypeError,ValueError):
|
|
pretty=str(amount)
|
|
return f"{cur} {pretty}"
|
|
|
|
|
|
def render_offer_email(
|
|
candidate_name,
|
|
base_salary,
|
|
currency="USD",
|
|
salary_period="year",
|
|
*,
|
|
annual_bonus_pct=None,
|
|
signing_bonus=None,
|
|
equity_units=None,
|
|
equity_instrument=None,
|
|
) -> tuple[str, str]:
|
|
name=(candidate_name or "").strip() or "Candidate"
|
|
salary=_money_line(base_salary,currency)
|
|
period=_period_label(salary_period)
|
|
html=(
|
|
f"<p>Dear {name},</p>"
|
|
"<p>We are pleased to extend an offer of employment from UtopiaBrands.</p>"
|
|
f"<p>Base salary: <strong>{salary} {period}</strong>.</p>"
|
|
)
|
|
if annual_bonus_pct not in (None,""):
|
|
html+=f"<p>Annual bonus: <strong>{annual_bonus_pct}%</strong>.</p>"
|
|
if signing_bonus not in (None,""):
|
|
html+=f"<p>Signing bonus: <strong>{_money_line(signing_bonus,currency)}</strong>.</p>"
|
|
if equity_units not in (None,"",0):
|
|
instrument=(equity_instrument or "RSU").strip() or "RSU"
|
|
html+=f"<p>Equity: <strong>{equity_units} {instrument}</strong>.</p>"
|
|
html+="<p>Please reply to this email if you have questions about the offer.</p>"
|
|
return OFFER_SUBJECT,html
|
|
|
|
|
|
async def send_offer_mail(to_email: str, subject: str, html: str) -> None:
|
|
if not TEAMS_MAIL_API_URL or not TEAMS_API_TOKEN:
|
|
raise RuntimeError("TEAMS_MAIL_API_URL and TEAMS_API_TOKEN must be set")
|
|
fields = [
|
|
("subject", (None, subject)),
|
|
("body", (None, html)),
|
|
("content_type", (None, "html")),
|
|
("save_to_sent_items", (None, "false")),
|
|
("to", (None, to_email)),
|
|
]
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
response = await client.post(
|
|
TEAMS_MAIL_API_URL,
|
|
files=fields,
|
|
headers={"Authorization": f"Bearer {TEAMS_API_TOKEN}"},
|
|
)
|
|
if response.status_code not in MAIL_ACCEPTED_STATUSES:
|
|
raise httpx.HTTPStatusError(
|
|
response.text,
|
|
request=response.request,
|
|
response=response,
|
|
)
|