gcivil-nyu-org / INET-Wednesday-Spring2024-Team-2 / 603

Committed 24 Apr 2024 12:36AM UTC coverage: 88.975% (+2.2%) from 86.788%

Build # 603

Build Type

cron

travis-pro

Committed by

web-flow

Commit Message

Merge pull request #163 from gcivil-nyu-org/develop

Merging into master from DEV

Run Details

199 of 235 new or added lines in 7 files covered. (84.68%)

58 existing lines in 3 files now uncovered.

1033 of 1161 relevant lines covered (88.98%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.72

/users/views.py

import ast
import logging
import os
import sys
import uuid

import boto3
from django.conf import settings
from django.contrib import messages
from django.contrib.auth import authenticate, login, logout
from django.contrib.auth.decorators import login_required
from django.contrib.postgres.search import SearchVector, SearchQuery, SearchRank
from django.contrib.sites.models import Site
from django.core import serializers
from django.core.paginator import Paginator
from django.core.serializers import serialize
from django.db.models import Min, Q
from django.http import JsonResponse
from django.shortcuts import render, redirect, get_object_or_404
from django.urls import reverse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.csrf import ensure_csrf_cookie
from django.utils import timezone
from users.decorators import user_type_required
from users.forms import UserSignUpForm, RentalListingForm
from .decorators import no_cache
from .forms import CustomLoginForm
from .forms import CustomUserEditForm
from .forms import LandlordSignupForm
from .models import Favorite, Rental_Listings
from .models import RentalImages
from .utils import send_email_to_admin

logger = logging.getLogger(__name__)

LOGGING = {
    "version": 1,
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "stream": sys.stdout,
        }
    },
    "root": {"handlers": ["console"], "level": "INFO"},
}

logging.config.dictConfig(LOGGING)


def custom_404_handler(request, exception):
    if request.user.is_authenticated:
        if getattr(request.user, "user_type", None) == "landlord":
            return redirect("landlord_homepage")
        else:
            return redirect("user_homepage")
    else:
        return redirect("index")


def login_process(request, user_type, this_page, destination_url_name):
    if request.method != "POST":
        form = CustomLoginForm()
        return render(request, this_page, {"form": form})

    form = CustomLoginForm(request,
                           request.POST)  # Pass the request to the form
    if form.is_valid():
        username = form.cleaned_data["username"]
        password = form.cleaned_data["password"]
        user = authenticate(request, username=username, password=password)

        if user is None:
            messages.error(request, "Invalid credentials!")
            return render(request, this_page, {"form": form})

        if getattr(user, "user_type", None) != user_type:
            messages.error(
                request,
                f"Please provide correct credentials to login as "
                f"{user_type.capitalize()}!",
                # noqa:<E501>
            )
            return render(request, this_page, {"form": form})
        # if user_type == "landlord" and user.verified is False:
        #     messages.error(
        #         request,
        #         "Your account has not been verified by the admin yet. Please wait!",
        #     )
        #     return render(request, this_page, {"form": form})
        # ..
        login(request, user)
        return redirect(reverse(destination_url_name))

    return render(request, this_page, {"form": form})


@no_cache
def landlord_login(request):
    return login_process(
        request,
        user_type="landlord",
        this_page="login/landlord_login.html",
        destination_url_name="landlord_homepage",
    )


@no_cache
def user_login(request):
    return login_process(
        request,
        user_type="user",
        this_page="login/user_login.html",
        destination_url_name="user_homepage",
        # URL pattern name for user's homepage
    )


@no_cache
@ensure_csrf_cookie
def user_signup(request):
    if request.method == "POST":
        form = UserSignUpForm(request.POST)
        if form.is_valid():
            email = form.cleaned_data["email"]
            user = form.save(commit=False)
            user.username = email
            user.save()
            if user.user_type == user.USER:
                user.verified = True
            else:
                user.verified = False
            login(request, user,
                  backend="django.contrib.auth.backends.ModelBackend")
            return redirect("user_homepage")
    else:
        form = UserSignUpForm()
    for field, errors in form.errors.items():
        for error in errors:
            messages.error(request, f"{error}")

    return render(request, "users/signup/signup.html", {"form": form})


@no_cache
def home(request):
    return render(request, "home.html")


@no_cache
def logout_view(request):
    logout(request)
    return redirect("/")


@no_cache
@user_type_required("user")
def user_home(request):
    return render(request, "user_homepage.html")


