#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Painel de apuração — Eleições 2026 (1º turno, 4 de outubro) Servidor local sem dependências (só biblioteca padrão do Python 3.8+). Busca os arquivos JSON públicos do TSE, normaliza e serve o painel em http://localhost:8000 . O navegador nunca fala direto com o TSE: o servidor faz cache (15 s), usa ETag e evita rajadas de requisições (limite do TSE: 100 req/s por IP; muitos 404 seguidos podem bloquear o IP por 10 min). Uso: python3 painel_eleicoes_2026.py # ambiente oficial python3 painel_eleicoes_2026.py --ambiente simulado # testes do TSE python3 painel_eleicoes_2026.py --check # testa a conexão e sai python3 painel_eleicoes_2026.py --host 0.0.0.0 # abrir para a rede/servidor Códigos de eleição (TSE, 1º turno 04/10/2026): federal 6257 (Presidente), estadual 6259 (Senador). Para o 2º turno, informe --fed e --est com os novos códigos (arquivo comum/config/ele-c.json do TSE). """ import argparse import json import os import sys import threading import time import urllib.error import urllib.request from concurrent.futures import ThreadPoolExecutor from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse UFS = { "AC": "Acre", "AL": "Alagoas", "AP": "Amapá", "AM": "Amazonas", "BA": "Bahia", "CE": "Ceará", "DF": "Distrito Federal", "ES": "Espírito Santo", "GO": "Goiás", "MA": "Maranhão", "MT": "Mato Grosso", "MS": "Mato Grosso do Sul", "MG": "Minas Gerais", "PA": "Pará", "PB": "Paraíba", "PR": "Paraná", "PE": "Pernambuco", "PI": "Piauí", "RJ": "Rio de Janeiro", "RN": "Rio Grande do Norte", "RS": "Rio Grande do Sul", "RO": "Rondônia", "RR": "Roraima", "SC": "Santa Catarina", "SP": "São Paulo", "SE": "Sergipe", "TO": "Tocantins", } AMBIENTES = { "oficial": { "base": "https://resultados.tse.jus.br/oficial/ele2026", "fed": "6257", "est": "6259", }, "simulado": { "base": "https://resultados-sim.tse.jus.br/simulado/simulado2026/ele2026", "fed": "21270", "est": "21272", }, } CFG = {"ambiente": "oficial", "base": "", "fed": "", "est": "", "ttl": 15, "refresh": 20} def url_presidente(): f = CFG["fed"] return f"{CFG['base']}/{f}/dados/br/br-c0001-e{int(f):06d}-u.json" def url_senador(uf): e = CFG["est"] u = uf.lower() return f"{CFG['base']}/{e}/dados/{u}/{u}-c0005-e{int(e):06d}-u.json" # ---------------------------------------------------------------- busca + cache class Fetcher: def __init__(self, ttl=15, neg_ttl=60, timeout=12): self.ttl, self.neg_ttl, self.timeout = ttl, neg_ttl, timeout self.cache, self.locks = {}, {} self.lock = threading.Lock() def _fresh(self, ent): if not ent: return False limite = self.ttl if ent["status"] == 200 and not ent.get("stale") else ( 5 if ent.get("stale") else self.neg_ttl) return time.time() - ent["ts"] < limite def get(self, url): with self.lock: ent = self.cache.get(url) lk = self.locks.setdefault(url, threading.Lock()) if self._fresh(ent): return ent with lk: ent = self.cache.get(url) if self._fresh(ent): return ent headers = {"User-Agent": "PainelEleicoes2026/1.0", "Accept": "application/json"} if ent and ent["status"] == 200 and ent.get("etag"): headers["If-None-Match"] = ent["etag"] req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, timeout=self.timeout) as r: new = {"status": 200, "body": r.read(), "etag": r.headers.get("ETag"), "ts": time.time(), "fetched": time.time()} except urllib.error.HTTPError as e: if e.code == 304 and ent: new = dict(ent, ts=time.time(), fetched=time.time(), stale=False) elif ent and ent["status"] == 200: new = dict(ent, ts=time.time(), stale=True, error=f"HTTP {e.code}") else: new = {"status": e.code, "body": b"", "ts": time.time(), "error": f"HTTP {e.code}"} except Exception as e: # rede fora, DNS, timeout... if ent and ent["status"] == 200: new = dict(ent, ts=time.time(), stale=True, error=str(e)) else: new = {"status": 0, "body": b"", "ts": time.time() - self.neg_ttl + 10, "error": str(e)} with self.lock: self.cache[url] = new return new F = Fetcher() # ------------------------------------------------------------- normalização def _i(x): try: return int(str(x).strip()) except Exception: return 0 def _p(x): try: return float(str(x).strip().replace(",", ".")) except Exception: return None def normalize(raw): """Reduz o arquivo EA20 do TSE ao que o painel usa.""" carg = (raw.get("carg") or [{}])[0] cands = [] for agr in carg.get("agr", []) or []: for par in agr.get("par", []) or []: for c in par.get("cand", []) or []: cands.append({ "n": c.get("n"), "nm": c.get("nm"), "nmu": c.get("nmu") or c.get("nm"), "sg": par.get("sg"), "vap": _i(c.get("vap")), "pvap": _p(c.get("pvapn") or c.get("pvap")), "dvt": c.get("dvt"), "e": c.get("e"), "st": c.get("st"), "seq": _i(c.get("seq")), }) s, e, v = raw.get("s") or {}, raw.get("e") or {}, raw.get("v") or {} return { "dt": raw.get("dt"), "ht": raw.get("ht"), "dg": raw.get("dg"), "hg": raw.get("hg"), "and": raw.get("and"), "vagas": _i(carg.get("nv")) or None, "s": {"ts": _i(s.get("ts")), "st": _i(s.get("st")), "pst": _p(s.get("pstn") or s.get("pst"))}, "e": {"te": _i(e.get("te")), "est": _i(e.get("est")), "c": _i(e.get("c")), "pc": _p(e.get("pcn") or e.get("pc")), "a": _i(e.get("a")), "pa": _p(e.get("pan") or e.get("pa"))}, "v": {"tv": _i(v.get("tv")), "vvc": _i(v.get("vvc")), "vb": _i(v.get("vb")), "pvb": _p(v.get("pvbn") or v.get("pvb")), "tvn": _i(v.get("tvn") or v.get("vn")), "ptvn": _p(v.get("ptvnn") or v.get("ptvn")), "vansj": _i(v.get("vansj")), "van": _i(v.get("van"))}, "cands": cands, } def wrap(ent): if ent["status"] != 200: return {"ok": False, "status": ent["status"], "error": ent.get("error")} if "norm" not in ent: try: ent["norm"] = normalize(json.loads(ent["body"].decode("utf-8-sig"))) except Exception as ex: return {"ok": False, "status": -1, "error": f"JSON inesperado: {ex}"} return {"ok": True, "data": ent["norm"], "stale": bool(ent.get("stale")), "fetchedAt": ent.get("fetched")} def api_presidente(): return wrap(F.get(url_presidente())) def api_senado(): ufs = list(UFS) with ThreadPoolExecutor(max_workers=9) as ex: res = list(ex.map(lambda u: wrap(F.get(url_senador(u))), ufs)) return {"ufs": {u: r for u, r in zip(ufs, res)}, "nomes": UFS} # ------------------------------------------------------------------ HTTP HTML = r""" Apuração 2026 — Presidente e Senado

