core_app.py

from __future__ import annotations

import json
import logging
import math
import os
import re
import unicodedata
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import gradio as gr
import numpy as np
import pandas as pd
import xgboost as xgb
from pyproj import Transformer

try:
    from .reporting import build_metric_summary, build_pdf_report, split_prediction_frames
except ImportError:
    from reporting import build_metric_summary, build_pdf_report, split_prediction_frames


LOGGER = logging.getLogger("avm_gradio_app")
if not LOGGER.handlers:
    logging.basicConfig(
        level=os.getenv("APP_LOG_LEVEL", "INFO").upper(),
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    )

BASE_DIR = Path(__file__).resolve().parent
WGS84_TO_WEBMERC = Transformer.from_crs("EPSG:4326", "EPSG:3857", always_xy=True)
ZONEAMENTO_SHP_ENV = "ZONEAMENTO_SHP_PATH"
IAPOND_MAP_ENV = "IAPOND_MAP_PATH"
IAPOND_PDF_ENV = "IAPOND_PDF_PATH"
POLYGON_CONTEXT_ENV = "POLYGON_CONTEXT_PATH"
RH_NUMBLOCO_CSV_ENV = "RH_NUMBLOCO_CSV_PATH"
TESTADA_NUMBLOCO_CSV_ENV = "TESTADA_NUMBLOCO_CSV_PATH"
IAPOND_DEFAULT = 1.0
AUTO_GLEBA_AREA_THRESHOLD = 3000.0
POLYGON_FALLBACK_MAX_DISTANCE_M = 250.0
POLYGON_TARGET_CRS = "EPSG:31982"
IAPOND_BY_INDICE = {
    "1": 1.0,
    "2A": 1.0,
    "2B": 1.0,
    "3": 1.3,
    "4": 1.3,
    "4A": 1.3,
    "5": 1.3,
    "6": 1.3,
    "7": 1.3,
    "9": 1.3,
    "11": 1.6,
    "13": 1.6,
    "15": 1.9,
    "17": 1.9,
    "19": 2.4,
    "21": 0.65,
    "23": 1.0,
    "25": 1.0,
    "31": 0.1,
    "33": 0.1,
    "35": 0.2,
    "37": 0.5,
    "39": 1.0,
    "41": 1.0,
}

RH_BINS = [-np.inf, 13.5, 31.5, 59.5, 155.0, np.inf]
RH_LABELS = ["RH_muito_baixo", "RH_baixo", "RH_medio", "RH_alto", "RH_muito_alto"]

DEFAULTS_NUMERIC = {
    "RH": 50.0,
    "IAPOND": 1.0,
    "APP": 0.0,
    "CP": 0.0,
    "LOTPOS": 0.0,
    "area_simpson": 0.0,
    "taxa_ocupacao": 0.0,
    "bldg_density": 0.0,
    "avg_bldg_footprint": 0.0,
    "circularity": 0.0,
    "dist_to_main_road": 0.0,
    "dist_to_park": 0.0,
    "Dist_weighted_Density": 0.0,
}

DEFAULTS_CATEGORICAL = {
    "FONTE": "0",
    "Ano_Dado": "2026",
    "faixa_rh": "RH_alto",
    "faixa_area_modelo": "desconhecido",
    "FINALIDADE": "desconhecido",
    "MZ": "desconhecido",
    "UEU": "desconhecido",
    "SUBUNIDADE": "desconhecido",
    "DENSIDADE": "desconhecido",
    "ATIVIDADE": "desconhecido",
    "INDICE": "desconhecido",
    "VOLUMETRIA": "desconhecido",
}

REQUESTED_MODEL_NOTEBOOKS = {
    "terreno": "TERRITORIAL_2026_TERRENO",
    "gleba": "TERRITORIAL_2026_GLEBA",
}

RELIABLE_MODEL_CONTEXT_SOURCES = {
    "rh_numbloco_csv",
    "testada_numbloco_csv",
    "poligono_numbloco",
    "poligono_intersect",
    "poligono_proximo",
    "zoneamento_shp",
    "zoneamento_shp_ia",
}

ZONEAMENTO_FIELDS = [
    "MZ",
    "UEU",
    "SUBUNIDADE",
    "DENSIDADE",
    "ATIVIDADE",
    "INDICE",
    "VOLUMETRIA",
]

ZONEAMENTO_COLUMN_ALIASES: dict[str, list[str]] = {
    "MZ": ["MZ", "MACROZONA"],
    "UEU": ["UEU"],
    "SUBUNIDADE": ["SUBUNIDADE", "SUBUNID", "SUB_UNIDADE", "SUBUNIDADE1"],
    "DENSIDADE": ["DENSIDADE", "DENSID"],
    "ATIVIDADE": ["ATIVIDADE", "ATIVID"],
    "INDICE": ["INDICE", "IND_APR", "INDAPROV", "INDICEAPR"],
    "VOLUMETRIA": ["VOLUMETRIA", "VOLUM"],
    "IA": ["IA", "IAPOND", "IA_POND", "INDICEAPROVEITAMENTO"],
    "CODIGO": ["CODIGO", "COD", "CODINDICE", "COD_INDICE", "INDICECOD"],
}

CONTEXT_FIELDS = sorted(
    {
        "FONTE",
        "Ano_Dado",
        "AREA",
        "AREA_bruta",
        "TESTADA",
        "TESTADA_bruta",
        "RH",
        "IAPOND",
        "APP",
        "CP",
        "LOTPOS",
        "FINALIDADE",
        "BAIRRO",
        "COD_BAIRRO",
        "area_simpson",
        "taxa_ocupacao",
        "bldg_density",
        "avg_bldg_footprint",
        "circularity",
        "dist_to_main_road",
        "dist_to_park",
        "MZ",
        "UEU",
        "SUBUNIDADE",
        "DENSIDADE",
        "ATIVIDADE",
        "INDICE",
        "VOLUMETRIA",
    }
)

POLYGON_CONTEXT_FIELDS = [
    "NUMBLOCO",
    "AREA",
    "TESTADA",
    "RH",
    "faixa_rh",
    "IAPOND",
    "APP",
    "PE",
    "CP",
    "log_area_modelo",
    "log_testada_modelo",
    "area_simpson",
    "taxa_ocupacao",
    "bldg_density",
    "avg_bldg_footprint",
    "circularity",
    "dist_to_main_road",
    "dist_to_park",
    "Dist_weighted_Density",
    "intersection_density",
    "MZ",
    "UEU",
    "SUBUNIDADE",
    "DENSIDADE",
    "ATIVIDADE",
    "INDICE",
    "VOLUMETRIA",
]


def _is_missing(value: Any) -> bool:
    if value is None:
        return True
    if isinstance(value, str) and value.strip() == "":
        return True
    try:
        return bool(pd.isna(value))
    except Exception:
        return False


def _to_float(value: Any, default: float | None = None) -> float | None:
    series = pd.to_numeric(pd.Series([value]), errors="coerce")
    number = series.iloc[0]
    if pd.isna(number):
        return default
    return float(number)


def _to_int(value: Any, default: int | None = None) -> int | None:
    number = _to_float(value)
    if number is None:
        return default
    return int(round(number))


def _normalize_token(value: Any) -> str:
    if _is_missing(value):
        return ""
    text = str(value).strip()
    text = re.sub(r"\s+", " ", text)
    return text.upper()


def _normalize_ascii_token(value: Any) -> str:
    token = _normalize_token(value)
    token = unicodedata.normalize("NFKD", token)
    token = "".join(ch for ch in token if not unicodedata.combining(ch))
    return token


def _normalize_column_token(value: Any) -> str:
    token = _normalize_ascii_token(value)
    return re.sub(r"[^A-Z0-9]", "", token)


def _normalize_code_token(value: Any) -> str:
    if _is_missing(value):
        return ""
    text = _normalize_ascii_token(value)
    text = text.replace(",", ".")
    text = re.sub(r"\s+", "", text).strip()
    if text == "":
        return ""

    code_match = re.fullmatch(r"0*([0-9]+)([A-Z]?)", text)
    if code_match:
        return f"{code_match.group(1)}{code_match.group(2)}"
    return text


def _normalize_numbloco_key(value: Any) -> str:
    if _is_missing(value):
        return ""
    text = _normalize_ascii_token(value)
    text = re.sub(r"\.0$", "", text)
    text = re.sub(r"[^A-Z0-9]", "", text)
    if text.isdigit():
        return text.lstrip("0") or "0"
    return text


def _find_column_by_alias(columns: list[str], aliases: list[str]) -> str | None:
    normalized = {_normalize_column_token(col): str(col) for col in columns}
    for alias in aliases:
        key = _normalize_column_token(alias)
        if key in normalized:
            return normalized[key]

    for col in columns:
        col_key = _normalize_column_token(col)
        for alias in aliases:
            alias_key = _normalize_column_token(alias)
            if alias_key and (col_key.startswith(alias_key) or alias_key.startswith(col_key)):
                return str(col)
    return None


def _safe_text(value: Any, default: str = "desconhecido") -> str:
    if _is_missing(value):
        return default
    text = str(value).strip()
    return text if text else default


def _currency_brl(value: float) -> str:
    if not np.isfinite(value):
        return "n/a"
    return f"R$ {value:,.2f}"


def _jsonable(value: Any) -> Any:
    if isinstance(value, (np.floating,)):
        return float(value)
    if isinstance(value, (np.integer,)):
        return int(value)
    if isinstance(value, (np.bool_,)):
        return bool(value)
    if isinstance(value, (pd.Timestamp,)):
        return value.isoformat()
    if isinstance(value, dict):
        return {str(k): _jsonable(v) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_jsonable(v) for v in value]
    return value


def _read_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        return json.load(f)


def _read_parquet_optional(path: Path | None) -> pd.DataFrame:
    if path is None or not path.exists():
        return pd.DataFrame()
    try:
        return pd.read_parquet(path)
    except Exception as exc:
        LOGGER.warning("Falha ao ler parquet %s: %s", path, exc)
        return pd.DataFrame()


def _read_csv_optional(path: Path | None) -> pd.DataFrame:
    if path is None or not path.exists():
        return pd.DataFrame()
    try:
        return pd.read_csv(path)
    except Exception as exc:
        LOGGER.warning("Falha ao ler csv %s: %s", path, exc)
        return pd.DataFrame()


def _find_first_file(directory: Path, patterns: list[str]) -> Path | None:
    for pattern in patterns:
        found = sorted(directory.glob(pattern))
        if found:
            return found[0]
    return None


def _candidate_base_dirs(base_dir: Path) -> list[Path]:
    candidates = [base_dir, base_dir.parent]
    unique: list[Path] = []
    seen: set[str] = set()
    for path in candidates:
        try:
            key = str(path.resolve())
        except Exception:
            key = str(path)
        if key in seen:
            continue
        seen.add(key)
        unique.append(path)
    return unique


def _candidate_rh_numbloco_paths(base_dir: Path) -> list[Path]:
    candidates: list[Path] = []
    env_path = os.getenv(RH_NUMBLOCO_CSV_ENV, "").strip()
    if env_path:
        candidates.append(Path(env_path).expanduser())

    for root in _candidate_base_dirs(base_dir):
        candidates.extend(
            [
                root / "data" / "reference" / "numbloco_rh.csv",
                root / "numbloco_rh.csv",
            ]
        )

    unique: list[Path] = []
    seen: set[str] = set()
    for path in candidates:
        try:
            key = str(path.resolve())
        except Exception:
            key = str(path)
        if key in seen:
            continue
        seen.add(key)
        unique.append(path)
    return unique


def _candidate_testada_numbloco_paths(base_dir: Path) -> list[Path]:
    candidates: list[Path] = []
    env_path = os.getenv(TESTADA_NUMBLOCO_CSV_ENV, "").strip()
    if env_path:
        candidates.append(Path(env_path).expanduser())

    for root in _candidate_base_dirs(base_dir):
        candidates.extend(
            [
                root / "data" / "reference" / "numbloco_testada.csv",
                root / "numbloco_testada.csv",
            ]
        )

    unique: list[Path] = []
    seen: set[str] = set()
    for path in candidates:
        try:
            key = str(path.resolve())
        except Exception:
            key = str(path)
        if key in seen:
            continue
        seen.add(key)
        unique.append(path)
    return unique


