#!/usr/bin/env python3
"""Social: coleta e tratamento dos indicadores pré-registrados (tema `social`).

Uso:
    python3 scripts/coletar_social.py              # coleta e trata
    python3 scripts/coletar_social.py --so-tratar  # só trata os brutos mais recentes
    python3 scripts/coletar_social.py --rebaixar   # baixa de novo as publicações de versão fechada
    python3 scripts/resumir.py --tema social       # depois: resumo por mandato

Grava:
    dados/brutos/<fonte>/...                 arquivo original, com linha no manifesto da pasta
    dados/tratados/social.csv                indicador,data,valor,unidade,fonte
    dados/tratados/social_pares_paises.csv   valor de cada país na comparação com pares
    seção "Extração" das fichas em fontes/fichas/social/

O tema é coletado e tratado só por este script. Só biblioteca padrão (D-004). Fontes,
decisões e lacunas nas fichas e em dados/tratados/social_notas.md.

Datas: AAAA (anual) e AAAA-MM (mensal). O mês corrente e o ano corrente, incompletos, ficam
fora das séries mensais e anuais.

Valores reais (D-015): reais de 2025. Dado anual: valor do ano × (IPCA médio de 2025 / IPCA
médio do ano). Dado mensal: valor do mês × (IPCA médio de 2025 / IPCA do mês). O índice é a
série `ipca_numero_indice` de dados/tratados/economia.csv (SIDRA 1737).

% do PIB: a mesma série do tema orçamento (D-027), soma dos quatro trimestres da tabela 1846
do SIDRA (Contas Nacionais Trimestrais, valores correntes).

Programa de transferência de renda (D-030): Bolsa Família até out/2021, Auxílio Brasil de
nov/2021 a fev/2023 e Bolsa Família desde mar/2023 formam uma série só, com o nome do programa
na coluna `fonte` de cada mês ou ano.

Dados pessoais: nenhum arquivo baixado traz nome, CPF, NIS ou endereço de pessoa. O MDS é
consultado por totais municipais (MI Social); o Minha Casa, Minha Vida, pelo arquivo sintético
por município (o analítico, com data de nascimento e sexo de cada mutuário, não é baixado).

Publicações de versão fechada (planilhas e relatórios da FJP, nota técnica do MDS sobre o BPC,
boletim do Ipea, edição de dezembro de 2022 da série histórica do RTN) não são baixadas de novo
se já houver bruto; --rebaixar força o download. As demais fontes são consultadas a cada dia
de coleta.

Pontos técnicos, testados em 6/10/2026:

- MDS: a API pública MI Social (Solr, núcleo matriz_informacoes2) tem um registro por município
  e mês (5.571 por mês, jan/2003 a dez/2026). Os totais nacionais vêm da soma dos municípios
  calculada pelo próprio servidor (parâmetro json.facet, um balde por mês). Os campos de valor
  com sufixo _f são float de 32 bits; a soma pode diferir em dezenas de reais da soma exata dos
  campos texto (_s), conferida em meses de amostra baixados por município.
- SIOP: a consulta SPARQL lista, por ano, as ações orçamentárias dos programas (código, título,
  função, subfunção, unidade, resultado primário) com empenhado, liquidado e pago. O título de
  cada ação é conferido com um padrão, para que um código reaproveitado não entre por engano.
- Banco Mundial: a API recusa rajadas (firewall); há pausa entre consultas, e a resposta pode
  vir com BOM (lida como utf-8-sig).
- Ministério das Cidades: os nomes dos arquivos do Minha Casa, Minha Vida mudam a cada
  atualização; o script lê os links na página de bases de dados. O HEAD no gov.br responde 403;
  o GET funciona.
- FJP: o repositório institucional (repositorio.fjp.mg.gov.br) respondeu 403 a qualquer acesso
  automatizado em 6/10/2026. As planilhas de 2016 em diante estão no site da FJP e no Google
  Drive da FJP (links na página do déficit habitacional); a série temporal 2007-2015 vem da
  cópia guardada pelo Wayback Machine do arquivo que a FJP publicava em fjp.mg.gov.br.
"""

import argparse
import csv
import datetime as dt
import gzip
import io
import json
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
import zipfile
from collections import defaultdict
from decimal import Decimal

from comum import (BRUTOS, HOJE, METODOLOGIA, RAIZ, TRATADOS, USER_AGENT, escrever_csv, ler_csv, ler_fichas,
                   ler_manifesto, salvar_bruto)
from tratar import mediana_por_ano

TEMA = "social"
ANO_CORRENTE = HOJE.year
MES_CORRENTE = f"{HOJE.year:04d}-{HOJE.month:02d}"
ANO_BASE_REAL = "2025"  # D-015
U_NOM = "R$ bilhões correntes"
U_REAL = f"R$ bilhões de {ANO_BASE_REAL} (IPCA, média do ano)"
U_PIB = "% do PIB"


# ================================================================ rede e brutos

def pedir(url, cabecalhos=None, tentativas=5, timeout=300):
    """GET com repetição em erro temporário. Devolve os bytes (copiada de coletar_educacao.pedir)."""
    erro = None
    for i in range(tentativas):
        req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, **(cabecalhos or {})})
        try:
            with urllib.request.urlopen(req, timeout=timeout) as r:
                return r.read()
        except urllib.error.HTTPError as e:
            erro = e
            if e.code not in (429, 500, 502, 503, 504):
                raise
        except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
            erro = e
        espera = min(10 * (i + 1), 60)
        print(f"    {erro}; nova tentativa em {espera} s", file=sys.stderr)
        time.sleep(espera)
    raise RuntimeError(f"falha ao baixar {url}: {erro}")


def brutos_exatos(fonte, nome):
    """Arquivos `<nome>_<AAAA-MM-DD>[-n].<ext>` da pasta, do mais antigo ao mais recente (prefixo exato;
    copiada de coletar_meio_ambiente)."""
    pasta = BRUTOS / fonte
    achados = []
    for p in pasta.glob(f"{nome}_*"):
        prefixo, carimbo = p.stem.rsplit("_", 1)
        if prefixo != nome or not re.fullmatch(r"\d{4}-\d{2}-\d{2}(-\d+)?", carimbo):
            continue
        achados.append(((carimbo[:10], int(carimbo[11:] or 1)), p))
    return [p for _, p in sorted(achados)]


def bruto(fonte, nome):
    arquivos = brutos_exatos(fonte, nome)
    if not arquivos:
        raise FileNotFoundError(f"sem bruto {fonte}/{nome}; rode sem --so-tratar")
    return arquivos[-1]


def obter(fonte, nome, ext, url, baixar_fn=None, reaproveitar=False, validar=None, consulta=None,
          mesma_url=False):
    """Baixa e grava um bruto com comum.salvar_bruto. Reaproveita o bruto de hoje; com `reaproveitar`,
    qualquer bruto já guardado (publicações de versão fechada); com `mesma_url`, o bruto mais recente
    se ele veio do mesmo endereço (arquivos cujo nome muda a cada atualização da fonte)."""
    existentes = brutos_exatos(fonte, nome)
    if existentes:
        ultimo = existentes[-1]
        mesma = mesma_url and ler_manifesto(fonte).get(ultimo.name, {}).get("consulta_url") == (consulta or url)
        if reaproveitar or mesma or HOJE.isoformat() in ultimo.name:
            print(f"  {fonte}/{ultimo.name}: reaproveitado")
            return ultimo
    conteudo = baixar_fn() if baixar_fn else pedir(url)
    if validar:
        validar(conteudo)
    caminho = salvar_bruto(fonte, nome, conteudo, consulta or url, ext)
    print(f"  {fonte}/{caminho.name}: {len(conteudo):,} bytes")
    return caminho


def validar_json(conteudo):
    json.loads(conteudo)


def validar_zip(conteudo):
    if conteudo[:2] != b"PK":
        raise RuntimeError("resposta não é um arquivo zip/xlsx")


def validar_pdf(conteudo):
    if conteudo[:5] != b"%PDF-":
        raise RuntimeError("resposta não é um PDF")


# ================================================================ saída e fichas

class Saida:
    def __init__(self):
        self.linhas = []
        self.paises = []
        self.extracao = defaultdict(list)  # id da ficha -> [(série, arquivos, dados)]
        self.notas = defaultdict(list)     # id da ficha -> parágrafos depois da tabela de extração
        self.conferencias = []             # resumo impresso no fim

    def add(self, ficha_id, serie_id, dados, unidade, fonte, arquivos=(), casas=4, fonte_por_data=None):
        """`fonte_por_data` ({data: texto}) troca a coluna fonte de períodos específicos (o nome do
        programa em cada mês, a pesquisa de cada ano)."""
        if not serie_id.startswith(ficha_id + "__"):
            raise ValueError(f"série {serie_id} fora da convenção <ficha>__<recorte>")
        for data, valor in sorted(dados.items()):
            v = round(valor, casas)
            self.linhas.append([serie_id, data, int(v) if casas == 0 else v, unidade,
                                (fonte_por_data or {}).get(data, fonte)])
        self.extracao[ficha_id].append((serie_id, sorted(set(arquivos)), dados))

    def conferiu(self, texto):
        print(f"  conferência: {texto}")
        self.conferencias.append(texto)


def atualizar_extracao(ficha, itens, notas=()):
    """Reescreve o bloco entre os marcadores de extração da ficha (tabela de tratar.atualizar_extracao;
    com mais de três arquivos, a célula diz quantos são e a lista vem depois da tabela, como em
    coletar_orcamento)."""
    manifestos = {}
    todos = []
    linhas = ["| Série | Arquivo bruto | Extraído em | Observações | Primeira | Última |",
              "|---|---|---|---|---|---|"]
    longa = False
    for serie_id, arquivos, dados in itens:
        datas = []
        for arq in arquivos:
            fonte = arq.parent.name
            manifestos.setdefault(fonte, ler_manifesto(fonte))
            datas.append(manifestos[fonte].get(arq.name, {}).get("data_extracao", "?")[:16].replace("T", " "))
            if arq not in todos:
                todos.append(arq)
        datas_txt = ", ".join(sorted(set(datas))) or "(derivada)"
        if len(set(datas)) > 1:
            datas_txt = f"{min(datas)} a {max(datas)}"
        if len(arquivos) > 3:
            longa = True
            nomes_txt = f"{len(arquivos)} arquivos (lista abaixo)"
        else:
            nomes_txt = "<br>".join(f"`{a.relative_to(RAIZ)}`" for a in arquivos) or "(derivada)"
        if dados:
            linhas.append(f"| `{serie_id}` | {nomes_txt} | {datas_txt} | {len(dados)} | {min(dados)} | {max(dados)} |")
        else:
            linhas.append(f"| `{serie_id}` | {nomes_txt} | {datas_txt} | 0 | | |")
    bloco = "\n".join(linhas)
    if longa:
        bloco += "\n\nArquivos brutos desta ficha: " + ", ".join(f"`{a.relative_to(RAIZ)}`" for a in todos) + "."
    if notas:
        bloco += "\n\n" + "\n\n".join(notas)
    caminho = ficha["_arquivo"]
    texto = caminho.read_text(encoding="utf-8")
    novo = re.sub(r"(<!-- extracao:inicio -->\n).*?(\n<!-- extracao:fim -->)",
                  lambda m: m.group(1) + bloco + m.group(2), texto, flags=re.S)
    caminho.write_text(novo, encoding="utf-8")


def br(x, casas=0):
    """Número no formato brasileiro (1.234,5)."""
    return f"{x:,.{casas}f}".replace(",", "X").replace(".", ",").replace("X", ".")


def num(x):
    """Texto ou número de planilha -> float; '-', '..', vazio -> None (copiada de coletar_educacao)."""
    if x is None:
        return None
    if isinstance(x, (int, float)):
        return float(x)
    s = str(x).strip().replace("\xa0", "")
    if s in ("", "-", "--", "...", "..", "x", "X", "*"):
        return None
    try:
        return float(s)
    except ValueError:
        return None


def limpo(x):
    return re.sub(r"\s+", " ", str(x or "")).strip()


def mes_seguinte(aaaa_mm):
    a, m = int(aaaa_mm[:4]), int(aaaa_mm[5:])
    return f"{a + (m == 12):04d}-{m % 12 + 1:02d}"


def meses_entre(ini, fim):
    """Lista de 'AAAA-MM' de ini a fim, inclusive."""
    saida, m = [], ini
    while m <= fim:
        saida.append(m)
        m = mes_seguinte(m)
    return saida


def anos_fechados(anual):
    return {a: v for a, v in anual.items() if a < str(ANO_CORRENTE)}


# ---------------------------------------------------------------- planilhas .xlsx

M_NS = "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}"
R_NS = "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}"
P_NS = "{http://schemas.openxmlformats.org/package/2006/relationships}"


def _coluna(ref):
    n = 0
    for ch in ref:
        if not ch.isalpha():
            break
        n = n * 26 + (ord(ch.upper()) - 64)
    return n - 1


class Xlsx:
    """Leitor mínimo de .xlsx (zipfile + xml.etree): sharedStrings e planilhas (copiado de coletar_educacao)."""

    def __init__(self, fonte):
        self.z = zipfile.ZipFile(io.BytesIO(fonte) if isinstance(fonte, bytes) else fonte)
        wb = ET.fromstring(self.z.read("xl/workbook.xml"))
        rels = ET.fromstring(self.z.read("xl/_rels/workbook.xml.rels"))
        alvo = {r.get("Id"): r.get("Target") for r in rels.iter(P_NS + "Relationship")}
        self.planilhas = {}
        for sh in wb.iter(M_NS + "sheet"):
            t = alvo[sh.get(R_NS + "id")]
            self.planilhas[sh.get("name")] = t.lstrip("/") if t.startswith("/") else "xl/" + t
        self._textos = None

    def textos(self):
        if self._textos is None:
            self._textos = []
            if "xl/sharedStrings.xml" in self.z.namelist():
                with self.z.open("xl/sharedStrings.xml") as f:
                    for _, el in ET.iterparse(f):
                        if el.tag == M_NS + "si":
                            self._textos.append("".join(t.text or "" for t in el.iter(M_NS + "t")))
                            el.clear()
        return self._textos

    def linhas(self, nome):
        """Lista de linhas (listas de valores; índice 0 = linha 1), como texto ou None."""
        textos = self.textos()
        out = []
        with self.z.open(self.planilhas[nome]) as f:
            for _, el in ET.iterparse(f):
                if el.tag != M_NS + "row":
                    continue
                r = int(el.get("r")) if el.get("r") else len(out) + 1
                vals, col = {}, -1
                for c in el.iter(M_NS + "c"):
                    t, v = c.get("t"), c.find(M_NS + "v")
                    if t == "s" and v is not None:
                        val = textos[int(v.text)]
                    elif t == "inlineStr":
                        val = "".join(x.text or "" for x in c.iter(M_NS + "t"))
                    else:
                        val = v.text if v is not None else None
                    col = _coluna(c.get("r")) if c.get("r") else col + 1
                    vals[col] = val
                linha = [None] * (max(vals) + 1) if vals else []
                for k, x in vals.items():
                    linha[k] = x
                while len(out) < r - 1:
                    out.append([])
                out.append(linha)
                el.clear()
        return out