Apuração — Eleições 20261º turno · domingo, 4 de outubro · Presidente e Senado

aguardando primeira leitura…
–seções apuradas no Brasil
""" def pagina(): cfg = {"ambiente": CFG["ambiente"], "refresh": CFG["refresh"]} return HTML.replace("__CONFIG__", json.dumps(cfg)).encode("utf-8") class Handler(BaseHTTPRequestHandler): server_version = "PainelEleicoes/1.0" def _send(self, code, body, ctype): self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _json(self, obj, code=200): self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8") def do_GET(self): path = urlparse(self.path).path try: if path in ("/", "/index.html"): self._send(200, pagina(), "text/html; charset=utf-8") elif path == "/api/presidente": self._json(api_presidente()) elif path == "/api/senado": self._json(api_senado()) elif path == "/healthz": self._json({"ok": True, "ambiente": CFG["ambiente"]}) else: self._send(404, b"nao encontrado", "text/plain; charset=utf-8") except (BrokenPipeError, ConnectionResetError): pass except Exception as ex: self._json({"ok": False, "status": -1, "error": str(ex)}, 500) def log_message(self, fmt, *args): if os.environ.get("PAINEL_LOG"): sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args)) def checar(): print(f"Ambiente: {CFG['ambiente']}") ok = True for nome, url in (("Presidente (BR)", url_presidente()), ("Senado (SP)", url_senador("SP"))): ent = F.get(url) print(f"- {nome}: {url}") if ent["status"] == 200: d = normalize(json.loads(ent["body"].decode("utf-8-sig"))) print(f" OK · {len(d['cands'])} candidatos · seções apuradas {d['s']['pst']}% · totalização {d['dt']} {d['ht']}") else: ok = False print(f" sem arquivo (status {ent['status']}) {ent.get('error') or ''}") if ent["status"] == 404: print(" Normal antes da divulgação: o TSE só publica após o fim da votação (17h de Brasília).") return ok def main(): ap = argparse.ArgumentParser(description="Painel de apuração das eleições 2026 (TSE)") ap.add_argument("--ambiente", choices=list(AMBIENTES), default="oficial") ap.add_argument("--host", default="127.0.0.1") ap.add_argument("--port", type=int, default=8000) ap.add_argument("--fed", help="código da eleição federal (Presidente)") ap.add_argument("--est", help="código da eleição estadual (Senador)") ap.add_argument("--ttl", type=int, default=15, help="cache no servidor, em segundos (mín. 5)") ap.add_argument("--refresh", type=int, default=20, help="intervalo de atualização no navegador, em segundos") ap.add_argument("--check", action="store_true", help="testa a conexão com o TSE e sai") a = ap.parse_args() amb = AMBIENTES[a.ambiente] CFG.update(ambiente=a.ambiente, base=os.environ.get("TSE_BASE", amb["base"]), fed=a.fed or amb["fed"], est=a.est or amb["est"], ttl=max(5, a.ttl), refresh=max(10, a.refresh)) F.ttl = CFG["ttl"] if a.check: sys.exit(0 if checar() else 1) srv = ThreadingHTTPServer((a.host, a.port), Handler) print(f"Painel no ar: http://{'localhost' if a.host in ('127.0.0.1', '0.0.0.0') else a.host}:{a.port}" f" (ambiente: {a.ambiente}; Presidente {CFG['fed']}, Senado {CFG['est']})") print("Ctrl+C para encerrar.") try: srv.serve_forever() except KeyboardInterrupt: print("\nEncerrado.") if __name__ == "__main__": main()