SICAR Data Processor: Processa dados de imóveis rurais do SICAR por município e consolida em nível estadual.

sicar_data_processor.py

# ==========================================================
# AUTENTICAÇÃO
# ==========================================================
from google.colab import auth
auth.authenticate_user()

# ==========================================================
# INSTALAÇÃO
# ==========================================================
!pip install basedosdados geopandas shapely pyogrio --upgrade --quiet

# ==========================================================
# IMPORTS
# ==========================================================
import os
import time
import pandas as pd
import basedosdados as bd
import geopandas as gpd
from shapely import wkt
from shapely.geometry import Polygon

# ==========================================================
# PARÂMETROS
# ==========================================================
UF = "SP"
BILLING_PROJECT_ID = "prefab-mapper-256812"

BASE_PATH = "/content/drive/MyDrive/SICAR"
MUN_PATH = f"{BASE_PATH}/{UF}/municipios"
ESTADO_PATH = f"{BASE_PATH}/{UF}"

os.makedirs(MUN_PATH, exist_ok=True)
os.makedirs(ESTADO_PATH, exist_ok=True)

# ==========================================================
# COLUNAS REAIS DO BIGQUERY (VALIDADAS)
# ==========================================================
BQ_COLUMNS = [
    "id_imovel",
    "id_municipio",
    "sigla_uf",
    "modulos_fiscais",
    "area",
    "status",
    "tipo",
    "condicao",
    "geometria"
]

# ==========================================================
# PADRONIZAÇÃO FINAL (SAÍDA)
# ==========================================================
COLUMN_RENAME = {
    "id_imovel": "cod_imovel",
    "modulos_fiscais": "num_modulo",
    "area": "num_area",
    "status": "situacao",
    "tipo": "tipo_imove",
    "condicao": "condicao_i",
    "geometry": "geom"
}

# ==========================================================
# FUNÇÕES AUXILIARES
# ==========================================================
def clean_geometry(geom):
    if geom is None or geom.is_empty or not geom.is_valid:
        return None
    return geom.buffer(0)

def ensure_polygon(geom):
    if geom is None:
        return None
    if geom.geom_type == "Polygon":
        return geom
    return None

def extract_cod_ibge_m(id_municipio):
    return str(id_municipio)

def extract_cod_ibge_e(cod_ibge_m):
    return str(cod_ibge_m)[:2]

# ==========================================================
# LISTA DE MUNICÍPIOS (DISTINCT NO BIGQUERY)
# ==========================================================
print(f"Buscando lista de municípios para {UF} no BigQuery...")
query_municipios = f"""
    SELECT DISTINCT id_municipio 
    FROM basedosdados.br_sfb_sicar.area_imovel 
    WHERE sigla_uf = '{UF}'
"""
df_municipios = bd.read_sql(
    query=query_municipios, 
    billing_project_id=BILLING_PROJECT_ID
)

lista_municipios = df_municipios["id_municipio"].tolist()
total = len(lista_municipios)

print(f"Total de municípios encontrados em {UF}: {total}")

# ==========================================================
# DOWNLOAD MUNICÍPIO A MUNICÍPIO (BIGQUERY OTIMIZADO)
# ==========================================================
baixados = 0

for i, id_municipio in enumerate(lista_municipios, start=1):
    out_file = f"{MUN_PATH}/{id_municipio}.gpkg"

    if os.path.exists(out_file):
        print(f"[{i}/{total}] Município {id_municipio} já existe → pulando")
        baixados += 1
        continue

    print(f"[{i}/{total}] Baixando município {id_municipio}")

    query = f"""
    SELECT
        {", ".join(BQ_COLUMNS)}
    FROM basedosdados.br_sfb_sicar.area_imovel
    WHERE sigla_uf = '{UF}'
      AND id_municipio = '{id_municipio}'
      AND data_extracao >= DATE '2025-01-01'
    """

    try:
        df = bd.read_sql(
            query=query,
            billing_project_id=BILLING_PROJECT_ID,
            reauth=False # Alterado para False para evitar prompts repetidos
        )

        if df.empty:
            print("    ⚠️ Nenhum registro encontrado")
            continue

        # Converte geometria
        df["geometry"] = df["geometria"].apply(wkt.loads)

        gdf = gpd.GeoDataFrame(df, geometry="geometry", crs="EPSG:4674")

        # Limpeza geométrica
        gdf["geometry"] = gdf["geometry"].apply(clean_geometry)
        gdf["geometry"] = gdf["geometry"].apply(ensure_polygon)
        gdf = gdf[gdf["geometry"].notnull()]

        # Remove duplicados
        gdf = gdf.drop_duplicates(subset=["id_imovel"])

        # Salva município
        gdf.to_file(out_file, driver="GPKG")
        baixados += 1

    except Exception as e:
        print(f"    ❌ Erro: {e}")
        time.sleep(5)

# ==========================================================
# JUNÇÃO ESTADUAL (LOCAL, SEM BIGQUERY)
# ==========================================================
arquivos = [
    os.path.join(MUN_PATH, f)
    for f in os.listdir(MUN_PATH)
    if f.endswith(".gpkg")
]

print(f"\nArquivos municipais válidos: {len(arquivos)}")