# ---------------------------------------------------------------- IPCA (D-015)

def indice_ipca():
    """{AAAA-MM: número-índice do IPCA}, {AAAA: média anual (só anos com 12 meses)} e a média de 2025,
    da série ipca_numero_indice de dados/tratados/economia.csv (SIDRA 1737)."""
    mensal = {l["data"]: float(l["valor"]) for l in ler_csv(TRATADOS / "economia.csv")
              if l["indicador"] == "ipca_numero_indice"}
    por_ano = defaultdict(list)
    for m, v in mensal.items():
        por_ano[m[:4]].append(v)
    anual = {a: sum(v) / 12 for a, v in por_ano.items() if len(v) == 12}
    if ANO_BASE_REAL not in anual:
        raise SystemExit(f"economia.csv não tem os 12 meses de {ANO_BASE_REAL} do IPCA; rode o tema economia")
    return mensal, anual, anual[ANO_BASE_REAL]


def real(nominal, ipca_anual):
    """{ano: R$} -> {ano: R$ de 2025}, pela média anual do IPCA (D-015)."""
    return {a: v * ipca_anual[ANO_BASE_REAL] / ipca_anual[a] for a, v in nominal.items() if a in ipca_anual}


# ================================================================ MDS: MI Social (Bolsa Família, Auxílio Brasil, BPC, CadÚnico, AE)

FONTE_MDS = "mds_misocial"
API_MISOCIAL = "https://aplicacoes.mds.gov.br/sagi/servicos/misocial"
PAGINA_MISOCIAL = "https://aplicacoes.mds.gov.br/sagi/vis/data3/"  # VIS DATA 3, que lê a mesma base
# grupo de campos -> lista de (chave no balde, função de agregação do Solr)
FACETAS_MISOCIAL = {
    "transferencia_renda_mensal": [
        ("bf_familias", "sum(qtd_familias_beneficiarias_bolsa_familia_i)"),
        ("bf_valor", "sum(valor_repassado_bolsa_familia_f)"),
        ("bf_valor_medio_x_familias", "sum(product(pbf_vlr_medio_benef_f,qtd_familias_beneficiarias_bolsa_familia_i))"),
        ("ab_familias", "sum(pab_qtd_fam_benef_i)"),
        ("ab_valor", "sum(pab_valor_pago_d)"),
        ("ab_extraordinario_familias", "sum(pab_extraordinario_qtd_fam_i)"),
        ("ab_extraordinario_valor", "sum(pab_extraordinario_valor_pago_d)"),
        ("bf_ae_familias_todas", "sum(qtd_total_fam_benef_pbf_ae_aer_todos_i)"),
        ("bf_ae_valor_todas", "sum(valor_total_fam_benef_pbf_ae_aer_todos_f)"),
        ("bf_sem_ae_familias", "sum(qtd_familias_benef_do_pbf_sem_ae_aer_i)"),
        ("bf_so_ae_familias", "sum(qtd_fam_benef_do_pbf_com_recebimento_apenas_de_ae_i)"),
        ("bf_so_ae_valor", "sum(valor_repassado_fam_benef_do_pbf_com_recebimento_apenas_ae_f)"),
    ],
    "bpc_mensal": [
        ("bpc_idoso", "sum(bpc_idoso_ben_i)"),
        ("bpc_pcd", "sum(bpc_pcd_ben_i)"),
        ("bpc_total", "sum(bpc_ben_i)"),
        ("bpc_idoso_valor", "sum(bpc_idoso_val_f)"),
        ("bpc_pcd_valor", "sum(bpc_pcd_val_f)"),
    ],
    "cadunico_mensal": [
        ("familias", "sum(cadun_qtd_familias_cadastradas_i)"),
        ("pessoas", "sum(cadun_qtd_pessoas_cadastradas_i)"),
        ("familias_ate_meio_sm", "sum(cadun_qtd_familias_cadastradas_rfpc_ate_meio_sm_i)"),
        ("familias_pobreza_pbf", "sum(cadun_qtd_familias_cadastradas_pobreza_pbf_i)"),
    ],
    "auxilio_emergencial_mensal": (
        [(f"p{i}_pessoas", f"sum(qtd_aux_emergencial_p{i}_i)") for i in range(1, 6)]
        + [(f"p{i}_valor", f"sum(vl_aux_emergencial_p{i}_f)") for i in range(1, 6)]
        + [("residual_set_pessoas", "sum(qtd_total_aux_residual_set_i)"),
           ("residual_dez_pessoas", "sum(qtd_total_aux_residual_dez_i)"),
           ("residual_set_valor", "sum(vl_total_aux_residual_set_f)"),
           ("residual_dez_valor", "sum(vl_total_aux_residual_dez_f)"),
           ("ae2021_pessoas", "sum(aux_emerg_21_tot_pessoas_distintas_i)"),
           ("ae2021_pessoas_pbf", "sum(aux_emerg_21_pbf_pessoas_distintas_i)"),
           ("ae2021_familias_pbf", "sum(aux_emerg_21_pbf_familias_distintas_i)")]),
}
# Meses baixados por município, para conferir a soma do servidor (campos de valor também em texto)
MESES_AMOSTRA = ["200612", "201012", "201812", "202105", "202110", "202212", "202305", "202512"]
CAMPOS_AMOSTRA = ["codigo_ibge", "anomes_s", "qtd_familias_beneficiarias_bolsa_familia_i", "valor_repassado_bolsa_familia_s",
                  "pab_qtd_fam_benef_i", "pab_valor_pago_d", "pab_extraordinario_valor_pago_d", "bpc_idoso_ben_i",
                  "bpc_pcd_ben_i", "cadun_qtd_familias_cadastradas_i", "cadun_qtd_pessoas_cadastradas_i"]

# Programa de transferência de renda (D-030)
INICIO_AUXILIO_BRASIL, FIM_AUXILIO_BRASIL = "2021-11", "2023-02"
PROGRAMAS = {
    "bf2004": "Bolsa Família (Lei 10.836/2004)",
    "ab": "Auxílio Brasil (MP 1.061/2021; Lei 14.284/2021)",
    "bf2023": "Bolsa Família (MP 1.164/2023; Lei 14.601/2023)",
}
# Meses em que o Auxílio Emergencial substituiu o benefício do Bolsa Família quando era mais vantajoso
# (Lei 13.982/2020, art. 2º, § 2º; MP 1.000/2020; MP 1.039/2021 e Decretos 10.661 e 10.740/2021): a
# família continua beneficiária, mas o valor do PBF só é pago às que não receberam o auxílio.
MESES_AE = meses_entre("2020-04", "2020-12") + meses_entre("2021-04", "2021-10")


def programa_do_mes(mes):
    if mes < INICIO_AUXILIO_BRASIL:
        return "bf2004"
    if mes <= FIM_AUXILIO_BRASIL:
        return "ab"
    return "bf2023"


def url_misocial_faceta(grupo):
    faceta = {"meses": {"type": "terms", "field": "anomes_s", "limit": -1, "sort": "index asc",
                        "facet": dict(FACETAS_MISOCIAL[grupo])}}
    return API_MISOCIAL + "?" + urllib.parse.urlencode(
        {"q": "*:*", "rows": 0, "wt": "json", "json.facet": json.dumps(faceta, separators=(",", ":"))})


def url_misocial_amostra(mes):
    return API_MISOCIAL + "?" + urllib.parse.urlencode(
        {"q": "*:*", "fq": f"anomes_s:{mes}", "rows": 6000, "wt": "csv", "fl": ",".join(CAMPOS_AMOSTRA),
         "sort": "codigo_ibge asc"})


def validar_faceta(conteudo):
    d = json.loads(conteudo)
    baldes = d.get("facets", {}).get("meses", {}).get("buckets", [])
    if d.get("response", {}).get("numFound", 0) < 1_000_000 or len(baldes) < 280:
        raise RuntimeError(f"MI Social: resposta inesperada ({len(baldes)} meses)")
    if any(not 5500 <= b["count"] <= 5600 for b in baldes):
        raise RuntimeError("MI Social: mês com número inesperado de municípios")


def validar_amostra(conteudo):
    linhas = conteudo.decode("utf-8").splitlines()
    if not linhas or not linhas[0].startswith("codigo_ibge,") or not 5500 <= len(linhas) - 1 <= 5600:
        raise RuntimeError(f"MI Social: amostra inesperada ({len(linhas)} linhas)")


def coletar_misocial(args):
    for grupo in FACETAS_MISOCIAL:
        obter(FONTE_MDS, grupo, "json", url_misocial_faceta(grupo), validar=validar_faceta)
        time.sleep(1)
    for mes in MESES_AMOSTRA:
        obter(FONTE_MDS, f"amostra_municipios_{mes}", "csv", url_misocial_amostra(mes), validar=validar_amostra)
        time.sleep(1)


def ler_faceta(grupo):
    """{AAAA-MM: {chave: soma}} da resposta do servidor, e o arquivo."""
    arq = bruto(FONTE_MDS, grupo)
    d = json.loads(arq.read_text(encoding="utf-8"))
    meses = {}
    for b in d["facets"]["meses"]["buckets"]:
        mes = f"{b['val'][:4]}-{b['val'][4:6]}"
        meses[mes] = {k: float(b.get(k, 0) or 0) for k, _ in FACETAS_MISOCIAL[grupo]}
        meses[mes]["municipios"] = b["count"]
    return meses, arq


def ler_amostra(mes):
    arq = bruto(FONTE_MDS, f"amostra_municipios_{mes}")
    with open(arq, encoding="utf-8", newline="") as f:
        linhas = list(csv.DictReader(f))
    somas = {}
    for campo in CAMPOS_AMOSTRA[2:]:
        if campo.endswith("_s"):
            somas[campo] = float(sum(Decimal(l[campo] or "0") for l in linhas))
        else:
            somas[campo] = sum(float(l[campo] or 0) for l in linhas)
    return somas, len(linhas), arq


def ultimo_mes_com_dado(serie):
    """Último mês anterior ao corrente com valor positivo."""
    meses = [m for m, v in serie.items() if v > 0 and m < MES_CORRENTE]
    return max(meses) if meses else None


