Speed up logradouro catalog loading

backend/app/core/shapefile_runtime.py

@@ -30,6 +30,14 @@ def _build_transformer(source_crs: Any, target_crs: str | None):
         return None
 
 
+def _field_name(field: Any) -> str:
+    if hasattr(field, "name"):
+        return str(getattr(field, "name") or "")
+    if isinstance(field, (list, tuple)) and field:
+        return str(field[0] or "")
+    return str(field or "")
+
+
 def _iter_features_pyshp(shapefile_path: Path, *, target_crs: str | None) -> list[tuple[dict[str, Any], Any]]:
     import shapefile
     from shapely.geometry import shape
@@ -73,7 +81,7 @@ def _iter_features_pyshp(shapefile_path: Path, *, target_crs: str | None) -> lis
             encoding="utf-8",
             encodingErrors="replace",
         )
-        field_names = [str(field[0]) for field in reader.fields[1:]]
+        field_names = [_field_name(field) for field in reader.fields[1:]]
         source_crs = None
         prj_path = shapefile_path.with_suffix(".prj")
         if prj_path.exists():
@@ -214,3 +222,42 @@ def load_vector_geojson(
         )
 
     return feature_collection
+
+
+def load_attribute_records(
+    shapefile_path: str | Path,
+    *,
+    property_fields: tuple[str, ...] | list[str] | None = None,
+) -> list[dict[str, Any]]:
+    path = Path(shapefile_path).expanduser().resolve()
+    dbf_path = path.with_suffix(".dbf")
+    if not dbf_path.exists():
+        raise FileNotFoundError(f"DBF nao encontrado: {dbf_path}")
+
+    wanted_fields = tuple(str(field).strip() for field in (property_fields or ()) if str(field).strip())
+
+    try:
+        import shapefile
+    except Exception:
+        records: list[dict[str, Any]] = []
+        for properties, _ in _load_features(path, target_crs=None):
+            if wanted_fields:
+                records.append({field: properties.get(field) for field in wanted_fields if field in properties})
+            else:
+                records.append(dict(properties))
+        return records
+
+    with open(dbf_path, "rb") as dbf_handle:
+        reader = shapefile.Reader(dbf=dbf_handle, encoding="utf-8", encodingErrors="replace")
+        field_names = [_field_name(field) for field in reader.fields[1:]]
+        records: list[dict[str, Any]] = []
+        for record in reader.iterRecords():
+            values = list(record)
+            row = {
+                field_name: values[index] if index < len(values) else None
+                for index, field_name in enumerate(field_names)
+            }
+            if wanted_fields:
+                row = {field: row.get(field) for field in wanted_fields if field in row}
+            records.append(row)
+        return records

backend/app/services/pesquisa_service.py

@@ -20,6 +20,7 @@ from fastapi import HTTPException
 from joblib import load
 
 from app.core.elaboracao import geocodificacao
+from app.core.shapefile_runtime import load_attribute_records
 from app.core.elaboracao.core import _migrar_pacote_v1_para_v2, normalizar_observacao_modelo
 from app.core.map_layers import (
     add_bairros_layer,
@@ -27,6 +28,7 @@ from app.core.map_layers import (
     add_zoom_responsive_circle_markers,
 )
 from app.portable_runtime_log import append_runtime_log
+from app.runtime_config import resolve_core_path
 from app.services import model_repository, trabalhos_tecnicos_service
 from app.services.serializers import sanitize_value
 
@@ -2566,6 +2568,44 @@ def _carregar_catalogo_vias() -> list[dict[str, Any]]:
     if _CATALOGO_VIAS_CACHE is not None:
         return list(_CATALOGO_VIAS_CACHE)
 
+    registros: list[dict[str, Any]] = []
+    try:
+        registros = load_attribute_records(
+            resolve_core_path("dados", "EixosLogradouros.shp"),
+            property_fields=("CDLOG", "NMIDELOG", "NMIDEPRE", "NMIDEABR"),
+        )
+        append_runtime_log(f"[mesa] pesquisa: catalogo leve de logradouros carregado com {len(registros)} registros")
+    except Exception as exc:
+        append_runtime_log(f"[mesa] pesquisa: falha no catalogo leve de logradouros: {exc}")
+
+    def _catalogo_from_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        vistos: set[int] = set()
+        catalogo_local: list[dict[str, Any]] = []
+        for row in rows:
+            cdlog = _to_int_or_none(row.get("CDLOG"))
+            if cdlog is None or cdlog in vistos:
+                continue
+            vistos.add(cdlog)
+            nome = str(row.get("NMIDELOG") or "").strip()
+            prefixo = str(row.get("NMIDEPRE") or "").strip()
+            abreviado = str(row.get("NMIDEABR") or "").strip()
+            logradouro = " ".join(item for item in [prefixo, nome] if item).strip() or nome or abreviado
+            aliases = _dedupe_strings([logradouro, nome, abreviado])
+            catalogo_local.append(
+                {
+                    "cdlog": cdlog,
+                    "logradouro": logradouro,
+                    "_aliases_norm": [_normalize(item) for item in aliases if _normalize(item)],
+                }
+            )
+        return catalogo_local
+
+    if registros:
+        catalogo = _catalogo_from_rows(registros)
+        _CATALOGO_VIAS_CACHE = list(catalogo)
+        append_runtime_log(f"[mesa] pesquisa: catalogo de logradouros carregado com {len(catalogo)} vias")
+        return list(catalogo)
+
     gdf = geocodificacao.carregar_eixos()
     if gdf is None or gdf.empty:
         append_runtime_log("[mesa] pesquisa: base de eixos indisponivel para catalogo de logradouros")