def load_rh_numbloco_context(base_dir: Path = BASE_DIR) -> RhNumblocoContext:
    for path in _candidate_rh_numbloco_paths(base_dir):
        if not path.exists():
            continue
        try:
            df = pd.read_csv(path)
        except Exception as exc:
            LOGGER.warning("Falha ao ler CSV RH por NUMBLOCO %s: %s", path, exc)
            continue

        numbloco_col = _find_column_by_alias(list(df.columns), ["NUMBLOCO", "NUM_BLOCO", "NUM BLOCO"])
        rh_col = _find_column_by_alias(list(df.columns), ["RHVALOR", "RH_VALOR", "RH", "VALOR_RH"])
        if numbloco_col is None or rh_col is None:
            LOGGER.warning("CSV %s sem colunas NUMBLOCO/RH reconheciveis.", path)
            continue

        work = pd.DataFrame(
            {
                "__numbloco_key": df[numbloco_col].map(_normalize_numbloco_key),
                "RH": pd.to_numeric(df[rh_col], errors="coerce"),
            }
        )
        work = work[(work["__numbloco_key"] != "") & work["RH"].notna()]
        if work.empty:
            LOGGER.warning("CSV %s nao contem pares NUMBLOCO/RH validos.", path)
            continue

        key_counts = work.groupby("__numbloco_key").size()
        duplicate_key_count = int((key_counts > 1).sum())
        conflicting_key_count = int(work.groupby("__numbloco_key")["RH"].nunique().gt(1).sum())
        rh_by_numbloco = work.groupby("__numbloco_key")["RH"].median().astype(float).to_dict()
        LOGGER.info(
            "RH por NUMBLOCO carregado: %s (linhas=%s, chaves=%s, duplicadas=%s, conflitos=%s)",
            path.name,
            len(work),
            len(rh_by_numbloco),
            duplicate_key_count,
            conflicting_key_count,
        )
        if conflicting_key_count:
            LOGGER.warning(
                "CSV RH por NUMBLOCO possui %s chaves com valores conflitantes; usando mediana por NUMBLOCO.",
                conflicting_key_count,
            )
        return RhNumblocoContext(
            path=path,
            rh_by_numbloco=rh_by_numbloco,
            duplicate_key_count=duplicate_key_count,
            conflicting_key_count=conflicting_key_count,
        )

    message = "CSV RH por NUMBLOCO nao encontrado."
    LOGGER.warning(message)
    return RhNumblocoContext(error=message)


def load_testada_numbloco_context(base_dir: Path = BASE_DIR) -> TestadaNumblocoContext:
    for path in _candidate_testada_numbloco_paths(base_dir):
        if not path.exists():
            continue
        try:
            df = pd.read_csv(path)
        except Exception as exc:
            LOGGER.warning("Falha ao ler CSV TESTADA por NUMBLOCO %s: %s", path, exc)
            continue

        numbloco_col = _find_column_by_alias(list(df.columns), ["NUMBLOCO", "NUM_BLOCO", "NUM BLOCO"])
        testada_col = _find_column_by_alias(
            list(df.columns),
            ["MTR_TESTADA", "TESTADA", "TESTADA_BRUTA", "FRENTE", "METROS_TESTADA"],
        )
        if numbloco_col is None or testada_col is None:
            LOGGER.warning("CSV %s sem colunas NUMBLOCO/TESTADA reconheciveis.", path)
            continue

        work = pd.DataFrame(
            {
                "__numbloco_key": df[numbloco_col].map(_normalize_numbloco_key),
                "TESTADA": pd.to_numeric(df[testada_col], errors="coerce"),
            }
        )
        work = work[(work["__numbloco_key"] != "") & work["TESTADA"].notna() & (work["TESTADA"] > 0)]
        if work.empty:
            LOGGER.warning("CSV %s nao contem pares NUMBLOCO/TESTADA validos.", path)
            continue

        key_counts = work.groupby("__numbloco_key").size()
        duplicate_key_count = int((key_counts > 1).sum())
        conflicting_key_count = int(work.groupby("__numbloco_key")["TESTADA"].nunique().gt(1).sum())
        testada_by_numbloco = work.groupby("__numbloco_key")["TESTADA"].median().astype(float).to_dict()
        LOGGER.info(
            "TESTADA por NUMBLOCO carregada: %s (linhas=%s, chaves=%s, duplicadas=%s, conflitos=%s)",
            path.name,
            len(work),
            len(testada_by_numbloco),
            duplicate_key_count,
            conflicting_key_count,
        )
        if conflicting_key_count:
            LOGGER.warning(
                "CSV TESTADA por NUMBLOCO possui %s chaves com valores conflitantes; usando mediana por NUMBLOCO.",
                conflicting_key_count,
            )
        return TestadaNumblocoContext(
            path=path,
            testada_by_numbloco=testada_by_numbloco,
            duplicate_key_count=duplicate_key_count,
            conflicting_key_count=conflicting_key_count,
        )

    message = "CSV TESTADA por NUMBLOCO nao encontrado."
    LOGGER.warning(message)
    return TestadaNumblocoContext(error=message)


def _parse_model_schema(model_path: Path) -> tuple[list[str], dict[str, str]]:
    payload = _read_json(model_path)
    learner = payload.get("learner", {})
    names = learner.get("feature_names") or []
    types = learner.get("feature_types") or []
    feature_types: dict[str, str] = {}
    for idx, name in enumerate(names):
        feature_types[str(name)] = str(types[idx]) if idx < len(types) else "float"
    return [str(name) for name in names], feature_types


def _parse_area_bins(raw_bins: list[Any]) -> list[float]:
    bins: list[float] = []
    for item in raw_bins:
        if item is None:
            bins.append(float("inf"))
            continue
        if isinstance(item, (int, float)):
            bins.append(float(item))
            continue
        text = str(item).strip().lower()
        if text in {"inf", "infinity", "+inf", "+infinity"}:
            bins.append(float("inf"))
        elif text in {"-inf", "-infinity"}:
            bins.append(float("-inf"))
        else:
            value = _to_float(item)
            bins.append(float(value) if value is not None else float("nan"))
    return bins


def _derive_ano_dado(df: pd.DataFrame) -> pd.DataFrame:
    if "Ano_Dado" in df.columns:
        return df

    out = df.copy()
    if "ANO" in out.columns:
        out["Ano_Dado"] = pd.to_numeric(out["ANO"], errors="coerce").astype("Int64")
        return out

    ano_cols = [col for col in out.columns if str(col).startswith("ANO_")]
    if not ano_cols:
        return out

    idx_max = out[ano_cols].idxmax(axis=1)
    out["Ano_Dado"] = (
        idx_max.astype("string")
        .str.replace("ANO_", "", regex=False)
        .astype("Int64")
    )
    return out


def _summarize_series(series: pd.Series, prefer_mode: bool) -> Any:
    clean = series.dropna()
    if clean.empty:
        return None
    if prefer_mode:
        mode = clean.astype("string").mode(dropna=True)
        return None if mode.empty else str(mode.iloc[0])

    as_num = pd.to_numeric(clean, errors="coerce")
    if float(as_num.notna().mean()) >= 0.8:
        return float(as_num.median())

    mode = clean.astype("string").mode(dropna=True)
    return None if mode.empty else str(mode.iloc[0])


def _normalize_category_values(values: list[str]) -> list[str]:
    out = sorted({str(v) for v in values if str(v).strip() != ""})
    if "desconhecido" not in out:
        out.append("desconhecido")
    return out


def _normalize_fonte(value: Any, categories: list[str]) -> str:
    if _is_missing(value):
        value = DEFAULTS_CATEGORICAL["FONTE"]
    text = str(value).strip().lower()

    if text in {"venda", "sale"}:
        if "0" in categories and "Venda" not in categories:
            return "0"
        return "Venda" if "Venda" in categories else "0"
    if text in {"oferta", "offer"}:
        if "1" in categories and "Oferta" not in categories:
            return "1"
        return "Oferta" if "Oferta" in categories else "1"
    if text in {"0", "1"}:
        return text
    return str(value).strip() or DEFAULTS_CATEGORICAL["FONTE"]


def _is_reliable_model_context_source(source: str) -> bool:
    return source in RELIABLE_MODEL_CONTEXT_SOURCES or str(source).startswith("mapa_ia:")


def _categorize_area(area: float, bins: list[float], labels: list[str]) -> str:
    if not bins or not labels:
        return "desconhecido"
    series = pd.Series([area], dtype="float64")
    cat = pd.cut(series, bins=bins, labels=labels, include_lowest=True).astype("string").iloc[0]
    if _is_missing(cat):
        return labels[-1]
    return str(cat)


def _categorize_rh(rh: float) -> str:
    series = pd.Series([rh], dtype="float64")
    cat = pd.cut(series, bins=RH_BINS, labels=RH_LABELS).astype("string").iloc[0]
    if _is_missing(cat):
        return RH_LABELS[-1]
    return str(cat)


def _compute_grid_id(x: float, y: float, cell_size: float) -> str:
    if not np.isfinite(x) or not np.isfinite(y) or cell_size <= 0:
        return ""
    gx = int(math.floor(x / cell_size))
    gy = int(math.floor(y / cell_size))
    return f"{gx}_{gy}"


def _infer_testada(area: float, local_context: dict[str, Any]) -> tuple[float, str]:
    for key in ("TESTADA", "TESTADA_bruta", "testada"):
        candidate = _to_float(local_context.get(key))
        if candidate is not None and candidate > 0:
            return float(candidate), "inferido_coord"

    # Heuristica geométrica simples para reduzir input manual quando testada não estiver disponível.
    # Assume razão profundidade/testada ~ 2.5 para lotes urbanos.
    heuristic = math.sqrt(max(area, 1.0) / 2.5)
    heuristic = min(max(heuristic, 5.0), 120.0)
    return float(heuristic), "heuristica_area"


def _derive_cp(area: float, testada: float) -> float:
    if area <= 0 or testada <= 0:
        raise ValueError("AREA e TESTADA devem ser positivas para calcular CP.")

    pe = area / testada
    if area > 5000 and pe > 90:
        return float(4.8 * (testada**0.2) * (area**-0.4))
    if pe < 20:
        return float(math.sqrt(pe / 20.0))
    if pe <= 33:
        return 1.0
    if pe < 90:
        return float(math.sqrt(33.0 / pe))
    return 0.6


def _parse_overrides_json(text: str) -> dict[str, Any]:
    if _is_missing(text):
        return {}
    try:
        payload = json.loads(str(text))
    except json.JSONDecodeError as exc:
        raise ValueError(f"JSON invalido em overrides: {exc.msg}") from exc
    if not isinstance(payload, dict):
        raise ValueError("Overrides deve ser um objeto JSON.")
    return {str(k).strip().lower(): v for k, v in payload.items()}


@dataclass
class ArtifactBundle:
    key: str
    label: str
    artifact_dir: Path
    metadata_path: Path
    model_path: Path
    metadata: dict[str, Any]
    model: xgb.Booster
    feature_names: list[str]
    feature_types: dict[str, str]
    area_bins: list[float]
    area_labels: list[str]
    grid_cell_size: float
    tempo_index_df: pd.DataFrame
    grid_surface_df: pd.DataFrame
    segment_surface_df: pd.DataFrame
    surface_global_df: pd.DataFrame
    tier_map_raw: dict[str, str]
    tier_map_norm: dict[str, str]
    search_results_df: pd.DataFrame
    holdout_predictions_df: pd.DataFrame
    oof_predictions_df: pd.DataFrame
    notebook_name: str
    grid_value_map: dict[tuple[str, str], float] = field(default_factory=dict)
    segment_value_map: dict[str, float] = field(default_factory=dict)
    global_surface_value: float | None = None
    tempo_index_map: dict[str, float] = field(default_factory=dict)
    reference_categories: dict[str, list[str]] = field(default_factory=dict)


@dataclass
class ReferenceContext:
    path: Path | None = None
    df: pd.DataFrame = field(default_factory=pd.DataFrame)
    x: np.ndarray = field(default_factory=lambda: np.array([], dtype=float))
    y: np.ndarray = field(default_factory=lambda: np.array([], dtype=float))

    @property
    def has_data(self) -> bool:
        return not self.df.empty and self.x.size > 0 and self.y.size > 0


@dataclass
class ZoneamentoContext:
    shp_path: Path | None = None
    gdf: pd.DataFrame = field(default_factory=pd.DataFrame)
    column_map: dict[str, str] = field(default_factory=dict)
    iapond_map: dict[str, float] = field(default_factory=dict)
    iapond_map_source: str = "fallback_default"
    error: str | None = None

    @property
    def has_data(self) -> bool:
        return self.shp_path is not None and not self.gdf.empty


@dataclass
class PolygonContext:
    path: Path | None = None
    gdf: pd.DataFrame = field(default_factory=pd.DataFrame)
    available_fields: list[str] = field(default_factory=list)
    numbloco_key_index: dict[str, list[int]] = field(default_factory=dict)
    error: str | None = None
    fallback_max_distance_m: float = POLYGON_FALLBACK_MAX_DISTANCE_M
    to_base_crs: Transformer | None = None

    @property
    def has_data(self) -> bool:
        return self.path is not None and not self.gdf.empty and "geometry" in self.gdf.columns


@dataclass
class RhNumblocoContext:
    path: Path | None = None
    rh_by_numbloco: dict[str, float] = field(default_factory=dict)
    duplicate_key_count: int = 0
    conflicting_key_count: int = 0
    error: str | None = None

    @property
    def has_data(self) -> bool:
        return self.path is not None and bool(self.rh_by_numbloco)


@dataclass
class TestadaNumblocoContext:
    path: Path | None = None
    testada_by_numbloco: dict[str, float] = field(default_factory=dict)
    duplicate_key_count: int = 0
    conflicting_key_count: int = 0
    error: str | None = None

    @property
    def has_data(self) -> bool:
        return self.path is not None and bool(self.testada_by_numbloco)


@dataclass
class AppState:
    bundles: dict[str, ArtifactBundle] = field(default_factory=dict)
    reference: ReferenceContext = field(default_factory=ReferenceContext)
    zoneamento: ZoneamentoContext = field(default_factory=ZoneamentoContext)
    polygon_context: PolygonContext = field(default_factory=PolygonContext)
    rh_numbloco: RhNumblocoContext = field(default_factory=RhNumblocoContext)
    testada_numbloco: TestadaNumblocoContext = field(default_factory=TestadaNumblocoContext)
    error: str | None = None
    k_neighbors: int = 100


