""" Camada de acesso a dados: a planilha Excel funciona como banco de dados. Toda leitura/escrita passa por aqui. Usamos openpyxl e um lock simples para evitar condição de corrida entre requisições concorrentes (suficiente para um sistema enxuto de uso interno de uma imobiliária). """ import json import os import threading from datetime import datetime from typing import Optional from openpyxl import Workbook, load_workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils import get_column_letter from app.config import COLUMNS, HEADERS_DISPLAY, EXCEL_PATH, DATA_DIR, SHEET_NAME _lock = threading.Lock() # -------------------------------------------------------------------------- # Inicialização / criação do arquivo # -------------------------------------------------------------------------- def _aplicar_estilo_cabecalho(ws): header_font = Font(name="Arial", bold=True, color="FFFFFF", size=11) header_fill = PatternFill(start_color="2F5496", end_color="2F5496", fill_type="solid") for col_idx, header in enumerate(HEADERS_DISPLAY, start=1): cell = ws.cell(row=1, column=col_idx, value=header) cell.font = header_font cell.fill = header_fill cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) ws.row_dimensions[1].height = 30 ws.freeze_panes = "A2" def _largura_colunas(ws): larguras = { "id": 6, "tipo_negocio": 14, "tipo_imovel": 16, "titulo": 30, "descricao": 40, "endereco": 28, "bairro": 16, "cidade": 16, "estado": 6, "cep": 11, "preco": 14, "condominio": 12, "iptu": 10, "area_util": 12, "area_total": 12, "quartos": 9, "suites": 8, "banheiros": 10, "vagas_garagem": 12, "status": 14, "destaque": 10, "corretor_responsavel": 20, "telefone_contato": 16, "midias": 50, "data_cadastro": 18, "data_atualizacao": 18, } for idx, col in enumerate(COLUMNS, start=1): ws.column_dimensions[get_column_letter(idx)].width = larguras.get(col, 14) def init_excel(with_demo_data: bool = True) -> None: """Cria o arquivo Excel com cabeçalhos (e dados de demonstração) se ele não existir.""" os.makedirs(DATA_DIR, exist_ok=True) if os.path.exists(EXCEL_PATH): return wb = Workbook() ws = wb.active ws.title = SHEET_NAME _aplicar_estilo_cabecalho(ws) _largura_colunas(ws) wb.save(EXCEL_PATH) if with_demo_data: from app.seed_data import DEMO_ANUNCIOS for anuncio in DEMO_ANUNCIOS: criar_anuncio(anuncio) # -------------------------------------------------------------------------- # Helpers internos # -------------------------------------------------------------------------- _CAMPOS_NUMERICOS_FLOAT = ("preco", "condominio", "iptu", "area_util", "area_total") _CAMPOS_NUMERICOS_INT = ("quartos", "suites", "banheiros", "vagas_garagem") _CAMPOS_DATA = ("data_cadastro", "data_atualizacao") _CAMPOS_ESPECIAIS = {"id", "destaque", *_CAMPOS_NUMERICOS_FLOAT, *_CAMPOS_NUMERICOS_INT, *_CAMPOS_DATA} def _row_to_dict(row_values) -> dict: d = dict(zip(COLUMNS, row_values)) # normaliza tipos d["id"] = int(d["id"]) if d["id"] is not None else None d["destaque"] = bool(d["destaque"]) if d["destaque"] is not None else False for campo in _CAMPOS_NUMERICOS_FLOAT: d[campo] = float(d[campo]) if d[campo] not in (None, "") else 0.0 for campo in _CAMPOS_NUMERICOS_INT: d[campo] = int(d[campo]) if d[campo] not in (None, "") else 0 for campo in _CAMPOS_DATA: if isinstance(d[campo], str) and d[campo]: d[campo] = datetime.fromisoformat(d[campo]) elif d[campo] is None: d[campo] = datetime.now() # campos de texto: o Excel/openpyxl devolve None para células vazias, # mas nosso schema espera string vazia "" nesses campos. for campo in COLUMNS: if campo not in _CAMPOS_ESPECIAIS and d[campo] is None: d[campo] = "" return d def _dict_to_row(d: dict) -> list: row = [] for col in COLUMNS: val = d.get(col) if isinstance(val, datetime): val = val.isoformat() row.append(val) return row def _next_id(ws) -> int: max_id = 0 for row in ws.iter_rows(min_row=2, values_only=True): if row[0] is not None: max_id = max(max_id, int(row[0])) return max_id + 1 # -------------------------------------------------------------------------- # CRUD # -------------------------------------------------------------------------- def _carregar_todos() -> list[dict]: """Carrega todos os anúncios da planilha como lista de dicts (uso interno).""" with _lock: wb = load_workbook(EXCEL_PATH) ws = wb[SHEET_NAME] return [ _row_to_dict(row) for row in ws.iter_rows(min_row=2, values_only=True) if row[0] is not None ] _CHAVES_ORDENACAO = { "mais_recentes": (lambda i: i["id"], True), "mais_antigos": (lambda i: i["id"], False), "preco_asc": (lambda i: i["preco"], False), "preco_desc": (lambda i: i["preco"], True), "area_asc": (lambda i: i["area_util"] or i["area_total"], False), "area_desc": (lambda i: i["area_util"] or i["area_total"], True), } def _ordenar(itens: list[dict], ordenar_por: str = "mais_recentes") -> list[dict]: chave, reverse = _CHAVES_ORDENACAO.get(ordenar_por, _CHAVES_ORDENACAO["mais_recentes"]) return sorted(itens, key=chave, reverse=reverse) def listar_anuncios( filtros: Optional[dict] = None, skip: int = 0, limit: int = 50, ordenar_por: str = "mais_recentes", ) -> tuple[list[dict], int]: """Retorna (lista_paginada, total_apos_filtro).""" filtrados = [item for item in _carregar_todos() if _passa_filtros(item, filtros or {})] ordenados = _ordenar(filtrados, ordenar_por) total = len(ordenados) return ordenados[skip: skip + limit], total def _passa_filtros(item: dict, filtros: dict) -> bool: if filtros.get("tipo_negocio") and item["tipo_negocio"] != filtros["tipo_negocio"]: return False if filtros.get("tipo_imovel") and item["tipo_imovel"] != filtros["tipo_imovel"]: return False if filtros.get("cidade") and filtros["cidade"].lower() not in item["cidade"].lower(): return False if filtros.get("bairro") and filtros["bairro"].lower() not in item["bairro"].lower(): return False if filtros.get("status") and item["status"] != filtros["status"]: return False if filtros.get("destaque") is not None and item["destaque"] != filtros["destaque"]: return False if filtros.get("preco_min") is not None and item["preco"] < filtros["preco_min"]: return False if filtros.get("preco_max") is not None and item["preco"] > filtros["preco_max"]: return False if filtros.get("quartos_min") is not None and item["quartos"] < filtros["quartos_min"]: return False busca = filtros.get("busca") if busca: alvo = f"{item['titulo']} {item['descricao']} {item['endereco']}".lower() if busca.lower() not in alvo: return False return True def buscar_anuncio(anuncio_id: int) -> Optional[dict]: for item in _carregar_todos(): if item["id"] == anuncio_id: return item return None def criar_anuncio(dados: dict) -> dict: with _lock: wb = load_workbook(EXCEL_PATH) ws = wb[SHEET_NAME] novo_id = _next_id(ws) agora = datetime.now() registro = dict(dados) registro["id"] = novo_id registro["data_cadastro"] = agora registro["data_atualizacao"] = agora registro.setdefault("destaque", False) ws.append(_dict_to_row(registro)) wb.save(EXCEL_PATH) return registro def atualizar_anuncio(anuncio_id: int, dados: dict) -> Optional[dict]: """Atualização parcial: só sobrescreve as chaves presentes em `dados`.""" with _lock: wb = load_workbook(EXCEL_PATH) ws = wb[SHEET_NAME] for row_cells in ws.iter_rows(min_row=2): if row_cells[0].value is not None and int(row_cells[0].value) == anuncio_id: atual = _row_to_dict([c.value for c in row_cells]) atual.update({k: v for k, v in dados.items() if v is not None}) atual["data_atualizacao"] = datetime.now() nova_linha = _dict_to_row(atual) for col_idx, valor in enumerate(nova_linha, start=1): row_cells[col_idx - 1].value = valor wb.save(EXCEL_PATH) return atual return None def deletar_anuncio(anuncio_id: int) -> bool: with _lock: wb = load_workbook(EXCEL_PATH) ws = wb[SHEET_NAME] for row_cells in ws.iter_rows(min_row=2): if row_cells[0].value is not None and int(row_cells[0].value) == anuncio_id: ws.delete_rows(row_cells[0].row, 1) wb.save(EXCEL_PATH) return True return False # -------------------------------------------------------------------------- # Mídias (fotos/vídeos) de um anúncio # -------------------------------------------------------------------------- def _parse_midias(raw) -> list[str]: if not raw: return [] try: valores = json.loads(raw) except (json.JSONDecodeError, TypeError): return [] return [str(u) for u in valores if u] if isinstance(valores, list) else [] def listar_midias(anuncio_id: int) -> Optional[list[str]]: """Retorna a lista de URLs de mídia de um anúncio, ou None se o anúncio não existir.""" anuncio = buscar_anuncio(anuncio_id) if anuncio is None: return None return _parse_midias(anuncio.get("midias")) def adicionar_midias(anuncio_id: int, novas_urls: list[str]) -> Optional[dict]: """Acrescenta URLs à lista de mídias existente do anúncio (sem duplicar).""" atuais = listar_midias(anuncio_id) if atuais is None: return None for url in novas_urls: if url and url not in atuais: atuais.append(url) return atualizar_anuncio(anuncio_id, {"midias": json.dumps(atuais)}) def remover_midia(anuncio_id: int, url: str) -> Optional[dict]: """Remove uma URL específica da lista de mídias do anúncio.""" atuais = listar_midias(anuncio_id) if atuais is None: return None restantes = [u for u in atuais if u != url] return atualizar_anuncio(anuncio_id, {"midias": json.dumps(restantes)}) def calcular_estatisticas() -> dict: itens = _carregar_todos() total = len(itens) vendas = [i for i in itens if i["tipo_negocio"] == "Venda"] locacoes = [i for i in itens if i["tipo_negocio"] == "Locação"] disponiveis = [i for i in itens if i["status"] == "Disponível"] por_tipo_imovel: dict = {} por_cidade: dict = {} for i in itens: por_tipo_imovel[i["tipo_imovel"]] = por_tipo_imovel.get(i["tipo_imovel"], 0) + 1 por_cidade[i["cidade"]] = por_cidade.get(i["cidade"], 0) + 1 preco_medio_venda = sum(i["preco"] for i in vendas) / len(vendas) if vendas else 0.0 preco_medio_locacao = sum(i["preco"] for i in locacoes) / len(locacoes) if locacoes else 0.0 return { "total_anuncios": total, "total_venda": len(vendas), "total_locacao": len(locacoes), "total_disponiveis": len(disponiveis), "preco_medio_venda": round(preco_medio_venda, 2), "preco_medio_locacao": round(preco_medio_locacao, 2), "por_tipo_imovel": por_tipo_imovel, "por_cidade": por_cidade, } # -------------------------------------------------------------------------- # Navegação exploratória: opções de filtro, busca facetada, destaques, similares # -------------------------------------------------------------------------- def _contar_por_campo(itens: list[dict], campo: str) -> list[dict]: """Conta ocorrências de cada valor de `campo`, ordenado do mais frequente para o menos.""" contagem: dict = {} for item in itens: valor = item.get(campo) if valor in (None, ""): continue contagem[valor] = contagem.get(valor, 0) + 1 return [ {"valor": valor, "quantidade": qtd} for valor, qtd in sorted(contagem.items(), key=lambda kv: (-kv[1], str(kv[0]))) ] def _faixa(itens: list[dict], campo: str) -> dict: valores = [i[campo] for i in itens if i.get(campo)] if not valores: return {"min": 0, "max": 0} return {"min": min(valores), "max": max(valores)} def obter_opcoes_filtro(tipo_negocio: Optional[str] = None, cidade: Optional[str] = None) -> dict: """ Monta as opções disponíveis para montar filtros de navegação exploratória (dropdowns, sliders de preço/área etc.), com a contagem de anúncios de cada opção. - `tipos_negocio` é sempre calculado sobre a base toda (não faz sentido escopar por si mesmo). - `cidades` é escopado por `tipo_negocio`, se informado. - `bairros` é escopado por `tipo_negocio` e também por `cidade`, se informados (permite montar um segundo dropdown "cidade -> bairro" que se atualiza em cascata). """ todos = _carregar_todos() disponiveis = [i for i in todos if i["status"] == "Disponível"] escopo = disponiveis if tipo_negocio: escopo = [i for i in escopo if i["tipo_negocio"] == tipo_negocio] escopo_bairros = escopo if cidade: escopo_bairros = [i for i in escopo if i["cidade"].lower() == cidade.lower()] return { "total_anuncios": len(escopo), "tipos_negocio": _contar_por_campo(disponiveis, "tipo_negocio"), "tipos_imovel": _contar_por_campo(escopo, "tipo_imovel"), "status": _contar_por_campo(todos, "status"), "cidades": _contar_por_campo(escopo, "cidade"), "bairros": _contar_por_campo(escopo_bairros, "bairro"), "faixa_preco": _faixa(escopo, "preco"), "faixa_area": _faixa(escopo, "area_util"), "quartos_disponiveis": sorted({i["quartos"] for i in escopo if i["quartos"]}), "vagas_disponiveis": sorted({i["vagas_garagem"] for i in escopo if i["vagas_garagem"]}), } def busca_exploratoria(filtros: dict, skip: int = 0, limit: int = 20, ordenar_por: str = "mais_recentes") -> dict: """ Busca com filtros para navegação exploratória: retorna a página de resultados E as facetas (contagens por tipo de imóvel, cidade, bairro, status, faixas de preço/área) calculadas sobre o CONJUNTO JÁ FILTRADO — assim o frontend pode mostrar "348 resultados, sendo 120 em Curitiba, 45 com 3 quartos..." e ir refinando a busca passo a passo, como em portais de imóveis. """ todos = _carregar_todos() filtrados = [item for item in todos if _passa_filtros(item, filtros)] ordenados = _ordenar(filtrados, ordenar_por) total = len(ordenados) pagina = ordenados[skip: skip + limit] facetas = { "tipos_imovel": _contar_por_campo(filtrados, "tipo_imovel"), "cidades": _contar_por_campo(filtrados, "cidade"), "bairros": _contar_por_campo(filtrados, "bairro"), "status": _contar_por_campo(filtrados, "status"), "faixa_preco": _faixa(filtrados, "preco"), "faixa_area": _faixa(filtrados, "area_util"), } return { "total": total, "skip": skip, "limit": limit, "ordenar_por": ordenar_por, "resultados": pagina, "facetas": facetas, } def listar_destaques(limite: int = 6) -> list[dict]: """Anúncios marcados como destaque e disponíveis, para vitrines/carrosséis iniciais.""" itens = [i for i in _carregar_todos() if i["destaque"] and i["status"] == "Disponível"] itens = _ordenar(itens, "mais_recentes") return itens[:limite] def anuncios_similares(anuncio_id: int, limite: int = 4) -> list[dict]: """ Sugere anúncios parecidos com o de referência: mesmo tipo de negócio e de imóvel, disponíveis, priorizando a mesma cidade e depois o preço mais próximo. Útil para navegação exploratória a partir da página de um anúncio. """ todos = _carregar_todos() alvo = next((i for i in todos if i["id"] == anuncio_id), None) if not alvo: return [] candidatos = [ i for i in todos if i["id"] != anuncio_id and i["tipo_negocio"] == alvo["tipo_negocio"] and i["tipo_imovel"] == alvo["tipo_imovel"] and i["status"] == "Disponível" ] def chave_similaridade(item: dict): mesma_cidade = 0 if item["cidade"].lower() == alvo["cidade"].lower() else 1 distancia_preco = abs(item["preco"] - alvo["preco"]) return (mesma_cidade, distancia_preco) candidatos.sort(key=chave_similaridade) return candidatos[:limite]