def tratar_transferencia_renda(saida, ipca_mensal, ipca_anual, base):
    tr, arq = ler_faceta("transferencia_renda_mensal")
    ultimo = ultimo_mes_com_dado({m: v["bf_familias"] + v["ab_familias"] for m, v in tr.items()})
    meses = meses_entre("2004-01", ultimo)
    familias, valor, prog = {}, {}, {}
    for m in meses:
        v = tr[m]
        p = programa_do_mes(m)
        if p == "ab":
            if v["bf_familias"] or v["bf_valor"]:
                raise RuntimeError(f"MI Social {m}: Bolsa Família no período do Auxílio Brasil")
            familias[m] = v["ab_familias"]
            valor[m] = v["ab_valor"] + v["ab_extraordinario_valor"]  # o benefício extraordinário faz parte do AB
        else:
            if v["ab_familias"] or v["ab_valor"]:
                raise RuntimeError(f"MI Social {m}: Auxílio Brasil fora do período do programa")
            familias[m] = v["bf_familias"]
            valor[m] = v["bf_valor"]
        if familias[m] <= 0:
            raise RuntimeError(f"MI Social {m}: mês sem famílias beneficiárias")
        prog[m] = p

    # conferência dos meses do Auxílio Emergencial: em 2020 o MDS marca as famílias do PBF que receberam o
    # auxílio; em 2021 o valor médio do PBF cai a menos da metade nos meses de pagamento do auxílio
    meses_ae_2020 = {m for m, v in tr.items() if v["bf_ae_familias_todas"] > 0}
    if meses_ae_2020 != {m for m in MESES_AE if m < "2021"}:
        raise RuntimeError(f"meses do Auxílio Emergencial em 2020 diferentes do esperado: {sorted(meses_ae_2020)}")
    ref_2021 = valor["2021-03"] / familias["2021-03"]
    for m in meses_entre("2021-01", "2021-10"):
        caiu = valor[m] / familias[m] < 0.75 * ref_2021
        if caiu != (m in MESES_AE):
            raise RuntimeError(f"MI Social {m}: valor médio do PBF não segue o calendário do Auxílio Emergencial")

    rotulo = {m: PROGRAMAS[p] for m, p in prog.items()}
    f_id = "transferencia_renda_familias"
    fonte = "MDS, MI Social (matriz de informações sociais), famílias beneficiárias no mês, soma dos municípios"
    fonte_mes = {m: f"{fonte}; {rotulo[m]}" + ("; inclui as famílias que receberam o Auxílio Emergencial no lugar "
                                              "do benefício do PBF" if m in MESES_AE else "") for m in meses}
    mensal = {m: familias[m] for m in meses if m < MES_CORRENTE}
    saida.add(f_id, f"{f_id}__mensal", mensal, "famílias", fonte, [arq], 0, fonte_por_data=fonte_mes)
    anual, fonte_ano, prog_ano = {}, {}, {}
    for ano in sorted({m[:4] for m in meses}):
        ms = [f"{ano}-{k:02d}" for k in range(1, 13)]
        if not all(m in familias for m in ms) or ano >= str(ANO_CORRENTE):
            continue
        anual[ano] = sum(familias[m] for m in ms) / 12
        progs = []
        for m in ms:
            if not progs or progs[-1][0] != rotulo[m]:
                progs.append([rotulo[m], m, m])
            progs[-1][2] = m
        partes = [p if len(progs) == 1 else f"{p} ({mes_pt(a)} a {mes_pt(b)})" for p, a, b in progs]
        prog_ano[ano] = " e ".join(partes)
        fonte_ano[ano] = f"{fonte}; média dos 12 meses; {prog_ano[ano]}"
    saida.add(f_id, f"{f_id}__media_anual", anual, "famílias (média mensal do ano)", fonte, [arq], 0,
              fonte_por_data=fonte_ano)

    # benefício médio: valor transferido no mês / famílias beneficiárias, fora dos meses do Auxílio Emergencial
    b_id = "transferencia_renda_beneficio_medio"
    fonte_b = ("MDS, MI Social, valor transferido às famílias no mês dividido pelas famílias beneficiárias (soma dos "
               "municípios)")
    nominal = {m: valor[m] / familias[m] for m in meses if m not in MESES_AE}
    real_m = {m: v * base / ipca_mensal[m] for m, v in nominal.items() if m in ipca_mensal}
    fonte_b_mes = {m: f"{fonte_b}; {rotulo[m]}" + ("; inclui o benefício extraordinário" if prog[m] == "ab" and
                                                  tr[m]["ab_extraordinario_valor"] > 0 else "") for m in nominal}
    saida.add(b_id, f"{b_id}__real_mensal", real_m, f"R$ de {ANO_BASE_REAL} por família (IPCA do mês)",
              fonte_b + f"; reais de {ANO_BASE_REAL} pelo IPCA do mês", [arq], 2,
              fonte_por_data={m: t + f"; reais de {ANO_BASE_REAL} pelo IPCA do mês" for m, t in fonte_b_mes.items()})
    saida.add(b_id, f"{b_id}__nominal_mensal", nominal, "R$ correntes por família", fonte_b, [arq], 2,
              fonte_por_data=fonte_b_mes)
    nom_a, real_a, fonte_a = {}, {}, {}
    for ano in anual:
        ms = [m for m in meses if m[:4] == ano and m not in MESES_AE]
        nom_a[ano] = sum(valor[m] for m in ms) / sum(familias[m] for m in ms)
        if all(m in ipca_mensal for m in ms):
            real_a[ano] = sum(valor[m] * base / ipca_mensal[m] for m in ms) / sum(familias[m] for m in ms)
        excl = [m for m in MESES_AE if m[:4] == ano]
        fonte_a[ano] = (f"{fonte_b}; média dos meses do ano ponderada pelas famílias; {prog_ano[ano]}"
                        + (f"; sem os meses do Auxílio Emergencial ({mes_pt(excl[0])} a {mes_pt(excl[-1])})" if excl else ""))
    saida.add(b_id, f"{b_id}__real_media_anual", real_a, f"R$ de {ANO_BASE_REAL} por família (IPCA do mês)",
              fonte_b, [arq], 2, fonte_por_data={a: t + f"; reais de {ANO_BASE_REAL} pelo IPCA de cada mês"
                                                  for a, t in fonte_a.items()})
    saida.add(b_id, f"{b_id}__nominal_media_anual", nom_a, "R$ correntes por família", fonte_b, [arq], 2,
              fonte_por_data=fonte_a)

    # notas: valor médio calculado pelo MDS (inclui famílias com benefício bloqueado) e meses do Auxílio Emergencial
    difs = []
    for m in ("2023-05", "2023-12", "2024-12", "2025-12"):
        if m in tr and tr[m]["bf_familias"]:
            vm = tr[m]["bf_valor_medio_x_familias"] / tr[m]["bf_familias"]
            difs.append(f"{mes_pt(m)}: {br(nominal[m], 2)} (transferido) e {br(vm, 2)} (valor médio do benefício)")
    saida.notas[b_id].append(
        "Valor médio do benefício calculado pelo MDS (média municipal `pbf_vlr_medio_benef_f` ponderada pelas famílias) "
        "contra o valor transferido por família desta série, em R$ correntes: " + "; ".join(difs) + ". Até 2021 os dois "
        "coincidem; desde 2023 o valor médio do MDS inclui o benefício calculado das famílias com pagamento bloqueado "
        "ou suspenso no mês.")
    ae = [f"{mes_pt(m)}: {br(tr[m]['bf_familias'])} famílias, das quais {br(tr[m]['bf_so_ae_familias'])} só com o "
          f"auxílio (R$ {br(tr[m]['bf_so_ae_valor'] / 1e6)} milhões); R$ {br(tr[m]['bf_valor'] / 1e6)} milhões pagos pelo PBF"
          for m in ("2020-05", "2020-10") if m in tr]
    saida.notas[b_id].append(
        "Meses fora da série por causa do Auxílio Emergencial: " + ", ".join(mes_pt(m) for m in MESES_AE) + ". Neles o "
        "valor do PBF só foi pago às famílias para as quais o benefício era maior que o auxílio. Exemplos (MI Social): "
        + "; ".join(ae) + ".")
    saida.notas[f_id].append(
        "Programa de cada mês na coluna `fonte`: Bolsa Família de jan/2004 a out/2021 e desde mar/2023; Auxílio Brasil "
        "de nov/2021 a fev/2023 (D-030). O MI Social não tem os meses de out a dez/2003 do Bolsa Família; a série "
        f"mensal começa em jan/2004. Último mês: {mes_pt(ultimo)}. Registros (municípios) por mês no MI Social: "
        + (br(min(v['municipios'] for v in tr.values())) if len({v['municipios'] for v in tr.values()}) == 1 else
           f"{br(min(v['municipios'] for v in tr.values()))} a {br(max(v['municipios'] for v in tr.values()))}") + ".")
    return tr, familias, valor, prog, arq


MESES_ABREV = ["jan", "fev", "mar", "abr", "mai", "jun", "jul", "ago", "set", "out", "nov", "dez"]


def mes_pt(aaaa_mm):
    return f"{MESES_ABREV[int(aaaa_mm[5:]) - 1]}/{aaaa_mm[:4]}"


# ---------------------------------------------------------------- programas anteriores ao Bolsa Família (Ipea)

FONTE_IPEA = "ipea_publicacoes"
NOME_BPS13 = "politicas_sociais_acompanhamento_analise_13_2007"
URL_BPS13 = "https://repositorio.ipea.gov.br/bitstreams/5b23f985-001f-4585-9c00-3d8021ff8b1f/download"
PAGINA_BPS13 = "https://repositorio.ipea.gov.br/handle/11058/4164"
# Políticas Sociais: acompanhamento e análise nº 13 (edição especial, 2007), tabela 6, p. 105 do PDF: famílias
# beneficiárias em dezembro. Fonte da tabela: MDS, relatório "Análise comparativa de programas de proteção social"
# (abril de 2004) para 2001-2004 e Sigplan (Ministério do Planejamento) para 2005-2006.
PROGRAMAS_ANTERIORES = {
    "bolsa_escola": ("Bolsa Escola (Lei 10.219/2001)", {"2002": 5106509, "2003": 3771199, "2004": 3042794, "2005": 1783874}),
    "bolsa_alimentacao": ("Bolsa Alimentação (MP 2.206-1/2001)", {"2002": 966553, "2003": 369463, "2004": 53507, "2005": 24175}),
    "auxilio_gas": ("Auxílio-Gás (Decreto 4.102/2002)", {"2002": 8556785, "2003": 8229144, "2004": 4220953, "2005": 3401114}),
    "cartao_alimentacao": ("Cartão Alimentação (Lei 10.689/2003)", {"2003": 349905, "2004": 107907, "2005": 83524}),
}
BF_BPS13 = {"2003-12": 3615596, "2004-12": 6571842, "2005-12": 8700451, "2006-08": 11120353}  # mesma tabela, conferência


def coletar_programas_anteriores(args):
    obter(FONTE_IPEA, NOME_BPS13, "pdf", URL_BPS13, validar=validar_pdf, reaproveitar=not args.rebaixar,
          baixar_fn=lambda: pedir(URL_BPS13, timeout=600))


def tratar_programas_anteriores(saida, familias):
    arq = bruto(FONTE_IPEA, NOME_BPS13)
    f_id = "transferencia_renda_familias"
    for chave, (nome, valores) in PROGRAMAS_ANTERIORES.items():
        saida.add(f_id, f"{f_id}__{chave}_dezembro", {a: float(v) for a, v in valores.items()}, "famílias (dezembro)",
                  f"Ipea, Políticas Sociais: acompanhamento e análise nº 13 (2007), tabela 6, a partir de dados do MDS "
                  f"(até 2004) e do Sigplan (2005); {nome}; famílias beneficiárias em dezembro", [arq], 0)
    conf = []
    for m, v in BF_BPS13.items():
        if m in familias:
            texto = f"{mes_pt(m)}: {br(v)} no Ipea e {br(familias[m])} no MI Social"
            anterior = f"{m[:4]}-{int(m[5:]) - 1:02d}"
            if familias.get(anterior) == v and familias[m] != v:
                texto += f" (o número do Ipea é o de {mes_pt(anterior)} no MI Social)"
            conf.append(texto)
    saida.conferiu("Bolsa Família, Ipea (tabela 6 do nº 13) x MI Social: " + "; ".join(conf))
    saida.notas[f_id].append(
        "Programas anteriores ao Bolsa Família, em séries à parte (D-030), com as famílias beneficiárias em dezembro da "
        f"tabela 6 do boletim Políticas Sociais nº 13 do Ipea (`{arq.relative_to(RAIZ)}`, p. 105), que cita o MDS (2001 a "
        "2004) e o Sigplan (2005). As famílias se sobrepõem entre programas (o Auxílio-Gás foi pago primeiro às famílias "
        "do Bolsa Escola e do Bolsa Alimentação) e não podem ser somadas. O Bolsa Família tinha 3.615.596 famílias em "
        "dezembro de 2003 na mesma tabela. Conferência do Bolsa Família com o MI Social: " + "; ".join(conf) + ".")


# ---------------------------------------------------------------- BPC

FONTE_MDS_PUB = "mds_publicacoes"
NOME_NT_BPC = "nota_tecnica_03_2016_dba_snas_judicializacao_bpc"
URL_NT_BPC = ("https://www.mds.gov.br/webarquivos/publicacao/assistencia_social/Normativas/"
              "NotaTecnica_n03_Judicializacao_BPC.PDF")
# Nota Técnica nº 03/2016/DBA/SNAS/MDS, tabela "Evolução do Quantitativo de Beneficiários do BPC" (p. 3-4 do PDF),
# benefícios ativos em dezembro, fonte Síntese/Dataprev: (pessoa com deficiência, pessoa idosa)
BPC_NOTA = {"2002": (976257, 584597), "2003": (1036365, 664875), "2004": (1127849, 933164), "2005": (1211761, 1065604),
            "2006": (1293645, 1183840), "2007": (1385107, 1295716), "2008": (1510682, 1423790),
            "2009": (1625625, 1541220), "2010": (1778345, 1623196), "2011": (1907511, 1687826),
            "2012": (2021721, 1750121), "2013": (2156581, 1830476), "2014": (2253822, 1876610),
            "2015": (2323794, 1918903)}
ANOS_BPC_DA_NOTA = ("2002", "2003")  # o MI Social começa em jan/2004


def coletar_nota_bpc(args):
    obter(FONTE_MDS_PUB, NOME_NT_BPC, "pdf", URL_NT_BPC, validar=validar_pdf, reaproveitar=not args.rebaixar)


def tratar_bpc(saida):
    bpc, arq = ler_faceta("bpc_mensal")
    arq_nota = bruto(FONTE_MDS_PUB, NOME_NT_BPC)
    ultimo = ultimo_mes_com_dado({m: v["bpc_total"] for m, v in bpc.items()})
    meses = meses_entre("2004-01", ultimo)
    for m in meses:
        v = bpc[m]
        if abs(v["bpc_idoso"] + v["bpc_pcd"] - v["bpc_total"]) > 0.5:
            raise RuntimeError(f"BPC {m}: idosos + pessoas com deficiência diferente do total")
    f_id = "bpc"
    fonte = "MDS, MI Social, beneficiários do BPC no mês (benefícios emitidos), soma dos municípios"
    fonte_nota = ("MDS, Nota Técnica nº 03/2016/DBA/SNAS/MDS, tabela de evolução dos beneficiários do BPC "
                  "(Síntese/Dataprev), benefícios ativos em dezembro")
    for recorte, chave, idx in (("idoso", "bpc_idoso", 1), ("pessoa_com_deficiencia", "bpc_pcd", 0), ("total", "bpc_total", None)):
        mensal = {m: bpc[m][chave] for m in meses}
        dezembro = {m[:4]: v for m, v in mensal.items() if m.endswith("-12")}
        fonte_dez = {}
        for a in ANOS_BPC_DA_NOTA:
            dezembro[a] = float(sum(BPC_NOTA[a]) if idx is None else BPC_NOTA[a][idx])
            fonte_dez[a] = fonte_nota
        saida.add(f_id, f"{f_id}__{recorte}_dezembro", anos_fechados(dezembro), "beneficiários (dezembro)",
                  fonte + "; dezembro", [arq, arq_nota], 0, fonte_por_data=fonte_dez)
        saida.add(f_id, f"{f_id}__{recorte}_mensal", mensal, "beneficiários", fonte, [arq], 0)
    # conferência com a nota técnica (2004-2015)
    iguais, diferentes = [], []
    for a, (pcd, ido) in BPC_NOTA.items():
        if a in ANOS_BPC_DA_NOTA:
            continue
        v = bpc[f"{a}-12"]
        if (round(v["bpc_pcd"]), round(v["bpc_idoso"])) == (pcd, ido):
            iguais.append(a)
        else:
            diferentes.append(f"{a}: nota {br(pcd)} e {br(ido)}; MI Social {br(v['bpc_pcd'])} e {br(v['bpc_idoso'])}")
    texto = (f"BPC em dezembro, MI Social x Nota Técnica 03/2016 do MDS (pessoa com deficiência e idosa): iguais em "
             f"{', '.join(iguais)}; diferentes em " + "; ".join(diferentes))
    saida.conferiu(texto)
    saida.notas[f_id].append(
        f"Dezembro de 2002 e de 2003: Nota Técnica nº 03/2016/DBA/SNAS/MDS (`{arq_nota.relative_to(RAIZ)}`, p. 3 e 4), "
        "porque o MI Social começa em janeiro de 2004. " + texto + ". Em 2013, os números da nota são os de fevereiro "
        f"de 2014 no MI Social ({br(bpc['2014-02']['bpc_pcd'])} e {br(bpc['2014-02']['bpc_idoso'])}).")
    saida.notas[f_id].append(
        f"Último mês no MI Social: {mes_pt(ultimo)}. Valor pago no mês (R$ bilhões correntes, MI Social): "
        + "; ".join(f"{mes_pt(m)}: {br((bpc[m]['bpc_idoso_valor'] + bpc[m]['bpc_pcd_valor']) / 1e9, 2)}"
                    for m in (f"{a}-12" for a in ("2006", "2010", "2018", "2022", "2025")) if m in bpc) + ".")
    return bpc, arq