@no_cache
@user_type_required("landlord")
def landlord_home(request):
    listings = (
        Rental_Listings.objects.filter(Landlord=request.user)
        .annotate(first_image=Min("images__image_url"))
        .order_by("-Submitted_date")
    )
    return render(request, "landlord_homepage.html", {"listings": listings})


@no_cache
def landlord_signup(request):
    if request.method == "POST":
        form = LandlordSignupForm(request.POST, request.FILES)
        if form.is_valid():
            print(request.FILES)
            user = form.save(commit=False)

            # Upload file to S3 if present check
            pdf_file = request.FILES.get("pdf_file")
            print(pdf_file)
            if pdf_file:
                file_name, file_extension = os.path.splitext(pdf_file.name)
                print(f"Received file: {pdf_file.name}")  # Debug print
                file_name = f"pdfs/{file_name}_{uuid.uuid4()}{file_extension}"
                print(file_name)
                try:
                    s3_client = boto3.client(
                        "s3",
                        aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
                        aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
                    )

                    # Check if file already exists in S3 bucket
                    existing_files = s3_client.list_objects_v2(
                        Bucket=settings.AWS_STORAGE_BUCKET_NAME,
                        Prefix=file_name
                    )

                    if "Contents" in existing_files:
                        messages.error(
                            request,
                            "A file with the same name already exists."
                            " Please rename your file and try again.",
                        )
                        return render(
                            request, "signup/landlord_signup.html",
                            {"form": form}
                        )

                    s3_client.upload_fileobj(
                        pdf_file, "landlord-verification-files", file_name
                    )
                    user.s3_doclink = (
                        f"https://landlord-verification-files."
                        f"s3.amazonaws.com/{file_name}"
                    )
                    print("File uploaded successfully")
                except Exception as e:
                    print(f"Error uploading file to S3: {e}")
            user.save()
            send_email_to_admin(user.username)

            messages.success(request, "Registration successful. Please log in.")
            return redirect("landlord_login")
        else:
            messages.error(
                request, "Registration failed. Please correct the errors below."
            )
    else:
        form = LandlordSignupForm()
    return render(request, "signup/landlord_signup.html", {"form": form})


@no_cache
def apply_filters(listings, filter_params):
    # TODO: fix in database
    listings = listings.exclude(neighborhood="Hell's Kitchen")
    # Apply filters
    if filter_params.get("borough"):
        borough = filter_params.get("borough")
        if borough == "All(NYC)":
            # No need to filter by borough if "All(NYC)" is selected
            pass
        else:
            # Filter by neighborhood or borough using exact match
            listings = listings.filter(
                Q(neighborhood=borough) | Q(borough=borough))
    if filter_params.get("min_price"):
        min_price = int(filter_params.get("min_price"))
        listings = listings.filter(price__gte=min_price)
    if filter_params.get("max_price"):
        max_price = int(filter_params.get("max_price"))
        listings = listings.filter(price__lte=max_price)
    if filter_params.get("bedrooms"):
        if filter_params.get("bedrooms") == "Any":
            listings = listings
        else:
            listings = listings.filter(beds=filter_params.get("bedrooms"))
    if filter_params.get("bathrooms"):
        if filter_params.get("bathrooms") == "Any":
            listings = listings
        else:
            listings = listings.filter(baths=filter_params.get("bathrooms"))
    if filter_params.get("elevator"):
        listings = listings.filter(elevator=True)
    if filter_params.get("laundry"):
        listings = listings.filter(washer_dryer_in_unit=True)
    if filter_params.get("no_fee"):
        listings = listings.filter(broker_fee=0)
    if filter_params.get("building_type"):
        if filter_params.get("building_type") == "Any":
            listings = listings
        else:
            listings = listings.filter(
                unit_type=filter_params.get("building_type"))
    if filter_params.get("parking"):
        listings = listings.filter(parking_available=True)
    if filter_params.get("search_query"):
        query = SearchQuery(filter_params.get("search_query"))
        listings = (
            listings.annotate(
                search=SearchVector("address"),
                rank=SearchRank(SearchVector("address"), query),
            )
            .filter(search=query)
            .order_by("-rank")
        )
    return listings