def _build_grid_and_temporal_maps(bundle: ArtifactBundle) -> None:
    if not bundle.grid_surface_df.empty and {"segmento_area", "grid_id", "valor"}.issubset(bundle.grid_surface_df.columns):
        for _, row in bundle.grid_surface_df.iterrows():
            seg_key = _normalize_token(row["segmento_area"])
            gid = str(row["grid_id"]).strip()
            val = _to_float(row["valor"])
            if seg_key and gid and val is not None:
                bundle.grid_value_map[(seg_key, gid)] = float(val)

    if not bundle.segment_surface_df.empty and {"segmento_area", "valor"}.issubset(bundle.segment_surface_df.columns):
        for _, row in bundle.segment_surface_df.iterrows():
            seg_key = _normalize_token(row["segmento_area"])
            val = _to_float(row["valor"])
            if seg_key and val is not None:
                bundle.segment_value_map[seg_key] = float(val)

    if not bundle.surface_global_df.empty and "mediana_global" in bundle.surface_global_df.columns:
        val = _to_float(bundle.surface_global_df["mediana_global"].iloc[0])
        bundle.global_surface_value = float(val) if val is not None else None

    if not bundle.tempo_index_df.empty and {"periodo_tempo", "indice_temporal_interno"}.issubset(bundle.tempo_index_df.columns):
        for _, row in bundle.tempo_index_df.iterrows():
            periodo = _safe_text(row["periodo_tempo"], "")
            indice = _to_float(row["indice_temporal_interno"])
            if periodo and indice is not None:
                bundle.tempo_index_map[periodo] = float(indice)


def _preferred_artifact_dirs(base_dir: Path) -> dict[str, list[Path]]:
    return {
        "gleba": [
            base_dir / "artifacts_gleba_cadastro",
            base_dir.parent / "artifacts_gleba_cadastro",
        ],
        "terreno": [
            base_dir / "artifacts_terreno_cadastro",
            base_dir.parent / "pddua_shp" / "artifacts_terreno_cadastro",
            base_dir.parent / "artifacts_terreno_cadastro",
        ],
    }


def _artifact_prediction_paths(artifact_dir: Path, key: str) -> tuple[Path | None, Path | None]:
    if key == "gleba":
        holdout = _find_first_file(artifact_dir, ["predictions_holdout_gleba_vunit.csv", "*predictions_holdout*.csv"])
        oof = _find_first_file(artifact_dir, ["predictions_oof_gleba_vunit.csv", "*predictions_oof*.csv"])
    else:
        holdout = _find_first_file(artifact_dir, ["predictions_holdout.csv", "*predictions_holdout*.csv"])
        oof = _find_first_file(artifact_dir, ["predictions_oof.csv", "*predictions_oof*.csv"])
    return holdout, oof


def load_artifacts(base_dir: Path = BASE_DIR) -> dict[str, ArtifactBundle]:
    bundles: dict[str, ArtifactBundle] = {}
    preferred_dirs = _preferred_artifact_dirs(base_dir)

    for key, candidates in preferred_dirs.items():
        artifact_dir = next((path for path in candidates if path.exists()), None)
        if artifact_dir is None:
            LOGGER.warning("Artefatos solicitados nao encontrados para %s.", key)
            continue
        metadata_path = _find_first_file(artifact_dir, ["*_metadata.json"])
        if metadata_path is None:
            LOGGER.warning("Sem metadata em %s", artifact_dir)
            continue

        metadata = _read_json(metadata_path)
        label = "Terreno" if key == "terreno" else "Gleba"

        model_path = _find_first_file(
            artifact_dir,
            [f"{key}_xgb_model.json", "*_xgb_model.json"],
        )
        if model_path is None:
            LOGGER.warning("Sem modelo em %s", artifact_dir)
            continue

        feature_names, feature_types = _parse_model_schema(model_path)
        model = xgb.Booster()
        model.load_model(str(model_path))

        area_bins = _parse_area_bins(list(metadata.get("area_bins", [])))
        area_labels = [str(label_item) for label_item in metadata.get("area_labels", [])]
        grid_cell_size = float(_to_float(metadata.get("grid_cell_size"), 1000.0) or 1000.0)

        tempo_index_df = _read_parquet_optional(_find_first_file(artifact_dir, [f"{key}_tempo_index.parquet", "*_tempo_index.parquet"]))
        grid_surface_df = _read_parquet_optional(_find_first_file(artifact_dir, [f"{key}_grid_surface.parquet", "*_grid_surface.parquet"]))
        segment_surface_df = _read_parquet_optional(_find_first_file(artifact_dir, [f"{key}_segment_surface.parquet", "*_segment_surface.parquet"]))
        surface_global_df = _read_parquet_optional(_find_first_file(artifact_dir, [f"{key}_surface_global.parquet", "*_surface_global.parquet"]))
        search_results_df = _read_csv_optional(_find_first_file(artifact_dir, [f"{key}_search_results.csv", "*_search_results.csv"]))
        holdout_path, oof_path = _artifact_prediction_paths(artifact_dir, key)
        holdout_predictions_df = _read_csv_optional(holdout_path)
        oof_predictions_df = _read_csv_optional(oof_path)

        tier_map_path = _find_first_file(artifact_dir, [f"{key}_tier_map.json", "*_tier_map.json"])
        tier_map_raw = _read_json(tier_map_path) if tier_map_path else {}
        tier_map_norm = {_normalize_token(name): str(tier) for name, tier in tier_map_raw.items()}

        bundle = ArtifactBundle(
            key=key,
            label=label,
            artifact_dir=artifact_dir,
            metadata_path=metadata_path,
            model_path=model_path,
            metadata=metadata,
            model=model,
            feature_names=feature_names,
            feature_types=feature_types,
            area_bins=area_bins,
            area_labels=area_labels,
            grid_cell_size=grid_cell_size,
            tempo_index_df=tempo_index_df,
            grid_surface_df=grid_surface_df,
            segment_surface_df=segment_surface_df,
            surface_global_df=surface_global_df,
            tier_map_raw=tier_map_raw,
            tier_map_norm=tier_map_norm,
            search_results_df=search_results_df,
            holdout_predictions_df=holdout_predictions_df,
            oof_predictions_df=oof_predictions_df,
            notebook_name=REQUESTED_MODEL_NOTEBOOKS[key],
        )
        _build_grid_and_temporal_maps(bundle)
        bundles[key] = bundle
        LOGGER.info("Artefatos carregados: %s (%s)", label, artifact_dir)

    if not bundles:
        raise RuntimeError("Nao foi possivel carregar artefatos de nenhum modelo.")
    return bundles


def load_reference_data(base_dir: Path = BASE_DIR) -> ReferenceContext:
    env_path = os.getenv("REFERENCE_DATA_PATH", "").strip()
    candidates: list[Path] = []
    if env_path:
        candidates.append(Path(env_path).expanduser())
    for root in _candidate_base_dirs(base_dir):
        candidates.extend(
            [
                root / "imoveis_v2_xgboost.parquet",
                root / "data" / "reference" / "base_referencia_final.parquet",
                root / "base_completa_quarteiroes.parquet",
            ]
        )

    for path in candidates:
        if not path.exists():
            continue
        try:
            df = pd.read_parquet(path)
        except Exception as exc:
            LOGGER.warning("Falha ao ler referencia %s: %s", path, exc)
            continue

        if df.empty:
            LOGGER.warning("Base de referencia vazia: %s", path)
            continue

        df = df.copy()
        df.columns = [str(col) for col in df.columns]
        df = _derive_ano_dado(df)

        if {"lat", "lon"}.issubset(df.columns):
            lon = pd.to_numeric(df["lon"], errors="coerce").to_numpy(dtype=float)
            lat = pd.to_numeric(df["lat"], errors="coerce").to_numpy(dtype=float)
            x, y = WGS84_TO_WEBMERC.transform(lon, lat)
            df["x"] = x
            df["y"] = y
        elif {"x", "y"}.issubset(df.columns):
            df["x"] = pd.to_numeric(df["x"], errors="coerce")
            df["y"] = pd.to_numeric(df["y"], errors="coerce")
        else:
            LOGGER.warning("Base %s sem lat/lon ou x/y. Inferencia por vizinho sera desativada.", path)
            return ReferenceContext(path=path, df=pd.DataFrame())

        mask = df["x"].notna() & df["y"].notna()
        df = df.loc[mask].reset_index(drop=True)
        if df.empty:
            LOGGER.warning("Base %s sem coordenadas validas apos limpeza.", path)
            return ReferenceContext(path=path, df=pd.DataFrame())

        ref = ReferenceContext(
            path=path,
            df=df,
            x=df["x"].to_numpy(dtype=float),
            y=df["y"].to_numpy(dtype=float),
        )
        LOGGER.info("Base de referencia carregada: %s (n=%s)", path.name, len(df))
        return ref

    LOGGER.warning("Nenhuma base de referencia disponivel. Sera usado fallback por defaults.")
    return ReferenceContext(path=None, df=pd.DataFrame())


def _candidate_polygon_paths(base_dir: Path) -> list[Path]:
    candidates: list[Path] = []
    env_path = os.getenv(POLYGON_CONTEXT_ENV, "").strip()
    if env_path:
        candidates.append(Path(env_path).expanduser())

    enriched_candidates: list[Path] = []
    legacy_candidates: list[Path] = []
    for root in _candidate_base_dirs(base_dir):
        enriched_candidates.extend(
            [
                root / "poligonos_enriquecidos_260317_com_features_modelos.parquet",
                root / "data" / "reference" / "poligonos_enriquecidos_260317_com_features_modelos.parquet",
                root / "data" / "geospatial" / "poligonos_enriquecidos_260317_com_features_modelos.parquet",
            ]
        )
        legacy_candidates.extend(
            [
                root / "poligonos_enriquecidos_260317.parquet",
                root / "data" / "reference" / "poligonos_enriquecidos_260317.parquet",
                root / "data" / "geospatial" / "poligonos_enriquecidos_260317.parquet",
                root / "data" / "reference" / "poligonos_enriquecidos.parquet",
                root / "data" / "geospatial" / "poligonos_enriquecidos.parquet",
            ]
        )
    candidates.extend(enriched_candidates)
    candidates.extend(legacy_candidates)

    unique: list[Path] = []
    seen: set[str] = set()
    for path in candidates:
        try:
            key = str(path.resolve())
        except Exception:
            key = str(path)
        if key in seen:
            continue
        seen.add(key)
        unique.append(path)
    return unique


def load_polygon_context(base_dir: Path = BASE_DIR) -> PolygonContext:
    context = PolygonContext()
    try:
        import geopandas as gpd
    except Exception as exc:
        context.error = f"geopandas indisponivel para geoparquet enriquecido: {exc}"
        LOGGER.warning("Camada de poligonos desativada: %s", context.error)
        return context

    for path in _candidate_polygon_paths(base_dir):
        if not path.exists():
            continue
        try:
            gdf = gpd.read_parquet(path)
        except Exception as exc:
            LOGGER.warning("Falha ao ler geoparquet %s: %s", path, exc)
            continue

        if gdf.empty:
            LOGGER.warning("Geoparquet vazio: %s", path)
            continue
        if gdf.crs is None:
            LOGGER.warning("Geoparquet sem CRS: %s", path)
            continue

        try:
            gdf = gdf.to_crs(POLYGON_TARGET_CRS)
        except Exception as exc:
            LOGGER.warning("Falha ao reprojetar geoparquet %s para %s: %s", path, POLYGON_TARGET_CRS, exc)
            continue

        available_fields = [field for field in POLYGON_CONTEXT_FIELDS if field in gdf.columns]
        model_context_fields = [field for field in available_fields if field != "NUMBLOCO"]
        if not model_context_fields:
            LOGGER.warning("Geoparquet %s sem campos espaciais esperados.", path)
            continue

        geometry_col = gdf.geometry.name
        work = gdf[available_fields + [geometry_col]].copy().reset_index(drop=True)
        work["__geom_area"] = work.geometry.area
        numbloco_key_index: dict[str, list[int]] = {}
        if "NUMBLOCO" in work.columns:
            work["__numbloco_key"] = work["NUMBLOCO"].map(_normalize_numbloco_key)
            for idx, key in work["__numbloco_key"].items():
                if key:
                    numbloco_key_index.setdefault(str(key), []).append(int(idx))
        LOGGER.info(
            "Geoparquet enriquecido carregado: %s (n=%s, campos=%s, NUMBLOCO=%s chaves)",
            path.name,
            len(work),
            ",".join(available_fields),
            len(numbloco_key_index),
        )
        return PolygonContext(
            path=path,
            gdf=work,
            available_fields=available_fields,
            numbloco_key_index=numbloco_key_index,
            fallback_max_distance_m=float(
                _to_float(
                    os.getenv("POLYGON_FALLBACK_MAX_DISTANCE_M", POLYGON_FALLBACK_MAX_DISTANCE_M),
                    POLYGON_FALLBACK_MAX_DISTANCE_M,
                )
                or POLYGON_FALLBACK_MAX_DISTANCE_M
            ),
            to_base_crs=Transformer.from_crs("EPSG:4326", gdf.crs, always_xy=True),
        )

    context.error = "Nenhum geoparquet enriquecido encontrado."
    LOGGER.warning(context.error)
    return context