# ---------------------------------------------------------------- Cadastro Único

def tratar_cadunico(saida):
    cad, arq = ler_faceta("cadunico_mensal")
    primeiro = min(m for m, v in cad.items() if v["familias"] > 0)
    ultimo = ultimo_mes_com_dado({m: v["familias"] for m, v in cad.items()})
    vazios = [m for m in meses_entre(primeiro, ultimo) if cad[m]["familias"] <= 0]
    if len(vazios) > 3:
        raise RuntimeError(f"Cadastro Único sem dado em {vazios}")
    meses = [m for m in meses_entre(primeiro, ultimo) if m not in vazios]  # mês sem dado no MI Social fica fora
    f_id = "cadunico"
    fonte = "MDS, MI Social, famílias cadastradas no Cadastro Único no mês, soma dos municípios"
    for recorte, chave, unidade, rotulo in (
            ("familias", "familias", "famílias", ""),
            ("pessoas", "pessoas", "pessoas", "; pessoas cadastradas"),
            ("familias_ate_meio_salario_minimo", "familias_ate_meio_sm", "famílias",
             "; famílias com renda mensal per capita de até meio salário mínimo")):
        mensal = {m: cad[m][chave] for m in meses}
        dezembro = {m[:4]: v for m, v in mensal.items() if m.endswith("-12")}
        saida.add(f_id, f"{f_id}__{recorte}_dezembro", anos_fechados(dezembro), f"{unidade} (dezembro)",
                  fonte + rotulo + "; dezembro", [arq], 0)
        saida.add(f_id, f"{f_id}__{recorte}_mensal", mensal, unidade, fonte + rotulo, [arq], 0)
    quedas = []
    for ant, m in zip(meses, meses[1:]):
        if mes_seguinte(ant) != m:
            continue  # não compara através de um mês sem dado
        var = cad[m]["familias"] / cad[ant]["familias"] - 1
        if var < -0.025:
            quedas.append(f"{mes_pt(m)}: {br(100 * var, 1)}%")
    saida.notas[f_id].append(
        f"O MI Social traz as famílias cadastradas a partir de {mes_pt(primeiro)} (último mês: {mes_pt(ultimo)})"
        + (f"; sem dado em {', '.join(mes_pt(m) for m in vazios)} (campo zerado em todos os municípios), mês que fica "
           "fora das séries mensais" if vazios else "") + ". Quedas mensais de mais de 2,5% no número de famílias, em "
        "geral exclusões de cadastros desatualizados em processos de revisão e averiguação cadastral: "
        + "; ".join(quedas) + ".")
    return cad, arq


# ---------------------------------------------------------------- Auxílio Emergencial (pessoas)

def pessoas_auxilio_emergencial():
    ae, arq = ler_faceta("auxilio_emergencial_mensal")
    p1 = sum(v["p1_pessoas"] for v in ae.values())
    ae21 = sum(v["ae2021_pessoas"] for v in ae.values())
    return ae, arq, {"2020": p1, "2021": ae21}


# ================================================================ despesa federal (SIOP, SPARQL) e PIB

FONTE_SIOP = "siop_sparql"
SPARQL_SIOP = "https://www1.siop.planejamento.gov.br/sparql/"  # espelho de http://orcamento.dados.gov.br/sparql/
# ação orçamentária -> (grupo, padrão que o título da ação precisa conter)
ACOES = {
    "006O": ("bf_ab", r"Transfer[eê]ncia de Renda Diretamente"),
    "099A": ("bf_ab", r"Pobreza Extrema"),
    "0B19": ("bf_ab", r"Pobreza Extrema"),
    "8442": ("bf_ab", r"Transfer[eê]ncia"),
    "21DP": ("bf_ab", r"Aux[ií]lio Brasil"),
    "0519": ("anteriores", r"Bolsa-Escola"),
    "0357": ("anteriores", r"Bolsa-Alimenta"),
    "2100": ("anteriores", r"Fome Zero"),
    "0789": ("anteriores", r"Aux[ií]lio-G[aá]s"),
    "00S4": ("auxilio_emergencial", r"Aux[ií]lio Emergencial"),
    "00SF": ("auxilio_emergencial", r"Aux[ií]lio Emergencial Residual"),
    "00SI": ("auxilio_emergencial", r"Aux[ií]lio Emergencial 2021"),
}
CONSULTA_SIOP = """PREFIX loa: <http://vocab.e.gov.br/2013/09/loa#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
SELECT ?acao ?nome ?funcao ?subfuncao ?uo ?rp (SUM(?emp) AS ?empenhado) (SUM(?liq) AS ?liquidado) (SUM(?pag) AS ?pago)
FROM <http://orcamento.dados.gov.br/{ano}/>
WHERE {{
  ?i a loa:ItemDespesa ; loa:temAcao ?a ; loa:temFuncao ?f ; loa:temSubfuncao ?s ; loa:temUnidadeOrcamentaria ?u ;
     loa:temResultadoPrimario ?r ; loa:valorEmpenhado ?emp ; loa:valorLiquidado ?liq ; loa:valorPago ?pag .
  ?a loa:codigo ?acao ; rdfs:label ?nome . ?f loa:codigo ?funcao . ?s loa:codigo ?subfuncao . ?u loa:codigo ?uo .
  ?r loa:codigo ?rp .
  FILTER(?acao IN ({acoes}))
}}
GROUP BY ?acao ?nome ?funcao ?subfuncao ?uo ?rp ORDER BY ?acao ?uo ?rp"""
ANO_INICIAL_DESPESA = 2002  # D-002
NOME_SIOP = "despesa_transferencia_renda_{ano}"


def url_siop(ano):
    consulta = CONSULTA_SIOP.format(ano=ano, acoes=",".join(f'"{a}"' for a in ACOES))
    return SPARQL_SIOP + "?" + urllib.parse.urlencode({"query": consulta, "format": "text/csv"})


def coletar_despesa(args):
    for ano in range(ANO_INICIAL_DESPESA, ANO_CORRENTE):  # só anos fechados
        def baixar_csv(ano=ano):
            corpo = pedir(url_siop(ano))
            if not corpo.startswith(b'"acao","nome","funcao","subfuncao","uo","rp"'):
                raise RuntimeError(f"SPARQL {ano}: resposta inesperada {corpo[:200]!r}")
            time.sleep(1)
            return corpo
        obter(FONTE_SIOP, NOME_SIOP.format(ano=ano), "csv", url_siop(ano), baixar_fn=baixar_csv)


def ler_despesa():
    """{ano: {grupo: R$ pago}}, {ano: {ação: (título, R$ pago)}} e os arquivos."""
    por_grupo, por_acao, arquivos = {}, {}, []
    for ano in range(ANO_INICIAL_DESPESA, ANO_CORRENTE):
        arq = bruto(FONTE_SIOP, NOME_SIOP.format(ano=ano))
        arquivos.append(arq)
        g, a = defaultdict(float), {}
        with open(arq, encoding="utf-8") as f:
            for l in csv.DictReader(f):
                grupo, padrao = ACOES[l["acao"]]
                if not re.search(padrao, l["nome"]):
                    raise RuntimeError(f"{arq.name}: ação {l['acao']} com título inesperado: {l['nome']}")
                pago = float(l["pago"] or 0)
                g[grupo] += pago
                titulo, soma = a.get(l["acao"], (limpo(l["nome"]), 0.0))
                a[l["acao"]] = (titulo, soma + pago)
        por_grupo[str(ano)] = g
        por_acao[str(ano)] = {k: v for k, v in a.items() if v[1]}
    return por_grupo, por_acao, arquivos


FONTE_SIDRA = "ibge_sidra"
SIDRA_PIB = "/t/1846/n1/all/v/585/p/all/c11255/90707"  # Contas Nacionais Trimestrais, valores correntes, PIB
NOME_PIB = "t1846_v585"


def coletar_pib(args):
    def validar(c):
        d = json.loads(c)
        if not isinstance(d, list) or len(d) < 2 or "V" not in d[0]:
            raise RuntimeError("SIDRA: resposta inesperada")
    obter(FONTE_SIDRA, NOME_PIB, "json", "https://apisidra.ibge.gov.br/values" + SIDRA_PIB, validar=validar)


def ler_pib():
    """{ano: PIB nominal em R$ milhões}, soma dos quatro trimestres (mesma série do tema orçamento, D-027)."""
    arq = bruto(FONTE_SIDRA, NOME_PIB)
    linhas = json.loads(arq.read_text(encoding="utf-8"))
    col = next(k for k, v in linhas[0].items() if v == "Trimestre (Código)")
    trimestres = defaultdict(dict)
    for l in linhas[1:]:
        if num(l["V"]) is not None:
            trimestres[l[col][:4]][l[col][4:]] = num(l["V"])
    pib = {a: sum(t.values()) for a, t in trimestres.items() if len(t) == 4}
    faltam = [str(a) for a in range(ANO_INICIAL_DESPESA, ANO_CORRENTE) if str(a) not in pib]
    if faltam:
        raise RuntimeError(f"PIB trimestral sem os quatro trimestres em {faltam}")
    return pib, arq


# conferência: Resultado do Tesouro Nacional (RTN), regime de caixa, com restos a pagar pagos
FONTE_RTN = "tesouro_rtn"
CKAN_RTN = "https://www.tesourotransparente.gov.br/ckan/api/3/action/package_show?id=resultado-do-tesouro-nacional"
RECURSO_RTN = "Resultado do Tesouro Nacional - Série Histórica - Mensal"
URL_RTN_DEZ22 = "https://thot-arquivos.tesouro.gov.br/publicacao-anexo/19092"  # edição de dez/2022, com a tabela 4.1
LINHA_RTN_BF = "4.4.1.2"  # "Bolsa Família e Auxílio Brasil", tabela 2.2 (anual)
LINHA_RTN_AE = "Auxílio Emergencial a Pessoas em Situação de Vulnerabilidade"  # tabela 4.1 (covid-19, mensal)


def coletar_rtn(args):
    """Série histórica do RTN (o endereço muda todo mês e vem do CKAN do Tesouro Transparente) e a edição de
    dezembro de 2022, a única com as despesas de combate à covid-19. Os mesmos arquivos do tema orçamento."""
    obter(FONTE_RTN, "rtn_serie_historica_dez22", "xlsx", URL_RTN_DEZ22, validar=validar_zip,
          reaproveitar=not args.rebaixar)
    arq = obter(FONTE_RTN, "ckan_resultado_do_tesouro_nacional", "json", CKAN_RTN, validar=validar_json)
    recursos = json.loads(arq.read_text(encoding="utf-8"))["result"]["resources"]
    (url,) = [r["url"] for r in recursos if limpo(r.get("name")) == RECURSO_RTN]
    obter(FONTE_RTN, "rtn_serie_historica", "xlsx", url, validar=validar_zip)


def rtn_linha_anual(prefixo):
    """{ano: R$ milhões} da linha da tabela 2.2 do RTN que começa com o código `prefixo`."""
    arq = bruto(FONTE_RTN, "rtn_serie_historica")
    livro = Xlsx(str(arq))
    (aba,) = [n for n in livro.planilhas if n.strip() == "2.2"]
    linhas = livro.linhas(aba)
    cab = next(l for l in linhas if l and limpo(l[0]) == "Discriminação")
    anos = {c: limpo(x) for c, x in enumerate(cab) if re.fullmatch(r"\d{4}", limpo(x))}
    (linha,) = [l for l in linhas if l and limpo(l[0]).startswith(prefixo + " ")]
    return {a: num(linha[c]) for c, a in anos.items() if c < len(linha) and num(linha[c]) is not None}, \
        limpo(linha[0]), arq


def rtn_auxilio_emergencial():
    """{ano: R$ milhões} da linha do Auxílio Emergencial na tabela 4.1 da edição de dez/2022 do RTN."""
    arq = bruto(FONTE_RTN, "rtn_serie_historica_dez22")
    livro = Xlsx(str(arq))
    (aba,) = [n for n in livro.planilhas if n.strip() == "4.1"]
    linhas = livro.linhas(aba)
    cab = next(l for l in linhas if l and limpo(l[0]) == "Discriminação")
    datas = {c: dt.date(1899, 12, 30) + dt.timedelta(days=int(num(x))) for c, x in enumerate(cab) if c and num(x)}
    (linha,) = [l for l in linhas if l and limpo(l[0]) == LINHA_RTN_AE]
    por_ano = defaultdict(float)
    for c, d in datas.items():
        if c < len(linha) and num(linha[c]):
            por_ano[str(d.year)] += num(linha[c])
    return dict(por_ano), arq


