262 lines
9.6 KiB
Python
262 lines
9.6 KiB
Python
import math
|
|
|
|
from flask import Blueprint, jsonify, request
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import and_, or_
|
|
from sqlalchemy.orm import aliased
|
|
|
|
from app.extensions import db
|
|
from app.models.catalog import Amenity, property_amenity
|
|
from app.models.location import Neighborhood
|
|
from app.models.property import Property
|
|
from app.schemas.property import PaginatedPropertiesOut, PropertyDetailOut, PropertyOut
|
|
|
|
properties_bp = Blueprint("properties", __name__, url_prefix="/api/v1")
|
|
|
|
|
|
def _parse_int(value: str | None, default: int | None = None) -> int | None:
|
|
if value is None:
|
|
return default
|
|
try:
|
|
return int(value)
|
|
except (ValueError, TypeError):
|
|
return default
|
|
|
|
|
|
def _parse_float(value: str | None, default: float | None = None) -> float | None:
|
|
if value is None:
|
|
return default
|
|
try:
|
|
return float(value)
|
|
except (ValueError, TypeError):
|
|
return default
|
|
|
|
|
|
@properties_bp.get("/properties")
|
|
def list_properties():
|
|
args = request.args
|
|
|
|
# ── Featured shortcut (homepage) ────────────────────────────────────────
|
|
if args.get("featured", "").lower() == "true":
|
|
from app.models.homepage import HomepageConfig
|
|
|
|
config = HomepageConfig.query.first()
|
|
limit = config.featured_properties_limit if config else 6
|
|
props = (
|
|
Property.query.filter_by(is_active=True, is_featured=True)
|
|
.order_by(Property.created_at.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return jsonify(
|
|
[PropertyOut.model_validate(p).model_dump(mode="json") for p in props]
|
|
)
|
|
|
|
# ── Base query ───────────────────────────────────────────────────────────
|
|
query = Property.query.filter_by(is_active=True)
|
|
|
|
# Listing type (venda | aluguel)
|
|
listing_type = args.get("listing_type", "").lower()
|
|
if listing_type in ("venda", "aluguel"):
|
|
query = query.filter(Property.type == listing_type)
|
|
|
|
# Subtype — supports subtype_ids (comma-sep) or legacy subtype_id
|
|
subtype_ids_raw = args.get("subtype_ids", "")
|
|
if subtype_ids_raw:
|
|
try:
|
|
subtype_ids_list = [int(x) for x in subtype_ids_raw.split(",") if x.strip()]
|
|
except ValueError:
|
|
subtype_ids_list = []
|
|
if subtype_ids_list:
|
|
query = query.filter(Property.subtype_id.in_(subtype_ids_list))
|
|
else:
|
|
subtype_id = _parse_int(args.get("subtype_id"))
|
|
if subtype_id is not None:
|
|
query = query.filter(Property.subtype_id == subtype_id)
|
|
|
|
# City
|
|
city_id = _parse_int(args.get("city_id"))
|
|
if city_id is not None:
|
|
query = query.filter(Property.city_id == city_id)
|
|
|
|
# Neighborhood — supports neighborhood_ids (comma-sep) or legacy neighborhood_id
|
|
neighborhood_ids_raw = args.get("neighborhood_ids", "")
|
|
if neighborhood_ids_raw:
|
|
try:
|
|
neighborhood_ids_list = [int(x) for x in neighborhood_ids_raw.split(",") if x.strip()]
|
|
except ValueError:
|
|
neighborhood_ids_list = []
|
|
if neighborhood_ids_list:
|
|
query = query.filter(Property.neighborhood_id.in_(neighborhood_ids_list))
|
|
else:
|
|
neighborhood_id = _parse_int(args.get("neighborhood_id"))
|
|
if neighborhood_id is not None:
|
|
query = query.filter(Property.neighborhood_id == neighborhood_id)
|
|
|
|
# Imobiliária
|
|
imobiliaria_id = _parse_int(args.get("imobiliaria_id"))
|
|
if imobiliaria_id is not None:
|
|
query = query.filter(Property.imobiliaria_id == imobiliaria_id)
|
|
|
|
# Price range
|
|
price_min = _parse_float(args.get("price_min"))
|
|
price_max = _parse_float(args.get("price_max"))
|
|
include_condo = args.get("include_condo", "").lower() == "true"
|
|
|
|
if price_min is not None or price_max is not None:
|
|
if include_condo:
|
|
# Effective price = price + coalesce(condo_fee, 0)
|
|
from sqlalchemy import func
|
|
|
|
effective = Property.price + func.coalesce(Property.condo_fee, 0)
|
|
else:
|
|
effective = Property.price
|
|
|
|
if price_min is not None:
|
|
query = query.filter(effective >= price_min)
|
|
if price_max is not None:
|
|
query = query.filter(effective <= price_max)
|
|
|
|
# Bedrooms
|
|
bedrooms_min = _parse_int(args.get("bedrooms_min"))
|
|
bedrooms_max = _parse_int(args.get("bedrooms_max"))
|
|
if bedrooms_min is not None:
|
|
query = query.filter(Property.bedrooms >= bedrooms_min)
|
|
if bedrooms_max is not None:
|
|
query = query.filter(Property.bedrooms <= bedrooms_max)
|
|
|
|
# Bathrooms
|
|
bathrooms_min = _parse_int(args.get("bathrooms_min"))
|
|
bathrooms_max = _parse_int(args.get("bathrooms_max"))
|
|
if bathrooms_min is not None:
|
|
query = query.filter(Property.bathrooms >= bathrooms_min)
|
|
if bathrooms_max is not None:
|
|
query = query.filter(Property.bathrooms <= bathrooms_max)
|
|
|
|
# Parking spots
|
|
parking_min = _parse_int(args.get("parking_min"))
|
|
parking_max = _parse_int(args.get("parking_max"))
|
|
if parking_min is not None:
|
|
query = query.filter(Property.parking_spots >= parking_min)
|
|
if parking_max is not None:
|
|
query = query.filter(Property.parking_spots <= parking_max)
|
|
|
|
# Area m²
|
|
area_min = _parse_int(args.get("area_min"))
|
|
area_max = _parse_int(args.get("area_max"))
|
|
if area_min is not None:
|
|
query = query.filter(Property.area_m2 >= area_min)
|
|
if area_max is not None:
|
|
query = query.filter(Property.area_m2 <= area_max)
|
|
|
|
# Amenities (AND logic — property must have ALL selected amenities)
|
|
amenity_ids_raw = args.get("amenity_ids", "")
|
|
if amenity_ids_raw:
|
|
try:
|
|
amenity_ids = [int(x) for x in amenity_ids_raw.split(",") if x.strip()]
|
|
except ValueError:
|
|
amenity_ids = []
|
|
|
|
for aid in amenity_ids:
|
|
query = query.filter(
|
|
Property.id.in_(
|
|
property_amenity.select()
|
|
.where(property_amenity.c.amenity_id == aid)
|
|
.with_only_columns(property_amenity.c.property_id)
|
|
)
|
|
)
|
|
|
|
# ── Text search ──────────────────────────────────────────────────────────
|
|
q_raw = args.get("q", "").strip()[:200]
|
|
if q_raw:
|
|
pattern = f"%{q_raw}%"
|
|
NeighborhoodAlias = aliased(Neighborhood)
|
|
query = (
|
|
query
|
|
.outerjoin(NeighborhoodAlias, Property.neighborhood_id == NeighborhoodAlias.id)
|
|
.filter(
|
|
or_(
|
|
Property.title.ilike(pattern),
|
|
Property.address.ilike(pattern),
|
|
Property.code.ilike(pattern),
|
|
NeighborhoodAlias.name.ilike(pattern),
|
|
)
|
|
)
|
|
)
|
|
|
|
# ── Sort ─────────────────────────────────────────────────────────────────
|
|
_sort_map = {
|
|
"price_asc": Property.price.asc(),
|
|
"price_desc": Property.price.desc(),
|
|
"area_desc": Property.area_m2.desc(),
|
|
"newest": Property.created_at.desc(),
|
|
"relevance": Property.created_at.desc(),
|
|
}
|
|
sort_key = args.get("sort", "relevance")
|
|
sort_order = _sort_map.get(sort_key, Property.created_at.desc())
|
|
|
|
# ── Pagination ───────────────────────────────────────────────────────────
|
|
page = max(1, _parse_int(args.get("page"), 1))
|
|
per_page = min(48, max(1, _parse_int(args.get("per_page"), 24)))
|
|
|
|
total = query.count()
|
|
pages = math.ceil(total / per_page) if total > 0 else 1
|
|
|
|
props = (
|
|
query.order_by(sort_order)
|
|
.offset((page - 1) * per_page)
|
|
.limit(per_page)
|
|
.all()
|
|
)
|
|
|
|
result = PaginatedPropertiesOut(
|
|
items=[PropertyOut.model_validate(p) for p in props],
|
|
total=total,
|
|
page=page,
|
|
per_page=per_page,
|
|
pages=pages,
|
|
)
|
|
return jsonify(result.model_dump(mode="json"))
|
|
|
|
|
|
@properties_bp.get("/properties/<slug>")
|
|
def get_property(slug: str):
|
|
prop = Property.query.filter_by(slug=slug, is_active=True).first()
|
|
if prop is None:
|
|
return jsonify({"error": "Im\u00f3vel n\u00e3o encontrado"}), 404
|
|
return jsonify(PropertyDetailOut.model_validate(prop).model_dump(mode="json"))
|
|
|
|
|
|
@properties_bp.post("/properties/<slug>/contact")
|
|
def contact_property(slug: str):
|
|
from app.models.lead import ContactLead
|
|
from app.schemas.lead import ContactLeadIn, ContactLeadCreatedOut
|
|
|
|
prop = Property.query.filter_by(slug=slug, is_active=True).first()
|
|
if prop is None:
|
|
return jsonify({"error": "Im\u00f3vel n\u00e3o encontrado"}), 404
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
try:
|
|
lead_in = ContactLeadIn.model_validate(data)
|
|
except ValidationError as exc:
|
|
return jsonify({"error": "Dados inv\u00e1lidos", "details": exc.errors()}), 422
|
|
|
|
lead = ContactLead(
|
|
property_id=prop.id,
|
|
name=lead_in.name,
|
|
email=lead_in.email,
|
|
phone=lead_in.phone,
|
|
message=lead_in.message,
|
|
)
|
|
db.session.add(lead)
|
|
db.session.commit()
|
|
|
|
return (
|
|
jsonify(
|
|
ContactLeadCreatedOut(
|
|
id=lead.id, message="Mensagem enviada com sucesso!"
|
|
).model_dump()
|
|
),
|
|
201,
|
|
)
|