def _candidate_zoneamento_paths(base_dir: Path) -> list[Path]:
    candidates: list[Path] = []
    env_path = os.getenv(ZONEAMENTO_SHP_ENV, "").strip()
    if env_path:
        path = Path(env_path).expanduser()
        if path.suffix.lower() == ".shp":
            candidates.append(path)

    search_roots: list[Path] = []
    for root in _candidate_base_dirs(base_dir):
        search_roots.extend([root / "pddua_shp", root])
    patterns = ["*SUBUNIDADE*.shp", "*subunidade*.shp", "*ZONEAMENTO*.shp", "*.shp"]
    for root in search_roots:
        if not root.exists():
            continue
        for pattern in patterns:
            for shp in sorted(root.glob(pattern)):
                if shp.is_file() and shp not in candidates:
                    candidates.append(shp)
    return candidates


def _candidate_iapond_shapefile_paths(base_dir: Path, exclude: Path | None = None) -> list[Path]:
    candidates: list[Path] = []
    for parent in _candidate_base_dirs(base_dir):
        for root in [parent / "pddua_shp", parent]:
            if not root.exists():
                continue
            for pattern in ["*ANEXO6*.shp", "*IA*.shp", "*.shp"]:
                for shp in sorted(root.glob(pattern)):
                    if not shp.is_file():
                        continue
                    if exclude is not None and shp.resolve() == exclude.resolve():
                        continue
                    if shp not in candidates:
                        candidates.append(shp)
    return candidates


def _build_iapond_map_from_dataframe(df: pd.DataFrame) -> dict[str, float]:
    if df.empty:
        return {}

    columns = [str(col) for col in df.columns]
    ia_col = _find_column_by_alias(columns, ZONEAMENTO_COLUMN_ALIASES["IA"])
    if ia_col is None:
        return {}

    code_columns: list[str] = []
    for canonical in ("CODIGO", "INDICE"):
        code_col = _find_column_by_alias(columns, ZONEAMENTO_COLUMN_ALIASES[canonical])
        if code_col and code_col not in code_columns:
            code_columns.append(code_col)
    if not code_columns:
        return {}

    out: dict[str, float] = {}
    for code_col in code_columns:
        codes = df[code_col].tolist()
        ia_values = df[ia_col].tolist()
        for code_raw, ia_raw in zip(codes, ia_values):
            code = _normalize_code_token(code_raw)
            ia_number = _to_float(ia_raw)
            if code == "" or ia_number is None:
                continue
            out[code] = float(ia_number)
    return out


def _load_iapond_map_from_json(path: Path) -> dict[str, float]:
    payload = _read_json(path)
    out: dict[str, float] = {}

    if isinstance(payload, dict):
        for key, value in payload.items():
            code = _normalize_code_token(key)
            ia_number = _to_float(value)
            if code and ia_number is not None:
                out[code] = float(ia_number)
        return out

    if isinstance(payload, list):
        try:
            frame = pd.DataFrame(payload)
        except Exception:
            return {}
        return _build_iapond_map_from_dataframe(frame)
    return {}


def _load_iapond_map_from_pdf(path: Path) -> dict[str, float]:
    try:
        from pypdf import PdfReader
    except Exception as exc:
        LOGGER.warning("pypdf indisponivel para leitura de %s: %s", path, exc)
        return {}

    try:
        reader = PdfReader(str(path))
    except Exception as exc:
        LOGGER.warning("Falha ao abrir PDF de mapeamento IA %s: %s", path, exc)
        return {}

    out: dict[str, float] = {}
    saw_header = False
    header_pattern = re.compile(r"\bCODIGO\b.*\bIA\b")
    row_pattern = re.compile(r"^\s*([0-9]+(?:[.,][0-9]+)*)\s*[\|;,:\t ]+\s*([0-9]+(?:[.,][0-9]+)?)\s*$")

    for page in reader.pages:
        text = page.extract_text() or ""
        for line in text.splitlines():
            ascii_line = _normalize_ascii_token(line)
            if header_pattern.search(ascii_line):
                saw_header = True
                continue
            if not saw_header:
                continue
            match = row_pattern.match(line.strip())
            if not match:
                continue
            code = _normalize_code_token(match.group(1))
            ia_number = _to_float(match.group(2))
            if code and ia_number is not None:
                out[code] = float(ia_number)

    if not out:
        LOGGER.warning("PDF %s nao trouxe linhas CODIGO->IA em formato tabular reconhecivel.", path.name)
    return out


def _candidate_iapond_map_paths(base_dir: Path) -> list[Path]:
    candidates: list[Path] = []

    env_map = os.getenv(IAPOND_MAP_ENV, "").strip()
    if env_map:
        candidates.append(Path(env_map).expanduser())

    for parent in _candidate_base_dirs(base_dir):
        defaults = [
            parent / "pddua_shp" / "iapond_map.csv",
            parent / "pddua_shp" / "iapond_map.parquet",
            parent / "pddua_shp" / "iapond_map.json",
            parent / "iapond_map.csv",
            parent / "iapond_map.parquet",
            parent / "iapond_map.json",
            parent / "data" / "reference" / "iapond_map.csv",
            parent / "data" / "reference" / "iapond_map.parquet",
            parent / "data" / "reference" / "iapond_map.json",
        ]
        for path in defaults:
            if path not in candidates:
                candidates.append(path)
    return candidates


def _candidate_iapond_pdf_paths(base_dir: Path) -> list[Path]:
    candidates: list[Path] = []
    env_pdf = os.getenv(IAPOND_PDF_ENV, "").strip()
    if env_pdf:
        candidates.append(Path(env_pdf).expanduser())

    for parent in _candidate_base_dirs(base_dir):
        defaults = [
            parent / "pddua_layout_anexo6.pdf",
            parent / "pddua_shp" / "pddua_layout_anexo6.pdf",
            parent / "docs" / "manuals" / "pddua_layout_anexo6.pdf",
            parent / "docs" / "technical" / "pddua_layout_anexo6.pdf",
        ]
        for path in defaults:
            if path not in candidates:
                candidates.append(path)
    return candidates


def _load_iapond_map_from_path(path: Path) -> dict[str, float]:
    suffix = path.suffix.lower()
    if suffix == ".json":
        return _load_iapond_map_from_json(path)
    if suffix in {".csv", ".txt"}:
        frame = pd.read_csv(path)
        return _build_iapond_map_from_dataframe(frame)
    if suffix == ".parquet":
        frame = pd.read_parquet(path)
        return _build_iapond_map_from_dataframe(frame)
    if suffix == ".pdf":
        return _load_iapond_map_from_pdf(path)
    return {}


def _load_iapond_map_from_shapefile(path: Path, gpd_module: Any) -> dict[str, float]:
    try:
        frame = gpd_module.read_file(path, ignore_geometry=True)
    except TypeError:
        frame = gpd_module.read_file(path)
    except Exception as exc:
        LOGGER.warning("Falha ao ler shapefile de IA %s: %s", path, exc)
        return {}

    frame_df = pd.DataFrame(frame)
    if "geometry" in frame_df.columns:
        frame_df = frame_df.drop(columns=["geometry"])
    return _build_iapond_map_from_dataframe(frame_df)


def _resolve_iapond_from_zone_row(row: pd.Series, zoneamento: ZoneamentoContext) -> tuple[float | None, str]:
    if "IA" in row.index:
        ia_number = _to_float(row.get("IA"))
        if ia_number is not None:
            return float(max(ia_number, 1e-6)), "zoneamento_shp_ia"

    code_candidates = [
        _normalize_code_token(row.get("INDICE")),
        _normalize_code_token(row.get("CODIGO")),
    ]
    for code in code_candidates:
        if code and code in IAPOND_BY_INDICE:
            return float(IAPOND_BY_INDICE[code]), "indice_anexo6"
        if code and code in zoneamento.iapond_map:
            return float(max(zoneamento.iapond_map[code], 1e-6)), f"mapa_ia:{zoneamento.iapond_map_source}"

    return None, "nao_disponivel"


def load_zoneamento_data(base_dir: Path = BASE_DIR) -> ZoneamentoContext:
    context = ZoneamentoContext()
    try:
        import geopandas as gpd
    except Exception as exc:
        context.error = f"geopandas indisponivel: {exc}"
        LOGGER.warning("Zoneamento desativado: %s", context.error)
        return context

    candidates = _candidate_zoneamento_paths(base_dir)
    if not candidates:
        context.error = "Nenhum shapefile de zoneamento encontrado."
        LOGGER.warning(context.error)
        return context

    best_score = -1
    selected_path: Path | None = None
    selected_gdf: pd.DataFrame | None = None
    selected_column_map: dict[str, str] = {}

    for path in candidates:
        if not path.exists():
            continue
        try:
            gdf = gpd.read_file(path)
        except Exception as exc:
            LOGGER.warning("Falha ao ler shapefile %s: %s", path, exc)
            continue

        if gdf.empty:
            continue

        columns = [str(col) for col in gdf.columns]
        canonical_map: dict[str, str] = {}
        used_source_columns: set[str] = set()
        for canonical, aliases in ZONEAMENTO_COLUMN_ALIASES.items():
            col = _find_column_by_alias(columns, aliases + [canonical])
            if col and col not in used_source_columns:
                canonical_map[canonical] = col
                used_source_columns.add(col)

        score = len([field for field in ZONEAMENTO_FIELDS if field in canonical_map])
        if score > best_score:
            best_score = score
            selected_path = path
            selected_gdf = gdf
            selected_column_map = canonical_map

        if score == len(ZONEAMENTO_FIELDS):
            break

    if selected_path is None or selected_gdf is None or best_score < 3:
        context.error = "Nenhum shapefile com colunas suficientes para zoneamento (MZ/UEU/SUBUNIDADE/...)"
        LOGGER.warning(context.error)
        return context

    if selected_gdf.crs is None:
        context.error = f"Shapefile sem CRS: {selected_path}"
        LOGGER.warning(context.error)
        return context

    try:
        selected_gdf = selected_gdf.to_crs(epsg=3857)
    except Exception as exc:
        context.error = f"Falha ao reprojetar shapefile {selected_path.name} para EPSG:3857: {exc}"
        LOGGER.warning(context.error)
        return context

    keep_columns = [selected_column_map[field] for field in selected_column_map if selected_column_map[field] in selected_gdf.columns]
    keep_columns = list(dict.fromkeys(keep_columns))
    geometry_col = selected_gdf.geometry.name
    work = selected_gdf[keep_columns + [geometry_col]].copy()
    rename_map = {source: canonical for canonical, source in selected_column_map.items() if source in work.columns}
    work = work.rename(columns=rename_map)
    work = work.reset_index(drop=True)
    work["__geom_area"] = work.geometry.area

    iapond_map = _build_iapond_map_from_dataframe(work.drop(columns=[geometry_col], errors="ignore"))
    iapond_source = "shapefile_ia"

    if not iapond_map:
        for shp_map_path in _candidate_iapond_shapefile_paths(base_dir, exclude=selected_path):
            external_map = _load_iapond_map_from_shapefile(shp_map_path, gpd)
            if external_map:
                iapond_map = external_map
                iapond_source = str(shp_map_path.name)
                break

    if not iapond_map:
        for map_path in _candidate_iapond_map_paths(base_dir):
            if not map_path.exists():
                continue
            try:
                external_map = _load_iapond_map_from_path(map_path)
            except Exception as exc:
                LOGGER.warning("Falha ao ler mapeamento IA %s: %s", map_path, exc)
                continue
            if external_map:
                iapond_map = external_map
                iapond_source = str(map_path.name)
                break

    if not iapond_map:
        for pdf_path in _candidate_iapond_pdf_paths(base_dir):
            if not pdf_path.exists():
                continue
            try:
                external_map = _load_iapond_map_from_path(pdf_path)
            except Exception as exc:
                LOGGER.warning("Falha ao ler IA de PDF %s: %s", pdf_path, exc)
                continue
            if external_map:
                iapond_map = external_map
                iapond_source = str(pdf_path.name)
                break

    context.shp_path = selected_path
    context.gdf = work
    context.column_map = selected_column_map
    context.iapond_map = iapond_map
    context.iapond_map_source = iapond_source if iapond_map else "fallback_default"

    LOGGER.info(
        "Zoneamento carregado de %s (n=%s, campos_zoneamento=%s, mapa_ia=%s)",
        selected_path.name,
        len(work),
        best_score,
        len(iapond_map),
    )
    if not iapond_map:
        LOGGER.warning("Mapeamento IA nao encontrado; IAPOND permanecera manual quando necessario.")
    return context