def tratar_despesa(saida, ipca_anual, valor_mensal=None):
    """`valor_mensal` ({AAAA-MM: R$ transferido no mês}, do MI Social) serve para comparar o pago no orçamento com os
    benefícios registrados pelo MDS nos anos do Auxílio Emergencial."""
    por_grupo, por_acao, arquivos = ler_despesa()
    pib, arq_pib = ler_pib()
    f_id = "transferencia_renda_despesa"
    fonte = "SIOP/SOF, Orçamento Federal em dados abertos (SPARQL), valor pago no exercício"
    deflator = f"; reais de {ANO_BASE_REAL} pelo IPCA médio do ano (SIDRA 1737)"
    fonte_pib = "; % do PIB nominal do IBGE (SIDRA 1846, soma dos quatro trimestres)"
    bf = {a: g["bf_ab"] for a, g in por_grupo.items() if g.get("bf_ab")}
    ant = {a: g["anteriores"] for a, g in por_grupo.items() if g.get("anteriores")}

    def acoes_do_ano(a, grupo):
        return ", ".join(f"{k} ({v[0]}: R$ {br(v[1] / 1e9, 3)} bi)" for k, v in sorted(por_acao[a].items())
                         if ACOES[k][0] == grupo)

    def programa_do_ano(a):
        if a < "2021":
            return "Bolsa Família"
        return {"2021": "Bolsa Família (jan–out) e Auxílio Brasil (nov–dez)", "2022": "Auxílio Brasil",
                "2023": "Auxílio Brasil (jan–fev) e Bolsa Família (mar–dez)"}.get(a, "Bolsa Família")

    fonte_ano = {a: f"{fonte}; {programa_do_ano(a)}; ações {acoes_do_ano(a, 'bf_ab')}" for a in bf}
    for sufixo, dados, unidade, extra, casas in (
            ("paga_real", {a: v / 1e9 for a, v in real(bf, ipca_anual).items()}, U_REAL, deflator, 3),
            ("paga_nominal", {a: v / 1e9 for a, v in bf.items()}, U_NOM, "", 3),
            ("paga_pct_pib", {a: 100 * v / 1e6 / pib[a] for a, v in bf.items()}, U_PIB, fonte_pib, 4)):
        saida.add(f_id, f"{f_id}__{sufixo}", dados, unidade, fonte + extra, arquivos + ([arq_pib] if "pib" in sufixo else []),
                  casas, fonte_por_data={a: t + extra for a, t in fonte_ano.items()})
    fonte_ant = {a: f"{fonte}; programas anteriores ao Bolsa Família; ações {acoes_do_ano(a, 'anteriores')}" for a in ant}
    for sufixo, dados, unidade, extra, casas in (
            ("programas_anteriores_real", {a: v / 1e9 for a, v in real(ant, ipca_anual).items()}, U_REAL, deflator, 3),
            ("programas_anteriores_nominal", {a: v / 1e9 for a, v in ant.items()}, U_NOM, "", 3),
            ("programas_anteriores_pct_pib", {a: 100 * v / 1e6 / pib[a] for a, v in ant.items()}, U_PIB, fonte_pib, 4)):
        saida.add(f_id, f"{f_id}__{sufixo}", dados, unidade, fonte + extra, arquivos + ([arq_pib] if "pib" in sufixo else []),
                  casas, fonte_por_data={a: t + extra for a, t in fonte_ant.items()})

    # conferência com o RTN (Tesouro Nacional, caixa)
    rtn, rotulo_rtn, arq_rtn = rtn_linha_anual(LINHA_RTN_BF)
    difs = {a: bf[a] / 1e6 - v for a, v in rtn.items() if a in bf and v}
    texto = "; ".join(f"{a}: SIOP {br(bf[a] / 1e9, 2)} x RTN {br(rtn[a] / 1e3, 2)}" for a in sorted(difs))
    saida.conferiu(f"despesa BF/AB, SIOP (pago no exercício) x RTN, linha {rotulo_rtn} (R$ bilhões): {texto}")
    grandes = {a: d for a, d in difs.items() if abs(d) / rtn[a] > 0.05 and a != "2022"}
    if grandes:
        raise RuntimeError(f"despesa do programa difere do RTN em mais de 5%: {grandes}")
    pior = max((abs(d) / rtn[a], a) for a, d in difs.items() if a != "2022")
    saida.notas[f_id].append(
        f"Conferência com o Resultado do Tesouro Nacional (série histórica, tabela 2.2, linha \"{rotulo_rtn}\", regime de "
        f"caixa, com restos a pagar pagos; `{arq_rtn.relative_to(RAIZ)}`), em R$ bilhões correntes: {texto}. Fora 2022, a "
        f"maior diferença é de {br(100 * pior[0], 1)}% ({pior[1]}), compatível com os restos a pagar e com a ação 8446 (apoio à "
        "gestão descentralizada, o IGD, de R$ 0,4 a 0,5 bilhão pagos por ano de 2015 a 2018 no SIOP), que esta série não "
        "inclui. Em 2022 o RTN deixa fora dessa linha o crédito extraordinário da EC 123/2022 (MP 1.130/2022, plano "
        "orçamentário MP01 da ação 21DP, R$ 24,93 bilhões pagos), contado à parte; sem ele, o pago no SIOP em 2022 é de "
        "R$ 88,10 bilhões.")
    if valor_mensal:
        comp = []
        for a in ("2019", "2020", "2021", "2022"):
            mi = sum(v for m, v in valor_mensal.items() if m[:4] == a)
            comp.append(f"{a}: orçamento R$ {br(bf[a] / 1e9, 2)} bilhões, MI Social R$ {br(mi / 1e9, 2)} bilhões")
        saida.notas[f_id].append(
            "Pago no orçamento (ações do programa) contra os benefícios transferidos às famílias registrados no MI "
            "Social (soma dos meses, Bolsa Família e Auxílio Brasil): " + "; ".join(comp) + ". Em 2020, a ação 8442 pagou "
            "cerca do dobro dos benefícios do PBF registrados no MI Social; nos meses do Auxílio Emergencial o MI Social só "
            "registra o PBF pago às famílias que não receberam o auxílio. A diferença não foi conciliada nesta coleta.")
    saida.notas[f_id].append(
        "PIB nominal usado (R$ bilhões correntes, SIDRA 1846, soma dos quatro trimestres; a mesma série do tema "
        f"orçamento, `{arq_pib.relative_to(RAIZ)}`): "
        + "; ".join(f"{a}: {br(pib[a] / 1e3, 1)}" for a in sorted(pib) if ANO_INICIAL_DESPESA <= int(a) < ANO_CORRENTE) + ".")
    return por_grupo, por_acao, arquivos, pib, arq_pib


def tratar_auxilio_emergencial(saida, por_grupo, por_acao, arquivos_siop, pib, arq_pib, ipca_anual):
    ae, arq_ae, pessoas = pessoas_auxilio_emergencial()
    f_id = "auxilio_emergencial"
    saida.add(f_id, f"{f_id}__pessoas", pessoas, "pessoas beneficiárias no ano do programa",
              "MDS, MI Social, soma dos municípios", [arq_ae], 0,
              fonte_por_data={
                  "2020": "MDS, MI Social, pessoas que receberam a 1ª parcela do Auxílio Emergencial de 2020 (Lei 13.982/2020), "
                          "somadas em todos os meses de pagamento; soma dos municípios",
                  "2021": "MDS, MI Social, pessoas distintas que receberam o Auxílio Emergencial 2021 (MP 1.039/2021); soma "
                          "dos municípios"})
    valor = {a: g["auxilio_emergencial"] for a, g in por_grupo.items() if g.get("auxilio_emergencial")}
    fonte = "SIOP/SOF, Orçamento Federal em dados abertos (SPARQL), valor pago no exercício"
    fonte_ano = {a: f"{fonte}; ações " + ", ".join(f"{k} ({v[0]}: R$ {br(v[1] / 1e9, 3)} bi)"
                                                   for k, v in sorted(por_acao[a].items())
                                                   if ACOES[k][0] == "auxilio_emergencial") for a in valor}
    deflator = f"; reais de {ANO_BASE_REAL} pelo IPCA médio do ano (SIDRA 1737)"
    fonte_pib = "; % do PIB nominal do IBGE (SIDRA 1846, soma dos quatro trimestres)"
    for sufixo, dados, unidade, extra, casas in (
            ("valor_pago_real", {a: v / 1e9 for a, v in real(valor, ipca_anual).items()}, U_REAL, deflator, 3),
            ("valor_pago_nominal", {a: v / 1e9 for a, v in valor.items()}, U_NOM, "", 3),
            ("valor_pago_pct_pib", {a: 100 * v / 1e6 / pib[a] for a, v in valor.items()}, U_PIB, fonte_pib, 4)):
        saida.add(f_id, f"{f_id}__{sufixo}", dados, unidade, fonte + extra,
                  [a for a in arquivos_siop if any(x in a.name for x in valor)] + ([arq_pib] if "pib" in sufixo else []),
                  casas, fonte_por_data={a: t + extra for a, t in fonte_ano.items()})
    rtn, arq_rtn = rtn_auxilio_emergencial()
    texto = "; ".join(f"{a}: SIOP {br(valor.get(a, 0) / 1e9, 2)} x RTN {br(v / 1e3, 2)}" for a, v in sorted(rtn.items()))
    saida.conferiu(f"Auxílio Emergencial, SIOP (pago no exercício) x RTN tabela 4.1 (caixa), R$ bilhões: {texto}")
    if abs(valor["2020"] / 1e6 / rtn["2020"] - 1) > 0.01:
        raise RuntimeError("Auxílio Emergencial de 2020: SIOP e RTN diferem em mais de 1%")
    p = {k: sum(v[k] for v in ae.values()) for k in ("p1_valor", "p2_valor", "p3_valor", "p4_valor", "p5_valor",
                                                       "residual_set_pessoas", "residual_dez_pessoas", "ae2021_pessoas_pbf")}
    saida.notas[f_id].append(
        f"Valor pago: ações 00S4 e 00SF em 2020 e 00SI (e restos da 00S4) em 2021, no SIOP. Conferência com a tabela 4.1 "
        f"(despesas de combate à covid-19) da série histórica do RTN de dezembro de 2022 (`{arq_rtn.relative_to(RAIZ)}`, "
        f"regime de caixa, com restos a pagar pagos), em R$ bilhões: {texto}. O valor de 2022 no RTN é de restos a pagar.")
    saida.notas[f_id].append(
        f"Pessoas (MI Social, `{arq_ae.relative_to(RAIZ)}`): em 2020, quem recebeu a 1ª parcela (todo beneficiário começa "
        f"por ela; inclui pagamentos atrasados feitos até 2022); em 2021, pessoas distintas, das quais "
        f"{br(p['ae2021_pessoas_pbf'])} em famílias do Bolsa Família. O MI Social traz o Auxílio Emergencial Residual só "
        f"em setembro ({br(p['residual_set_pessoas'])} pessoas) e dezembro de 2020 ({br(p['residual_dez_pessoas'])}), sem "
        f"série completa. Valor das cinco parcelas de R$ 600 pelo MI Social: R$ {br(sum(p[f'p{i}_valor'] for i in range(1, 6)) / 1e9, 1)} "
        "bilhões (todos os meses de pagamento).")
    return pessoas, valor


# ================================================================ insegurança alimentar (IBGE, EBIA)

SIDRA_EBIA = {
    # ODS 2.1.2: % de domicílios e de moradores por situação de segurança alimentar; PNAD 2004, 2009 e 2013 (escala
    # de 14 itens), POF 2017-2018 (período 2018) e PNAD Contínua 2023 e 2024
    "ods": "/t/6665/n1/all/v/800,2133/p/all/c12404/all",
    # PNAD Contínua: domicílios (mil) e distribuição percentual, só conferência
    "pnadc": "/t/9552/n1/all/v/162,9784/p/all/c1/6795/c12404/all",
    # PNAD 2004, como publicada em 2006 (escala de 15 itens), só conferência
    "pnad2004": "/t/3017/n1/all/v/800,1000800/p/all/c12404/all",
}
TABELAS_EBIA = ("6665", "9552", "3017")
IBGE_AGREGADOS = "https://servicodados.ibge.gov.br/api/v3/agregados/{tabela}/{recurso}"
CAT_MODERADA, CAT_GRAVE, CAT_LEVE, CAT_IA = "109101", "109102", "109100", "109099"


def nome_sidra(caminho):
    """/t/6784/n1/all/v/9814,93/p/all -> t6784_v9814-93 (mesma regra de coletar.nome_sidra)."""
    partes = caminho.strip("/").split("/")
    tabela = partes[partes.index("t") + 1]
    variaveis = partes[partes.index("v") + 1].replace(",", "-") if "v" in partes else "all"
    return f"t{tabela}_v{variaveis}"


def baixar_ibge_api(url):
    """servicodados.ibge.gov.br responde em gzip mesmo sem pedido; grava-se o JSON descomprimido (a compressão
    é do transporte)."""
    conteudo = pedir(url)
    if conteudo[:2] == b"\x1f\x8b":
        conteudo = gzip.decompress(conteudo)
    json.loads(conteudo)
    return conteudo


def coletar_ebia(args):
    def validar(c):
        d = json.loads(c)
        if not (isinstance(d, list) and len(d) > 1 and "V" in d[0]):
            raise RuntimeError(f"SIDRA: resposta inesperada {c[:200]!r}")
    for caminho in SIDRA_EBIA.values():
        obter(FONTE_SIDRA, nome_sidra(caminho), "json", "https://apisidra.ibge.gov.br/values" + caminho, validar=validar)
    for tabela in TABELAS_EBIA:
        for recurso in ("metadados", "periodos"):
            url = IBGE_AGREGADOS.format(tabela=tabela, recurso=recurso)
            obter(FONTE_SIDRA, f"t{tabela}_{recurso}", "json", url, baixar_fn=lambda u=url: baixar_ibge_api(u))


def ler_sidra(chave):
    """{(variável, ano, categoria de segurança alimentar): valor} e o arquivo."""
    arq = bruto(FONTE_SIDRA, nome_sidra(SIDRA_EBIA[chave]))
    linhas = json.loads(arq.read_text(encoding="utf-8"))
    cab = linhas[0]
    col = {v: k[:-1] + "C" for k, v in cab.items() if k.endswith("N") and k.startswith("D")}
    c_ano, c_var = col["Ano"], col["Variável"]
    c_cat = col["Situação de segurança alimentar existente no domicílio"]
    pontos = {}
    for l in linhas[1:]:
        if num(l["V"]) is not None:
            pontos[(l[c_var], l[c_ano], l[c_cat])] = num(l["V"])
    return pontos, arq


def pesquisa_ebia(ano):
    a = int(ano)
    if a in (2004, 2009, 2013):
        return "pnad", "IBGE, PNAD, suplemento de segurança alimentar"
    if a == 2018:
        return "pof", "IBGE, POF 2017-2018 (coleta de julho de 2017 a julho de 2018)"
    if a >= 2023:
        return "pnadc", "IBGE, PNAD Contínua, 4º trimestre"
    raise RuntimeError(f"edição da EBIA não classificada: {ano}")


