385 lines
15 KiB
Python
385 lines
15 KiB
Python
"""
|
|
Importa imóveis de imoveis.csv + fotos.csv para o banco de dados.
|
|
|
|
Uso:
|
|
cd backend
|
|
uv run python seeds/import_from_csv.py [--csv-dir CAMINHO]
|
|
|
|
Padrão de CAMINHO: ../../aluguel_helper (relativo ao diretório backend/)
|
|
Idempotente: pula imóveis cujo `code` já existe no banco.
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import os
|
|
import re
|
|
import sys
|
|
import unicodedata
|
|
from collections import defaultdict
|
|
from decimal import Decimal
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from app import create_app
|
|
from app.extensions import db
|
|
from app.models.location import City, Neighborhood
|
|
from app.models.property import Property, PropertyPhoto
|
|
from app.models.catalog import Amenity, PropertyType
|
|
from app.models.imobiliaria import Imobiliaria
|
|
|
|
# ── Configurações ─────────────────────────────────────────────────────────────
|
|
|
|
CITY_NAME = "Franca"
|
|
CITY_STATE = "SP"
|
|
|
|
# Mapeamento tipo_imovel (CSV) → slug do subtipo (já criado pelo seed.py)
|
|
SUBTYPE_MAP: dict[str, str] = {
|
|
"casa": "casa",
|
|
"apartamento": "apartamento",
|
|
"flat/studio": "flat",
|
|
"store": "loja-salao-ponto-comercial",
|
|
"landform": "terreno-lote-condominio",
|
|
}
|
|
|
|
# Mapeamento amenidade CSV (inglês) → slug português do banco
|
|
# Valores sem correspondência no sistema são ignorados (None).
|
|
AMENITY_MAP: dict[str, str | None] = {
|
|
# ── Características ───────────────────────────────────────────────────────
|
|
"PETS_ALLOWED": "aceita-animais",
|
|
"AIR_CONDITIONING": "ar-condicionado",
|
|
"SERVICE_AREA": "area-de-servico",
|
|
"BUILTIN_WARDROBE": "armario-embutido",
|
|
"BEDROOM_WARDROBE": "armario-embutido-no-quarto",
|
|
"KITCHEN_CABINETS": "armario-na-cozinha",
|
|
"BATHROOM_CABINETS": "armario-no-banheiro",
|
|
"BLINDEX_BOX": "box-blindex",
|
|
"CLOSET": "closet",
|
|
"DRESS_ROOM2": "closet",
|
|
"INTERNET_ACCESS": "conexao-a-internet",
|
|
"AMERICAN_KITCHEN": "cozinha-americana",
|
|
"LARGE_KITCHEN": "cozinha-americana",
|
|
"GOURMET_KITCHEN": "cozinha-gourmet",
|
|
"COOKER": "fogao",
|
|
"INTERCOM": "interfone",
|
|
"LARGE_WINDOW": "janela-grande",
|
|
"FURNISHED": "mobiliado",
|
|
"PLANNED_FURNITURE": "mobiliado",
|
|
"BACKYARD": "quintal",
|
|
"CABLE_TV": "tv-a-cabo",
|
|
"BALCONY": "varanda",
|
|
"WALL_BALCONY": "varanda",
|
|
"GOURMET_BALCONY": "varanda-gourmet",
|
|
"BARBECUE_BALCONY": "varanda-gourmet",
|
|
"NATURAL_VENTILATION": "ventilacao-natural",
|
|
"PANORAMIC_VIEW": "vista-panoramica",
|
|
"EXTERIOR_VIEW": "vista-panoramica",
|
|
"GAS_SHOWER": "aquecedor-a-gas",
|
|
"HEATING": "aquecimento-central",
|
|
"BATHTUB": "banheira",
|
|
"SERVICE_BATHROOM": "banheiro-de-servico",
|
|
"COPA": "copa",
|
|
"LUNCH_ROOM": "copa",
|
|
"EMPLOYEE_DEPENDENCY": "dependencia-de-empregada",
|
|
"DEPOSIT": "deposito",
|
|
"EDICULE": "edicula",
|
|
"SOLAR_ENERGY": "energia-solar",
|
|
"PET_SPACE": "espaco-pet",
|
|
"ENTRANCE_HALL": "hall-de-entrada",
|
|
"HOME_OFFICE": "home-office",
|
|
"CORNER_PROPERTY": "imovel-de-esquina",
|
|
"SOUNDPROOFING": "isolamento-acustico",
|
|
"ALUMINUM_WINDOW": "janela-de-aluminio",
|
|
"LAVABO": "lavabo",
|
|
"MEZZANINE": "mezanino",
|
|
"WOOD_FLOOR": "piso-de-madeira",
|
|
"LAMINATED_FLOOR": "piso-laminado",
|
|
"VINYL_FLOOR": "piso-vinilico",
|
|
"PORCELAIN": "porcelanato",
|
|
"DINNER_ROOM": "sala-de-jantar",
|
|
"PANTRY": "despensa",
|
|
"SERVICE_ENTRANCE": "entrada-de-servico",
|
|
"INTEGRATED_ENVIRONMENTS": "ambientes-integrados",
|
|
# ── Lazer ─────────────────────────────────────────────────────────────────
|
|
"FITNESS_ROOM": "academia",
|
|
"GYM": "academia",
|
|
"BAR": "bar",
|
|
"CINEMA": "cinema",
|
|
"BARBECUE_GRILL": "churrasqueira",
|
|
"PIZZA_OVEN": "churrasqueira",
|
|
"GOURMET_SPACE": "espaco-gourmet",
|
|
"GREEN_SPACE": "espaco-verde-parque",
|
|
"RECREATION_AREA": "espaco-verde-parque",
|
|
"WHIRLPOOL": "hidromassagem",
|
|
"GARDEN": "jardim",
|
|
"POOL": "piscina",
|
|
"PRIVATE_POOL": "piscina",
|
|
"ADULT_POOL": "piscina",
|
|
"HEATED_POOL": "piscina",
|
|
"CHILDRENS_POOL": "piscina",
|
|
"PLAYGROUND": "playground",
|
|
"TOYS_PLACE": "playground",
|
|
"SPORTS_COURT": "quadra-poliesportiva",
|
|
"FOOTBALL_FIELD": "quadra-poliesportiva",
|
|
"GAMES_ROOM": "sala-de-jogos",
|
|
"ADULT_GAME_ROOM": "sala-de-jogos",
|
|
"YOUTH_GAME_ROOM": "sala-de-jogos",
|
|
"TEEN_SPACE": "sala-de-jogos",
|
|
"PARTY_HALL": "salao-de-festas",
|
|
"SAUNA": "sauna",
|
|
"SPA": "spa",
|
|
"MASSAGE_ROOM": "spa",
|
|
# ── Condomínio ────────────────────────────────────────────────────────────
|
|
"DISABLED_ACCESS": "acesso-para-deficientes",
|
|
"BICYCLES_PLACE": "bicicletario",
|
|
"ELEVATOR": "elevador",
|
|
"ELECTRIC_GENERATOR": "gerador-eletrico",
|
|
"GRASS": "gramado",
|
|
"LAUNDRY": "lavanderia",
|
|
"RECEPTION": "recepcao",
|
|
"MEETING_ROOM": "sala-de-reunioes",
|
|
"COVENTION_HALL": "salao-de-convencoes",
|
|
"GUEST_PARKING": "vaga-para-visitante",
|
|
"RESTAURANT": "restaurante",
|
|
# ── Segurança ─────────────────────────────────────────────────────────────
|
|
"FENCE": "cerca-muro",
|
|
"SAFETY_CIRCUIT": "circuito-de-seguranca",
|
|
"SECURITY_CAMERA": "circuito-de-seguranca",
|
|
"GATED_COMMUNITY": "condominio-fechado",
|
|
"ELECTRONIC_GATE": "portao-eletronico",
|
|
"CONCIERGE_24H": "portaria-24h",
|
|
"ALARM_SYSTEM": "sistema-de-alarme",
|
|
"WATCHMAN": "vigia",
|
|
"PATROL": "vigia",
|
|
"SECURITY_CABIN": "vigia",
|
|
}
|
|
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _slugify(text: str) -> str:
|
|
text = unicodedata.normalize("NFD", text)
|
|
text = "".join(c for c in text if unicodedata.category(c) != "Mn")
|
|
text = text.lower()
|
|
text = re.sub(r"[^a-z0-9\s-]", "", text)
|
|
text = re.sub(r"[\s/]+", "-", text.strip())
|
|
return text
|
|
|
|
|
|
def _decimal_or_none(value: str) -> Decimal | None:
|
|
v = value.strip()
|
|
if not v:
|
|
return None
|
|
try:
|
|
return Decimal(v)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _int_or_zero(value: str) -> int:
|
|
v = value.strip()
|
|
if not v:
|
|
return 0
|
|
try:
|
|
return int(float(v))
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
# ── Leitura dos CSVs ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def load_imoveis(csv_dir: str) -> list[dict]:
|
|
path = os.path.join(csv_dir, "imoveis.csv")
|
|
with open(path, encoding="utf-8", newline="") as f:
|
|
return list(csv.DictReader(f))
|
|
|
|
|
|
def load_fotos(csv_dir: str) -> dict[str, list[str]]:
|
|
"""Retorna {id_imovel: [url_ordenada_por_indice]}"""
|
|
path = os.path.join(csv_dir, "fotos.csv")
|
|
grouped: dict[str, list[tuple[int, str]]] = defaultdict(list)
|
|
with open(path, encoding="utf-8", newline="") as f:
|
|
for row in csv.DictReader(f):
|
|
grouped[row["id_imovel"]].append((int(row["indice"]), row["url"]))
|
|
return {
|
|
imovel_id: [url for _, url in sorted(fotos)]
|
|
for imovel_id, fotos in grouped.items()
|
|
}
|
|
|
|
|
|
# ── Importação ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def importar(csv_dir: str) -> None:
|
|
app = create_app()
|
|
with app.app_context():
|
|
imoveis = load_imoveis(csv_dir)
|
|
fotos_map = load_fotos(csv_dir)
|
|
|
|
# ── Cidade Franca ─────────────────────────────────────────────────────
|
|
city_slug = _slugify(CITY_NAME)
|
|
city = City.query.filter_by(slug=city_slug).first()
|
|
if not city:
|
|
city = City(name=CITY_NAME, slug=city_slug, state=CITY_STATE)
|
|
db.session.add(city)
|
|
db.session.flush()
|
|
print(f" Cidade criada: {CITY_NAME}")
|
|
else:
|
|
print(f" Cidade existente: {CITY_NAME}")
|
|
|
|
# ── Cache de bairros ──────────────────────────────────────────────────
|
|
neighborhood_cache: dict[str, Neighborhood] = {
|
|
nbh.slug: nbh for nbh in Neighborhood.query.filter_by(city_id=city.id).all()
|
|
}
|
|
|
|
def get_or_create_neighborhood(name: str) -> Neighborhood:
|
|
slug = _slugify(name)
|
|
if slug not in neighborhood_cache:
|
|
nbh = Neighborhood(name=name, slug=slug, city_id=city.id)
|
|
db.session.add(nbh)
|
|
db.session.flush()
|
|
neighborhood_cache[slug] = nbh
|
|
return neighborhood_cache[slug]
|
|
|
|
# ── Cache de subtipos ─────────────────────────────────────────────────
|
|
subtype_cache: dict[str, PropertyType] = {
|
|
pt.slug: pt for pt in PropertyType.query.all()
|
|
}
|
|
# ── Cache de amenidades ───────────────────────────────────────────────
|
|
amenity_cache: dict[str, Amenity] = {a.slug: a for a in Amenity.query.all()}
|
|
# ── Cache de imobiliárias (cria se não existir) ───────────────────────
|
|
imob_cache: dict[str, Imobiliaria] = {
|
|
i.name: i for i in Imobiliaria.query.all()
|
|
}
|
|
|
|
def get_or_create_imobiliaria(name: str) -> Imobiliaria:
|
|
if name not in imob_cache:
|
|
imob = Imobiliaria(name=name, is_active=True, display_order=0)
|
|
db.session.add(imob)
|
|
db.session.flush()
|
|
imob_cache[name] = imob
|
|
print(f" Imobiliária criada: {name}")
|
|
return imob_cache[name]
|
|
|
|
# ── Imóveis existentes (por code) ─────────────────────────────────────
|
|
existing_codes: set[str] = {
|
|
p.code for p in Property.query.with_entities(Property.code).all() if p.code
|
|
}
|
|
|
|
created = 0
|
|
skipped = 0
|
|
|
|
updated = 0
|
|
|
|
for row in imoveis:
|
|
code = row["id"].strip()
|
|
titulo = row["titulo"].strip()
|
|
listing_type = "aluguel" if "alugar" in titulo.lower() else "venda"
|
|
|
|
if code in existing_codes:
|
|
# Corrige o type de registros já importados
|
|
rows_updated = (
|
|
Property.query.filter_by(code=code)
|
|
.update({"type": listing_type})
|
|
)
|
|
if rows_updated:
|
|
updated += 1
|
|
skipped += 1
|
|
continue
|
|
|
|
# Bairro
|
|
bairro_nome = row["bairro"].strip()
|
|
nbh = get_or_create_neighborhood(bairro_nome) if bairro_nome else None
|
|
|
|
# Subtipo
|
|
tipo_raw = row["tipo_imovel"].strip().lower()
|
|
subtype_slug = SUBTYPE_MAP.get(tipo_raw, "casa")
|
|
subtype = subtype_cache.get(subtype_slug)
|
|
if subtype is None:
|
|
# Fallback: pega qualquer subtipo residencial
|
|
subtype = subtype_cache.get("casa")
|
|
base_slug = _slugify(titulo)[:200]
|
|
slug = base_slug
|
|
suffix = 1
|
|
while Property.query.filter_by(slug=slug).first():
|
|
slug = f"{base_slug}-{suffix}"
|
|
suffix += 1
|
|
|
|
# Amenidades
|
|
amenidade_raw = row.get("amenidades", "").strip()
|
|
prop_amenities: list[Amenity] = []
|
|
if amenidade_raw:
|
|
for code_en in amenidade_raw.split("|"):
|
|
slug_pt = AMENITY_MAP.get(code_en.strip())
|
|
if slug_pt and slug_pt in amenity_cache:
|
|
amenity = amenity_cache[slug_pt]
|
|
if amenity not in prop_amenities:
|
|
prop_amenities.append(amenity)
|
|
|
|
# Imobiliária
|
|
imob_name = row.get("imobiliaria", "").strip()
|
|
imob = get_or_create_imobiliaria(imob_name) if imob_name else None
|
|
|
|
listing_type = "aluguel" if "alugar" in titulo.lower() else "venda"
|
|
|
|
prop = Property(
|
|
title=titulo,
|
|
slug=slug,
|
|
code=code,
|
|
description=row["descricao"].strip() or None,
|
|
address=row["endereco"].strip() or None,
|
|
price=Decimal(row["preco_aluguel"].strip()),
|
|
condo_fee=_decimal_or_none(row["condominio"]),
|
|
iptu_anual=_decimal_or_none(row["iptu"]),
|
|
type=listing_type,
|
|
subtype_id=subtype.id if subtype else None,
|
|
city_id=city.id,
|
|
neighborhood_id=nbh.id if nbh else None,
|
|
bedrooms=_int_or_zero(row["quartos"]),
|
|
bathrooms=_int_or_zero(row["banheiros"]),
|
|
parking_spots=_int_or_zero(row["vagas_garagem"]),
|
|
area_m2=_int_or_zero(row["area_m2"]) or 1,
|
|
is_featured=False,
|
|
is_active=True,
|
|
imobiliaria_id=imob.id if imob else None,
|
|
)
|
|
prop.amenities = prop_amenities
|
|
db.session.add(prop)
|
|
db.session.flush()
|
|
|
|
# Fotos
|
|
for order, url in enumerate(fotos_map.get(code, [])):
|
|
db.session.add(
|
|
PropertyPhoto(
|
|
property_id=prop.id,
|
|
url=url,
|
|
alt_text=titulo,
|
|
display_order=order,
|
|
)
|
|
)
|
|
|
|
existing_codes.add(code)
|
|
created += 1
|
|
|
|
db.session.commit()
|
|
print(
|
|
f"\nImportação concluída: {created} imóveis criados, "
|
|
f"{updated} tipo(s) atualizado(s), {skipped - updated} já existiam sem alteração."
|
|
)
|
|
|
|
|
|
# ── Entrypoint ─────────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Importa imóveis do aluguel_helper.")
|
|
parser.add_argument(
|
|
"--csv-dir",
|
|
default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "data"),
|
|
help="Diretório contendo imoveis.csv e fotos.csv",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
csv_dir = os.path.abspath(args.csv_dir)
|
|
print(f"==> Lendo CSVs de: {csv_dir}")
|
|
importar(csv_dir)
|