from django.contrib import messages from django.contrib.auth import authenticate, login, logout from django.contrib.auth.decorators import login_required from django.core.exceptions import PermissionDenied from django.db import transaction from django.shortcuts import get_object_or_404, redirect, render from django.utils import timezone from accounts.models import CustomUser from accounts.permissions import admin_required from core.models import Project, ProjectApplication, ProjectMembership from core.tenant_db import delete_company_database, provision_company_database from core.project_utils import get_current_project, get_membership COMPANY_USER_ROLES = ( (CustomUser.Role.ACCOUNTANT, "Company Accountant"), (CustomUser.Role.MEMBER, "Company Member"), (CustomUser.Role.VIEWER, "Viewer"), ) def login_view(request): if request.user.is_authenticated: return redirect("dashboard:home") if request.method == "POST": username = request.POST.get("username", "").strip() password = request.POST.get("password", "") user = authenticate(request, username=username, password=password) if user is not None: login(request, user) return redirect("dashboard:home") messages.error(request, "Invalid username or password, or the account is disabled.") return render(request, "accounts/login.html") def forgot_password_view(request): """Lets a user recover access by proving they know their own username AND phone number, then set a brand new password immediately — no email/SMS service required. Both fields must match the SAME account, and a generic error is shown either way so this can't be used to check whether a given username exists.""" if request.user.is_authenticated: return redirect("dashboard:home") # Simple abuse throttle: cap attempts per browser session. Not a # substitute for server-side rate limiting in front of a public # deployment, but stops casual guessing. attempts = request.session.get("pw_reset_attempts", 0) if request.method == "POST": if attempts >= 8: messages.error(request, "Too many attempts. Please try again later.") return render(request, "accounts/forgot_password.html") username = request.POST.get("username", "").strip() phone = request.POST.get("phone", "").strip() new_password = request.POST.get("new_password", "") confirm_password = request.POST.get("confirm_password", "") request.session["pw_reset_attempts"] = attempts + 1 generic_error = "No account matches that username and phone number together. Please check both and try again." if not username or not phone: messages.error(request, generic_error) return render(request, "accounts/forgot_password.html", {"username": username}) user = CustomUser.objects.filter(username__iexact=username, phone=phone).first() if not user: messages.error(request, generic_error) return render(request, "accounts/forgot_password.html", {"username": username}) if not user.is_active: messages.error( request, "This account is currently disabled. Contact your Company Admin or the Application Owner." ) return render(request, "accounts/forgot_password.html", {"username": username}) if not new_password or new_password != confirm_password: messages.error(request, "The two new passwords don't match.") return render(request, "accounts/forgot_password.html", {"username": username, "verified": True}) from django.contrib.auth.password_validation import validate_password from django.core.exceptions import ValidationError as DjangoValidationError try: validate_password(new_password, user=user) except DjangoValidationError as exc: for err in exc.messages: messages.error(request, err) return render(request, "accounts/forgot_password.html", {"username": username, "verified": True}) user.set_password(new_password) user.save(update_fields=["password"]) request.session.pop("pw_reset_attempts", None) messages.success(request, "Password updated. You can now log in with your new password.") return redirect("accounts:login") return render(request, "accounts/forgot_password.html") @login_required def logout_view(request): if request.method != 'POST': return redirect('dashboard:home') logout(request) return redirect('accounts:login') @login_required def profile_view(request): return render(request, "accounts/profile.html", {"user_obj": request.user}) @admin_required def user_list(request): project = get_current_project(request) if not project and not request.user.is_superuser: return render(request, "accounts/user_list.html", {"users": [], "company": None}) if request.user.is_superuser: users = CustomUser.objects.all().order_by("username") else: users = CustomUser.objects.filter( project_memberships__project=project, project_memberships__is_active=True, ).distinct().order_by("username") access_by_user = {} if project: access_by_user = {m.user_id: m.is_active for m in ProjectMembership.objects.filter(project=project)} user_rows = [(u, access_by_user.get(u.pk, u.is_active)) for u in users] return render(request, "accounts/user_list.html", {"user_rows": user_rows, "company": project}) @admin_required def user_create(request): project = get_current_project(request) if request.user.is_superuser: messages.info(request, "Super Admin manages companies. Create company users from inside a company as Company Admin.") return redirect("accounts:project_applications") if not project or not project.is_active: messages.error(request, "No active company is selected.") return redirect("dashboard:home") if request.method == "POST": username = request.POST.get("username", "").strip() password = request.POST.get("password", "") role = request.POST.get("role", "") if role not in {r[0] for r in COMPANY_USER_ROLES}: messages.error(request, "Only Accountant, Member, or Viewer can be created here.") elif not username or not password: messages.error(request, "Username and password are required.") elif CustomUser.objects.filter(username=username).exists(): messages.error(request, "That username already exists. Use a unique username.") else: user = CustomUser.objects.create_user( username=username, password=password, role=role, first_name=request.POST.get("first_name", "").strip(), last_name=request.POST.get("last_name", "").strip(), email=request.POST.get("email", "").strip(), phone=request.POST.get("phone", "").strip(), is_active=True, ) ProjectMembership.objects.create( project=project, user=user, role=role, is_active=True ) messages.success(request, f"{user.username} was added to {project.name} as {dict(COMPANY_USER_ROLES)[role]}.") return redirect("accounts:user_list") return render(request, "accounts/user_form.html", {"roles": COMPANY_USER_ROLES, "company": project}) @admin_required def user_toggle_active(request, pk): if request.method != "POST": return redirect("accounts:user_list") project = get_current_project(request) user = get_object_or_404(CustomUser, pk=pk) if user == request.user: messages.warning(request, "You cannot disable your own Company Admin account.") return redirect("accounts:user_list") if not project or request.user.is_superuser: raise PermissionDenied membership = get_object_or_404(ProjectMembership, project=project, user=user) membership.is_active = not membership.is_active membership.save(update_fields=["is_active"]) messages.success(request, f"{user.username} is now {'active' if membership.is_active else 'disabled'} for {project.name}.") return redirect("accounts:user_list") def project_register(request): if request.user.is_authenticated: return redirect("dashboard:home") if request.method == "POST": username = request.POST.get("username", "").strip() password = request.POST.get("password", "") if not username or not password: messages.error(request, "Username and password are required.") return render(request, "accounts/project_register.html") if CustomUser.objects.filter(username=username).exists(): messages.error(request, "This username already exists.") return render(request, "accounts/project_register.html") user = CustomUser.objects.create_user( username=username, password=password, first_name=request.POST.get("first_name", "").strip(), last_name=request.POST.get("last_name", "").strip(), email=request.POST.get("email", "").strip(), phone=request.POST.get("phone", "").strip(), role=CustomUser.Role.VIEWER, is_active=False, ) ProjectApplication.objects.create( applicant=user, project_name=request.POST.get("project_name", "").strip(), location=request.POST.get("location", "").strip(), description=request.POST.get("description", "").strip(), start_date=request.POST.get("start_date") or None, total_land_area=request.POST.get("total_land_area", "").strip(), ) messages.success(request, "Company registration submitted. The Application Owner must approve it before you can log in.") return redirect("accounts:login") return render(request, "accounts/project_register.html") @login_required @admin_required def project_applications(request): if not request.user.is_superuser: messages.error(request, "Only the Application Owner (Super Admin) can manage company registrations.") return redirect("dashboard:home") applications = ProjectApplication.objects.select_related("applicant", "approved_project").all() counts = { "total": applications.count(), "pending": applications.filter(status="pending").count(), "approved": applications.filter(status="approved").count(), "paused": applications.filter(status="paused").count(), } return render(request, "accounts/project_applications.html", {"applications": applications, "counts": counts}) # Kept as a separate URL name so it can grow into a richer landing page # later without touching the login/redirect wiring that points here. super_admin_home = project_applications @login_required @admin_required def approve_project(request, pk): if not request.user.is_superuser: raise PermissionDenied application = get_object_or_404(ProjectApplication, pk=pk) if request.method != "POST" or application.status not in ("pending", "paused"): return redirect("accounts:project_applications") project = application.approved_project created_project = False provisioned_now = False try: if not project: project = Project.objects.create( name=application.project_name, location=application.location, description=application.description, start_date=application.start_date, total_land_area=application.total_land_area, is_active=False, owner=application.applicant, database_key=f"company_{application.applicant_id}_{timezone.now().strftime('%Y%m%d%H%M%S')}", database_status="pending", created_by=request.user, ) created_project = True application.approved_project = project application.save(update_fields=["approved_project"]) # Provision the physical MySQL database before granting access. if project.database_status != "provisioned": credentials = provision_company_database(project) provisioned_now = True project.database_name = credentials["database_name"] project.database_user = credentials["database_user"] project.database_password = credentials["database_password"] project.database_host = credentials["database_host"] project.database_port = credentials["database_port"] project.database_status = credentials["database_status"] project.database_provisioned_at = credentials["provisioned_at"] project.database_error = "" # Register the connection and create this company's tables. This # is the step that makes the dedicated database actually usable — # without it, the database exists but is empty forever, and all # queries would silently keep hitting the shared platform database. from django.core.management import call_command from core.db_router import ensure_tenant_connection alias = ensure_tenant_connection(project) if not alias: raise RuntimeError("Tenant database connection could not be registered.") call_command("migrate", database=alias, verbosity=0, interactive=False) project.is_active = True project.owner = application.applicant project.save() application.applicant.is_active = True application.applicant.role = CustomUser.Role.ADMIN application.applicant.save(update_fields=["is_active", "role"]) ProjectMembership.objects.update_or_create( project=project, user=application.applicant, defaults={"role": "admin", "is_active": True}, ) # Resuming a paused company should undo the pause fully — re-enable # every one of its users' logins and memberships too (Accountant, # Member, Viewer accounts the Company Admin created), not just the # original applicant's own account. ProjectMembership.objects.filter(project=project).update(is_active=True) CustomUser.objects.filter(project_memberships__project=project).update(is_active=True) application.status = "approved" application.reviewed_at = timezone.now() application.save(update_fields=["status", "reviewed_at"]) except Exception as exc: if project: # Never leave a partially approved company behind. If this approval # created a physical tenant DB/user, remove them when the following # migration/connection step fails. Otherwise a retry can hit an # orphaned MySQL account and produce error 1045. try: if created_project or provisioned_now: delete_company_database(project) if created_project: project.delete() else: project.database_status = "failed" project.database_error = str(exc)[:2000] project.save(update_fields=["database_status", "database_error"]) except Exception: pass messages.error(request, f"Company approval failed because its database could not be provisioned: {exc}") return redirect("accounts:project_applications") messages.success(request, f"Company '{project.name}' approved. A dedicated MySQL database was created and the applicant is now Company Admin.") return redirect("accounts:project_applications") @login_required @admin_required def pause_project(request, pk): if not request.user.is_superuser: raise PermissionDenied application = get_object_or_404(ProjectApplication, pk=pk) if request.method == "POST" and application.approved_project: project = application.approved_project project.is_active = False project.save(update_fields=["is_active", "updated_at"]) ProjectMembership.objects.filter(project=project).update(is_active=False) # Fully block login too, not just data access — "Pause" should mean # the whole company is disabled, matching Approve/Resume/Disable # as one consistent on/off switch. CustomUser.objects.filter(project_memberships__project=project).update(is_active=False) application.status = "paused" application.save(update_fields=["status"]) messages.success(request, f"Company '{project.name}' is paused. Its users can no longer log in.") return redirect("accounts:project_applications") @login_required @admin_required def delete_project(request, pk): if not request.user.is_superuser: raise PermissionDenied application = get_object_or_404(ProjectApplication, pk=pk) if request.method == "POST": project = application.approved_project if project: try: delete_company_database(project) except Exception as exc: messages.error(request, f"Company database could not be deleted: {exc}") return redirect("accounts:project_applications") project.delete() application.status = "deleted" application.approved_project = None application.save(update_fields=["status", "approved_project"]) application.applicant.is_active = False application.applicant.save(update_fields=["is_active"]) messages.success(request, "Company registration and its dedicated database were deleted.") return redirect("accounts:project_applications")