def tratar_ebia(saida):
    ods, arq = ler_sidra("ods")
    pnadc, arq_c = ler_sidra("pnadc")
    p2004, arq_04 = ler_sidra("pnad2004")
    arqs_meta = [bruto(FONTE_SIDRA, f"t{t}_{r}") for t in TABELAS_EBIA for r in ("metadados", "periodos")]
    anos = sorted({a for (_, a, _) in ods})
    f_id = "inseguranca_alimentar"
    series = defaultdict(dict)
    fontes = {}
    for ano in anos:
        chave, fonte = pesquisa_ebia(ano)
        mod, grave = ods.get(("800", ano, CAT_MODERADA)), ods.get(("800", ano, CAT_GRAVE))
        series[f"moderada_ou_grave_{chave}"][ano] = mod + grave
        series[f"grave_{chave}"][ano] = grave
        fontes[ano] = fonte
    rotulos = {"moderada_ou_grave": "domicílios em insegurança alimentar moderada ou grave (soma das duas categorias, "
                                    "cada uma com uma casa decimal)",
               "grave": "domicílios em insegurança alimentar grave"}
    for serie, dados in series.items():
        tipo = "moderada_ou_grave" if serie.startswith("moderada") else "grave"
        saida.add(f_id, f"{f_id}__{serie}", dados, "% dos domicílios particulares",
                  f"IBGE, SIDRA 6665 (ODS 2.1.2), EBIA; {rotulos[tipo]}", [arq] + arqs_meta, 1,
                  fonte_por_data={a: f"{fontes[a]}; SIDRA 6665 (ODS 2.1.2), EBIA de 14 itens; {rotulos[tipo]}" for a in dados})
    # conferências: PNAD Contínua em números absolutos (9552) e PNAD 2004 como publicada em 2006 (3017)
    conf = []
    for ano in sorted({a for (_, a, _) in pnadc}):
        tot = pnadc.get(("162", ano, "109106"))
        mg = (pnadc[("162", ano, CAT_MODERADA)] + pnadc[("162", ano, CAT_GRAVE)]) / tot * 100
        conf.append(f"{ano}: {br(mg, 2)}% pelos domicílios em mil da tabela 9552 e {br(series['moderada_ou_grave_pnadc'][ano], 1)}% "
                    "pela soma da tabela 6665")
        if abs(mg - series["moderada_ou_grave_pnadc"][ano]) > 0.15:
            raise RuntimeError(f"EBIA {ano}: tabelas 9552 e 6665 não batem")
    original = p2004[("1000800", "2004", CAT_MODERADA)] + p2004[("1000800", "2004", CAT_GRAVE)]
    saida.conferiu("insegurança alimentar moderada ou grave: " + "; ".join(conf)
                   + f"; PNAD 2004 publicada em 2006 (15 itens, SIDRA 3017): {br(original, 2)}%")
    moradores = {a: ods.get(("2133", a, CAT_MODERADA), 0) + ods.get(("2133", a, CAT_GRAVE), 0) for a in anos}
    saida.notas[f_id].append(
        "Conferência: " + "; ".join(conf) + f". A PNAD 2004 como publicada em 2006, com a escala de 15 itens (SIDRA 3017, "
        f"`{arq_04.relative_to(RAIZ)}`), dava {br(p2004[('1000800', '2004', CAT_MODERADA)], 2)}% de domicílios em insegurança "
        f"moderada e {br(p2004[('1000800', '2004', CAT_GRAVE)], 2)}% em grave ({br(original, 2)}% somadas), contra "
        f"{br(ods[('800', '2004', CAT_MODERADA)], 1)}% e {br(ods[('800', '2004', CAT_GRAVE)], 1)}% na tabela 6665, que "
        "reprocessa as edições da PNAD com a escala de 14 itens.")
    saida.notas[f_id].append(
        "Moradores em domicílios com insegurança moderada ou grave (% dos moradores, mesma tabela 6665, variável 2133): "
        + "; ".join(f"{a}: {br(v, 1)}" for a, v in moradores.items()) + f". Arquivos de conferência: `{arq_c.relative_to(RAIZ)}`. "
        "Metadados e datas de atualização das tabelas: " + ", ".join(f"`{a.relative_to(RAIZ)}`" for a in arqs_meta) + ".")
    return series


# ================================================================ comparação com pares (Banco Mundial, FAO)

FONTE_BM = "bm_wdi"
INDICADORES_BM = {"bm_subalimentacao": "SN.ITK.DEFC.ZS", "bm_inseguranca_alimentar": "SN.ITK.MSFI.ZS"}
NOME_BM = "wdi_seguranca_alimentar"
PAUSA_BM = 15  # segundos entre consultas ao Banco Mundial (o firewall bloqueia rajadas)


def ler_grupos():
    grupos = defaultdict(list)
    with open(METODOLOGIA / "pares.csv", encoding="utf-8") as f:
        for linha in csv.DictReader(f):
            grupos[linha["grupo"]].append(linha["iso3"])
    return grupos


def paises_bm():
    return ["BRA"] + sorted({p for ps in ler_grupos().values() for p in ps})


def url_wdi():
    return (f"https://api.worldbank.org/v2/country/{';'.join(paises_bm())}/indicator/"
            f"{';'.join(INDICADORES_BM.values())}?source=2&format=json&per_page=20000")


def json_bom(conteudo):
    return json.loads(conteudo.decode("utf-8-sig"))


def coletar_bm(args):
    def validar(c):
        d = json_bom(c)
        if not (isinstance(d, list) and len(d) == 2 and isinstance(d[1], list) and d[0].get("total") == len(d[1])):
            raise RuntimeError(f"resposta inesperada do Banco Mundial: {c[:300]!r}")

    def validar_meta(c):
        d = json_bom(c)
        if not (isinstance(d, list) and len(d) == 2 and d[1]):
            raise RuntimeError(f"metadado inesperado do Banco Mundial: {c[:300]!r}")
    pedidos = [(NOME_BM, url_wdi(), validar)] + [
        (f"wdi_metadados_{c.replace('.', '-')}", f"https://api.worldbank.org/v2/indicator/{c}?format=json", validar_meta)
        for c in INDICADORES_BM.values()]
    for i, (nome, url, val) in enumerate(pedidos):
        existentes = brutos_exatos(FONTE_BM, nome)
        if i and not (existentes and HOJE.isoformat() in existentes[-1].name):
            time.sleep(PAUSA_BM)
        obter(FONTE_BM, nome, "json", url, validar=val)


# conferência: banco de dados global dos ODS da ONU (custodiante: FAO), Brasil
FONTE_ONU = "onu_sdg"
SERIES_ONU = {"bm_subalimentacao": "SN_ITK_DEFC", "bm_inseguranca_alimentar": "AG_PRD_FIESMS"}
URL_ONU = "https://unstats.un.org/sdgapi/v1/sdg/Series/Data?seriesCode={serie}&areaCode=76&pageSize=500"


def coletar_onu(args):
    def validar(c):
        if "data" not in json.loads(c):
            raise RuntimeError("ONU/SDG: resposta inesperada")
    for serie in SERIES_ONU.values():
        obter(FONTE_ONU, f"sdg_{serie}_brasil", "json", URL_ONU.format(serie=serie), validar=validar)


def onu_brasil(serie):
    """{ano: valor} da série no banco dos ODS da ONU (todas as idades e ambos os sexos), e o arquivo."""
    arq = bruto(FONTE_ONU, f"sdg_{serie}_brasil")
    saida = {}
    for x in json.loads(arq.read_text(encoding="utf-8"))["data"]:
        dims = x.get("dimensions", {})
        if dims.get("Age") not in (None, "ALLAGE") or dims.get("Sex") not in (None, "BOTHSEX"):
            continue
        valor = str(x.get("value", "")).strip()
        if valor and valor.lower() != "nan":
            saida[str(int(float(x["timePeriodStart"])))] = valor
    return saida, arq


def tratar_pares(saida):
    arq = bruto(FONTE_BM, NOME_BM)
    cab, registros = json_bom(arq.read_bytes())
    arqs_meta = [bruto(FONTE_BM, f"wdi_metadados_{c.replace('.', '-')}") for c in INDICADORES_BM.values()]
    por_ind = defaultdict(lambda: defaultdict(dict))
    nomes = {}
    for r in registros:
        nomes[r["indicator"]["id"]] = r["indicator"]["value"]
        if r["value"] is not None and "2000" <= r["date"] < str(ANO_CORRENTE):
            por_ind[r["indicator"]["id"]][r["countryiso3code"]][r["date"]] = float(r["value"])
    grupos = ler_grupos()
    resultado = {}
    for ficha_id, codigo in INDICADORES_BM.items():
        por_pais = {p: obs for p, obs in por_ind[codigo].items() if obs}
        fonte = f"Banco Mundial, WDI {codigo} (FAO), média de três anos centrada no ano (WDI atualizado em {cab.get('lastupdated')})"
        brasil = por_pais.get("BRA", {})
        saida.add(ficha_id, f"{ficha_id}__brasil", brasil, "% da população", fonte, [arq], 1)
        for grupo, membros in grupos.items():
            med, n = mediana_por_ano(por_pais, membros)  # D-022: só com pelo menos metade do grupo
            saida.add(ficha_id, f"{ficha_id}__mediana_{grupo}", med, "% da população", f"{fonte}, mediana de {grupo}",
                      [arq], 2)
            dif = {a: brasil[a] - med[a] for a in brasil if a in med}
            saida.add(ficha_id, f"{ficha_id}__brasil_menos_mediana_{grupo}", dif, "p.p.",
                      f"{fonte}, Brasil menos mediana de {grupo}", [arq], 2)
            for a in sorted(n):
                saida.paises.append([f"{ficha_id}__n_{grupo}", grupo, a, n[a], "países", fonte])
        for pais, obs in sorted(por_pais.items()):
            for a, v in sorted(obs.items()):
                saida.paises.append([ficha_id, pais, a, round(v, 4), "% da população", fonte])
        sem_dado = [p for p in paises_bm() if p not in por_pais]
        piso = sorted({p for p, obs in por_pais.items() if any(v == 2.5 for v in obs.values())})
        saida.notas[ficha_id].append(
            f"Indicador na fonte: {nomes.get(codigo, '?')} (WDI atualizado em {cab.get('lastupdated')}; definição em "
            + ", ".join(f"`{a.relative_to(RAIZ)}`" for a in arqs_meta if codigo.replace('.', '-') in a.name) + "). "
            f"Países sem nenhum dado de 2000 a {ANO_CORRENTE - 1}: {', '.join(sem_dado) or 'nenhum'}. "
            + (f"Países com algum ano igual a 2,5 (a FAO publica \"<2,5\" e o WDI grava 2,5): {', '.join(piso)}. " if piso else "")
            + f"Valores por país e número de países de cada mediana (`{ficha_id}__n_<grupo>`) em "
            f"`dados/tratados/{TEMA}_pares_paises.csv`.")
        onu, arq_onu = onu_brasil(SERIES_ONU[ficha_id])
        iguais, difs = [], []
        for a, v in sorted(brasil.items()):
            o = onu.get(a)
            if o is None:
                continue
            o_num = 2.5 if o.startswith("<") else float(o)
            (iguais if abs(o_num - v) < 0.05 else difs).append(
                a if abs(o_num - v) < 0.05 else f"{a}: WDI {br(v, 1)}, ONU {o.replace('.', ',')}")
        novos = {a: v for a, v in onu.items() if a not in brasil}
        texto = (f"{ficha_id}, Brasil no WDI x banco dos ODS da ONU ({SERIES_ONU[ficha_id]}, custodiante FAO): iguais em "
                 f"{iguais[0]}–{iguais[-1]} ({len(iguais)} anos)" + (f"; diferentes: {'; '.join(difs)}" if difs else "")
                 + (f"; só no banco da ONU: {'; '.join(f'{a}: ' + v.replace('.', ',') for a, v in sorted(novos.items()))}"
                    if novos else ""))
        saida.conferiu(texto)
        saida.notas[ficha_id].append(
            f"Conferência: {texto} (`{arq_onu.relative_to(RAIZ)}`). O banco da ONU já traz a revisão seguinte da FAO; o "
            "projeto usa o WDI, como no pré-registro, e registra a diferença.")
        resultado[ficha_id] = por_pais
    return resultado


# ================================================================ habitação: Minha Casa, Minha Vida (Ministério das Cidades)

FONTE_MCIDADES = "mcidades"
PAGINA_MCMV = ("https://www.gov.br/cidades/pt-br/acesso-a-informacao/acoes-e-programas/habitacao/"
               "programa-minha-casa-minha-vida/bases-de-dados-do-programa-minha-casa-minha-vida")
ARQUIVOS_MCMV = {  # nome do bruto -> padrão do link na página (o arquivo analítico, com dados pessoais, fica fora)
    "mcmv_subsidiado": r"mcmv_subsidiado_\d+\.zip",
    "mcmv_financiado_sintetico": r"mcmv_financ_sintetico_[\w.]+\.zip",
}
ANO_INICIAL_MCMV = 2009  # Lei 11.977/2009 (MP 459, de 25/03/2009)
FAIXAS_FINANCIADO = ("1", "2", "3", "4")


def programa_habitacao(ano):
    a = int(ano)
    if a <= 2019:
        return "Minha Casa, Minha Vida (Lei 11.977/2009)"
    if a == 2020:
        return "Minha Casa, Minha Vida (até 25/08) e Casa Verde e Amarela (MP 996/2020, desde 26/08)"
    if a in (2021, 2022):
        return "Casa Verde e Amarela (Lei 14.118/2021)"
    if a == 2023:
        return "Casa Verde e Amarela (até 13/02) e Minha Casa, Minha Vida (MP 1.162/2023; Lei 14.620/2023)"
    return "Minha Casa, Minha Vida (Lei 14.620/2023)"


def links_mcmv():
    pagina = pedir(PAGINA_MCMV).decode("utf-8", "replace")
    links = {}
    for nome, padrao in ARQUIVOS_MCMV.items():
        achados = sorted(set(re.findall(r'href="(https?://[^"]*?' + padrao + r')"', pagina)))
        if len(achados) != 1:
            raise RuntimeError(f"página do MCMV: {len(achados)} links para {nome}: {achados}")
        links[nome] = achados[0]
    return links


