import logging
from datetime import timedelta
from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.utils import timezone
from django.utils.html import escape
from django.db.models import Q
from apps.users.models import User
from apps.jobs.models import JobDescription, JDRecruiterAssignment
from apps.pipeline.models import JobApplication, InterviewSchedule, InterviewFeedback, Shortlist
from apps.llm.service import call_llm
from .models import MorningEfficiencyLog
logger = logging.getLogger(__name__)
def get_user_action_points(user):
"""
Extracts real-time daily action items and metrics for a user based on their role
(Recruiter, Hiring Manager, Project Manager, or Admin).
"""
today = timezone.now().date()
today_start = timezone.make_aware(timezone.datetime.combine(today, timezone.datetime.min.time()))
today_end = timezone.make_aware(timezone.datetime.combine(today, timezone.datetime.max.time()))
action_items = []
interviews_today = []
pending_approvals = []
user_role = (user.role or "").upper()
# Determine primary responsibilities
is_recruiter = (
user_role == User.Role.RECRUITER
or JDRecruiterAssignment.objects.filter(recruiter=user).exists()
or user.groups.filter(name__icontains="Recruiter").exists()
)
is_hiring_manager = (
user_role in [User.Role.ADMIN, User.Role.INTERVIEWER]
or JobDescription.objects.filter(created_by=user).exists()
or JobDescription.objects.filter(approved_by=user).exists()
or JobDescription.objects.filter(current_approver=user).exists()
)
# 1. Check Today's Scheduled Interviews for this user (either as interviewer or recruiter)
# Search by email or name matching
schedules = InterviewSchedule.objects.filter(
status=InterviewSchedule.Status.SCHEDULED,
scheduled_at__range=(today_start, today_end)
).select_related('application', 'application__candidate', 'application__job')
for sched in schedules:
is_my_interview = False
if sched.interviewer_email and sched.interviewer_email.lower() == user.email.lower():
is_my_interview = True
elif user.full_name and user.full_name.lower() in (sched.interviewer or "").lower():
is_my_interview = True
elif sched.created_by == user:
is_my_interview = True
elif is_recruiter and JDRecruiterAssignment.objects.filter(jd=sched.application.job, recruiter=user).exists():
is_my_interview = True
if is_my_interview:
time_str = sched.scheduled_at.strftime("%I:%M %p")
cand_name = sched.application.candidate.full_name if sched.application.candidate else "Candidate"
job_title = sched.application.job.title if sched.application.job else "Job"
interviews_today.append({
"time": time_str,
"candidate": cand_name,
"job": job_title,
"round": sched.round or "Technical Round",
"location": sched.location or "Online Video Link",
})
action_items.append({
"priority": "URGENT",
"category": "Scheduled Interview",
"title": f"Interview with {cand_name} at {time_str}",
"detail": f"Round: {sched.round or 'Technical'} for position '{job_title}'. Venue: {sched.location or 'Online'}.",
"url": f"/jobs/{sched.application.job_id}/pipeline",
})
# 2. Recruiter Action Points
if is_recruiter:
# Assigned JDs
assigned_jds = JobDescription.objects.filter(
assigned_recruiters=user,
status="Published"
).distinct()
# Un-emailed top candidates
unemailed_apps = JobApplication.objects.filter(
job__in=assigned_jds,
email_status=JobApplication.EmailStatus.NOT_SENT
).count()
if unemailed_apps > 0:
action_items.append({
"priority": "URGENT",
"category": "Candidate Outreach",
"title": f"{unemailed_apps} screened candidates awaiting email outreach",
"detail": "Candidates have been ranked/screened. Send interview invites or screening updates to keep momentum.",
"url": "/candidates",
})
# Candidates in early screening stages needing action
screening_apps = JobApplication.objects.filter(
job__in=assigned_jds,
stage__outcome="IN_PROGRESS"
).count()
if screening_apps > 0:
action_items.append({
"priority": "PENDING",
"category": "Pipeline Sourcing",
"title": f"{screening_apps} active candidates in pipeline for your assigned JDs",
"detail": "Review candidate scores, screening responses, and advance eligible candidates to next rounds.",
"url": "/jobs",
})
# Pending feedback to be submitted for completed interviews in past 3 days
three_days_ago = today_start - timedelta(days=3)
past_interviews = InterviewSchedule.objects.filter(
scheduled_at__range=(three_days_ago, today_start),
application__job__in=assigned_jds
).exclude(application__feedbacks__created_at__gte=three_days_ago)
if past_interviews.exists():
count = past_interviews.count()
action_items.append({
"priority": "URGENT",
"category": "Feedback Collection",
"title": f"{count} interview(s) from past 3 days pending interviewer feedback",
"detail": "Follow up with technical interviewers to collect candidate feedback scores before pipeline stalls.",
"url": "/jobs",
})
# 3. Hiring Manager / Project Manager Action Points
if is_hiring_manager:
# JDs Pending Approval
jds_pending = JobDescription.objects.filter(
Q(approval_status="PENDING_APPROVAL") | Q(current_approver=user) | Q(approval_status="SUBMITTED")
).distinct()
for jd in jds_pending:
pending_approvals.append({
"id": jd.id,
"title": jd.title,
"department": jd.department or "General",
"priority": jd.priority,
})
action_items.append({
"priority": "URGENT",
"category": "JD Approval Required",
"title": f"Approve Job Description: '{jd.title}'",
"detail": f"Priority: {jd.priority} | Department: {jd.department or 'N/A'}. Awaiting your review to publish.",
"url": f"/jobs/{jd.id}",
})
# Shortlists needing review
shortlists = Shortlist.objects.filter(job__created_by=user, is_active=True).count()
if shortlists > 0:
action_items.append({
"priority": "PENDING",
"category": "Candidate Shortlists",
"title": f"{shortlists} candidate shortlist link(s) active for your positions",
"detail": "Review shared candidate profiles and share final round interview decisions with the recruiting team.",
"url": "/jobs",
})
# 4. If no specific action points found, provide general high-efficiency checklist items
if not action_items:
action_items.append({
"priority": "INFO",
"category": "Daily Clean Slate",
"title": "All primary queue items are up to date!",
"detail": "Great job! Spend time today proactively reviewing active job descriptions, refreshing candidate pipelines, or updating screening criteria.",
"url": "/dashboard",
})
return {
"is_recruiter": is_recruiter,
"is_hiring_manager": is_hiring_manager,
"action_items": action_items,
"interviews_today": interviews_today,
"pending_approvals": pending_approvals,
}
def generate_ai_efficiency_insight(user, summary_data):
"""
Uses LLM service (`call_llm`) to generate personalized AI efficiency tips for the morning email.
Includes smart fallbacks if no LLM key is configured.
"""
role_str = "Recruiter & Talent Acquisition Specialist" if summary_data["is_recruiter"] else "Hiring Manager & Project Lead"
action_summary = "\n".join([f"- [{item['priority']}] {item['title']}: {item['detail']}" for item in summary_data["action_items"][:5]])
prompt = f"""You are an elite AI Efficiency & Productivity Coach for Talent Acquisition and Engineering Hiring Teams.
Target User: {user.full_name or user.email} ({role_str})
Date: {timezone.now().strftime('%A, %B %d, %Y')}
User's Pending Action Points Today:
{action_summary}
Task: Write 2 to 3 concise, highly strategic, and practical daily efficiency tips (total under 120 words) tailored to this user's day to help them eliminate hiring friction, automate outreach, and close positions faster. Use crisp bullet points with bullet emojis. No markdown code blocks, just raw formatted text."""
try:
res = call_llm(prompt=prompt, purpose="morning_efficiency_agent", user=user)
if res.get("ok") and res.get("text"):
return res["text"].strip()
except Exception as e:
logger.warning(f"LLM generation for morning efficiency agent failed: {e}")
# Fallback tips tailored to role
if summary_data["is_recruiter"]:
return (
"💡 **Morning Efficiency Tip:** Batch your candidate emails in 30-minute time blocks before noon. "
"Prompt response rates increase by 42% when interview invites are sent before 11:00 AM!"
)
else:
return (
"💡 **Morning Efficiency Tip:** Review pending JD approvals and interview feedback at the start of your day. "
"Quick feedback turnarounds maintain high candidate engagement and prevent top talent from taking competing offers."
)
def build_beautiful_morning_email_html(user, summary_data, ai_insight):
"""
Builds an exceptionally beautiful, responsive, modern HTML email template for the Morning Briefing.
Designed with inline styles, gradients, visual metric badges, and interactive CTA buttons.
"""
today_str = timezone.now().strftime("%A, %B %d, %Y")
display_name = escape(user.full_name or user.email.split("@")[0])
action_count = len(summary_data["action_items"])
interviews_count = len(summary_data["interviews_today"])
approvals_count = len(summary_data["pending_approvals"])
# 1. Action Items HTML Blocks
action_rows_html = ""
for item in summary_data["action_items"]:
priority = item.get("priority", "PENDING")
if priority == "URGENT":
badge_bg = "#fee2e2"
badge_color = "#991b1b"
border_left = "#ef4444"
elif priority == "PENDING":
badge_bg = "#fef3c7"
badge_color = "#92400e"
border_left = "#f59e0b"
else:
badge_bg = "#e0e7ff"
badge_color = "#3730a3"
border_left = "#6366f1"
action_rows_html += f"""
{escape(item.get('category', 'Action Item'))}
{priority}
{escape(item['title'])}
{escape(item['detail'])}
"""
# 2. Today's Interviews HTML Block
interviews_html = ""
if summary_data["interviews_today"]:
rows = ""
for i in summary_data["interviews_today"]:
rows += f"""
| {escape(i['time'])} |
{escape(i['candidate'])} |
{escape(i['job'])} ({escape(i['round'])}) |
"""
interviews_html = f"""
📅 Scheduled Interviews & Meetings Today ({interviews_count})
| Time |
Candidate |
Position & Round |
{rows}
"""
# 3. AI Efficiency Insight Block
formatted_insight = escape(ai_insight).replace("\n", "
")
html_body = f"""\
TA-ATS Morning Efficiency Briefing
âš¡ TA-ATS EFFICIENCY AGENT
Good Morning, {display_name}!
Here is your daily action point briefing for {today_str}
|
|
{action_count}
Action Points
|
{interviews_count}
Interviews Today
|
{approvals_count}
Pending Reviews
|
|
|
🎯 Your High-Priority Action Items
{action_rows_html}
{interviews_html}
🚀
AI Efficiency Strategy of the Day
{formatted_insight}
|
|
TA-ATS Automated Efficiency Agent
This morning briefing was personalized & generated for {escape(user.email)}. You are receiving this because morning summary agents are active for your workspace.
|
|
"""
return html_body
def send_morning_efficiency_email(user, dry_run=False):
"""
Triggers the Morning Efficiency Agent for a single user.
Generates action points, AI strategy tip, email HTML, and sends the email (if dry_run is False).
Creates a MorningEfficiencyLog entry in DB.
"""
if not user.email:
return {"ok": False, "error": "User has no email address."}
summary_data = get_user_action_points(user)
ai_insight = generate_ai_efficiency_insight(user, summary_data)
html_body = build_beautiful_morning_email_html(user, summary_data, ai_insight)
subject = f"[Morning Action Briefing] {timezone.now().strftime('%b %d')}: {len(summary_data['action_items'])} Action Point(s)"
# Plain text version for non-HTML clients
plain_text = f"Good morning {user.full_name or user.email}!\n\nHere are your action items for today:\n\n"
for item in summary_data["action_items"]:
plain_text += f"- [{item['priority']}] {item['title']}: {item['detail']}\n"
plain_text += f"\nAI Efficiency Tip:\n{ai_insight}\n\nLog into TA-ATS to complete your tasks."
status = MorningEfficiencyLog.Status.DRY_RUN if dry_run else MorningEfficiencyLog.Status.FAILED
error_msg = ""
if not dry_run:
try:
msg = EmailMultiAlternatives(
subject=subject,
body=plain_text,
from_email=settings.DEFAULT_FROM_EMAIL,
to=[user.email],
)
msg.attach_alternative(html_body, "text/html")
msg.send(fail_silently=False)
status = MorningEfficiencyLog.Status.SENT
logger.info(f"Morning efficiency email successfully sent to {user.email}")
except Exception as e:
error_msg = str(e)
logger.exception(f"Failed sending morning efficiency email to {user.email}: {e}")
log_entry = MorningEfficiencyLog.objects.create(
user=user,
role=user.role or "USER",
email_sent_to=user.email,
subject=subject,
action_point_count=len(summary_data["action_items"]),
interviews_today_count=len(summary_data["interviews_today"]),
pending_approvals_count=len(summary_data["pending_approvals"]),
email_status=status,
ai_insight=ai_insight,
generated_html=html_body,
error_message=error_msg,
)
return {
"ok": status in [MorningEfficiencyLog.Status.SENT, MorningEfficiencyLog.Status.DRY_RUN],
"log_id": log_entry.id,
"user_id": user.id,
"user_email": user.email,
"role": user.role,
"subject": subject,
"action_point_count": len(summary_data["action_items"]),
"status": status,
"error": error_msg,
"generated_html": html_body,
"ai_insight": ai_insight,
"summary_data": summary_data,
}
def run_morning_efficiency_agent(target_roles=None, dry_run=False):
"""
Runs the morning efficiency agent across all relevant users (Recruiters, Hiring Managers, Project Managers, Admins).
"""
users = User.objects.filter(is_active=True)
if target_roles:
users = users.filter(role__in=target_roles)
results = []
for u in users:
# Only process users with role in RECRUITER, ADMIN, INTERVIEWER or assigned JDs
res = send_morning_efficiency_email(u, dry_run=dry_run)
results.append(res)
return results