from __future__ import annotations import os import platform from functools import wraps from django import get_version as django_version from django.contrib import messages from django.contrib.auth import get_user_model, login from django.contrib.auth.decorators import login_required from django.contrib.auth.views import ( LoginView, PasswordResetCompleteView, PasswordResetConfirmView, PasswordResetDoneView, PasswordResetView, ) from django.core.exceptions import PermissionDenied from django.core.mail import send_mail from django.db import transaction from django.db.models import Count, Q from django.http import HttpRequest, HttpResponse from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse, reverse_lazy from django.utils import timezone from django.utils.text import slugify from .forms import ( BusinessOnboardingForm, BusinessSettingsForm, JobIntakeForm, ProfileSettingsForm, ProofCardForm, PublicFeedbackForm, SignUpForm, TeamMemberInviteForm, TrustForgeAuthenticationForm, TrustForgePasswordResetForm, TrustForgeSetPasswordForm, ) from .models import Business, BusinessMembership, Customer, Feedback, Job, JobMedia, ProofCard, ReviewRequest User = get_user_model() ACTIVE_BUSINESS_SESSION_KEY = 'trustforge_active_business_id' POSITIVE_EXPERIENCES = {Feedback.Experience.GREAT, Feedback.Experience.GOOD} RATING_MAP = { Feedback.Experience.GREAT: 5, Feedback.Experience.GOOD: 4, Feedback.Experience.OKAY: 3, Feedback.Experience.BAD: 2, } def _theme_context() -> dict: return { 'project_name': 'TrustForge', 'project_description': ( 'TrustForge turns completed service jobs into proof cards, testimonials, and conversion assets ' 'for contractors, HVAC teams, roofers, plumbers, and local service businesses.' ), 'project_image_url': os.getenv('PROJECT_IMAGE_URL', ''), } class ThemedAuthContextMixin: auth_page_title = 'TrustForge Account' auth_page_description = 'Secure access to your proof pipeline, review engine, and published proof assets.' def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context.update(_theme_context()) context.setdefault('auth_page_title', self.auth_page_title) context.setdefault('auth_page_description', self.auth_page_description) return context class TrustForgeLoginView(ThemedAuthContextMixin, LoginView): template_name = 'registration/login.html' authentication_form = TrustForgeAuthenticationForm redirect_authenticated_user = True auth_page_title = 'Welcome back to your trust engine' auth_page_description = 'Sign in to manage completed jobs, review requests, proof cards, and every asset that helps the next customer say yes.' def get_success_url(self): redirect_url = self.get_redirect_url() if redirect_url: return redirect_url if _get_active_membership(self.request) is None: return reverse('business_onboarding') return reverse('dashboard') class TrustForgePasswordResetView(ThemedAuthContextMixin, PasswordResetView): template_name = 'registration/password_reset_form.html' email_template_name = 'registration/password_reset_email.txt' subject_template_name = 'registration/password_reset_subject.txt' success_url = reverse_lazy('password_reset_done') form_class = TrustForgePasswordResetForm auth_page_title = 'Reset your TrustForge password' auth_page_description = 'Enter your work email and we will send a secure reset link so you can get back into your proof pipeline.' class TrustForgePasswordResetDoneView(ThemedAuthContextMixin, PasswordResetDoneView): template_name = 'registration/password_reset_done.html' auth_page_title = 'Check your email' auth_page_description = 'If that email is tied to an account, a secure reset link is on its way.' class TrustForgePasswordResetConfirmView(ThemedAuthContextMixin, PasswordResetConfirmView): template_name = 'registration/password_reset_confirm.html' form_class = TrustForgeSetPasswordForm success_url = reverse_lazy('password_reset_complete') auth_page_title = 'Create a new password' auth_page_description = 'Set a new password for your account and return to the TrustForge dashboard securely.' class TrustForgePasswordResetCompleteView(ThemedAuthContextMixin, PasswordResetCompleteView): template_name = 'registration/password_reset_complete.html' auth_page_title = 'Password updated' auth_page_description = 'Your password has been changed successfully. You can sign back into TrustForge now.' def _get_memberships_queryset(user): return BusinessMembership.objects.select_related('business').filter(user=user, business__is_active=True) def _get_user_memberships(request: HttpRequest) -> list[BusinessMembership]: if not request.user.is_authenticated: return [] cached = getattr(request, '_trustforge_memberships', None) if cached is None: cached = list(_get_memberships_queryset(request.user)) request._trustforge_memberships = cached return cached def _get_active_membership(request: HttpRequest) -> BusinessMembership | None: if not request.user.is_authenticated: return None cached = getattr(request, '_trustforge_active_membership', None) if cached is not None: return cached memberships = _get_user_memberships(request) active_business_id = request.session.get(ACTIVE_BUSINESS_SESSION_KEY) membership = next((item for item in memberships if item.business_id == active_business_id), None) if membership is None and memberships: membership = memberships[0] request._trustforge_active_membership = membership return membership def _set_active_membership(request: HttpRequest, business_id: int) -> None: request.session[ACTIVE_BUSINESS_SESSION_KEY] = business_id request._trustforge_active_membership = None def _generate_unique_business_slug(name: str) -> str: base_slug = slugify(name)[:45] or 'business' candidate = base_slug counter = 2 while Business.objects.filter(slug=candidate).exists(): candidate = f'{base_slug}-{counter}'[:55] counter += 1 return candidate def business_required(view_func): @wraps(view_func) def wrapped(request: HttpRequest, *args, **kwargs): if _get_active_membership(request) is None: messages.info(request, 'Create or join a business workspace to unlock your protected TrustForge pipeline.') return redirect('business_onboarding') return view_func(request, *args, **kwargs) return wrapped def membership_role_required(*allowed_roles: str): def decorator(view_func): @wraps(view_func) def wrapped(request: HttpRequest, *args, **kwargs): membership = _get_active_membership(request) if membership is None: messages.info(request, 'Create or join a business workspace to continue.') return redirect('business_onboarding') if membership.role not in allowed_roles: raise PermissionDenied('Your role does not allow this action in the current workspace.') return view_func(request, *args, **kwargs) return wrapped return decorator def _build_review_link(request: HttpRequest, review_request: ReviewRequest) -> str: return request.build_absolute_uri(reverse('review_request', args=[str(review_request.token)])) @transaction.atomic def _create_review_request(request: HttpRequest, job: Job, channel: str) -> ReviewRequest: review_request, created = ReviewRequest.objects.get_or_create( job=job, defaults={ 'channel': channel, 'status': ReviewRequest.Status.SENT, }, ) if created: review_request.delivery_note = 'Share this link manually from the field app.' if channel == ReviewRequest.Channel.EMAIL and job.customer.email: review_link = _build_review_link(request, review_request) email_sent = send_mail( subject=f'How was your {job.service_type.lower()} experience?', message=( f'Hi {job.customer.full_name}\n\n' f'Thanks for choosing {job.business.name}. Please share your feedback here: {review_link}\n\n' 'Your response helps us build verified proof of work for future customers.' ), from_email=None, recipient_list=[job.customer.email], fail_silently=True, ) review_request.delivery_note = 'Email sent automatically.' if email_sent else 'Email backend unavailable — copy link manually.' review_request.status = ReviewRequest.Status.SENT review_request.sent_at = timezone.now() review_request.save() job.status = Job.Status.REVIEW_REQUESTED job.save(update_fields=['status']) return review_request @transaction.atomic def signup(request: HttpRequest) -> HttpResponse: if request.user.is_authenticated: return redirect('dashboard' if _get_active_membership(request) else 'business_onboarding') if request.method == 'POST': form = SignUpForm(request.POST) if form.is_valid(): user = form.save() login(request, user) messages.success(request, 'Your TrustForge account is ready. Now let’s create your first business workspace.') return redirect('business_onboarding') else: form = SignUpForm() context = { **_theme_context(), 'auth_page_title': 'Create your TrustForge account', 'auth_page_description': 'Start with secure account access, then connect your business workspace, team roles, and protected proof pipeline.', 'form': form, } return render(request, 'registration/signup.html', context) @login_required @transaction.atomic def business_onboarding(request: HttpRequest) -> HttpResponse: current_membership = _get_active_membership(request) if current_membership is not None: return redirect('dashboard') if request.method == 'POST': form = BusinessOnboardingForm(request.POST) if form.is_valid(): business = form.save(commit=False) business.slug = _generate_unique_business_slug(business.name) business.save() membership = BusinessMembership.objects.create( business=business, user=request.user, role=BusinessMembership.Role.OWNER, ) _set_active_membership(request, membership.business_id) messages.success(request, 'Workspace created. Your jobs, proof cards, and reviews are now scoped to this business.') return redirect('dashboard') else: form = BusinessOnboardingForm() context = { **_theme_context(), 'form': form, } return render(request, 'core/business_onboarding.html', context) @login_required @transaction.atomic def switch_workspace(request: HttpRequest, business_id: int) -> HttpResponse: membership = get_object_or_404(_get_memberships_queryset(request.user), business_id=business_id) _set_active_membership(request, membership.business_id) messages.success(request, f'Workspace switched to {membership.business.name}.') next_url = request.POST.get('next') or reverse('dashboard') return redirect(next_url) @login_required @transaction.atomic def profile_settings(request: HttpRequest) -> HttpResponse: current_membership = _get_active_membership(request) memberships = _get_user_memberships(request) if request.method == 'POST': form = ProfileSettingsForm(request.POST, instance=request.user) if form.is_valid(): form.save() messages.success(request, 'Profile settings updated.') return redirect('profile_settings') else: form = ProfileSettingsForm(instance=request.user) context = { **_theme_context(), 'form': form, 'current_membership': current_membership, 'memberships': memberships, } return render(request, 'core/profile_settings.html', context) @login_required @membership_role_required(BusinessMembership.Role.OWNER, BusinessMembership.Role.ADMIN) @transaction.atomic def workspace_settings(request: HttpRequest) -> HttpResponse: current_membership = _get_active_membership(request) business = current_membership.business team_members = BusinessMembership.objects.select_related('user').filter(business=business).order_by('created_at', 'id') if request.method == 'POST': action = request.POST.get('action') if action == 'update_business': business_form = BusinessSettingsForm(request.POST, instance=business) invite_form = TeamMemberInviteForm() if business_form.is_valid(): business_form.save() messages.success(request, 'Workspace settings updated.') return redirect('workspace_settings') elif action == 'invite_member': business_form = BusinessSettingsForm(instance=business) invite_form = TeamMemberInviteForm(request.POST) if invite_form.is_valid(): email = invite_form.cleaned_data['email'] user, created = User.objects.get_or_create( email=email, defaults={ 'username': email, 'email': email, 'first_name': invite_form.cleaned_data.get('first_name', '').strip(), 'last_name': invite_form.cleaned_data.get('last_name', '').strip(), }, ) if created: user.set_unusable_password() user.save(update_fields=['password']) else: updated_fields = [] first_name = invite_form.cleaned_data.get('first_name', '').strip() last_name = invite_form.cleaned_data.get('last_name', '').strip() if first_name and not user.first_name: user.first_name = first_name updated_fields.append('first_name') if last_name and not user.last_name: user.last_name = last_name updated_fields.append('last_name') if updated_fields: user.save(update_fields=updated_fields) membership, membership_created = BusinessMembership.objects.update_or_create( business=business, user=user, defaults={'role': invite_form.cleaned_data['role']}, ) if membership_created or created: messages.success(request, 'Team member added. If this is a brand-new user, they can use “Forgot password” to set access.') else: messages.success(request, 'Team member role updated for this workspace.') return redirect('workspace_settings') else: business_form = BusinessSettingsForm(instance=business) invite_form = TeamMemberInviteForm() else: business_form = BusinessSettingsForm(instance=business) invite_form = TeamMemberInviteForm() context = { **_theme_context(), 'business_form': business_form, 'invite_form': invite_form, 'current_membership': current_membership, 'team_members': team_members, } return render(request, 'core/workspace_settings.html', context) @transaction.atomic def home(request: HttpRequest) -> HttpResponse: businesses = Business.objects.count() stats = Job.objects.aggregate( completed_jobs=Count('id'), review_requests=Count('review_request'), proof_cards=Count('proof_card'), published_proof=Count('proof_card', filter=Q(proof_card__status=ProofCard.Status.PUBLISHED)), ) featured_proofs = ProofCard.objects.select_related('job__customer', 'job__business').prefetch_related('job__media').filter( is_featured=True, status=ProofCard.Status.PUBLISHED, job__business__is_active=True, )[:3] recent_jobs = Job.objects.select_related('customer', 'business').prefetch_related('media')[:4] context = { **_theme_context(), 'django_version': django_version(), 'python_version': platform.python_version(), 'current_time': timezone.now(), 'business_count': businesses, 'stats': stats, 'featured_proofs': featured_proofs, 'recent_jobs': recent_jobs, } return render(request, 'core/index.html', context) @transaction.atomic def public_proof_gallery(request: HttpRequest, slug: str) -> HttpResponse: business = get_object_or_404(Business, slug=slug, is_active=True) proof_cards = ProofCard.objects.select_related('job__customer', 'job__business').prefetch_related('job__media').filter( job__business=business, status=ProofCard.Status.PUBLISHED, ) featured_proofs = proof_cards.filter(is_featured=True)[:3] context = { **_theme_context(), 'business': business, 'proof_cards': proof_cards, 'featured_proofs': featured_proofs, } return render(request, 'core/public_proof_gallery.html', context) @transaction.atomic def public_proof_detail(request: HttpRequest, slug: str, card_id: int) -> HttpResponse: proof_card = get_object_or_404( ProofCard.objects.select_related('job__customer', 'job__business').prefetch_related('job__media'), id=card_id, job__business__slug=slug, job__business__is_active=True, status=ProofCard.Status.PUBLISHED, ) related_proofs = ProofCard.objects.select_related('job__customer', 'job__business').prefetch_related('job__media').filter( job__business=proof_card.job.business, status=ProofCard.Status.PUBLISHED, ).exclude(id=proof_card.id)[:3] context = { **_theme_context(), 'business': proof_card.job.business, 'proof_card': proof_card, 'related_proofs': related_proofs, } return render(request, 'core/public_proof_detail.html', context) @login_required @business_required @transaction.atomic def dashboard(request: HttpRequest) -> HttpResponse: current_membership = _get_active_membership(request) business = current_membership.business jobs = Job.objects.filter(business=business) stats = jobs.aggregate( completed_jobs=Count('id'), review_requests=Count('review_request'), proof_cards=Count('proof_card'), published_cards=Count('proof_card', filter=Q(proof_card__status=ProofCard.Status.PUBLISHED)), ) feedback_qs = Feedback.objects.filter(review_request__job__business=business) positive_feedback = feedback_qs.filter(experience__in=POSITIVE_EXPERIENCES).count() total_feedback = feedback_qs.count() conversion_rate = round((positive_feedback / total_feedback) * 100, 1) if total_feedback else 0 recent_jobs = jobs.select_related('customer', 'business').prefetch_related('media')[:5] recent_proofs = ProofCard.objects.select_related('job__customer', 'job__business').prefetch_related('job__media').filter(job__business=business)[:4] context = { **_theme_context(), 'current_membership': current_membership, 'stats': stats, 'conversion_rate': conversion_rate, 'recent_jobs': recent_jobs, 'recent_proofs': recent_proofs, } return render(request, 'core/dashboard.html', context) @login_required @business_required @transaction.atomic def jobs_list(request: HttpRequest) -> HttpResponse: current_membership = _get_active_membership(request) jobs = Job.objects.select_related('customer', 'business').prefetch_related('media').filter(business=current_membership.business) context = {**_theme_context(), 'jobs': jobs, 'current_membership': current_membership} return render(request, 'core/jobs_list.html', context) @login_required @business_required @transaction.atomic def job_create(request: HttpRequest) -> HttpResponse: current_membership = _get_active_membership(request) business = current_membership.business if request.method == 'POST': form = JobIntakeForm(request.POST, request.FILES, business=business) if form.is_valid(): customer = Customer.objects.create( business=business, full_name=form.cleaned_data['customer_name'], email=form.cleaned_data['customer_email'], phone=form.cleaned_data['customer_phone'], city=form.cleaned_data['customer_city'], state=form.cleaned_data['customer_state'], ) job = Job.objects.create( business=business, customer=customer, service_type=form.cleaned_data['service_type'], description=form.cleaned_data['description'], technician_name=form.cleaned_data['technician_name'], city=form.cleaned_data['customer_city'], state=form.cleaned_data['customer_state'], completed_at=form.cleaned_data['completion_date'], project_value=form.cleaned_data['project_value'], status=Job.Status.COMPLETED, ) for media_type, upload in ( (JobMedia.MediaType.BEFORE, form.cleaned_data.get('before_photo')), (JobMedia.MediaType.AFTER, form.cleaned_data.get('after_photo')), ): if upload: JobMedia.objects.create(job=job, media_type=media_type, file=upload) display_name = 'Verified homeowner' if form.cleaned_data['anonymize_customer'] else customer.full_name ProofCard.objects.create( job=job, customer_display_name=display_name, is_anonymized=form.cleaned_data['anonymize_customer'], attached_widget_label='Homepage proof gallery', attached_pages='Homepage, Service pages', status=ProofCard.Status.DRAFT, ) if form.cleaned_data['send_review_request']: _create_review_request(request, job, form.cleaned_data['review_channel']) messages.success(request, 'Job logged inside your workspace. Proof card drafted and ready for review workflow.') return redirect('job_detail', job_id=job.id) else: form = JobIntakeForm( business=business, initial={ 'business': business, 'customer_city': business.primary_city, 'customer_state': business.primary_state, 'technician_name': request.user.get_full_name(), }, ) context = {**_theme_context(), 'form': form, 'current_membership': current_membership} return render(request, 'core/job_form.html', context) @login_required @business_required @transaction.atomic def job_detail(request: HttpRequest, job_id: int) -> HttpResponse: current_membership = _get_active_membership(request) job = get_object_or_404( Job.objects.select_related('customer', 'business', 'proof_card', 'review_request').prefetch_related('media'), id=job_id, business=current_membership.business, ) if request.method == 'POST' and request.POST.get('action') == 'send_review_request': channel = request.POST.get('channel', ReviewRequest.Channel.EMAIL) review_request = _create_review_request(request, job, channel) messages.success(request, f'Review request sent. Share link: {_build_review_link(request, review_request)}') return redirect('job_detail', job_id=job.id) context = {**_theme_context(), 'job': job, 'current_membership': current_membership} return render(request, 'core/job_detail.html', context) @login_required @business_required @transaction.atomic def proof_cards_list(request: HttpRequest) -> HttpResponse: current_membership = _get_active_membership(request) proof_cards = ProofCard.objects.select_related('job__customer', 'job__business').prefetch_related('job__media').filter( job__business=current_membership.business ) context = {**_theme_context(), 'proof_cards': proof_cards, 'current_membership': current_membership} return render(request, 'core/proof_cards_list.html', context) @login_required @business_required @transaction.atomic def proof_card_detail(request: HttpRequest, card_id: int) -> HttpResponse: current_membership = _get_active_membership(request) proof_card = get_object_or_404( ProofCard.objects.select_related('job__customer', 'job__business').prefetch_related('job__media'), id=card_id, job__business=current_membership.business, ) if request.method == 'POST': if not current_membership.can_manage_proof: raise PermissionDenied('Your role does not allow proof publishing controls.') action = request.POST.get('action') if action == 'publish': proof_card.status = ProofCard.Status.PUBLISHED proof_card.published_at = timezone.now() proof_card.save(update_fields=['status', 'published_at', 'updated_at']) proof_card.job.status = Job.Status.PROOF_READY proof_card.job.save(update_fields=['status']) messages.success(request, 'Proof card published to the trust gallery.') elif action == 'hide': proof_card.status = ProofCard.Status.HIDDEN proof_card.save(update_fields=['status', 'updated_at']) messages.success(request, 'Proof card hidden from public display.') elif action == 'toggle_featured': proof_card.is_featured = not proof_card.is_featured proof_card.save(update_fields=['is_featured', 'updated_at']) messages.success(request, 'Featured flag updated.') return redirect('proof_card_detail', card_id=proof_card.id) context = {**_theme_context(), 'proof_card': proof_card, 'current_membership': current_membership} return render(request, 'core/proof_card_detail.html', context) @login_required @membership_role_required(BusinessMembership.Role.OWNER, BusinessMembership.Role.ADMIN, BusinessMembership.Role.MANAGER) @transaction.atomic def proof_card_edit(request: HttpRequest, card_id: int) -> HttpResponse: current_membership = _get_active_membership(request) proof_card = get_object_or_404( ProofCard.objects.select_related('job__customer', 'job__business'), id=card_id, job__business=current_membership.business, ) if request.method == 'POST': form = ProofCardForm(request.POST, instance=proof_card) if form.is_valid(): proof_card = form.save(commit=False) if proof_card.status == ProofCard.Status.PUBLISHED and not proof_card.published_at: proof_card.published_at = timezone.now() proof_card.save() messages.success(request, 'Proof card updated.') return redirect('proof_card_detail', card_id=proof_card.id) else: form = ProofCardForm(instance=proof_card) context = {**_theme_context(), 'form': form, 'proof_card': proof_card, 'current_membership': current_membership} return render(request, 'core/proof_card_form.html', context) @transaction.atomic def review_request_view(request: HttpRequest, token: str) -> HttpResponse: review_request = get_object_or_404( ReviewRequest.objects.select_related('job__customer', 'job__business').prefetch_related('job__media'), token=token, ) proof_card = review_request.job.proof_card if review_request.status == ReviewRequest.Status.SENT: review_request.status = ReviewRequest.Status.VIEWED review_request.last_opened_at = timezone.now() review_request.save(update_fields=['status', 'last_opened_at']) submitted = False positive = False redirect_url = review_request.job.business.google_review_url if request.method == 'POST': form = PublicFeedbackForm(request.POST) if form.is_valid(): experience = form.cleaned_data['experience'] testimonial = form.cleaned_data['testimonial'].strip() positive = experience in POSITIVE_EXPERIENCES feedback, _ = Feedback.objects.update_or_create( review_request=review_request, defaults={ 'experience': experience, 'rating': RATING_MAP[experience], 'testimonial': testimonial, 'follow_up_required': not positive, 'is_public_approved': positive, }, ) review_request.status = ReviewRequest.Status.RESPONDED review_request.reviewed_at = timezone.now() review_request.save(update_fields=['status', 'reviewed_at']) proof_card.rating = feedback.rating proof_card.testimonial_quote = testimonial if positive: proof_card.status = ProofCard.Status.PUBLISHED proof_card.published_at = timezone.now() else: proof_card.status = ProofCard.Status.DRAFT proof_card.save(update_fields=['rating', 'testimonial_quote', 'status', 'published_at', 'updated_at']) review_request.job.status = Job.Status.PROOF_READY if positive else Job.Status.REVIEW_REQUESTED review_request.job.save(update_fields=['status']) submitted = True form = PublicFeedbackForm() else: form = PublicFeedbackForm() context = { **_theme_context(), 'review_request': review_request, 'job': review_request.job, 'proof_card': proof_card, 'form': form, 'submitted': submitted, 'positive': positive, 'redirect_url': redirect_url, } return render(request, 'core/review_request.html', context)