def _infer_zoneamento_context(zoneamento: ZoneamentoContext, x: float, y: float) -> tuple[dict[str, Any], dict[str, str], list[str]]:
    if not zoneamento.has_data:
        return {}, {}, []

    try:
        from shapely.geometry import Point
    except Exception as exc:
        LOGGER.warning("Shapely indisponivel para inferencia de zoneamento: %s", exc)
        return {}, {}, []

    point = Point(float(x), float(y))
    alerts: list[str] = []

    try:
        idx = zoneamento.gdf.sindex.query(point, predicate="intersects")
        idx_list = list(idx)
    except Exception:
        mask = zoneamento.gdf.geometry.intersects(point)
        idx_list = zoneamento.gdf.index[mask].tolist()

    if not idx_list:
        alerts.append("Coordenada fora do shapefile de zoneamento; campos de zoneamento permanecem manuais.")
        return {}, {}, alerts

    matches = zoneamento.gdf.iloc[idx_list]
    if "__geom_area" in matches.columns:
        row = matches.sort_values("__geom_area", ascending=True).iloc[0]
    else:
        row = matches.iloc[0]

    values: dict[str, Any] = {}
    sources: dict[str, str] = {}
    for field in ZONEAMENTO_FIELDS:
        if field in row.index and not _is_missing(row.get(field)):
            values[field] = row.get(field)
            sources[field] = "zoneamento_shp"

    iapond_value, iapond_source = _resolve_iapond_from_zone_row(row, zoneamento)
    if iapond_value is not None:
        values["IAPOND"] = float(iapond_value)
        sources["IAPOND"] = iapond_source
    else:
        alerts.append("INDICE sem IA mapeado; IAPOND permanece manual.")

    return values, sources, alerts


def _polygon_row_to_context(
    polygons: PolygonContext,
    row: pd.Series,
    source_label: str,
    polygon_distance: float,
) -> tuple[dict[str, Any], dict[str, str], list[str]]:
    values: dict[str, Any] = {}
    sources: dict[str, str] = {}
    for field in polygons.available_fields:
        if field in row.index and not _is_missing(row.get(field)):
            values[field] = row.get(field)
            sources[field] = source_label
    values["_polygon_distance_m"] = polygon_distance
    values["_polygon_lookup_mode"] = source_label
    if "NUMBLOCO" in row.index and not _is_missing(row.get("NUMBLOCO")):
        values["_polygon_numbloco"] = row.get("NUMBLOCO")
    return values, sources, []


def _select_best_polygon_row(
    polygons: PolygonContext,
    matches: pd.DataFrame,
    point: Any | None = None,
) -> tuple[pd.Series | None, float, list[str]]:
    if matches.empty:
        return None, float("nan"), []

    alerts: list[str] = []
    if point is not None and "geometry" in matches.columns:
        try:
            intersecting = matches.loc[matches.geometry.intersects(point)]
        except Exception:
            intersecting = pd.DataFrame()
        if not intersecting.empty:
            if "__geom_area" in intersecting.columns:
                return intersecting.sort_values("__geom_area", ascending=True).iloc[0], 0.0, alerts
            return intersecting.iloc[0], 0.0, alerts

        try:
            distances = matches.geometry.distance(point)
            nearest_idx = distances.idxmin()
            nearest_distance = _to_float(distances.loc[nearest_idx], float("nan"))
        except Exception:
            nearest_idx = None
            nearest_distance = float("nan")
        if nearest_idx is not None and np.isfinite(nearest_distance):
            alerts.append(
                "NUMBLOCO encontrado, mas a coordenada nao intersecta seus poligonos; "
                f"selecionado registro do mesmo NUMBLOCO mais proximo a {nearest_distance:.1f} m."
            )
            return matches.loc[nearest_idx], float(nearest_distance), alerts

    rank = matches.copy()
    feature_cols = [field for field in matches.columns if field in polygons.available_fields]
    rank["__non_null_count"] = rank[feature_cols].notna().sum(axis=1) if feature_cols else 0
    if "__geom_area" not in rank.columns:
        rank["__geom_area"] = float("inf")
    rank = rank.sort_values(["__non_null_count", "__geom_area"], ascending=[False, True])
    return rank.iloc[0], float("nan"), alerts


def _select_polygon_row_by_numbloco(
    polygons: PolygonContext,
    numbloco: Any,
    x: float | None = None,
    y: float | None = None,
) -> tuple[pd.Series | None, float, list[str], int]:
    if not polygons.has_data or not polygons.numbloco_key_index:
        return None, float("nan"), [], 0

    key = _normalize_numbloco_key(numbloco)
    if not key:
        return None, float("nan"), [], 0

    candidate_idx = polygons.numbloco_key_index.get(key)
    if not candidate_idx:
        return (
            None,
            float("nan"),
            [f"NUMBLOCO `{_safe_text(numbloco, '')}` nao encontrado no geoparquet enriquecido; usada busca espacial."],
            0,
        )

    point = None
    x_number = _to_float(x)
    y_number = _to_float(y)
    if x_number is not None and y_number is not None and np.isfinite(x_number) and np.isfinite(y_number):
        try:
            from shapely.geometry import Point

            point = Point(x_number, y_number)
        except Exception as exc:
            LOGGER.warning("Shapely indisponivel para desempate por NUMBLOCO: %s", exc)

    matches = polygons.gdf.loc[candidate_idx]
    row, polygon_distance, alerts = _select_best_polygon_row(polygons, matches, point)
    if row is None:
        return (
            None,
            float("nan"),
            [f"NUMBLOCO `{_safe_text(numbloco, '')}` sem linha utilizavel no geoparquet; usada busca espacial."],
            len(candidate_idx),
        )

    if len(candidate_idx) > 1:
        alerts.append(f"NUMBLOCO encontrado em {len(candidate_idx)} poligonos; aplicado desempate deterministico.")

    return row, polygon_distance, alerts, len(candidate_idx)


def _infer_polygon_context_by_numbloco(
    polygons: PolygonContext,
    numbloco: Any,
    x: float | None = None,
    y: float | None = None,
) -> tuple[dict[str, Any], dict[str, str], list[str]]:
    row, polygon_distance, alerts, _ = _select_polygon_row_by_numbloco(polygons, numbloco, x, y)
    if row is None:
        return {}, {}, alerts

    values, sources, _ = _polygon_row_to_context(polygons, row, "poligono_numbloco", polygon_distance)
    return values, sources, alerts


def _infer_polygon_context(polygons: PolygonContext, x: float, y: float) -> tuple[dict[str, Any], dict[str, str], list[str]]:
    if not polygons.has_data:
        return {}, {}, []

    try:
        from shapely.geometry import Point
    except Exception as exc:
        LOGGER.warning("Shapely indisponivel para geoparquet enriquecido: %s", exc)
        return {}, {}, []

    point = Point(float(x), float(y))
    alerts: list[str] = []

    try:
        idx = polygons.gdf.sindex.query(point, predicate="intersects")
        idx_list = list(idx)
    except Exception:
        mask = polygons.gdf.geometry.intersects(point)
        idx_list = polygons.gdf.index[mask].tolist()

    source_label = "poligono_intersect"
    row = None
    polygon_distance = 0.0
    if idx_list:
        matches = polygons.gdf.iloc[idx_list]
        row, polygon_distance, _ = _select_best_polygon_row(polygons, matches, point)
    else:
        try:
            distances = polygons.gdf.geometry.distance(point)
            nearest_idx = distances.idxmin()
            nearest_distance = _to_float(distances.loc[nearest_idx], float("nan"))
        except Exception:
            nearest_idx = None
            nearest_distance = float("nan")

        if nearest_idx is not None and np.isfinite(nearest_distance):
            row = polygons.gdf.loc[nearest_idx]
            source_label = "poligono_proximo"
            polygon_distance = float(nearest_distance)
            if nearest_distance > polygons.fallback_max_distance_m:
                alerts.append(
                    f"Coordenada sem interseccao exata com a malha enriquecida; "
                    f"usado poligono mais proximo a {nearest_distance:.1f} m "
                    "(baixa confiabilidade espacial)."
                )
            else:
                alerts.append(
                    f"Coordenada sem interseccao exata com a malha enriquecida; "
                    f"usado poligono mais proximo a {nearest_distance:.1f} m."
                )
        else:
            alerts.append(
                "Camada de poligonos enriquecidos sem interseccao para a coordenada; "
                "mantido fallback por base de referencia."
            )
            return {}, {}, alerts

    if row is None:
        return {}, {}, alerts

    values, sources, _ = _polygon_row_to_context(polygons, row, source_label, polygon_distance)
    return values, sources, alerts


def _infer_rh_context_by_numbloco(rh_context: RhNumblocoContext, numbloco: Any) -> tuple[dict[str, Any], dict[str, str], list[str]]:
    if not rh_context.has_data:
        return {}, {}, []

    key = _normalize_numbloco_key(numbloco)
    if not key:
        return {}, {}, []

    rh_value = rh_context.rh_by_numbloco.get(key)
    if rh_value is None or not np.isfinite(float(rh_value)):
        return {}, {}, [f"NUMBLOCO `{_safe_text(numbloco, '')}` sem RH no CSV; usado fallback espacial quando disponivel."]

    return {"RH": float(rh_value)}, {"RH": "rh_numbloco_csv"}, []


def _infer_testada_context_by_numbloco(
    testada_context: TestadaNumblocoContext,
    numbloco: Any,
) -> tuple[dict[str, Any], dict[str, str], list[str]]:
    if not testada_context.has_data:
        return {}, {}, []

    key = _normalize_numbloco_key(numbloco)
    if not key:
        return {}, {}, []

    testada_value = testada_context.testada_by_numbloco.get(key)
    if testada_value is None or not np.isfinite(float(testada_value)):
        return {}, {}, [f"NUMBLOCO `{_safe_text(numbloco, '')}` sem TESTADA no CSV; usado fallback atual."]

    return (
        {"TESTADA": float(testada_value), "TESTADA_bruta": float(testada_value)},
        {"TESTADA": "testada_numbloco_csv", "TESTADA_bruta": "testada_numbloco_csv"},
        [],
    )


def _build_reference_categories(bundle: ArtifactBundle, reference: ReferenceContext) -> dict[str, list[str]]:
    maps: dict[str, list[str]] = {}

    for feature in bundle.feature_names:
        if str(bundle.feature_types.get(feature, "")).lower() != "c":
            continue

        values: list[str] = []
        if reference.has_data and feature in reference.df.columns:
            values.extend(
                _safe_text(value, "")
                for value in reference.df[feature].dropna().tolist()
                if _safe_text(value, "") != ""
            )

        if feature == "Ano_Dado" and reference.has_data:
            if "Ano_Dado" in reference.df.columns:
                values.extend(
                    str(int(value))
                    for value in pd.to_numeric(reference.df["Ano_Dado"], errors="coerce").dropna().tolist()
                )
            elif "ANO" in reference.df.columns:
                values.extend(
                    str(int(value))
                    for value in pd.to_numeric(reference.df["ANO"], errors="coerce").dropna().tolist()
                )

        if feature == "faixa_area_modelo":
            values.extend(bundle.area_labels)
        if feature == "faixa_rh":
            values.extend(RH_LABELS)
        if feature == "FONTE":
            values.extend(["0", "1", "Venda", "Oferta"])

        maps[feature] = _normalize_category_values(values)
    return maps


def _infer_local_context(reference: ReferenceContext, x: float, y: float, k_neighbors: int) -> dict[str, Any]:
    if not reference.has_data:
        return {}

    dx = reference.x - x
    dy = reference.y - y
    dist2 = dx * dx + dy * dy
    if dist2.size == 0:
        return {}

    k = max(1, min(k_neighbors, dist2.size))
    idx = np.argpartition(dist2, k - 1)[:k]
    idx = idx[np.argsort(dist2[idx])]
    local = reference.df.iloc[idx]

    context: dict[str, Any] = {}
    for field in CONTEXT_FIELDS:
        if field not in local.columns:
            continue
        prefer_mode = field in {
            "FONTE",
            "Ano_Dado",
            "BAIRRO",
            "COD_BAIRRO",
            "FINALIDADE",
            "MZ",
            "UEU",
            "SUBUNIDADE",
            "DENSIDADE",
            "ATIVIDADE",
            "INDICE",
            "VOLUMETRIA",
        }
        context[field] = _summarize_series(local[field], prefer_mode=prefer_mode)

    context["_nearest_distance_m"] = float(np.sqrt(dist2[idx[0]]))
    return context


def select_model(
    tipo_imovel: str,
    bundles: dict[str, ArtifactBundle],
    area_m2: float | None = None,
) -> tuple[ArtifactBundle, str]:
    token = _normalize_token(tipo_imovel)
    if token.startswith("AUTO"):
        if area_m2 is None or not np.isfinite(area_m2) or area_m2 <= 0:
            raise ValueError("Area valida e obrigatoria para roteamento automatico.")
        key = "terreno" if float(area_m2) < AUTO_GLEBA_AREA_THRESHOLD else "gleba"
        source = f"auto_area_{AUTO_GLEBA_AREA_THRESHOLD:.0f}"
    elif token.startswith("TER"):
        key = "terreno"
        source = "manual"
    elif token.startswith("GLE"):
        key = "gleba"
        source = "manual"
    else:
        raise ValueError("Tipo de imovel invalido. Use Auto (area), Terreno ou Gleba.")

    if key not in bundles:
        raise ValueError(f"Artefatos nao encontrados para {key}.")
    return bundles[key], source