def coletar_mcmv(args):
    for nome, url in links_mcmv().items():
        obter(FONTE_MCIDADES, nome, "zip", url, validar=validar_zip, mesma_url=not args.rebaixar,
              baixar_fn=lambda u=url: pedir(u, timeout=900))


def ler_csv_zip(arq):
    with zipfile.ZipFile(arq) as z:
        (membro,) = [n for n in z.namelist() if n.lower().endswith(".csv")]
        texto = z.read(membro).decode("utf-8-sig")
    return list(csv.DictReader(io.StringIO(texto), delimiter=";")), membro


def inteiro(x):
    """'1.234' ou '1234' -> 1234 (quantidades de unidades habitacionais)."""
    s = (x or "").strip().replace(".", "")
    return int(s.split(",")[0]) if s else 0


def tratar_mcmv(saida):
    arq_s = bruto(FONTE_MCIDADES, "mcmv_subsidiado")
    arq_f = bruto(FONTE_MCIDADES, "mcmv_financiado_sintetico")
    sub, membro_s = ler_csv_zip(arq_s)
    fin, membro_f = ler_csv_zip(arq_f)
    ref_s = sorted({l["data_referencia"] for l in sub})
    ref_f = sorted({l["data_referencia"] for l in fin})
    subsidiado, modalidades = defaultdict(float), defaultdict(lambda: defaultdict(float))
    distratadas, entregues = defaultdict(float), defaultdict(float)
    for l in sub:
        ano = l["dt_assinatura"].strip()[-4:]
        if not re.fullmatch(r"\d{4}", ano):
            raise RuntimeError(f"{membro_s}: data de contratação inválida {l['dt_assinatura']!r}")
        q = inteiro(l["qtd_uh"])
        subsidiado[ano] += q
        modalidades[ano][limpo(l["txt_modalidade"]).upper()] += q
        distratadas[ano] += inteiro(l["qtd_uh_distratadas"])
        entregues[ano] += inteiro(l["qtd_uh_entregues"])
    financiado = defaultdict(lambda: defaultdict(float))
    for l in fin:
        faixa = limpo(l["txt_compatibilidade_faixa_renda"])
        if faixa not in FAIXAS_FINANCIADO + ("",):
            raise RuntimeError(f"{membro_f}: faixa de renda desconhecida {faixa!r}")
        financiado[l["num_ano"]][faixa] += float(l["qtd_uh_financiadas"] or 0)
    anos = [str(a) for a in range(ANO_INICIAL_MCMV, ANO_CORRENTE)]
    if min(subsidiado) < str(ANO_INICIAL_MCMV) or min(financiado) < str(ANO_INICIAL_MCMV):
        raise RuntimeError("MCMV: contratos anteriores a 2009")
    f_id = "habitacao_unidades"
    fonte_s = (f"Ministério das Cidades, MCMV Subsidiado (OGU), empreendimentos, unidades contratadas por ano de "
               f"assinatura (dados de {', '.join(ref_s)})")
    fonte_f = (f"Ministério das Cidades, MCMV Financiado (FGTS), contratos sintéticos por município, unidades "
               f"financiadas por ano de contratação e faixa de renda compatível (dados de {', '.join(ref_f)})")
    prog = {a: programa_habitacao(a) for a in anos}
    serie_sub = {a: subsidiado.get(a, 0.0) for a in anos}
    saida.add(f_id, f"{f_id}__subsidiado_faixa_1", serie_sub, "unidades contratadas", fonte_s, [arq_s], 0,
              fonte_por_data={a: f"{fonte_s}; {prog[a]}" for a in anos})
    for faixa in FAIXAS_FINANCIADO:
        dados = {a: financiado[a].get(faixa, 0.0) for a in anos if any(financiado[x].get(faixa) for x in anos if x <= a)}
        saida.add(f_id, f"{f_id}__financiado_faixa_{faixa}", dados, "unidades contratadas", fonte_f + f"; faixa {faixa}",
                  [arq_f], 0, fonte_por_data={a: f"{fonte_f}; faixa {faixa}; {prog[a]}" for a in dados})
    fin_total = {a: sum(financiado[a].get(fx, 0.0) for fx in FAIXAS_FINANCIADO) for a in anos}
    saida.add(f_id, f"{f_id}__financiado_total", fin_total, "unidades contratadas",
              fonte_f + "; faixas 1 a 4 (sem os contratos fora das faixas do programa)", [arq_f], 0,
              fonte_por_data={a: f"{fonte_f}; faixas 1 a 4; {prog[a]}" for a in anos})
    total = {a: serie_sub[a] + fin_total[a] for a in anos}
    saida.add(f_id, f"{f_id}__total", total, "unidades contratadas",
              "Ministério das Cidades, MCMV Subsidiado (OGU) + MCMV Financiado (FGTS, faixas 1 a 4)", [arq_s, arq_f], 0,
              fonte_por_data={a: f"Ministério das Cidades, MCMV Subsidiado (OGU) + Financiado (FGTS, faixas 1 a 4); {prog[a]}"
                              for a in anos})
    corrente = str(ANO_CORRENTE)
    saida.notas[f_id].append(
        f"Arquivos: `{arq_s.relative_to(RAIZ)}` (membro `{membro_s}`, {br(len(sub))} empreendimentos) e "
        f"`{arq_f.relative_to(RAIZ)}` (membro `{membro_f}`, {br(len(fin))} linhas por município, mês e faixa). O ano corrente "
        f"({corrente}) fica fora: até a data dos dados, {br(subsidiado.get(corrente, 0))} unidades subsidiadas e "
        f"{br(sum(financiado[corrente].get(fx, 0) for fx in FAIXAS_FINANCIADO))} financiadas nas faixas 1 a 4.")
    saida.notas[f_id].append(
        "Unidades subsidiadas por modalidade e ano de contratação: " + "; ".join(
            f"{a}: " + ", ".join(f"{m} {br(q)}" for m, q in sorted(modalidades[a].items())) for a in anos if modalidades[a])
        + ". Anos sem nenhuma contratação subsidiada no arquivo: " + ", ".join(a for a in anos if not subsidiado.get(a)) + ".")
    saida.notas[f_id].append(
        "Das unidades subsidiadas contratadas em cada ano, já distratadas (canceladas) / já entregues na data dos dados: "
        + "; ".join(f"{a}: {br(distratadas[a])} / {br(entregues[a])}" for a in anos if subsidiado.get(a)) + ".")
    saida.notas[f_id].append(
        "Contratos do FGTS fora das faixas do programa (faixa em branco no arquivo; sem subsídio, financiamento médio "
        "acima do teto da faixa 3), fora das séries: " + "; ".join(f"{a}: {br(financiado[a].get('', 0))}" for a in anos) + ".")
    return total, serie_sub, financiado


# ================================================================ déficit habitacional (Fundação João Pinheiro)

FONTE_FJP = "fjp"
PAGINA_FJP = "https://fjp.mg.gov.br/deficit-habitacional-no-brasil/"
# Série temporal 2007-2015 (base de dados da FJP, versão alterada em 19/10/2018), cópia do Wayback Machine de 08/02/2019
URL_FJP_2007_2015 = ("http://web.archive.org/web/20190208132731id_/http://fjp.mg.gov.br/index.php/docman/direi-2018/"
                     "861-serie-temporaldeficit-habitacional-2007-2015-2610/file")
URL_FJP_2016_2019_ORIGINAL = "https://fjp.mg.gov.br/wp-content/uploads/2020/12/04.03_Dados-Deficit-Habitacional-2016-2019.xlsx"
# Planilhas de estimativas no Google Drive da FJP (links da página do déficit habitacional, 6/10/2026)
FJP_DRIVE_PADRAO = {"2016_2019_reponderado": "1y0c_oDcNpcq29fJyPAA4jtjDTAzij2MW",
                    "2022": "1OqAuggLve9HoPrI8jJcaAfSu1q5DCvdc",
                    "2023": "1lQzsAuzhRCEJufiIEI_tVzO4vllcBT38",
                    "2024": "1AwlL17JBmASv8Jurnkb6zhfXc0fzGQwp"}
URL_DRIVE = "https://drive.google.com/uc?export=download&id={id}"
FJP_RELATORIO_2022 = "13FL-MVsULmFMjkQb1nQsYzO2JVhLQvwz"  # relatório de 2022, com a tabela 13 (2016-2022)
FJP_TABELA13_2022 = {"2016": 5768482, "2017": 6068566, "2018": 5973218, "2019": 5964993, "2022": 6215313}


def drive_fjp():
    """{edição: id no Google Drive} das planilhas de estimativas do déficit, lidas na página da FJP; se a página
    falhar ou mudar, valem os ids conhecidos."""
    ids = dict(FJP_DRIVE_PADRAO)
    try:
        pagina = pedir(PAGINA_FJP, tentativas=2).decode("utf-8", "replace")
    except Exception as e:  # a página só serve para achar edições novas
        print(f"  ! página da FJP indisponível ({e}); usando os links conhecidos", file=sys.stderr)
        return ids
    for m in re.finditer(r'(?is)<a[^>]+href="https://docs\.google\.com/spreadsheets/d/([\w-]+)[^"]*"[^>]*>(.*?)</a>', pagina):
        rotulo = limpo(re.sub(r"<[^>]+>", "", m.group(2)))
        if "Estimativas" not in rotulo or "Déficit" not in rotulo or "Inadequação" in rotulo:
            continue
        if "REPONDERADOS" in rotulo.upper():
            ids["2016_2019_reponderado"] = m.group(1)
        else:
            ano = re.search(r"(20\d\d)\)?\s*$", rotulo)
            if ano and int(ano.group(1)) >= 2022:
                ids[ano.group(1)] = m.group(1)
    return ids


def coletar_fjp(args):
    obter(FONTE_FJP, "serie_temporal_deficit_2007_2015", "xlsx", URL_FJP_2007_2015, validar=validar_zip,
          reaproveitar=not args.rebaixar)
    obter(FONTE_FJP, "deficit_2016_2019_original", "xlsx", URL_FJP_2016_2019_ORIGINAL, validar=validar_zip,
          reaproveitar=not args.rebaixar)
    for edicao, ident in sorted(drive_fjp().items()):
        obter(FONTE_FJP, f"deficit_{edicao}", "xlsx", URL_DRIVE.format(id=ident), validar=validar_zip,
              reaproveitar=not args.rebaixar)
    obter(FONTE_FJP, "relatorio_deficit_2022", "pdf", URL_DRIVE.format(id=FJP_RELATORIO_2022), validar=validar_pdf,
          reaproveitar=not args.rebaixar)


def fjp_2007_2015(arq):
    """{ano: (total, relativo em %)} da linha Brasil da aba 1 da série temporal."""
    linhas = Xlsx(str(arq)).linhas("1 - Déficit Habitacional")
    i_anos = next(i for i, l in enumerate(linhas) if l and limpo(l[0]) == "Especificação")
    anos = {c: limpo(x) for c, x in enumerate(linhas[i_anos]) if re.fullmatch(r"\d{4}", limpo(x))}
    sub = linhas[i_anos + 1]
    brasil = next(l for l in linhas if l and limpo(l[0]) == "Brasil")
    saida = {}
    for c, ano in anos.items():
        if [limpo(sub[c + k]) for k in range(4)] != ["Total", "Urbano", "Rural", "Total Relativo"]:
            raise RuntimeError(f"{arq.name}: colunas inesperadas em {ano}")
        total, urb, rur, rel = (num(brasil[c + k]) for k in range(4))
        if abs(urb + rur - total) > 1:
            raise RuntimeError(f"{arq.name}: urbano + rural diferente do total em {ano}")
        saida[ano] = (total, rel)
    return saida


def fjp_2016_2019(arq, reponderado):
    linhas = Xlsx(str(arq)).linhas("Déficit")
    cab = [limpo(x) for x in linhas[0]]
    i_ano, i_reg = (cab.index("Ano"), cab.index("regiao" if reponderado else "Região"))
    i_tot = cab.index("Total")
    i_rel = next(i for i, c in enumerate(cab) if c.startswith("% do total de domicílios"))
    saida = {}
    for l in linhas[1:]:
        if len(l) > i_rel and limpo(l[i_reg]) == "Brasil":
            saida[str(int(num(l[i_ano])))] = (num(l[i_tot]), 100 * num(l[i_rel]))
    if sorted(saida) != ["2016", "2017", "2018", "2019"]:
        raise RuntimeError(f"{arq.name}: anos inesperados {sorted(saida)}")
    return saida


def fjp_edicao(arq):
    """(ano, total, relativo em %, improvisados, habitação precária) de uma planilha de 2022 em diante."""
    livro = Xlsx(str(arq))
    linhas = livro.linhas("Deficit_habitacional")
    cab = [limpo(x) for x in linhas[0]]
    i_tot = cab.index("Déficit Habitacional")
    i_rel = next(i for i, c in enumerate(cab) if c.startswith("Déficit Habitacional Rela"))
    (l,) = [x for x in linhas[1:] if len(x) > i_rel and limpo(x[1]) == "Brasil"]
    comp = livro.linhas("DH_Componentes")
    (c,) = [x for x in comp[1:] if len(x) > 4 and limpo(x[1]) == "Brasil"]
    return limpo(l[0]), num(l[i_tot]), 100 * num(l[i_rel]), num(c[2]), num(c[4])


