from django.contrib import messages from django.contrib.auth.decorators import login_required from django.db.models import Count, Prefetch, Q from django.shortcuts import get_object_or_404, redirect, render from django.utils import timezone from django.views.decorators.http import require_POST, require_http_methods from accounts.decorators import admin_required from accounts.models import User from ats.forms import ApplicationNoteForm, ApplicationStatusForm, InterviewForm from ats.models import ApplicationEvent, ApplicationNote, Interview from home.pagination import paginate from ats.pipeline_utils import PIPELINE_ORDER, build_job_boards, pipeline_stats from ats.services import change_application_status, explain_match_score, log_event from blog.models import BlogPost from companies.models import Company from home.models import ContactMessage from jobs.models import Job, JobApplication from jobs.attachments import save_job_attachments from employers.access import apply_company_to_job, get_editable_attachment, get_editable_job from employers.forms import AdminJobPostForm from messaging.models import Conversation from messaging.services import get_application_messaging_context from podcast.models import PodcastEpisode from seekers.models import SeekerCategory, SeekerCVRecord, SeekerProfile from seekers.cv_library import get_cv_library_stats from seekers.cv_library_views import cv_library_context, handle_cv_download, handle_single_cv_download from seekers.forms import SeekerCategoryForm from subscriptions.models import CompanySubscription, SeekerSubscription from testimonials.models import Testimonial from .utils import admin_render, paginate_for_admin @login_required @admin_required def platform_dashboard(request): pending_testimonials = Testimonial.objects.filter(is_approved=False).count() unread_messages = ContactMessage.objects.filter(is_read=False).count() stats = { 'users': User.objects.count(), 'seekers': User.objects.filter(role=User.Role.SEEKER).count(), 'employers': User.objects.filter(role=User.Role.COMPANY).count(), 'companies_count': Company.objects.filter(is_active=True).count(), 'jobs': Job.objects.filter(is_active=True).count(), 'total_jobs': Job.objects.count(), 'applications': JobApplication.objects.count(), 'profiles': SeekerProfile.objects.filter(is_public=True).count(), 'cv_library_count': SeekerCVRecord.objects.filter(is_current=True).count(), 'cv_categories_count': SeekerCategory.objects.filter(is_active=True).count(), 'blog_posts': BlogPost.objects.filter(is_published=True).count(), 'testimonials': Testimonial.objects.filter(is_approved=True).count(), 'pending_testimonials': pending_testimonials, 'contact_messages': ContactMessage.objects.count(), 'unread_messages': unread_messages, } stat_cards = [ {'icon': 'fa-users', 'value': stats['users'], 'label': 'Total Users'}, {'icon': 'fa-user-graduate', 'value': stats['seekers'], 'label': 'Job Seekers'}, {'icon': 'fa-building', 'value': stats['companies_count'], 'label': 'Active Companies'}, {'icon': 'fa-briefcase', 'value': stats['jobs'], 'label': 'Active Jobs'}, {'icon': 'fa-file-signature', 'value': stats['applications'], 'label': 'Applications'}, {'icon': 'fa-id-badge', 'value': stats['profiles'], 'label': 'Public Profiles'}, ] secondary_stats = [ {'value': stats['cv_library_count'], 'label': 'Indexed CVs', 'url_name': 'platform_admin:cv_library'}, {'value': stats['cv_categories_count'], 'label': 'CV Categories', 'url_name': 'platform_admin:cv_categories'}, {'value': stats['blog_posts'], 'label': 'Published Posts'}, {'value': stats['testimonials'], 'label': 'Approved Reviews'}, {'value': stats['pending_testimonials'], 'label': 'Pending Reviews'}, {'value': stats['contact_messages'], 'label': 'Contact Messages'}, ] cv_library_stats = get_cv_library_stats() hour = timezone.localtime().hour if hour < 12: greeting = 'Good morning' elif hour < 17: greeting = 'Good afternoon' else: greeting = 'Good evening' return admin_render(request, 'platform_admin/dashboard.html', 'dashboard', **{ 'stats': stats, 'stat_cards': stat_cards, 'secondary_stats': secondary_stats, 'recent_jobs': Job.objects.order_by('-created_at')[:4], 'recent_messages': ContactMessage.objects.order_by('-created_at')[:4], 'recent_users': User.objects.order_by('-date_joined')[:4], 'greeting': greeting, 'cv_library_stats': cv_library_stats, }) def _search_q(request, *fields): q = request.GET.get('q', '').strip() if not q: return Q() clause = Q() for field in fields: clause |= Q(**{f'{field}__icontains': q}) return clause @login_required @admin_required def users_list(request): qs = User.objects.order_by('-date_joined') role = request.GET.get('role', '') if role in dict(User.Role.choices): qs = qs.filter(role=role) qs = qs.filter(_search_q(request, 'username', 'email', 'first_name', 'last_name')) return admin_render(request, 'platform_admin/users.html', 'users', **{ **paginate_for_admin(qs, request), 'role_filter': role, 'role_choices': User.Role.choices, 'page_title': 'Users', 'page_subtitle': 'View registered accounts and roles', }) @login_required @admin_required @require_http_methods(['GET']) def user_detail(request, pk): user_obj = get_object_or_404(User, pk=pk) seeker_profile = SeekerProfile.objects.filter(user=user_obj).first() company = Company.objects.filter(owner=user_obj).first() return admin_render(request, 'platform_admin/user_detail.html', 'users', **{ 'user_obj': user_obj, 'seeker_profile': seeker_profile, 'company': company, 'page_title': user_obj.get_full_name() or user_obj.username, 'page_subtitle': f'@{user_obj.username} · {user_obj.email}', }) @login_required @admin_required def seekers_list(request): qs = SeekerProfile.objects.select_related('user').order_by('-updated_at') if request.GET.get('public') == '1': qs = qs.filter(is_public=True) elif request.GET.get('public') == '0': qs = qs.filter(is_public=False) qs = qs.filter(_search_q(request, 'full_name', 'headline', 'location', 'user__email')) return admin_render(request, 'platform_admin/seekers.html', 'seekers', **{ **paginate_for_admin(qs, request), 'public_filter': request.GET.get('public', ''), 'page_title': 'Seeker Profiles', 'page_subtitle': 'Talent directory profiles', }) @login_required @admin_required def companies_list(request): qs = Company.objects.select_related('owner').order_by('-created_at') if request.GET.get('active') == '1': qs = qs.filter(is_active=True) elif request.GET.get('active') == '0': qs = qs.filter(is_active=False) qs = qs.filter(_search_q(request, 'name', 'industry', 'location', 'owner__email')) return admin_render(request, 'platform_admin/companies.html', 'companies', **{ **paginate_for_admin(qs, request), 'active_filter': request.GET.get('active', ''), 'page_title': 'Companies', 'page_subtitle': 'Employer profiles on the platform', }) @login_required @admin_required def jobs_list(request): qs = Job.objects.order_by('-created_at') if request.GET.get('active') == '1': qs = qs.filter(is_active=True) elif request.GET.get('active') == '0': qs = qs.filter(is_active=False) qs = qs.filter(_search_q(request, 'title', 'company_name', 'location')) return admin_render(request, 'platform_admin/jobs.html', 'jobs', **{ **paginate_for_admin(qs, request, per_page=10), 'active_filter': request.GET.get('active', ''), 'page_title': 'Jobs', 'page_subtitle': 'All job listings on the platform', }) @login_required @admin_required @require_http_methods(['GET', 'POST']) def post_job(request): company = Company.objects.filter(owner=request.user).first() form = AdminJobPostForm( request.POST or None, request.FILES or None, company=company, user=request.user, ) if request.method == 'POST' and form.is_valid(): job = form.save(commit=False) job.posted_by = request.user apply_company_to_job( job, request.user, company, company_name=form.cleaned_data.get('company_name'), ) job.is_active = True job.is_featured = True job.save() files = form.cleaned_data.get('attachment_files') or [] if files: save_job_attachments(job, files) messages.success(request, f'"{job.title}" was published successfully.') return redirect('platform_admin:job_detail', slug=job.slug) return admin_render(request, 'platform_admin/post_job.html', 'post_job', **{ 'form': form, 'page_title': 'Post a Job', 'page_subtitle': 'Create a listing for any company on the platform', }) @login_required @admin_required @require_http_methods(['GET', 'POST']) def edit_job(request, slug): job = get_editable_job(request.user, slug) company = Company.objects.filter(owner=request.user).first() form = AdminJobPostForm( request.POST or None, request.FILES or None, instance=job, company=company, user=request.user, ) if request.method == 'POST' and form.is_valid(): job = form.save(commit=False) apply_company_to_job( job, request.user, company, company_name=form.cleaned_data.get('company_name'), ) job.save() files = form.cleaned_data.get('attachment_files') or [] if files: save_job_attachments(job, files, start_order=job.attachments.count()) messages.success(request, f'"{job.title}" was updated successfully.') return redirect('platform_admin:job_detail', slug=job.slug) return admin_render(request, 'platform_admin/edit_job.html', 'jobs', **{ 'form': form, 'job': job, 'attachments': job.attachments.all(), 'page_title': 'Edit Job', 'page_subtitle': job.title, }) @login_required @admin_required @require_POST def delete_job_attachment(request, pk): attachment = get_editable_attachment(request.user, pk) job_slug = attachment.job.slug attachment.file.delete(save=False) attachment.delete() messages.success(request, 'Attachment removed.') return redirect('platform_admin:edit_job', slug=job_slug) @login_required @admin_required def job_detail(request, slug): job = get_object_or_404( Job.objects.select_related('posted_by', 'category'), slug=slug, ) applications = ( job.applications .select_related('applicant', 'applicant__seeker_profile') .order_by('-match_score', '-created_at')[:5] ) return admin_render(request, 'platform_admin/job_detail.html', 'jobs', **{ 'job': job, 'applications': applications, 'application_count': job.applications.count(), 'page_title': job.title, 'page_subtitle': f'{job.company_name} · {job.location}', }) @login_required @admin_required @require_POST def job_toggle(request, slug): job = get_object_or_404(Job, slug=slug) if job.is_active and job.is_ongoing and job.accepts_applications: messages.error( request, f'"{job.title}" is an ongoing listing and can only be closed by the posting company.', ) return redirect(request.POST.get('next') or 'platform_admin:jobs') job.is_active = not job.is_active job.save(update_fields=['is_active']) state = 'reopened' if job.is_active else 'closed' messages.success(request, f'"{job.title}" {state}.') return redirect(request.POST.get('next') or 'platform_admin:jobs') BOARDS_PER_PAGE = 9 @login_required @admin_required def ats_pipeline(request): base_qs = JobApplication.objects.select_related( 'job', 'applicant', 'applicant__seeker_profile', ).prefetch_related( Prefetch( 'conversation', queryset=Conversation.objects.annotate( unread_count=Count( 'messages', filter=Q(messages__read_at__isnull=True) & ~Q(messages__sender_id=request.user.id), ), ), ), ) qs = base_qs.order_by(*PIPELINE_ORDER) status = request.GET.get('status', '') if status in dict(JobApplication.Status.choices): qs = qs.filter(status=status) current_job = request.GET.get('job', '') if current_job: qs = qs.filter(job__slug=current_job) job_boards = build_job_boards() selected_board = next((b for b in job_boards if b.slug == current_job), None) show_board = not (current_job or status or request.GET.get('q', '').strip() or request.GET.get('view') == 'all') board_page = board_qs = board_range = None if show_board: board_page, board_qs, board_range = paginate(request, job_boards, per_page=BOARDS_PER_PAGE) qs = qs.filter(_search_q( request, 'job__title', 'applicant__email', 'applicant__first_name', 'applicant__last_name', )) _, total_count, status_tabs = pipeline_stats(base_qs) return admin_render(request, 'platform_admin/ats_pipeline.html', 'ats', **{ 'job_boards': job_boards, 'selected_board': selected_board, 'show_board': show_board, 'board_pager': 'platform_admin/partials/pagination.html', 'board_page': board_page, 'board_qs': board_qs, 'board_range': board_range, 'current_job': current_job, **paginate_for_admin(qs, request, per_page=25), 'status_filter': status, 'status_choices': JobApplication.Status.choices, 'status_tabs': status_tabs, 'total_count': total_count, 'page_title': 'ATS Pipeline', 'page_subtitle': 'Review and manage all job applications - highest match scores first', }) @login_required @admin_required def application_detail(request, pk): application = get_object_or_404( JobApplication.objects.select_related('job', 'applicant', 'applicant__seeker_profile'), pk=pk, ) profile = getattr(application.applicant, 'seeker_profile', None) chat = get_application_messaging_context(request.user, application) match_breakdown = explain_match_score(application) if request.method == 'POST': action = request.POST.get('action') if action == 'status': old_status = application.status old_rating = application.rating form = ApplicationStatusForm(request.POST, instance=application) if form.is_valid(): new_status = form.cleaned_data['status'] new_rating = form.cleaned_data.get('rating') if new_status != old_status: change_application_status(application, new_status, request.user, old_status=old_status) if new_rating != old_rating: JobApplication.objects.filter(pk=application.pk).update(rating=new_rating) if new_rating: log_event( application, request.user, ApplicationEvent.EventType.RATING_SET, new_value=str(new_rating), message=f'Rated {new_rating}/5', ) messages.success(request, 'Application updated.') else: messages.error(request, 'Could not update application.') return redirect('platform_admin:application_detail', pk=pk) if action == 'note': form = ApplicationNoteForm(request.POST) if form.is_valid(): note = form.save(commit=False) note.application = application note.author = request.user note.save() log_event( application, request.user, ApplicationEvent.EventType.NOTE_ADDED, message=note.body[:120], ) messages.success(request, 'Note added.') else: messages.error(request, 'Note could not be saved.') return redirect('platform_admin:application_detail', pk=pk) if action == 'interview': form = InterviewForm(request.POST) if form.is_valid(): interview = form.save(commit=False) interview.application = application interview.created_by = request.user interview.save() if application.status != JobApplication.Status.INTERVIEW: change_application_status(application, JobApplication.Status.INTERVIEW, request.user) log_event( application, request.user, ApplicationEvent.EventType.INTERVIEW_SCHEDULED, message=f'Interview on {interview.scheduled_at:%d %b %Y at %H:%M}', ) messages.success(request, 'Interview scheduled.') else: messages.error(request, 'Could not schedule interview.') return redirect('platform_admin:application_detail', pk=pk) return admin_render(request, 'platform_admin/application_detail.html', 'ats', **{ 'application': application, 'profile': profile, 'chat': chat, 'match_breakdown': match_breakdown, 'status_form': ApplicationStatusForm(instance=application), 'note_form': ApplicationNoteForm(), 'interview_form': InterviewForm(), 'notes': application.notes.select_related('author').all(), 'events': application.events.select_related('actor').all(), 'interviews': application.interviews.all(), 'page_title': application.applicant.get_full_name() or application.applicant.email, 'page_subtitle': f'Applied for {application.job.title}', }) @login_required @admin_required def interviews_list(request): qs = Interview.objects.select_related('application__job', 'application__applicant').order_by('-scheduled_at') return admin_render(request, 'platform_admin/interviews.html', 'interviews', **{ **paginate_for_admin(qs, request), 'page_title': 'Interviews', 'page_subtitle': 'Scheduled candidate interviews', }) @login_required @admin_required def notes_list(request): qs = ApplicationNote.objects.select_related('application__job', 'author').order_by('-created_at') return admin_render(request, 'platform_admin/notes.html', 'notes', **{ **paginate_for_admin(qs, request), 'page_title': 'Application Notes', 'page_subtitle': 'Internal notes on candidates', }) @login_required @admin_required def activity_list(request): qs = ApplicationEvent.objects.select_related('application__job', 'actor').order_by('-created_at') return admin_render(request, 'platform_admin/activity.html', 'activity', **{ **paginate_for_admin(qs, request, per_page=30), 'page_title': 'ATS Activity Log', 'page_subtitle': 'Application status and event history', }) @login_required @admin_required def blog_list(request): qs = BlogPost.objects.order_by('-published_at') qs = qs.filter(_search_q(request, 'title', 'author_name', 'category')) return admin_render(request, 'platform_admin/blog.html', 'blog', **{ **paginate_for_admin(qs, request), 'page_title': 'Blog Posts', 'page_subtitle': 'Published and draft articles', }) @login_required @admin_required def podcast_list(request): qs = PodcastEpisode.objects.order_by('-episode_number') return admin_render(request, 'platform_admin/podcast.html', 'podcast', **{ **paginate_for_admin(qs, request), 'page_title': 'Podcast Episodes', 'page_subtitle': 'Audio content library', }) @login_required @admin_required def testimonials_list(request): qs = Testimonial.objects.order_by('-created_at') if request.GET.get('approved') == '1': qs = qs.filter(is_approved=True) elif request.GET.get('approved') == '0': qs = qs.filter(is_approved=False) return admin_render(request, 'platform_admin/testimonials.html', 'testimonials', **{ **paginate_for_admin(qs, request), 'approved_filter': request.GET.get('approved', ''), 'page_title': 'Testimonials', 'page_subtitle': 'Review and approve user stories', }) @login_required @admin_required @require_POST def testimonial_approve(request, pk): testimonial = get_object_or_404(Testimonial, pk=pk) testimonial.is_approved = True testimonial.save(update_fields=['is_approved']) messages.success(request, f'Approved testimonial from {testimonial.full_name}.') return redirect(request.POST.get('next') or 'platform_admin:testimonials') @login_required @admin_required @require_POST def testimonial_reject(request, pk): testimonial = get_object_or_404(Testimonial, pk=pk) testimonial.is_approved = False testimonial.save(update_fields=['is_approved']) messages.success(request, f'Unapproved testimonial from {testimonial.full_name}.') return redirect(request.POST.get('next') or 'platform_admin:testimonials') @login_required @admin_required def messages_list(request): qs = ContactMessage.objects.order_by('-created_at') if request.GET.get('read') == '1': qs = qs.filter(is_read=True) elif request.GET.get('read') == '0': qs = qs.filter(is_read=False) qs = qs.filter(_search_q(request, 'first_name', 'last_name', 'email', 'message')) return admin_render(request, 'platform_admin/messages.html', 'messages', **{ **paginate_for_admin(qs, request), 'read_filter': request.GET.get('read', ''), 'page_title': 'Contact Messages', 'page_subtitle': 'Inbound messages from the contact form', }) @login_required @admin_required def message_detail(request, pk): msg = get_object_or_404(ContactMessage, pk=pk) if not msg.is_read: msg.is_read = True msg.save(update_fields=['is_read']) if request.method == 'POST': msg.is_read = request.POST.get('is_read') == 'on' msg.save(update_fields=['is_read']) messages.success(request, 'Message updated.') return redirect('platform_admin:message_detail', pk=pk) return admin_render(request, 'platform_admin/message_detail.html', 'messages', **{ 'msg': msg, 'page_title': msg.full_name, 'page_subtitle': msg.get_subject_display(), }) @login_required @admin_required def subscriptions_list(request): qs = SeekerSubscription.objects.select_related('user', 'plan').order_by('-started_at') return admin_render(request, 'platform_admin/subscriptions.html', 'subscriptions', **{ **paginate_for_admin(qs, request), 'page_title': 'Seeker Subscriptions', 'page_subtitle': 'Paid membership plans for job seekers', }) @login_required @admin_required def company_subscriptions_list(request): qs = CompanySubscription.objects.select_related('company', 'plan').order_by('-started_at') return admin_render(request, 'platform_admin/company_subscriptions.html', 'company_subscriptions', **{ **paginate_for_admin(qs, request), 'page_title': 'Company Subscriptions', 'page_subtitle': 'Employer recruitment plans', }) @login_required @admin_required def cv_library(request): ctx = cv_library_context(request, admin_mode=True) ctx.update(paginate_for_admin(ctx['records_queryset'], request, per_page=50)) ctx['page_title'] = 'CV Library' ctx['page_subtitle'] = f'{ctx["total_cvs"]} indexed CV(s) - CV Search subscribers only (employers)' ctx['download_url'] = 'platform_admin:cv_library_download' ctx['single_download_url_name'] = 'platform_admin:cv_library_download_one' ctx['cv_library_stats'] = get_cv_library_stats() return admin_render(request, 'platform_admin/cv_library.html', 'cv_library', **ctx) @login_required @admin_required @require_POST def cv_library_download(request): response = handle_cv_download(request, admin_mode=True) if response: return response return redirect('platform_admin:cv_library') @login_required @admin_required def cv_library_download_one(request, pk): return handle_single_cv_download(request, pk, admin_mode=True) @login_required @admin_required def cv_categories(request): categories = SeekerCategory.objects.annotate( cv_count=Count('cv_records', filter=Q(cv_records__is_current=True)), ).order_by('sort_order', 'name') form = SeekerCategoryForm() return admin_render(request, 'platform_admin/cv_categories.html', 'cv_categories', **{ 'categories': categories, 'form': form, 'page_title': 'CV Categories', 'page_subtitle': 'Manage profession categories used to organise the CV library', }) @login_required @admin_required @require_POST def cv_category_create(request): form = SeekerCategoryForm(request.POST) if form.is_valid(): form.save() messages.success(request, f'Category "{form.cleaned_data["name"]}" created.') else: messages.error(request, form.errors.as_text()) return redirect('platform_admin:cv_categories') @login_required @admin_required @require_http_methods(['GET', 'POST']) def cv_category_edit(request, pk): category = get_object_or_404(SeekerCategory, pk=pk) if request.method == 'POST': form = SeekerCategoryForm(request.POST, instance=category) if form.is_valid(): form.save() messages.success(request, 'Category updated.') return redirect('platform_admin:cv_categories') else: form = SeekerCategoryForm(instance=category) return admin_render(request, 'platform_admin/cv_category_edit.html', 'cv_categories', **{ 'form': form, 'category': category, 'page_title': f'Edit category - {category.name}', }) @login_required @admin_required @require_POST def cv_category_toggle(request, pk): category = get_object_or_404(SeekerCategory, pk=pk) category.is_active = not category.is_active category.save(update_fields=['is_active']) state = 'activated' if category.is_active else 'deactivated' messages.success(request, f'Category "{category.name}" {state}.') return redirect('platform_admin:cv_categories')