def _pick_value(
    field: str,
    overrides_ci: dict[str, Any],
    direct_inputs: dict[str, Any],
    local_context: dict[str, Any],
    default: Any,
    aliases: list[str] | None = None,
    context_sources: dict[str, str] | None = None,
    reliable_only: bool = False,
) -> tuple[Any, str]:
    aliases = aliases or []
    context_sources = context_sources or {}
    keys = [field.lower(), *(alias.lower() for alias in aliases)]
    candidate_names = [field, *aliases]

    for key in keys:
        if key in overrides_ci and not _is_missing(overrides_ci[key]):
            return overrides_ci[key], "override_json"

    for name in candidate_names:
        if name in direct_inputs and not _is_missing(direct_inputs[name]):
            return direct_inputs[name], "input_ui"

    for name in candidate_names:
        if name in local_context and not _is_missing(local_context[name]):
            source = context_sources.get(name, context_sources.get(field, "inferido_coord"))
            if reliable_only and not _is_reliable_model_context_source(source):
                continue
            return local_context[name], source

    return default, "default"


def _pick_testada_value(
    overrides_ci: dict[str, Any],
    direct_inputs: dict[str, Any],
    local_context: dict[str, Any],
    context_sources: dict[str, str],
) -> tuple[Any, str]:
    keys = ["testada_bruta", "testada"]
    for key in keys:
        if key in overrides_ci and not _is_missing(overrides_ci[key]):
            return overrides_ci[key], "override_json"

    for name in ("TESTADA_bruta", "TESTADA"):
        if name in local_context and not _is_missing(local_context[name]):
            source = context_sources.get(name, context_sources.get("TESTADA_bruta", "inferido_coord"))
            if source == "testada_numbloco_csv":
                return local_context[name], source

    return _pick_value(
        "TESTADA_bruta",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        aliases=["TESTADA", "testada", "testada_bruta"],
        context_sources=context_sources,
        reliable_only=True,
    )


def _resolve_temporal_index(bundle: ArtifactBundle, ano_dado: int) -> tuple[float, str | None, str]:
    if not bundle.tempo_index_map:
        return 1.0, None, "sem_artifato"

    periodos = sorted(bundle.tempo_index_map.keys())
    candidatos = [periodo for periodo in periodos if str(ano_dado) in periodo]
    if candidatos:
        periodo = sorted(candidatos)[-1]
        return float(bundle.tempo_index_map[periodo]), periodo, "ano_match"

    if not bundle.tempo_index_df.empty and {"periodo_tempo", "n_obs_periodo"}.issubset(bundle.tempo_index_df.columns):
        tmp = bundle.tempo_index_df.copy()
        tmp["n_obs_periodo"] = pd.to_numeric(tmp["n_obs_periodo"], errors="coerce")
        tmp = tmp.dropna(subset=["n_obs_periodo"])
        if not tmp.empty:
            row = tmp.sort_values("n_obs_periodo", ascending=False).iloc[0]
            periodo = _safe_text(row["periodo_tempo"], "")
            if periodo in bundle.tempo_index_map:
                return float(bundle.tempo_index_map[periodo]), periodo, "maior_n_obs"

    periodo = periodos[-1]
    return float(bundle.tempo_index_map[periodo]), periodo, "fallback_ultimo"


def _lookup_spatial_surface(bundle: ArtifactBundle, segmento_area: str, grid_id: str) -> tuple[float | None, str]:
    seg_key = _normalize_token(segmento_area)
    if seg_key and grid_id:
        val = bundle.grid_value_map.get((seg_key, grid_id))
        if val is not None:
            return float(val), "grid"

    if seg_key:
        val = bundle.segment_value_map.get(seg_key)
        if val is not None:
            return float(val), "segmento"

    if bundle.global_surface_value is not None:
        return float(bundle.global_surface_value), "global"

    return None, "sem_artifato"