@no_cache
@user_type_required("user")
def rentals_page(request):
    filter_params = {
        "borough": request.GET.get("borough"),
        "min_price": request.GET.get("min_price"),
        "max_price": request.GET.get("max_price"),
        "bedrooms": request.GET.get("bedrooms"),
        "bathrooms": request.GET.get("bathrooms"),
        "elevator": request.GET.get("elevator") == "on",
        "laundry": request.GET.get("laundry") == "on",
        "no_fee": request.GET.get("no_fee") == "on",
        "building_type": request.GET.get("building_type"),
        "parking": request.GET.get("parking") == "on",
        "search_query": request.GET.get("search_query", ""),
    }

    listings = Rental_Listings.objects.all()
    listings = apply_filters(listings, filter_params)
    listings = listings.annotate(first_image=Min("images__image_url"))
    sort_option = request.GET.get("sort")
    if sort_option == "recent":
        listings = listings.order_by(
            "-Submitted_date"
        )  # Assuming you have a created_at field
    elif sort_option == "high_to_low":
        listings = listings.order_by("-price")
    else:
        listings = listings.order_by("price")
    favorite_listings_ids = Favorite.objects.filter(
        user=request.user).values_list(
        "listing__id", flat=True
    )
    filter_params = {k: v for k, v in filter_params.items() if v is not None}
    paginator = Paginator(listings, 5)  # Show 5 listings per page
    page_number = request.GET.get("page")
    page_obj = paginator.get_page(page_number)
    favorite_listings_ids = Favorite.objects.filter(
        user=request.user).values_list(
        "listing__id", flat=True
    )
    context = {
        "page_obj": page_obj,
        "listings": listings,
        "favorite_listings_ids": list(favorite_listings_ids),
        "filter_params": filter_params,  # Ensure it's converted to a list
    }
    return render(request, "users/searchRental/rentalspage.html", context)


@user_type_required("landlord")
def add_rental_listing(request):
    if request.method == "POST":
        form = RentalListingForm(request.POST, request.FILES)
        images = request.FILES.getlist("photo")
        if form.is_valid():
            rental_listing = form.save(commit=False)
            rental_listing.Landlord = request.user
            rental_listing.Submitted_date = timezone.now().date()
            apt_no = form.cleaned_data.get('apt_no', '')
            if apt_no:
                base_address = rental_listing.address.split(',')[0]
                full_address = f"{base_address} #{apt_no}"
                rental_listing.address = full_address
            rental_listing.save()
            AWS_STORAGE_BUCKET_NAME = "landlord-verification-files"
            for image in images:
                file_name, file_extension = os.path.splitext(image.name)
                unique_file_name = f"pdfs/{uuid.uuid4()}{file_extension}"
                try:
                    s3_client = boto3.client(
                        "s3",
                        aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
                        aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
                    )
                    s3_client.upload_fileobj(
                        image,
                        AWS_STORAGE_BUCKET_NAME,
                        unique_file_name,
                    )
                    image_url = f"https://{AWS_STORAGE_BUCKET_NAME}.s3.amazonaws.com/{unique_file_name}"
                    RentalImages.objects.create(
                        rental_listing=rental_listing, image_url=image_url
                    )
                except Exception as e:
                    print(f"Error uploading file to S3: {e}")
            return redirect("landlord_homepage")
    else:
        form = RentalListingForm()
    return render(request, "add_rental_listing.html", {"form": form})


@no_cache
@login_required
def placeholder_view(request):
    return render(request, "users/searchRental/placeholder.html")


@user_type_required("landlord")
def landlord_placeholder_view(request):
    return render(request, "users/searchRental/landlord_placeholder.html")


@no_cache
def listing_detail(request, listing_id):
    # Retrieve the specific listing based on the ID provided in the URL parameter
    listing = get_object_or_404(Rental_Listings, id=listing_id)

    context = {"listing": listing}
    return render(request, "users/searchRental/listing_detail.html", context)


@no_cache
@csrf_exempt
@login_required
@user_type_required("user")
def toggle_favorite(request):
    if request.method == "POST":
        listing_id = request.POST.get("listing_id")
        if not listing_id:
            return JsonResponse({"error": "Listing ID is required"}, status=400)

        try:
            listing = Rental_Listings.objects.get(id=listing_id)
            favorite, created = Favorite.objects.get_or_create(
                user=request.user, listing=listing
            )
            if not created:
                favorite.delete()
                return JsonResponse({"status": "removed"})
            return JsonResponse({"status": "added"})
        except Rental_Listings.DoesNotExist:
            return JsonResponse({"error": "Listing not found"}, status=404)
        except Exception as e:
            logger.error(f"Internal server error: {e}", exc_info=True)
            return JsonResponse({"error": "Internal server error"}, status=500)