def tratar_fjp(saida):
    arq_0715 = bruto(FONTE_FJP, "serie_temporal_deficit_2007_2015")
    arq_orig = bruto(FONTE_FJP, "deficit_2016_2019_original")
    arq_rep = bruto(FONTE_FJP, "deficit_2016_2019_reponderado")
    arq_rel22 = bruto(FONTE_FJP, "relatorio_deficit_2022")
    edicoes = sorted({p.stem.rsplit("_", 1)[0] for p in (BRUTOS / FONTE_FJP).glob("deficit_20[2-9]*")
                      if re.fullmatch(r"deficit_20\d\d", p.stem.rsplit("_", 1)[0])})
    f_id = "deficit_habitacional"
    antiga = fjp_2007_2015(arq_0715)
    fonte_0715 = ("FJP, Série temporal do déficit habitacional 2007-2015 (base de dados publicada em fjp.mg.gov.br, versão "
                  "de 19/10/2018; cópia do Wayback Machine de 08/02/2019)")
    pnad = {a: v[0] for a, v in antiga.items() if a != "2010"}
    saida.add(f_id, f"{f_id}__pnad_2007_2015", pnad, "domicílios", fonte_0715 + "; PNAD anual (IBGE)", [arq_0715], 0)
    saida.add(f_id, f"{f_id}__pnad_2007_2015_relativo", {a: v[1] for a, v in antiga.items() if a != "2010"},
              "% dos domicílios particulares permanentes e improvisados", fonte_0715 + "; PNAD anual (IBGE)", [arq_0715], 1)
    saida.add(f_id, f"{f_id}__censo_2010", {"2010": antiga["2010"][0]}, "domicílios",
              fonte_0715 + "; Censo Demográfico 2010 (IBGE)", [arq_0715], 0)
    saida.add(f_id, f"{f_id}__censo_2010_relativo", {"2010": antiga["2010"][1]},
              "% dos domicílios particulares permanentes e improvisados", fonte_0715 + "; Censo Demográfico 2010 (IBGE)",
              [arq_0715], 1)
    rep = fjp_2016_2019(arq_rep, True)
    orig = fjp_2016_2019(arq_orig, False)
    nova = dict(rep)
    fonte_ano = {a: "FJP, Estimativas do Déficit Habitacional no Brasil 2016-2019, dados reponderados (PNAD Contínua "
                    "e CadÚnico)" for a in rep}
    arqs_nova = [arq_rep]
    componentes = {}
    for nome in edicoes:
        arq = bruto(FONTE_FJP, nome)
        ano, tot, rel, improv, precaria = fjp_edicao(arq)
        if ano != nome[-4:]:
            raise RuntimeError(f"{arq.name}: planilha da edição {nome[-4:]} traz o ano {ano}")
        nova[ano] = (tot, rel)
        componentes[ano] = (improv, precaria)
        fonte_ano[ano] = f"FJP, Estimativas do Déficit Habitacional no Brasil {ano} (PNAD Contínua e CadÚnico)"
        arqs_nova.append(arq)
    saida.add(f_id, f"{f_id}__pnadc", {a: v[0] for a, v in nova.items()}, "domicílios",
              "FJP, metodologia de 2016 em diante (PNAD Contínua e CadÚnico)", arqs_nova, 0, fonte_por_data=fonte_ano)
    saida.add(f_id, f"{f_id}__pnadc_relativo", {a: v[1] for a, v in nova.items()},
              "% dos domicílios particulares permanentes e improvisados",
              "FJP, metodologia de 2016 em diante (PNAD Contínua e CadÚnico)", arqs_nova, 1, fonte_por_data=fonte_ano)
    difs = {a: nova[a][0] - v for a, v in FJP_TABELA13_2022.items()}
    if max(abs(d) for d in difs.values()) > 1:
        raise RuntimeError(f"déficit difere da tabela 13 do relatório de 2022 da FJP: {difs}")
    saida.conferiu("déficit habitacional 2016-2019 (reponderado) e 2022 iguais à tabela 13 do relatório de 2022 da FJP")
    saida.notas[f_id].append(
        "Conferência: os totais de 2016 a 2019 (dados reponderados) e de 2022 são iguais aos da tabela 13 do relatório "
        f"\"Déficit Habitacional no Brasil 2022\" (`{arq_rel22.relative_to(RAIZ)}`): "
        + "; ".join(f"{a}: {br(v)}" for a, v in FJP_TABELA13_2022.items()) + ". Totais de 2016 a 2019 na primeira "
        f"divulgação (pesos originais da PNAD Contínua, `{arq_orig.relative_to(RAIZ)}`), fora das séries: "
        + "; ".join(f"{a}: {br(v[0])} (o reponderado é {br(100 * (rep[a][0] / v[0] - 1), 1)}% maior)" for a, v in orig.items()) + ".")
    saida.notas[f_id].append(
        "Domicílios improvisados (do CadÚnico) e habitação precária na metodologia nova, por edição: "
        + "; ".join(f"{a}: {br(i)} improvisados de {br(p)} em habitação precária" for a, (i, p) in sorted(componentes.items()))
        + ". O cálculo dos improvisados mudou em 2022 (CadÚnico de 2021 sem as famílias unipessoais cadastradas depois de "
        "novembro de 2021), em 2023 (Nota Técnica FJP nº 10/2023) e em 2024 (correção dos municípios com proporção extrema, "
        "Nota Técnica FJP nº 3/2026).")
    return pnad, nova


# ================================================================ conferências do MI Social

# Números publicados pelo MDS (ou pelo antigo Ministério da Cidadania): (grupo, mês, chave, valor, tolerância, documento)
PUBLICADO_MDS = [
    ("transferencia_renda_mensal", "2023-05", "bf_familias", 21249015, 0, "21.249.015 famílias",
     "Informe Bolsa Família nº 8, de 23/05/2023, https://www.mds.gov.br/webarquivos/MDS/2_Acoes_e_Programas/Bolsa_Familia/"
     "Informes/2023/Informe_Bolsa_Familia_N_8.pdf"),
    ("transferencia_renda_mensal", "2023-05", "bf_valor", 14100224092.00, 100, "R$ 14.100.224.092,00 repassados",
     "mesmo informe"),
    ("transferencia_renda_mensal", "2020-06", "bf_so_ae_familias", 13633180, 0,
     "13.633.180 famílias do PBF recebendo o Auxílio Emergencial",
     "Boletim BF Informa nº 718, julho de 2020, https://www.mds.gov.br/webarquivos/sala_de_imprensa/boletins/"
     "boletim_bolsa_familia/2020/julho/boletim_BFInforma718.html"),
    ("transferencia_renda_mensal", "2021-06", "bf_familias", 14.7e6, 0.05e6, "aproximadamente 14,7 milhões de famílias",
     "Boletim Bolsa Família e Cadastro Único nº 73, julho de 2021 (folha de junho)"),
    ("cadunico_mensal", "2021-05", "familias", 30.1e6, 0.05e6, "30,1 milhões de famílias",
     "mesmo boletim, cadastro extraído em 15/05/2021"),
    ("cadunico_mensal", "2021-05", "pessoas", 77.4e6, 0.05e6, "77,4 milhões de pessoas", "mesmo boletim"),
    ("auxilio_emergencial_mensal", "2021-04", "ae2021_pessoas_pbf", 10013426, 0,
     "10.013.426 pessoas do público do Bolsa Família no Auxílio Emergencial 2021",
     "MDS, notícia sobre a 5ª parcela do Auxílio Emergencial 2021 em gov.br/mds; sem acesso direto em 06/10/2026, "
     "número lido no resumo do buscador"),
]


def conferir_misocial(saida):
    """(1) soma do servidor x soma local dos municípios nos meses de amostra; (2) números publicados pelo MDS."""
    tr, _ = ler_faceta("transferencia_renda_mensal")
    bpc, _ = ler_faceta("bpc_mensal")
    cad, _ = ler_faceta("cadunico_mensal")
    pares = {"qtd_familias_beneficiarias_bolsa_familia_i": (tr, "bf_familias"),
             "valor_repassado_bolsa_familia_s": (tr, "bf_valor"), "pab_qtd_fam_benef_i": (tr, "ab_familias"),
             "pab_valor_pago_d": (tr, "ab_valor"), "pab_extraordinario_valor_pago_d": (tr, "ab_extraordinario_valor"),
             "bpc_idoso_ben_i": (bpc, "bpc_idoso"), "bpc_pcd_ben_i": (bpc, "bpc_pcd"),
             "cadun_qtd_familias_cadastradas_i": (cad, "familias"), "cadun_qtd_pessoas_cadastradas_i": (cad, "pessoas")}
    pior_contagem, pior_valor, arquivos = 0.0, 0.0, []
    for mes in MESES_AMOSTRA:
        somas, n, arq = ler_amostra(mes)
        arquivos.append(arq)
        m = f"{mes[:4]}-{mes[4:]}"
        for campo, (faceta, chave) in pares.items():
            d = abs(somas[campo] - faceta[m][chave])
            if campo.startswith(("valor", "pab_valor", "pab_extra")):
                pior_valor = max(pior_valor, d)
            else:
                pior_contagem = max(pior_contagem, d)
    texto = (f"MI Social, soma local dos municípios ({', '.join(mes_pt(f'{m[:4]}-{m[4:]}') for m in MESES_AMOSTRA)}) x soma "
             f"do servidor: maior diferença de {br(pior_contagem)} em contagens e de R$ {br(pior_valor, 2)} em valores")
    saida.conferiu(texto)
    if pior_contagem > 0 or pior_valor > 1000:
        raise RuntimeError("MI Social: soma do servidor difere da soma dos municípios")
    grupos = {"transferencia_renda_mensal": tr, "cadunico_mensal": cad,
              "auxilio_emergencial_mensal": ler_faceta("auxilio_emergencial_mensal")[0]}
    linhas = []
    for grupo, mes, chave, valor, tol, publicado, doc in PUBLICADO_MDS:
        obtido = grupos[grupo][mes][chave]
        ok = abs(obtido - valor) <= tol
        linhas.append(f"{mes_pt(mes)}: {publicado} ({doc}); MI Social: {br(obtido, 2 if 'valor' in chave else 0)}"
                      + ("" if ok else " [DIFERENTE]"))
        if not ok:
            print(f"  ! conferência com o MDS: {linhas[-1]}", file=sys.stderr)
    ab = tr["2021-11"]
    linhas.append("nov/2021: benefício médio do Auxílio Brasil de R$ 224,41 (Ministério da Cidadania, página de "
                  "esclarecimentos sobre o Auxílio Brasil em gov.br/cidadania; sem acesso direto em 06/10/2026, número "
                  f"lido no resumo do buscador); MI Social: R$ {br(ab['ab_valor'] / ab['ab_familias'], 2)}")
    for l in linhas:
        saida.conferiu(l)
    nota = (texto + f" (amostras: {', '.join(f'`{a.relative_to(RAIZ)}`' for a in arquivos)}). Números publicados pelo MDS: "
            + "; ".join(linhas) + ".")
    for ficha in ("transferencia_renda_familias", "cadunico"):
        saida.notas[ficha].append(nota)


# ================================================================ orquestração

def etapas():
    return [
        ("MDS, MI Social: Bolsa Família, Auxílio Brasil, BPC, Cadastro Único e Auxílio Emergencial", coletar_misocial),
        ("Ipea, boletim Políticas Sociais nº 13 (programas anteriores ao Bolsa Família)", coletar_programas_anteriores),
        ("MDS, Nota Técnica 03/2016 (BPC 1996-2015)", coletar_nota_bpc),
        ("Despesa federal por ação orçamentária (SIOP, SPARQL)", coletar_despesa),
        ("PIB nominal (IBGE, SIDRA 1846)", coletar_pib),
        ("Resultado do Tesouro Nacional (conferência da despesa e do Auxílio Emergencial)", coletar_rtn),
        ("Insegurança alimentar (IBGE, SIDRA)", coletar_ebia),
        ("Pares: subalimentação e insegurança alimentar (Banco Mundial, FAO)", coletar_bm),
        ("Conferência dos pares: banco de dados dos ODS da ONU", coletar_onu),
        ("Minha Casa, Minha Vida (Ministério das Cidades)", coletar_mcmv),
        ("Déficit habitacional (Fundação João Pinheiro)", coletar_fjp),
    ]


def coletar(args):
    falhas = []
    for titulo, funcao in etapas():
        print(titulo)
        try:
            funcao(args)
        except Exception as e:  # segue para a próxima fonte e reporta no fim
            print(f"  ! {e}", file=sys.stderr)
            falhas.append(titulo)
    if falhas:
        print(f"\nFalharam: {'; '.join(falhas)}", file=sys.stderr)


def tratar():
    fichas = {f["id"]: f for f in ler_fichas(TEMA)}
    saida = Saida()
    ipca_mensal, ipca_anual, base = indice_ipca()
    print("Tratamento")
    _, familias, valor_mensal, _, _ = tratar_transferencia_renda(saida, ipca_mensal, ipca_anual, base)
    tratar_programas_anteriores(saida, familias)
    tratar_bpc(saida)
    tratar_cadunico(saida)
    conferir_misocial(saida)
    por_grupo, por_acao, arquivos, pib, arq_pib = tratar_despesa(saida, ipca_anual, valor_mensal)
    tratar_auxilio_emergencial(saida, por_grupo, por_acao, arquivos, pib, arq_pib, ipca_anual)
    tratar_ebia(saida)
    tratar_pares(saida)
    tratar_mcmv(saida)
    tratar_fjp(saida)
    saida.linhas.sort(key=lambda l: (l[0], l[1]))
    escrever_csv(TRATADOS / f"{TEMA}.csv", ["indicador", "data", "valor", "unidade", "fonte"], saida.linhas)
    escrever_csv(TRATADOS / f"{TEMA}_pares_paises.csv", ["indicador", "pais", "data", "valor", "unidade", "fonte"],
                 saida.paises)
    for ficha_id, itens in saida.extracao.items():
        if ficha_id not in fichas:
            raise RuntimeError(f"série sem ficha: {ficha_id}")
        atualizar_extracao(fichas[ficha_id], itens, saida.notas.get(ficha_id, ()))
    sem_serie = sorted(set(fichas) - set(saida.extracao))
    if sem_serie:
        raise RuntimeError(f"fichas sem série: {sem_serie}")
    series = sorted({l[0] for l in saida.linhas})
    print(f"{len(saida.linhas)} linhas, {len(series)} séries -> dados/tratados/{TEMA}.csv; "
          f"{len(saida.paises)} linhas -> dados/tratados/{TEMA}_pares_paises.csv")
    for s in series:
        datas = [l[1] for l in saida.linhas if l[0] == s]
        print(f"  {s:72s} {min(datas)} a {max(datas)} ({len(datas)})")


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--so-tratar", action="store_true", help="não baixa nada; trata os brutos mais recentes")
    ap.add_argument("--rebaixar", action="store_true",
                    help="baixa de novo as publicações de versão fechada já guardadas (FJP, MDS, Ipea, RTN de 2022)")
    args = ap.parse_args()
    if not args.so_tratar:
        coletar(args)
    tratar()


if __name__ == "__main__":
    main()