def apply_preprocessing(
    bundle: ArtifactBundle,
    payload: dict[str, Any],
    local_context: dict[str, Any],
    overrides_ci: dict[str, Any],
    context_sources: dict[str, str] | None = None,
) -> tuple[dict[str, Any], dict[str, str], list[str]]:
    alerts: list[str] = []
    source_map: dict[str, str] = {}
    context_sources = context_sources or {}

    area = _to_float(payload.get("area"))
    if area is None or area <= 0:
        raise ValueError("Area deve ser numerica e maior que zero.")

    lat = _to_float(payload.get("lat"))
    lon = _to_float(payload.get("lon"))
    if lat is None or lon is None:
        raise ValueError("Latitude e longitude sao obrigatorias.")
    if not (-90.0 <= lat <= 90.0 and -180.0 <= lon <= 180.0):
        raise ValueError("Latitude/longitude fora da faixa valida.")

    x, y = WGS84_TO_WEBMERC.transform(lon, lat)
    nearest_distance = _to_float(local_context.get("_nearest_distance_m"), float("nan"))
    if np.isfinite(nearest_distance):
        threshold = bundle.grid_cell_size * 8.0
        if nearest_distance > threshold:
            alerts.append(
                f"Coordenada distante da base de referencia ({nearest_distance:.0f} m). "
                "Revisar localizacao e dados cadastrais."
            )

    direct_inputs = {
        "BAIRRO": payload.get("bairro"),
        "Ano_Dado": payload.get("ano_dado"),
        "IAPOND": payload.get("iapond"),
        "APP": payload.get("app"),
        "LOTPOS": payload.get("lotpos"),
        "TESTADA": payload.get("testada"),
        "TESTADA_bruta": payload.get("testada"),
    }

    mercado_default = "TERRENO" if bundle.key == "terreno" else "GLEBA"

    rh_raw, source_map["RH"] = _pick_value(
        "RH",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    iapond_raw, source_map["IAPOND"] = _pick_value(
        "IAPOND",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    app_raw, source_map["APP"] = _pick_value(
        "APP",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    cp_raw, source_map["CP"] = _pick_value(
        "CP",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    lotpos_raw, source_map["LOTPOS"] = _pick_value(
        "LOTPOS",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    ano_raw, source_map["Ano_Dado"] = _pick_value(
        "Ano_Dado",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        aliases=["ano", "ano_dado"],
        context_sources=context_sources,
        reliable_only=True,
    )
    fonte_raw = DEFAULTS_CATEGORICAL["FONTE"]
    source_map["FONTE"] = "fixo_zero"
    bairro_raw, source_map["BAIRRO"] = _pick_value(
        "BAIRRO",
        overrides_ci,
        direct_inputs,
        local_context,
        "",
        aliases=["bairro"],
        context_sources=context_sources,
    )
    finalidade_raw, source_map["FINALIDADE"] = _pick_value(
        "FINALIDADE",
        overrides_ci,
        direct_inputs,
        local_context,
        mercado_default,
        aliases=["finalidade"],
        context_sources=context_sources,
    )

    testada_raw, source_map["TESTADA_bruta"] = _pick_testada_value(
        overrides_ci,
        direct_inputs,
        local_context,
        context_sources,
    )
    testada = _to_float(testada_raw)
    if testada is None or testada <= 0:
        raise ValueError("TESTADA ausente; informe manualmente ou use uma coordenada com TESTADA disponivel no poligono.")

    rh = _to_float(rh_raw)
    if rh is not None:
        rh = max(rh, 1e-6)

    iapond = _to_float(iapond_raw)
    if iapond is not None:
        iapond = max(iapond, 1e-6)

    app = _to_float(app_raw)
    if app is not None:
        app = min(max(app, 0.0), 1.0)

    cp_override = _to_float(cp_raw)
    if source_map.get("CP") == "override_json" and cp_override is not None:
        cp = cp_override
    else:
        cp = _derive_cp(area=area, testada=testada)
        source_map["CP"] = "regra_area_testada"
    pe = area / testada
    source_map["PE"] = "regra_area_testada"

    lotpos = _to_float(lotpos_raw)
    ano_dado = _to_int(ano_raw)
    if ano_dado is not None:
        ano_dado = min(max(ano_dado, 2010), 2100)

    area_simpson_raw, source_map["area_simpson"] = _pick_value(
        "area_simpson",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    taxa_ocupacao_raw, source_map["taxa_ocupacao"] = _pick_value(
        "taxa_ocupacao",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    bldg_density_raw, source_map["bldg_density"] = _pick_value(
        "bldg_density",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    avg_bldg_raw, source_map["avg_bldg_footprint"] = _pick_value(
        "avg_bldg_footprint",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    circularity_raw, source_map["circularity"] = _pick_value(
        "circularity",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    dist_road_raw, source_map["dist_to_main_road"] = _pick_value(
        "dist_to_main_road",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )
    dist_park_raw, source_map["dist_to_park"] = _pick_value(
        "dist_to_park",
        overrides_ci,
        direct_inputs,
        local_context,
        None,
        context_sources=context_sources,
        reliable_only=True,
    )

    area_simpson = _to_float(area_simpson_raw)
    taxa_ocupacao = _to_float(taxa_ocupacao_raw)
    bldg_density = _to_float(bldg_density_raw)
    avg_bldg = _to_float(avg_bldg_raw)
    circularity = _to_float(circularity_raw)
    dist_road = _to_float(dist_road_raw)
    if dist_road is not None:
        dist_road = max(dist_road, 0.0)
    dist_park = _to_float(dist_park_raw)
    if dist_park is not None:
        dist_park = max(dist_park, 0.0)

    dist_weighted_density = None
    if bldg_density is not None and dist_road is not None:
        dist_weighted_density = bldg_density / (dist_road + 1.0)
        source_map["Dist_weighted_Density"] = "calculado"

    faixa_area = _categorize_area(area, bundle.area_bins, bundle.area_labels)
    faixa_rh = _categorize_rh(rh) if rh is not None else None

    area_total_construivel = None
    if app is not None and iapond is not None:
        area_total_construivel = max(area * (1.0 - app) * iapond, 1e-6)
    log_area = math.log1p(max(area, 0.0))
    log_testada = math.log1p(max(testada, 0.0))

    finite_bins = [value for value in bundle.area_bins if np.isfinite(value)]
    if finite_bins:
        area_min = min(finite_bins)
        area_max = max(finite_bins)
        if area < area_min or area > area_max:
            alerts.append(
                f"Area informada ({area:.2f} m2) fora da faixa observada nos bins do modelo "
                f"({area_min:.2f} a {area_max:.2f} m2)."
            )

    grid_id = _compute_grid_id(x, y, bundle.grid_cell_size)
    spatial_idx, spatial_source = _lookup_spatial_surface(bundle, faixa_area, grid_id)
    source_map["indice_espacial_grid"] = spatial_source
    if spatial_idx is None:
        spatial_idx = float("nan")

    temporal_idx, periodo_tempo, temporal_source = _resolve_temporal_index(bundle, ano_dado)
    source_map["indice_temporal_interno"] = temporal_source

    bairro_text = _safe_text(bairro_raw, "")
    bairro_key = _normalize_token(bairro_text)
    tier_localizacao = bundle.tier_map_norm.get(bairro_key, "localizacao_desconhecida")
    source_map["tier_localizacao"] = "tier_map" if bairro_key in bundle.tier_map_norm else "fallback"

    zone_values: dict[str, str] = {}
    for field in ZONEAMENTO_FIELDS:
        value, source = _pick_value(
            field,
            overrides_ci,
            direct_inputs,
            local_context,
            None,
            aliases=[field.lower()],
            context_sources=context_sources,
            reliable_only=True,
        )
        zone_values[field] = None if _is_missing(value) else _safe_text(value)
        source_map[field] = source
        if bundle.key == "gleba" and zone_values[field] is None:
            alerts.append(f"{field} nao inferido automaticamente; informe manualmente via override quando necessario.")

    prepared = {
        "AREA_bruta": area,
        "TESTADA_bruta": testada,
        "lat": lat,
        "lon": lon,
        "x": float(x),
        "y": float(y),
        "RH": rh,
        "IAPOND": iapond,
        "APP": app,
        "PE": pe,
        "CP": cp,
        "LOTPOS": lotpos,
        "Ano_Dado": str(ano_dado),
        "FONTE": fonte_raw,
        "FINALIDADE": _safe_text(finalidade_raw, mercado_default),
        "BAIRRO": bairro_text,
        "faixa_area_modelo": faixa_area,
        "faixa_rh": faixa_rh,
        "log_area_modelo": log_area,
        "log_testada_modelo": log_testada,
        "Area_Total_Construivel": area_total_construivel,
        "area_simpson": area_simpson,
        "taxa_ocupacao": taxa_ocupacao,
        "bldg_density": bldg_density,
        "avg_bldg_footprint": avg_bldg,
        "circularity": circularity,
        "dist_to_main_road": dist_road,
        "dist_to_park": dist_park,
        "Dist_weighted_Density": dist_weighted_density,
        "indice_espacial_grid": spatial_idx,
        "periodo_tempo": periodo_tempo or "",
        "indice_temporal_interno": temporal_idx,
        "tier_localizacao": tier_localizacao,
    }
    prepared.update(zone_values)

    if not _is_missing(prepared["FONTE"]):
        fonte_categories = bundle.reference_categories.get("FONTE", ["0", "1", "desconhecido"])
        prepared["FONTE"] = _normalize_fonte(prepared["FONTE"], fonte_categories)

    return prepared, source_map, alerts


def _default_for_feature(feature: str, feature_type: str) -> Any:
    if feature in DEFAULTS_NUMERIC:
        return DEFAULTS_NUMERIC[feature]
    if feature in DEFAULTS_CATEGORICAL:
        return DEFAULTS_CATEGORICAL[feature]
    if "int" in feature_type:
        return 0
    if feature_type == "c":
        return "desconhecido"
    return 0.0


def _prepare_categorical_value(feature: str, value: Any, categories: list[str]) -> str:
    if feature == "FONTE":
        return _normalize_fonte(value, categories)
    if _is_missing(value):
        return "desconhecido"
    number = _to_float(value)
    if number is not None and float(number).is_integer():
        text = str(int(number))
    else:
        text = _safe_text(value, "desconhecido")
    return text


def _missing_required_features(bundle: ArtifactBundle, prepared_values: dict[str, Any]) -> list[str]:
    return [
        feature
        for feature in bundle.feature_names
        if feature not in prepared_values or _is_missing(prepared_values.get(feature))
    ]


def build_features(bundle: ArtifactBundle, prepared_values: dict[str, Any]) -> pd.DataFrame:
    missing = _missing_required_features(bundle, prepared_values)
    if missing:
        raise ValueError(
            "Features obrigatorias sem preenchimento confiavel: "
            f"{', '.join(missing)}. Informe manualmente os campos disponiveis ou use overrides JSON."
        )

    data: dict[str, Any] = {}

    for feature in bundle.feature_names:
        feature_type = str(bundle.feature_types.get(feature, "float")).lower()
        value = prepared_values[feature]

        if feature_type == "c":
            categories = bundle.reference_categories.get(feature, ["desconhecido"])
            cat_value = _prepare_categorical_value(feature, value, categories)
            if cat_value not in categories:
                categories = categories + [cat_value]
            data[feature] = pd.Series(pd.Categorical([cat_value], categories=categories))
            continue

        if "int" in feature_type:
            number = _to_float(value, 0.0) or 0.0
            data[feature] = [int(round(number))]
            continue

        number = _to_float(value, 0.0) or 0.0
        data[feature] = [float(number)]

    return pd.DataFrame(data, columns=bundle.feature_names)


def predict_value(bundle: ArtifactBundle, x_input: pd.DataFrame, area_m2: float) -> dict[str, Any]:
    dmatrix = xgb.DMatrix(
        x_input,
        enable_categorical=True,
        feature_names=bundle.feature_names,
    )
    pred_raw = float(bundle.model.predict(dmatrix)[0])

    use_log_backtransform = bool(bundle.metadata.get("model_target_log_col"))
    pred_level = math.expm1(pred_raw) if use_log_backtransform else pred_raw
    pred_level = max(pred_level, 0.0)

    target_col = str(bundle.metadata.get("model_target_col", "")).upper()
    if "VUNIT" in target_col and "VTOTAL" not in target_col:
        vunit = pred_level
        vtotal = pred_level * area_m2
        target_mode = "VUNIT"
    else:
        vtotal = pred_level
        vunit = vtotal / area_m2 if area_m2 > 0 else float("nan")
        target_mode = "VTOTAL"

    return {
        "pred_raw_model": pred_raw,
        "pred_level": pred_level,
        "target_mode": target_mode,
        "valor_total": float(vtotal),
        "valor_unitario": float(vunit),
    }


def format_output(
    bundle: ArtifactBundle,
    prediction: dict[str, Any],
    prepared_values: dict[str, Any],
    source_map: dict[str, str],
    alerts: list[str],
    local_context: dict[str, Any],
    routing_source: str,
) -> tuple[str, dict[str, Any]]:
    resultado_lines = [
        "### Resultado de Inferencia",
        f"- **Modelo aplicado:** {bundle.label} (`{bundle.model_path.name}`)",
        f"- **Roteamento macro:** `{routing_source}`",
        f"- **Escala do alvo no treino:** `{bundle.metadata.get('model_target_log_col', 'sem_log')}`",
        f"- **Valor unitario estimado:** **{_currency_brl(prediction['valor_unitario'])} / m2**",
        f"- **Valor total estimado:** **{_currency_brl(prediction['valor_total'])}**",
        f"- **Modo da saida:** `{prediction['target_mode']}`",
    ]

    if alerts:
        resultado_lines.append("- **Alertas:**")
        for msg in sorted(set(alerts)):
            resultado_lines.append(f"  - {msg}")

    resumo = {
        "status": "ok",
        "modelo": {
            "tipo": bundle.label,
            "roteamento_macro": routing_source,
            "notebook_origem": bundle.notebook_name,
            "artifact_dir": str(bundle.artifact_dir),
            "model_file": bundle.model_path.name,
            "metadata_file": bundle.metadata_path.name,
            "n_features": len(bundle.feature_names),
        },
        "predicao": prediction,
        "variaveis_principais": {
            "area_m2": prepared_values.get("AREA_bruta"),
            "testada_m": prepared_values.get("TESTADA_bruta"),
            "lat": prepared_values.get("lat"),
            "lon": prepared_values.get("lon"),
            "x": prepared_values.get("x"),
            "y": prepared_values.get("y"),
            "RH": prepared_values.get("RH"),
            "IAPOND": prepared_values.get("IAPOND"),
            "APP": prepared_values.get("APP"),
            "PE": prepared_values.get("PE"),
            "CP": prepared_values.get("CP"),
            "Area_Total_Construivel": prepared_values.get("Area_Total_Construivel"),
            "faixa_area_modelo": prepared_values.get("faixa_area_modelo"),
            "faixa_rh": prepared_values.get("faixa_rh"),
            "FONTE": prepared_values.get("FONTE"),
            "Ano_Dado": prepared_values.get("Ano_Dado"),
            "FINALIDADE": prepared_values.get("FINALIDADE"),
            "MZ": prepared_values.get("MZ"),
            "UEU": prepared_values.get("UEU"),
            "SUBUNIDADE": prepared_values.get("SUBUNIDADE"),
            "DENSIDADE": prepared_values.get("DENSIDADE"),
            "ATIVIDADE": prepared_values.get("ATIVIDADE"),
            "INDICE": prepared_values.get("INDICE"),
            "VOLUMETRIA": prepared_values.get("VOLUMETRIA"),
            "indice_espacial_grid": prepared_values.get("indice_espacial_grid"),
            "periodo_tempo": prepared_values.get("periodo_tempo"),
            "indice_temporal_interno": prepared_values.get("indice_temporal_interno"),
            "tier_localizacao": prepared_values.get("tier_localizacao"),
        },
        "origem_variaveis": source_map,
        "diagnostico": {
            "distancia_vizinho_m": _to_float(local_context.get("_nearest_distance_m")),
            "distancia_poligono_m": _to_float(local_context.get("_polygon_distance_m")),
            "modo_busca_poligono": local_context.get("_polygon_lookup_mode"),
            "numbloco_poligono": local_context.get("_polygon_numbloco"),
            "alerts": sorted(set(alerts)),
        },
    }
    return "\n".join(resultado_lines), _jsonable(resumo)


def _build_report_limitations(
    bundle: ArtifactBundle,
    alerts: list[str],
    train_metrics: dict[str, float] | None,
) -> list[str]:
    limitations = [
        "A estimativa depende da qualidade das coordenadas informadas e da cobertura da base espacial de apoio.",
        "A busca espacial usa o poligono intersectante; fora da malha, usa o poligono mais proximo e registra a distancia.",
        "A inferencia aplica artefatos congelados; o app nao reestima nem retreina modelos.",
    ]
    if train_metrics is None:
        limitations.append(
            "Predicoes de treino nao foram exportadas para este modelo; por isso R2 (log) e grafico de treino aparecem como n/d."
        )
    metadata_warnings = bundle.metadata.get("warnings")
    if isinstance(metadata_warnings, list):
        limitations.extend(
            str(item)
            for item in metadata_warnings
            if "cod" not in str(item).lower() and "prd" not in str(item).lower()
        )
    limitations.extend(alerts)
    return list(dict.fromkeys(limitations))


def initialize_app_state(base_dir: Path = BASE_DIR) -> AppState:
    state = AppState()
    try:
        state.k_neighbors = max(1, int(os.getenv("K_NEIGHBORS", "120")))
    except Exception:
        state.k_neighbors = 120

    try:
        state.reference = load_reference_data(base_dir)
        state.bundles = load_artifacts(base_dir)
        state.rh_numbloco = load_rh_numbloco_context(base_dir)
        state.testada_numbloco = load_testada_numbloco_context(base_dir)
        state.polygon_context = load_polygon_context(base_dir)
        state.zoneamento = load_zoneamento_data(base_dir)
        for bundle in state.bundles.values():
            bundle.reference_categories = _build_reference_categories(bundle, state.reference)
    except Exception as exc:
        LOGGER.exception("Falha na inicializacao da aplicacao")
        state.error = str(exc)
    return state


APP_STATE = initialize_app_state(BASE_DIR)


def obter_lat_lon_centroide_por_numbloco(
    numbloco: Any,
    lat_atual: Any = None,
    lon_atual: Any = None,
) -> tuple[float, float] | None:
    polygon_x = None
    polygon_y = None
    lon_number = _to_float(lon_atual)
    lat_number = _to_float(lat_atual)
    if (
        APP_STATE.polygon_context.to_base_crs is not None
        and lon_number is not None
        and lat_number is not None
        and -180.0 <= lon_number <= 180.0
        and -90.0 <= lat_number <= 90.0
    ):
        try:
            polygon_x, polygon_y = APP_STATE.polygon_context.to_base_crs.transform(lon_number, lat_number)
        except Exception as exc:
            LOGGER.warning("Falha ao transformar coordenada atual para desempate por NUMBLOCO: %s", exc)

    row, _, _, _ = _select_polygon_row_by_numbloco(APP_STATE.polygon_context, numbloco, polygon_x, polygon_y)
    if row is None:
        return None

    geometry = row.get("geometry")
    if geometry is None or getattr(geometry, "is_empty", True):
        return None
    if hasattr(geometry, "is_valid") and not geometry.is_valid:
        return None

    source_crs = getattr(APP_STATE.polygon_context.gdf, "crs", None)
    if source_crs is None:
        return None

    try:
        centroid = geometry.centroid
        if centroid is None or getattr(centroid, "is_empty", True):
            return None
        lon_centroid, lat_centroid = Transformer.from_crs(source_crs, "EPSG:4326", always_xy=True).transform(
            float(centroid.x),
            float(centroid.y),
        )
    except Exception as exc:
        LOGGER.warning("Falha ao calcular centroide por NUMBLOCO: %s", exc)
        return None

    lat_result = _to_float(lat_centroid)
    lon_result = _to_float(lon_centroid)
    if (
        lat_result is None
        or lon_result is None
        or not np.isfinite(lat_result)
        or not np.isfinite(lon_result)
        or not (-90.0 <= lat_result <= 90.0 and -180.0 <= lon_result <= 180.0)
    ):
        return None
    return float(lat_result), float(lon_result)


def atualizar_lat_lon_por_numbloco(numbloco: str, lat_atual: Any, lon_atual: Any) -> tuple[Any, Any]:
    centroide = obter_lat_lon_centroide_por_numbloco(numbloco, lat_atual, lon_atual)
    if centroide is None:
        return gr.update(), gr.update()

    lat, lon = centroide
    return gr.update(value=lat), gr.update(value=lon)


def predict(
    area: float,
    numbloco: str,
    testada: float | None,
    lat: float,
    lon: float,
    bairro: str,
    ano_dado: str,
    iapond: float | None,
    app: float | None,
    lotpos: float | None,
    overrides_json: str,
) -> tuple[str, dict[str, Any], str | None]:
    if APP_STATE.error:
        message = f"Erro de inicializacao: {APP_STATE.error}"
        return message, {"status": "erro", "mensagem": APP_STATE.error}, None

    try:
        area_value = _to_float(area)
        bundle, routing_source = select_model("Auto (area)", APP_STATE.bundles, area_m2=area_value)

        payload = {
            "area": area,
            "numbloco": None if numbloco is None or str(numbloco).strip() == "" else numbloco,
            "testada": testada,
            "lat": lat,
            "lon": lon,
            "bairro": None if bairro is None or str(bairro).strip() == "" else bairro,
            "ano_dado": None if ano_dado in {None, ""} else ano_dado,
            "fonte": "0",
            "iapond": iapond,
            "app": app,
            "lotpos": lotpos,
        }
        overrides_ci = _parse_overrides_json(overrides_json)

        lon_number = _to_float(payload["lon"])
        lat_number = _to_float(payload["lat"])
        if lon_number is None or lat_number is None:
            raise ValueError("Latitude/longitude invalidas.")
        x, y = WGS84_TO_WEBMERC.transform(lon_number, lat_number)
        local_context = _infer_local_context(APP_STATE.reference, x, y, APP_STATE.k_neighbors)
        context_sources = {
            key: "inferido_coord"
            for key in local_context.keys()
            if not str(key).startswith("_")
        }
        if APP_STATE.polygon_context.to_base_crs is not None:
            polygon_x, polygon_y = APP_STATE.polygon_context.to_base_crs.transform(lon_number, lat_number)
            numbloco_lookup_value = payload.get("numbloco")
            if _is_missing(numbloco_lookup_value):
                numbloco_lookup_value = overrides_ci.get("numbloco")
            polygon_context, polygon_sources, polygon_alerts = _infer_polygon_context_by_numbloco(
                APP_STATE.polygon_context,
                numbloco_lookup_value,
                polygon_x,
                polygon_y,
            )
            if not polygon_context:
                spatial_context, spatial_sources, spatial_alerts = _infer_polygon_context(
                    APP_STATE.polygon_context,
                    polygon_x,
                    polygon_y,
                )
                polygon_context = spatial_context
                polygon_sources = spatial_sources
                polygon_alerts.extend(spatial_alerts)
        else:
            polygon_context, polygon_sources, polygon_alerts = {}, {}, []
        zone_context, zone_sources, zone_alerts = _infer_zoneamento_context(APP_STATE.zoneamento, x, y)
        rh_context, rh_sources, rh_alerts = _infer_rh_context_by_numbloco(
            APP_STATE.rh_numbloco,
            payload.get("numbloco") if not _is_missing(payload.get("numbloco")) else overrides_ci.get("numbloco"),
        )
        testada_context, testada_sources, testada_alerts = _infer_testada_context_by_numbloco(
            APP_STATE.testada_numbloco,
            payload.get("numbloco") if not _is_missing(payload.get("numbloco")) else overrides_ci.get("numbloco"),
        )
        merged_context = dict(local_context)
        merged_context.update(zone_context)
        merged_context.update(polygon_context)
        merged_context.update(rh_context)
        merged_context.update(testada_context)
        context_sources.update(zone_sources)
        context_sources.update(polygon_sources)
        context_sources.update(rh_sources)
        context_sources.update(testada_sources)

        prepared, source_map, alerts = apply_preprocessing(
            bundle=bundle,
            payload=payload,
            local_context=merged_context,
            overrides_ci=overrides_ci,
            context_sources=context_sources,
        )
        alerts.extend(polygon_alerts)
        alerts.extend(zone_alerts)
        alerts.extend(rh_alerts)
        alerts.extend(testada_alerts)
        x_input = build_features(bundle, prepared)
        prediction = predict_value(bundle, x_input, area_m2=float(prepared["AREA_bruta"]))
        markdown, details = format_output(
            bundle=bundle,
            prediction=prediction,
            prepared_values=prepared,
            source_map=source_map,
            alerts=alerts,
            local_context=merged_context,
            routing_source=routing_source,
        )
        train_predictions, test_predictions = split_prediction_frames(bundle.holdout_predictions_df)
        metrics = build_metric_summary(bundle.holdout_predictions_df, bundle.metadata)
        report_path = build_pdf_report(
            model_label=bundle.label.upper(),
            model_file=bundle.model_path.name,
            user_inputs={
                "Area territorial (m2)": area,
                "NUMBLOCO": payload.get("numbloco"),
                "Latitude": lat,
                "Longitude": lon,
                "Ano_Dado": payload["ano_dado"],
                "FON": "0",
                "LOTPOS": payload["lotpos"],
                "Testada manual": testada,
            },
            prepared_values={feature: prepared.get(feature) for feature in bundle.feature_names},
            source_map={feature: source_map.get(feature, "calculado/derivado") for feature in bundle.feature_names},
            prediction=prediction,
            metrics=metrics,
            train_predictions=train_predictions,
            test_predictions=test_predictions,
            limitations=_build_report_limitations(bundle, alerts, metrics.get("train")),
        )
        details["relatorio_pdf"] = report_path
        details["metricas_relatorio"] = _jsonable(metrics)
        return markdown, details, report_path
    except Exception as exc:
        LOGGER.exception("Falha na predicao")
        return (
            f"Erro na inferencia: {exc}",
            {
                "status": "erro",
                "mensagem": str(exc),
                "recomendacao": (
                    "Verifique area/lat/lon (testada e opcional) e, para casos de Gleba com falta "
                    "de dados cadastrais, informe overrides JSON."
                ),
            },
            None,
        )


def preencher_por_coordenadas(
    numbloco: str,
    lat: float,
    lon: float,
) -> tuple[Any, Any, Any, Any, Any, dict[str, Any]]:
    if APP_STATE.error:
        return gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), {
            "status": "erro",
            "mensagem": APP_STATE.error,
        }
    if APP_STATE.polygon_context.to_base_crs is None:
        return gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), {
            "status": "indisponivel",
            "mensagem": "Camada de poligonos enriquecidos indisponivel.",
        }

    lon_number = _to_float(lon)
    lat_number = _to_float(lat)
    if lon_number is None or lat_number is None:
        return gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), {
            "status": "erro",
            "mensagem": "Latitude/longitude invalidas.",
        }

    polygon_x, polygon_y = APP_STATE.polygon_context.to_base_crs.transform(lon_number, lat_number)
    values, sources, alerts = _infer_polygon_context_by_numbloco(
        APP_STATE.polygon_context,
        numbloco,
        polygon_x,
        polygon_y,
    )
    if not values:
        spatial_values, spatial_sources, spatial_alerts = _infer_polygon_context(APP_STATE.polygon_context, polygon_x, polygon_y)
        values = spatial_values
        sources = spatial_sources
        alerts.extend(spatial_alerts)
    rh_context, rh_sources, rh_alerts = _infer_rh_context_by_numbloco(APP_STATE.rh_numbloco, numbloco)
    testada_context, testada_sources, testada_alerts = _infer_testada_context_by_numbloco(APP_STATE.testada_numbloco, numbloco)
    values.update(rh_context)
    values.update(testada_context)
    sources.update(rh_sources)
    sources.update(testada_sources)
    alerts.extend(rh_alerts)
    alerts.extend(testada_alerts)
    area_value = _to_float(values.get("AREA"))
    testada_value = _to_float(values.get("TESTADA"))
    rh_value = _to_float(values.get("RH"))
    iapond_value = _to_float(values.get("IAPOND"))
    app_value = _to_float(values.get("APP"))
    lotpos_value = _to_float(values.get("LOTPOS"))
    payload = {
        "status": "ok" if values else "sem_resultado",
        "area_referencia_poligono": area_value,
        "testada": testada_value,
        "RH": rh_value,
        "IAPOND": iapond_value,
        "APP": app_value,
        "LOTPOS": lotpos_value,
        "NUMBLOCO_informado": None if _is_missing(numbloco) else _safe_text(numbloco, ""),
        "NUMBLOCO_poligono": values.get("_polygon_numbloco"),
        "distancia_poligono_m": _to_float(values.get("_polygon_distance_m")),
        "modo_busca_poligono": values.get("_polygon_lookup_mode"),
        "fontes": sources,
        "alerts": alerts,
    }
    return (
        gr.update(value=testada_value) if testada_value is not None else gr.update(),
        gr.update(value=rh_value) if rh_value is not None else gr.update(),
        gr.update(value=iapond_value) if iapond_value is not None else gr.update(),
        gr.update(value=app_value) if app_value is not None else gr.update(),
        gr.update(value=lotpos_value) if lotpos_value is not None else gr.update(),
        _jsonable(payload),
    )


def _loaded_models_text(state: AppState) -> str:
    if state.error:
        return f"**Falha ao iniciar app:** `{state.error}`"
    lines = ["**Modelos carregados:**"]
    for key in sorted(state.bundles.keys()):
        bundle = state.bundles[key]
        lines.append(
            f"- `{bundle.label}` -> `{bundle.notebook_name}` / `{bundle.model_path.name}` "
            f"em `{bundle.artifact_dir.name}` "
            f"({len(bundle.feature_names)} features)"
        )
    if state.reference.path:
        lines.append(f"**Base de referencia:** `{state.reference.path.name}`")
    else:
        lines.append("**Base de referencia:** nao encontrada (fallback por defaults)")
    if state.rh_numbloco.has_data and state.rh_numbloco.path is not None:
        lines.append(
            f"**RH por NUMBLOCO:** `{state.rh_numbloco.path.name}` "
            f"({len(state.rh_numbloco.rh_by_numbloco)} chaves; "
            f"duplicadas: {state.rh_numbloco.duplicate_key_count}; "
            f"conflitos: {state.rh_numbloco.conflicting_key_count})"
        )
    elif state.rh_numbloco.error:
        lines.append(f"**RH por NUMBLOCO:** indisponivel ({state.rh_numbloco.error})")
    if state.testada_numbloco.has_data and state.testada_numbloco.path is not None:
        lines.append(
            f"**TESTADA por NUMBLOCO:** `{state.testada_numbloco.path.name}` "
            f"({len(state.testada_numbloco.testada_by_numbloco)} chaves; "
            f"duplicadas: {state.testada_numbloco.duplicate_key_count}; "
            f"conflitos: {state.testada_numbloco.conflicting_key_count})"
        )
    elif state.testada_numbloco.error:
        lines.append(f"**TESTADA por NUMBLOCO:** indisponivel ({state.testada_numbloco.error})")
    if state.polygon_context.has_data and state.polygon_context.path is not None:
        lines.append(
            f"**Poligonos enriquecidos:** `{state.polygon_context.path.name}` "
            f"(campos: {', '.join(state.polygon_context.available_fields)}; "
            f"NUMBLOCO: {len(state.polygon_context.numbloco_key_index)} chaves)"
        )
    elif state.polygon_context.error:
        lines.append(f"**Poligonos enriquecidos:** indisponivel ({state.polygon_context.error})")
    else:
        lines.append("**Poligonos enriquecidos:** nao configurado")
    if state.zoneamento.has_data and state.zoneamento.shp_path is not None:
        lines.append(
            f"**Zoneamento:** `{state.zoneamento.shp_path.name}` "
            f"(mapa IA: {len(state.zoneamento.iapond_map)} chaves, fonte: `{state.zoneamento.iapond_map_source}`)"
        )
    elif state.zoneamento.error:
        lines.append(f"**Zoneamento:** indisponivel ({state.zoneamento.error})")
    else:
        lines.append("**Zoneamento:** nao configurado (fallback por contexto local)")
    return "\n".join(lines)


APP_TITLE = "AVM Territorial 2026 - Inferencia XGBoost (Terreno/Gleba)"

with gr.Blocks(title=APP_TITLE) as demo:
    gr.Markdown(f"# {APP_TITLE}")
    gr.Markdown(
        "Informe area, NUMBLOCO quando disponivel, e coordenadas do imovel. O app seleciona automaticamente "
        "`TERRENO` para area menor que `3000 m2` e `GLEBA` para area maior ou igual a `3000 m2`."
    )
    gr.Markdown(_loaded_models_text(APP_STATE))

    with gr.Row():
        area_input = gr.Number(label="Area territorial (m2)", value=405.0, precision=6)
        numbloco_input = gr.Textbox(
            label="NUMBLOCO",
            value="",
            placeholder="Opcional; usado antes da busca espacial",
        )

    with gr.Row():
        lat_input = gr.Number(label="Latitude", value=-30.0315, precision=8)
        lon_input = gr.Number(label="Longitude", value=-51.1645, precision=8)

    autofill_button = gr.Button("Buscar atributos por NUMBLOCO/coordenada")
    autofill_json = gr.JSON(label="Diagnostico do autopreenchimento espacial")

    with gr.Accordion("Opcional - ajustes e overrides de variaveis inferidas", open=False):
        testada_input = gr.Textbox(
            label="Testada (m)",
            value="",
            placeholder="Opcional",
        )
        bairro_input = gr.Textbox(label="Bairro (opcional)", placeholder="Ex.: PETROPOLIS")
        ano_input = gr.Dropdown(
            label="Ano_Dado",
            choices=[str(year) for year in range(2016, 2031)],
            value="2026",
        )
        with gr.Row():
            rh_input = gr.Textbox(label="RH", value="", placeholder="Preenchido por NUMBLOCO", interactive=False)
            iapond_input = gr.Textbox(label="IAPOND", value="", placeholder="Opcional")
            app_input = gr.Textbox(label="APP", value="", placeholder="Opcional")
            lotpos_input = gr.Number(label="LOTPOS", value=0.0, precision=6)
        overrides_input = gr.Code(
            label="Overrides JSON (opcional, prioridade maxima)",
            language="json",
            value='{\n  "RH": null,\n  "APP": null,\n  "CP": null,\n  "LOTPOS": null,\n  "FINALIDADE": null\n}',
            lines=9,
        )

    predict_button = gr.Button("Calcular estimativa", variant="primary")
    result_markdown = gr.Markdown()
    details_json = gr.JSON(label="Resumo tecnico da inferencia")
    report_file = gr.File(label="Relatorio PDF")

    predict_button.click(
        fn=predict,
        inputs=[
            area_input,
            numbloco_input,
            testada_input,
            lat_input,
            lon_input,
            bairro_input,
            ano_input,
            iapond_input,
            app_input,
            lotpos_input,
            overrides_input,
        ],
        outputs=[result_markdown, details_json, report_file],
    )
    numbloco_input.change(
        fn=atualizar_lat_lon_por_numbloco,
        inputs=[numbloco_input, lat_input, lon_input],
        outputs=[lat_input, lon_input],
    )
    autofill_button.click(
        fn=preencher_por_coordenadas,
        inputs=[numbloco_input, lat_input, lon_input],
        outputs=[
            testada_input,
            rh_input,
            iapond_input,
            app_input,
            lotpos_input,
            autofill_json,
        ],
    )


if __name__ == "__main__":
    port = int(os.getenv("PORT", "7860"))
    demo.launch(server_name="0.0.0.0", server_port=port)