if len(arquivos) > 0:
    lista_gdfs = [gpd.read_file(f) for f in arquivos]
    car = gpd.GeoDataFrame(
        pd.concat(lista_gdfs, ignore_index=True),
        crs="EPSG:4674"
    )

    # ==========================================================
    # PADRONIZAÇÃO FINAL
    # ==========================================================
    car = car.rename(columns=COLUMN_RENAME)

    car["cod_ibge_m"] = car["id_municipio"].apply(extract_cod_ibge_m)
    car["cod_ibge_e"] = car["cod_ibge_m"].apply(extract_cod_ibge_e)

    car = car[
        list(COLUMN_RENAME.values()) +
        ["cod_ibge_m", "cod_ibge_e"]
    ]

    # ==========================================================
    # SALVA GEOPACKAGE ESTADUAL
    # ==========================================================
    estado_file = f"{ESTADO_PATH}/sicar_{UF}.gpkg"
    car.to_file(estado_file, driver="GPKG")

    print("\n✅ PROCESSO FINALIZADO COM SUCESSO")
    print(f"Municípios baixados: {baixados}/{total}")
    print(f"Arquivo final: {estado_file}")
else:
    print("\n❌ Nenhum arquivo municipal foi gerado para realizar a junção.")

sicar_state.py

# ==========================================================
# AUTENTICAÇÃO
# ==========================================================
from google.colab import auth
auth.authenticate_user()

# ==========================================================
# INSTALAÇÃO
# ==========================================================
!pip install basedosdados geopandas shapely pyogrio --upgrade --quiet

# ==========================================================
# IMPORTS
# ==========================================================
import os
import pandas as pd
import basedosdados as bd
import geopandas as gpd
from shapely import wkt

# ==========================================================
# PARÂMETROS
# ==========================================================
UF = "SP"
BILLING_PROJECT_ID = "sig-iluminacao-1506686571559"

BASE_PATH = "/content/drive/MyDrive/SICAR"
ESTADO_PATH = f"{BASE_PATH}/{UF}"
os.makedirs(ESTADO_PATH, exist_ok=True)

OUTPUT_FILE = f"{ESTADO_PATH}/sicar_{UF}.gpkg"

# ==========================================================
# COLUNAS REAIS DO BIGQUERY (VALIDADAS)
# ==========================================================
BQ_COLUMNS = [
    "id_imovel",
    "id_municipio",
    "sigla_uf",
    "modulos_fiscais",
    "area",
    "status",
    "tipo",
    "condicao",
    "geometria"
]

# ==========================================================
# PADRONIZAÇÃO FINAL (SAÍDA)
# ==========================================================
COLUMN_RENAME = {
    "id_imovel": "cod_imovel",
    "modulos_fiscais": "num_modulo",
    "area": "num_area",
    "status": "situacao",
    "tipo": "tipo_imove",
    "condicao": "condicao_i",
    "geometry": "geom"
}

# ==========================================================
# FUNÇÕES AUXILIARES
# ==========================================================
def clean_geometry(geom):
    if geom is None or geom.is_empty or not geom.is_valid:
        return None
    return geom.buffer(0)

def ensure_polygon(geom):
    if geom is None:
        return None
    if geom.geom_type == "Polygon":
        return geom
    return None

def extract_cod_ibge_m(id_municipio):
    return str(int(id_municipio))

def extract_cod_ibge_e(cod_ibge_m):
    return cod_ibge_m[:2]

# ==========================================================
# QUERY ÚNICA POR ESTADO (OTIMIZADA)
# ==========================================================
print(f"⬇️ Baixando SICAR da UF {UF}")

query = f"""
    SELECT
        {", ".join(BQ_COLUMNS)}
    FROM basedosdados.br_sfb_sicar.area_imovel
    WHERE sigla_uf = '{UF}'
      AND data_extracao >= DATE '2025-01-01'
"""

df = bd.read_sql(
    query=query,
    billing_project_id=BILLING_PROJECT_ID,
    reauth=False
)

if df.empty:
    raise RuntimeError("❌ Nenhum dado retornado para esta UF")

print(f"📦 Registros baixados: {len(df):,}")

# ==========================================================
# CONVERSÃO PARA GEODATAFRAME
# ==========================================================
df["geometry"] = df["geometria"].apply(wkt.loads)

gdf = gpd.GeoDataFrame(
    df,
    geometry="geometry",
    crs="EPSG:4674"
)

# ==========================================================
# LIMPEZA GEOMÉTRICA
# ==========================================================
gdf["geometry"] = gdf["geometry"].apply(clean_geometry)
gdf["geometry"] = gdf["geometry"].apply(ensure_polygon)
gdf = gdf[gdf["geometry"].notnull()]

print(f"🧹 Registros após limpeza geométrica: {len(gdf):,}")

# ==========================================================
# REMOVE DUPLICADOS
# ==========================================================
gdf = gdf.drop_duplicates(subset=["id_imovel"])

# ==========================================================
# PADRONIZAÇÃO FINAL
# ==========================================================
gdf = gdf.rename(columns=COLUMN_RENAME)

gdf["cod_ibge_m"] = gdf["id_municipio"].apply(extract_cod_ibge_m)
gdf["cod_ibge_e"] = gdf["cod_ibge_m"].apply(extract_cod_ibge_e)

gdf = gdf[
    list(COLUMN_RENAME.values()) +
    ["cod_ibge_m", "cod_ibge_e"]
]

# ==========================================================
# SALVA GEOPACKAGE ESTADUAL
# ==========================================================
gdf.to_file(OUTPUT_FILE, driver="GPKG")

print("\n✅ PROCESSO FINALIZADO COM SUCESSO")
print(f"📍 UF: {UF}")
print(f"📦 Total de imóveis: {len(gdf):,}")
print(f"💾 Arquivo gerado: {OUTPUT_FILE}")