@user_type_required("user")
@login_required
@no_cache
def favorites_page(request):
    favorite_listings = Favorite.objects.filter(
        user=request.user).select_related(
        "listing"
    )
    listings = [fav.listing for fav in favorite_listings]

    listings_json = serializers.serialize("json", listings)
    favorite_listings_ids = [listing.id for listing in listings]

    context = {
        "listings_json": listings_json,
        "favorite_listings_ids": favorite_listings_ids,
    }
    return render(request, "users/searchRental/favorites.html", context)


@no_cache
def map_view(request):
    filter_params = ast.literal_eval(request.GET.get("filter_params"))
    rental_listings = Rental_Listings.objects.all()
    # http://127.0.0.1:8000/map/?filter_params={%27borough%27:%20%27Manhattan%27,%20%27min_price%27:%20%27%27,%20%27max_price%27:%20%27%27}
    rental_listings = apply_filters(rental_listings, filter_params)
    rental_listings_json = serialize("json", rental_listings)
    current_site = Site.objects.get_current()
    current_site.domain
    context = {
        "rental_listings": rental_listings_json,
        "this_domain": current_site.domain,
    }
    return render(request, "users/searchRental/map_view.html", context)

@login_required
@no_cache
@user_type_required("user")
def profile_view_edit(request):
    if request.method == "POST":
        form = CustomUserEditForm(request.POST, instance=request.user)
        if form.is_valid():
            form.save()
            messages.success(request, "Your profile was successfully updated!")
            return redirect("profile_view_edit")
    else:
        form = CustomUserEditForm(instance=request.user)

    return render(
        request,
        "users/Profile/profile_view_edit.html",
        {"form": form, "user": request.user},
    )


@login_required
@no_cache
@user_type_required("landlord")
def landlord_profile_update(request):
    if request.method == "POST":
        form = CustomUserEditForm(request.POST, instance=request.user)
        if form.is_valid():
            form.save()
            messages.success(request, "Your profile was successfully updated!")
            return redirect("landlord_profile_update")
    else:
        form = CustomUserEditForm(instance=request.user)

    return render(
        request,
        "users/Profile/landlord_profile_update.html",
        {"form": form, "user": request.user},
    )

1	import ast	1✔
2	import logging	1✔
3	import os	1✔
4	import sys	1✔
5	import uuid	1✔
6
7	import boto3	1✔
8	from django.conf import settings	1✔
9	from django.contrib import messages	1✔
10	from django.contrib.auth import authenticate, login, logout	1✔
11	from django.contrib.auth.decorators import login_required	1✔
12	from django.contrib.postgres.search import SearchVector, SearchQuery, SearchRank	1✔
13	from django.contrib.sites.models import Site	1✔
14	from django.core import serializers	1✔
15	from django.core.paginator import Paginator	1✔
16	from django.core.serializers import serialize	1✔
17	from django.db.models import Min, Q	1✔
18	from django.http import JsonResponse	1✔
19	from django.shortcuts import render, redirect, get_object_or_404	1✔
20	from django.urls import reverse	1✔
21	from django.views.decorators.csrf import csrf_exempt	1✔
22	from django.views.decorators.csrf import ensure_csrf_cookie	1✔
23	from django.utils import timezone	1✔
24	from users.decorators import user_type_required	1✔
25	from users.forms import UserSignUpForm, RentalListingForm	1✔
26	from .decorators import no_cache	1✔
27	from .forms import CustomLoginForm	1✔
28	from .forms import CustomUserEditForm	1✔
29	from .forms import LandlordSignupForm	1✔
30	from .models import Favorite, Rental_Listings	1✔
31	from .models import RentalImages	1✔
32	from .utils import send_email_to_admin	1✔
33
34	logger = logging.getLogger(__name__)	1✔
35
36	LOGGING = {	1✔
37	"version": 1,
38	"handlers": {
39	"console": {
40	"class": "logging.StreamHandler",
41	"stream": sys.stdout,
42	}
43	},
44	"root": {"handlers": ["console"], "level": "INFO"},
45	}
46
47	logging.config.dictConfig(LOGGING)	1✔
48
49
50	def custom_404_handler(request, exception):	1✔
51	if request.user.is_authenticated:	1✔
52	if getattr(request.user, "user_type", None) == "landlord":	1✔
53	return redirect("landlord_homepage")	×
54	else:
55	return redirect("user_homepage")	1✔
56	else:
57	return redirect("index")	1✔
58
59
60	def login_process(request, user_type, this_page, destination_url_name):	1✔
61	if request.method != "POST":	1✔
62	form = CustomLoginForm()	1✔
63	return render(request, this_page, {"form": form})	1✔
64
65	form = CustomLoginForm(request,	1✔
66	request.POST) # Pass the request to the form	1✔
67	if form.is_valid():	1✔
68	username = form.cleaned_data["username"]	1✔
69	password = form.cleaned_data["password"]	1✔
70	user = authenticate(request, username=username, password=password)
71		1✔
UNCOV 72	if user is None:	×
73	messages.error(request, "Invalid credentials!")	×
74	return render(request, this_page, {"form": form})
75		1✔
UNCOV 76	if getattr(user, "user_type", None) != user_type:	×
77	messages.error(
78	request,
79	f"Please provide correct credentials to login as "
80	f"{user_type.capitalize()}!",
81	# noqa:<E501>
UNCOV 82	)	×
83	return render(request, this_page, {"form": form})
84	# if user_type == "landlord" and user.verified is False:
85	# messages.error(
86	# request,
87	# "Your account has not been verified by the admin yet. Please wait!",
88	# )
89	# return render(request, this_page, {"form": form})
90	# ..	1✔
91	login(request, user)	1✔
92	return redirect(reverse(destination_url_name))
UNCOV 93		×
94	return render(request, this_page, {"form": form})
95
96		1✔
97	@no_cache	1✔
98	def landlord_login(request):	1✔
99	return login_process(
100	request,
101	user_type="landlord",
102	this_page="login/landlord_login.html",
103	destination_url_name="landlord_homepage",
104	)
105
106		1✔
107	@no_cache	1✔
108	def user_login(request):	1✔
109	return login_process(
110	request,
111	user_type="user",
112	this_page="login/user_login.html",
113	destination_url_name="user_homepage",
114	# URL pattern name for user's homepage
115	)
116
117		1✔
118	@no_cache	1✔
119	@ensure_csrf_cookie	1✔
120	def user_signup(request):	1✔
121	if request.method == "POST":	1✔
122	form = UserSignUpForm(request.POST)	1✔
123	if form.is_valid():	1✔
124	email = form.cleaned_data["email"]	1✔
125	user = form.save(commit=False)	1✔
126	user.username = email	1✔
127	user.save()	1✔
128	if user.user_type == user.USER:	1✔
129	user.verified = True
UNCOV 130	else:	×
131	user.verified = False	1✔
132	login(request, user,	1✔
133	backend="django.contrib.auth.backends.ModelBackend")
134	return redirect("user_homepage")	1✔
135	else:	1✔
136	form = UserSignUpForm()	1✔
137	for field, errors in form.errors.items():	1✔
138	for error in errors:
139	messages.error(request, f"{error}")	1✔
140
141	return render(request, "users/signup/signup.html", {"form": form})
142		1✔
143		1✔
144	@no_cache	1✔
145	def home(request):
146	return render(request, "home.html")
147		1✔
148		1✔
UNCOV 149	@no_cache	×
UNCOV 150	def logout_view(request):	×
151	logout(request)
152	return redirect("/")
153		1✔
154		1✔
155	@no_cache	1✔
156	@user_type_required("user")	1✔
157	def user_home(request):
158	return render(request, "user_homepage.html")
159		1✔
160		1✔
161	@no_cache	1✔
UNCOV 162	@user_type_required("landlord")	×
163	def landlord_home(request):
164	listings = (
165	Rental_Listings.objects.filter(Landlord=request.user)
166	.annotate(first_image=Min("images__image_url"))
NEW 167	.order_by("-Submitted_date")	×
168	)
169	return render(request, "landlord_homepage.html", {"listings": listings})
170		1✔
171		1✔
172	@no_cache	1✔
173	def landlord_signup(request):	1✔
174	if request.method == "POST":	1✔
175	form = LandlordSignupForm(request.POST, request.FILES)	1✔
176	if form.is_valid():	1✔
177	print(request.FILES)
178	user = form.save(commit=False)
179		1✔
180	# Upload file to S3 if present check	1✔
181	pdf_file = request.FILES.get("pdf_file")	1✔
182	print(pdf_file)	1✔
183	if pdf_file:	1✔
184	file_name, file_extension = os.path.splitext(pdf_file.name)	1✔
185	print(f"Received file: {pdf_file.name}") # Debug print	1✔
186	file_name = f"pdfs/{file_name}_{uuid.uuid4()}{file_extension}"	1✔
187	print(file_name)	1✔
188	try:
189	s3_client = boto3.client(
190	"s3",
191	aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
192	aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
193	)
194		1✔
195	# Check if file already exists in S3 bucket
196	existing_files = s3_client.list_objects_v2(
197	Bucket=settings.AWS_STORAGE_BUCKET_NAME,
198	Prefix=file_name	1✔
UNCOV 199	)	×
200
201	if "Contents" in existing_files:
202	messages.error(
203	request,
UNCOV 204	"A file with the same name already exists."	×
205	" Please rename your file and try again.",
206	)
207	return render(
208	request, "signup/landlord_signup.html",	1✔
209	{"form": form}
210	)
211		1✔
212	s3_client.upload_fileobj(
213	pdf_file, "landlord-verification-files", file_name
214	)
215	user.s3_doclink = (	1✔
UNCOV 216	f"https://landlord-verification-files."	×
UNCOV 217	f"s3.amazonaws.com/{file_name}"	×
218	)	1✔
219	print("File uploaded successfully")	1✔
220	except Exception as e:
221	print(f"Error uploading file to S3: {e}")	1✔
222	user.save()	1✔
223	send_email_to_admin(user.username)
224		1✔
225	messages.success(request, "Registration successful. Please log in.")
226	return redirect("landlord_login")
227	else:
228	messages.error(	1✔
229	request, "Registration failed. Please correct the errors below."	1✔
230	)
231	else:
232	form = LandlordSignupForm()	1✔
233	return render(request, "signup/landlord_signup.html", {"form": form})	1✔
234
235		1✔
236	@no_cache
237	def apply_filters(listings, filter_params):	1✔
238	# TODO: fix in database	1✔
239	listings = listings.exclude(neighborhood="Hell's Kitchen")	1✔
240	# Apply filters
241	if filter_params.get("borough"):	×
242	borough = filter_params.get("borough")
243	if borough == "All(NYC)":
244	# No need to filter by borough if "All(NYC)" is selected	1✔
245	pass	1✔
246	else:	1✔
247	# Filter by neighborhood or borough using exact match	1✔
248	listings = listings.filter(	1✔
249	Q(neighborhood=borough) \| Q(borough=borough))	1✔
250	if filter_params.get("min_price"):	1✔
251	min_price = int(filter_params.get("min_price"))	1✔
252	listings = listings.filter(price__gte=min_price)	1✔
253	if filter_params.get("max_price"):	×
254	max_price = int(filter_params.get("max_price"))
255	listings = listings.filter(price__lte=max_price)	1✔
256	if filter_params.get("bedrooms"):	1✔
257	if filter_params.get("bedrooms") == "Any":	1✔
258	listings = listings	1✔
259	else:
260	listings = listings.filter(beds=filter_params.get("bedrooms"))	1✔
261	if filter_params.get("bathrooms"):	1✔
262	if filter_params.get("bathrooms") == "Any":	1✔
263	listings = listings	1✔
264	else:	1✔
265	listings = listings.filter(baths=filter_params.get("bathrooms"))	1✔
266	if filter_params.get("elevator"):	1✔
267	listings = listings.filter(elevator=True)	1✔
268	if filter_params.get("laundry"):	1✔
269	listings = listings.filter(washer_dryer_in_unit=True)	×
270	if filter_params.get("no_fee"):
271	listings = listings.filter(broker_fee=0)	1✔
272	if filter_params.get("building_type"):	1✔
273	if filter_params.get("building_type") == "Any":	1✔
274	listings = listings	1✔
275	else:	1✔
276	listings = listings.filter(	1✔
277	unit_type=filter_params.get("building_type"))
278	if filter_params.get("parking"):
279	listings = listings.filter(parking_available=True)
280	if filter_params.get("search_query"):
281	query = SearchQuery(filter_params.get("search_query"))
282	listings = (
283	listings.annotate(
284	search=SearchVector("address"),	1✔
285	rank=SearchRank(SearchVector("address"), query),
286	)
287	.filter(search=query)	1✔
288	.order_by("-rank")	1✔
289	)	1✔
290	return listings	×
291
292
293	@no_cache
294	@user_type_required("user")
295	def rentals_page(request):
296	filter_params = {
297	"borough": request.GET.get("borough"),
298	"min_price": request.GET.get("min_price"),
299	"max_price": request.GET.get("max_price"),
300	"bedrooms": request.GET.get("bedrooms"),
301	"bathrooms": request.GET.get("bathrooms"),
302	"elevator": request.GET.get("elevator") == "on",
303	"laundry": request.GET.get("laundry") == "on",
UNCOV 304	"no_fee": request.GET.get("no_fee") == "on",	×
UNCOV 305	"building_type": request.GET.get("building_type"),	×
UNCOV 306	"parking": request.GET.get("parking") == "on",	×
UNCOV 307	"search_query": request.GET.get("search_query", ""),	×
UNCOV 308	}	×
UNCOV 309		×
310	listings = Rental_Listings.objects.all()
311	listings = apply_filters(listings, filter_params)
UNCOV 312	listings = listings.annotate(first_image=Min("images__image_url"))	×
UNCOV 313	sort_option = request.GET.get("sort")	×
314	if sort_option == "recent":
315	listings = listings.order_by(	×
UNCOV 316	"-Submitted_date"	×
317	) # Assuming you have a created_at field
318	elif sort_option == "high_to_low":
319	listings = listings.order_by("-price")	×
UNCOV 320	else:	×
321	listings = listings.order_by("price")	×
NEW 322	favorite_listings_ids = Favorite.objects.filter(	×
NEW 323	user=request.user).values_list(	×
324	"listing__id", flat=True
325	)
326	filter_params = {k: v for k, v in filter_params.items() if v is not None}	×
327	paginator = Paginator(listings, 5) # Show 5 listings per page
328	page_number = request.GET.get("page")
329	page_obj = paginator.get_page(page_number)
330	favorite_listings_ids = Favorite.objects.filter(
331	user=request.user).values_list(
UNCOV 332	"listing__id", flat=True	×
333	)
334	context = {
335	"page_obj": page_obj,	1✔
336	"listings": listings,	1✔
UNCOV 337	"favorite_listings_ids": list(favorite_listings_ids),	×
UNCOV 338	"filter_params": filter_params, # Ensure it's converted to a list	×
UNCOV 339	}	×
UNCOV 340	return render(request, "users/searchRental/rentalspage.html", context)	×
341		×
UNCOV 342		×
NEW 343	@user_type_required("landlord")	×
NEW 344	def add_rental_listing(request):	×
NEW 345	if request.method == "POST":	×
NEW 346	form = RentalListingForm(request.POST, request.FILES)	×
NEW 347	images = request.FILES.getlist("photo")	×
NEW 348	if form.is_valid():	×
NEW 349	rental_listing = form.save(commit=False)	×
NEW 350	rental_listing.Landlord = request.user	×
NEW 351	rental_listing.Submitted_date = timezone.now().date()	×
NEW 352	apt_no = form.cleaned_data.get('apt_no', '')	×
NEW 353	if apt_no:	×
NEW 354	base_address = rental_listing.address.split(',')[0]	×
NEW 355	full_address = f"{base_address} #{apt_no}"	×
356	rental_listing.address = full_address
357	rental_listing.save()
358	AWS_STORAGE_BUCKET_NAME = "landlord-verification-files"
359	for image in images:
NEW 360	file_name, file_extension = os.path.splitext(image.name)	×
361	unique_file_name = f"pdfs/{uuid.uuid4()}{file_extension}"
362	try:
363	s3_client = boto3.client(
364	"s3",
NEW 365	aws_access_key_id=settings.AWS_ACCESS_KEY_ID,	×
NEW 366	aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,	×
367	)
368	s3_client.upload_fileobj(
NEW 369	image,	×
NEW 370	AWS_STORAGE_BUCKET_NAME,	×
NEW 371	unique_file_name,	×
372	)
NEW 373	image_url = f"https://{AWS_STORAGE_BUCKET_NAME}.s3.amazonaws.com/{unique_file_name}"	×
NEW 374	RentalImages.objects.create(	×
375	rental_listing=rental_listing, image_url=image_url
376	)
377	except Exception as e:	1✔
378	print(f"Error uploading file to S3: {e}")	1✔
379	return redirect("landlord_homepage")	1✔
NEW 380	else:	×
381	form = RentalListingForm()
382	return render(request, "add_rental_listing.html", {"form": form})
383		1✔
384		1✔
UNCOV 385	@no_cache	×
386	@login_required
387	def placeholder_view(request):
388	return render(request, "users/searchRental/placeholder.html")	1✔
389		1✔
390
391	@user_type_required("landlord")	1✔
392	def landlord_placeholder_view(request):
NEW 393	return render(request, "users/searchRental/landlord_placeholder.html")	×
NEW 394		×
395
396	@no_cache
397	def listing_detail(request, listing_id):	1✔
398	# Retrieve the specific listing based on the ID provided in the URL parameter	1✔
399	listing = get_object_or_404(Rental_Listings, id=listing_id)	1✔
400		1✔
401	context = {"listing": listing}	1✔
402	return render(request, "users/searchRental/listing_detail.html", context)	1✔
403		1✔
404		1✔
405	@no_cache	1✔
406	@csrf_exempt
407	@login_required	1✔
408	@user_type_required("user")	1✔
409	def toggle_favorite(request):	1✔
410	if request.method == "POST":
411	listing_id = request.POST.get("listing_id")
412	if not listing_id:	1✔
413	return JsonResponse({"error": "Listing ID is required"}, status=400)	1✔
414		1✔
415	try:	1✔
416	listing = Rental_Listings.objects.get(id=listing_id)	1✔
417	favorite, created = Favorite.objects.get_or_create(	1✔
UNCOV 418	user=request.user, listing=listing	×
UNCOV 419	)	×
UNCOV 420	if not created:	×
421	favorite.delete()
422	return JsonResponse({"status": "removed"})
423	return JsonResponse({"status": "added"})	1✔
424	except Rental_Listings.DoesNotExist:	1✔
425	return JsonResponse({"error": "Listing not found"}, status=404)	1✔
426	except Exception as e:	1✔
427	logger.error(f"Internal server error: {e}", exc_info=True)	×
428	return JsonResponse({"error": "Internal server error"}, status=500)
429
UNCOV 430		×
431	@user_type_required("user")
UNCOV 432	@login_required	×
UNCOV 433	@no_cache	×
434	def favorites_page(request):
NEW 435	favorite_listings = Favorite.objects.filter(	×
436	user=request.user).select_related(
437	"listing"
438	)
439	listings = [fav.listing for fav in favorite_listings]	×
440
441	listings_json = serializers.serialize("json", listings)
442	favorite_listings_ids = [listing.id for listing in listings]	1✔
443		1✔
444	context = {	×
UNCOV 445	"listings_json": listings_json,	×
446	"favorite_listings_ids": favorite_listings_ids,
UNCOV 447	}	×
448	return render(request, "users/searchRental/favorites.html", context)	×
UNCOV 449		×
UNCOV 450		×
UNCOV 451	@no_cache	×
452	def map_view(request):
453	filter_params = ast.literal_eval(request.GET.get("filter_params"))
454	rental_listings = Rental_Listings.objects.all()
UNCOV 455	# http://127.0.0.1:8000/map/?filter_params={%27borough%27:%20%27Manhattan%27,%20%27min_price%27:%20%27%27,%20%27max_price%27:%20%27%27}	×
456	rental_listings = apply_filters(rental_listings, filter_params)
457	rental_listings_json = serialize("json", rental_listings)
458	current_site = Site.objects.get_current()	1✔
459	current_site.domain	1✔
460	context = {	1✔
461	"rental_listings": rental_listings_json,	1✔
462	"this_domain": current_site.domain,	1✔
463	}	1✔
464	return render(request, "users/searchRental/map_view.html", context)	1✔
465		1✔
466	@login_required	1✔
467	@no_cache	1✔
468	@user_type_required("user")
469	def profile_view_edit(request):	1✔
470	if request.method == "POST":
471	form = CustomUserEditForm(request.POST, instance=request.user)	1✔
472	if form.is_valid():
473	form.save()
474	messages.success(request, "Your profile was successfully updated!")
475	return redirect("profile_view_edit")
476	else:
477	form = CustomUserEditForm(instance=request.user)
478		1✔
479	return render(	1✔
480	request,	1✔
481	"users/Profile/profile_view_edit.html",	1✔
482	{"form": form, "user": request.user},	1✔
483	)	1✔
484		1✔
485		1✔
486	@login_required	1✔
487	@no_cache	1✔
488	@user_type_required("landlord")
489	def landlord_profile_update(request):	1✔
490	if request.method == "POST":
491	form = CustomUserEditForm(request.POST, instance=request.user)	1✔
492	if form.is_valid():
493	form.save()
494	messages.success(request, "Your profile was successfully updated!")
495	return redirect("landlord_profile_update")
496	else:
497	form = CustomUserEditForm(instance=request.user)
498
499	return render(
500	request,
501	"users/Profile/landlord_profile_update.html",
502	{"form": form, "user": request.user},
503	)

gcivil-nyu-org / INET-Wednesday-Spring2024-Team-2 / 603

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous