#!/usr/bin/env python3 """ scraper_trabalhista.py — SAVE Company / Cristiano Vasconcelos Coleta decisões onde o EMPREGADOR VENCEU no contencioso trabalhista. Fontes: - TST: Recurso de Revista / Agravo não provido (employee appeal denied) - TRTs: Reclamações Trabalhistas — sentença definitiva / parcialmente improcedente - TRTs: Ação Civil Pública do MPT — extinta ou improcedente Escopo INCLUI: contencioso trabalhista (TST/TRTs) + MPT (ACP, TAC) Escopo NÃO INCLUI: MTE autos de infração → use scraper_mte_trf.py Códigos DataJud verificados em campo (junho/2026): TST classes : 1002 (AIRR), 1008 (RR), 11882 (RR c/ Agravo) TRT classes : 985 (Ação Trabalhista Rito Ordinário), 1125 (Sumaríssimo) MPT classes : 65 (Ação Civil Pública) Movimento TST: 239 (Não-Provimento — employee appeal denied = employer wins) Movimento TRT: 246 (Definitivo), 221 (Procedência em Parte) Movimento MPT: 460 (Extinção), 196 (Extinção sem mérito), 221 """ import json import time import requests from pathlib import Path from collections import Counter OUTPUT = Path("/home/cristiano/meus-projetos/pesquisa-juridica/trabalhista_empregador_venceu.json") LOG = Path("/home/cristiano/meus-projetos/pesquisa-juridica/trabalhista_run.log") DATAJUD_KEY = "cDZHYzlZa0JadVREZDJCendQbXY6SkJlTzNjLV9TRENyQk1RdnFKZGRQdw==" BASE = "https://api-publica.datajud.cnj.jus.br" # ── Janelas de data para passar o limite de 10.000 por query ────────────────── DATE_WINDOWS = [ ("2023-01-01", "2023-06-30"), ("2023-07-01", "2023-12-31"), ("2024-01-01", "2024-06-30"), ("2024-07-01", "2024-12-31"), ("2025-01-01", "2025-06-30"), ("2025-07-01", "2025-12-31"), ("2026-01-01", "2026-12-31"), ] # ── Tribunais ───────────────────────────────────────────────────────────────── TST_INDEX = "api_publica_tst" TRTS = { "TRT1": "api_publica_trt1", "TRT2": "api_publica_trt2", "TRT3": "api_publica_trt3", "TRT4": "api_publica_trt4", "TRT5": "api_publica_trt5", "TRT9": "api_publica_trt9", "TRT10": "api_publica_trt10", "TRT12": "api_publica_trt12", "TRT15": "api_publica_trt15", } # ── Classes e movimentos (verificados em campo) ─────────────────────────────── CLASSES_TST = [1002, 1008, 11882] # AIRR, RR, RR c/ Agravo CLASSES_TRT = [985, 1125] # Ação Trabalhista Rito Ordinário / Sumaríssimo CLASSES_MPT = [65] # Ação Civil Pública MOV_TST = [239] # Não-Provimento (employee's appeal denied) MOV_TRT = [246, 221] # Definitivo / Procedência em Parte (partial for employee) MOV_MPT = [460, 196, 221] # Extinção / Sentença sem mérito / Parcialmente improcedente # ── Mapeamento assunto → (categoria, motivo) ───────────────────────────────── MOTIVO_MAPA = [ # (fragmento no assunto.lower(), categoria CSS, motivo legível) ("vínculo", "vinculo", "Vínculo empregatício negado — autonomia reconhecida"), ("vinculo", "vinculo", "Vínculo empregatício negado — autonomia reconhecida"), ("reconhecimento de rel", "vinculo", "Vínculo empregatício negado — relação autônoma confirmada"), ("horas extras", "horas", "Horas extras negadas — banco de horas ou cargo de confiança"), ("hora extra", "horas", "Horas extras negadas — banco de horas ou cargo de confiança"), ("banco de horas", "horas", "Banco de horas por ACT válido — horas extras negadas"), ("cargo de confiança", "horas", "Cargo de confiança (art. 62 II CLT) — horas extras indevidas"), ("jornada de trabalho", "horas", "Jornada regular comprovada — horas extras indeferidas"), ("assédio moral", "assedio", "Assédio moral improcedente — cobrança de metas ≠ assédio"), ("assedio moral", "assedio", "Assédio moral improcedente — cobrança de metas ≠ assédio"), ("dano moral", "assedio", "Dano moral improcedente — ausência de prova de conduta ilícita"), ("danos morais", "assedio", "Dano moral improcedente — ausência de prova de conduta ilícita"), ("discriminação", "assedio", "Dispensa discriminatória não comprovada"), ("justa causa", "justa", "Justa causa mantida — gradação punitiva documentada"), ("improbidade", "justa", "Justa causa por improbidade — prova material robusta"), ("insubordinação", "justa", "Justa causa por insubordinação — gradação e imediatidade comprovadas"), ("insalubridade", "insalub", "Insalubridade negada — EPI eficaz (Súmula 289 TST)"), ("periculosidade", "insalub", "Periculosidade negada — exposição eventual, não permanente"), ("adicional de insalub", "insalub", "Adicional de insalubridade negado — laudo e EPI comprovados"), ("adicional de periculosidade", "insalub", "Adicional de periculosidade negado — sem exposição permanente"), ("acidente de trabalho", "doenca", "Acidente — culpa exclusiva da vítima reconhecida"), ("doença ocupacional", "doenca", "Doença ocupacional — nexo causal não estabelecido"), ("doença profissional", "doenca", "Doença profissional — pré-existência no ASO admissional"), ("nexo causal", "doenca", "Nexo causal com o trabalho não estabelecido"), ("ler ", "doenca", "LER/DORT — pré-existência comprovada no ASO admissional"), ("dort", "doenca", "LER/DORT — pré-existência comprovada no ASO admissional"), ("terceirização", "terc", "Terceirização lícita — responsabilidade solidária afastada"), ("terceirizacao", "terc", "Terceirização lícita — responsabilidade solidária afastada"), ("responsabilidade subsidiária", "terc", "Responsabilidade subsidiária limitada — tomadora fiscalizou"), ("sobreaviso", "outros", "Sobreaviso negado — celular corporativo não configura (OJ 49 TST)"), ("teletrabalho", "outros", "Teletrabalho — sem controle de jornada (arts. 75-A CLT)"), ("home office", "outros", "Teletrabalho — sem controle de jornada (arts. 75-A CLT)"), ("equiparação salarial", "outros", "Equiparação salarial negada — distinção de produtividade"), ("ação civil pública", "mpt", "ACP do MPT julgada improcedente — empresa demonstrou conformidade"), ("ministério público", "mpt", "Ação do MPT improcedente — documentação e compliance regulares"), ("multa do artigo 477", "outros", "Multa rescisória (art. 477 CLT) — pagamento tempestivo comprovado"), ("multa do artigo 467", "outros", "Multa salarial (art. 467 CLT) — contestação fundamentada"), ("fgts", "outros", "FGTS — recolhimento regular comprovado por extrato"), ("verbas rescisórias", "outros", "Verbas rescisórias — quitação documentada no TRCT"), ("rescisão", "outros", "Verbas rescisórias — quitação documentada no TRCT"), ] def classificar(assunto_str: str): a = assunto_str.lower() for chave, categoria, motivo in MOTIVO_MAPA: if chave in a: return motivo, categoria return "Reclamação trabalhista — empregador venceu no mérito", "outros" def _log(msg: str): print(msg, flush=True) with open(LOG, "a") as f: f.write(f"{msg}\n") def buscar_janela(endpoint: str, classes: list, movimentos: list, dt_ini: str, dt_fim: str, existentes: set, tribunal_key: str) -> list: """Coleta até 1.000 registros de uma janela de data específica.""" url = f"{BASE}/{endpoint}/_search" headers = {"Authorization": f"APIKey {DATAJUD_KEY}", "Content-Type": "application/json"} resultados = [] body = { "size": 100, "from": 0, "query": {"bool": {"must": [ {"terms": {"classe.codigo": classes}}, {"terms": {"movimentos.codigo": movimentos}}, {"range": {"dataHoraUltimaAtualizacao": {"gte": dt_ini, "lte": dt_fim}}}, ]}}, "_source": [ "numeroProcesso", "classe", "assuntos", "movimentos", "orgaoJulgador", "dataAjuizamento", "dataHoraUltimaAtualizacao", ], "sort": [{"dataHoraUltimaAtualizacao": {"order": "asc"}}], } offset = 0 while offset < 1000: body["from"] = offset try: r = requests.post(url, json=body, headers=headers, timeout=30) if r.status_code != 200: _log(f" ERRO {tribunal_key} HTTP {r.status_code}") break hits = r.json().get("hits", {}).get("hits", []) if not hits: break novos = 0 for h in hits: s = h["_source"] num = s.get("numeroProcesso", "") if not num or num in existentes: continue existentes.add(num) novos += 1 assunto_str = " | ".join( a.get("nome", "") for a in (s.get("assuntos") or []) if isinstance(a, dict) ) tipo = "" for m in (s.get("movimentos") or []): if isinstance(m, dict) and m.get("codigo", 0) in movimentos: tipo = m.get("nome", "") break motivo, categoria = classificar(assunto_str) resultados.append({ "numero": num, "tribunal": tribunal_key, "classe": (s.get("classe") or {}).get("nome", "") if isinstance(s.get("classe"), dict) else "", "orgao_julgador": (s.get("orgaoJulgador") or {}).get("nome", "") if isinstance(s.get("orgaoJulgador"), dict) else "", "data_ajuizamento": (s.get("dataAjuizamento") or "")[:10], "data_julgamento": (s.get("dataHoraUltimaAtualizacao") or "")[:10].replace("-", "/"), "assuntos": assunto_str, "tipo_decisao": tipo, "motivo": motivo, "categoria": categoria, "resultado": "empregador_venceu", "fonte": "DataJud", }) if len(hits) < 100: break if novos == 0: break offset += 100 time.sleep(0.3) except Exception as e: _log(f" ERRO {tribunal_key}: {e}") break return resultados def coletar_tst(existentes: set) -> list: todos = [] _log("\n[TST] Recurso de Revista / AIRR — Não-Provimento (employee appeal denied)") for dt_ini, dt_fim in DATE_WINDOWS: r = buscar_janela(TST_INDEX, CLASSES_TST, MOV_TST, dt_ini, dt_fim, existentes, "TST") todos.extend(r) if r: _log(f" TST {dt_ini}→{dt_fim}: {len(r)} decisões") time.sleep(0.5) _log(f" TST TOTAL: {len(todos)}") return todos def coletar_trts(existentes: set) -> list: todos = [] for trib_key, endpoint in TRTS.items(): _log(f"\n[{trib_key}] Reclamações Trabalhistas...") trib_total = 0 for dt_ini, dt_fim in DATE_WINDOWS: r = buscar_janela(endpoint, CLASSES_TRT, MOV_TRT, dt_ini, dt_fim, existentes, trib_key) todos.extend(r) trib_total += len(r) time.sleep(0.3) _log(f" {trib_key} contencioso: {trib_total}") # MPT — Ação Civil Pública _log(f"[{trib_key}] ACP do MPT...") mpt_total = 0 for dt_ini, dt_fim in DATE_WINDOWS: r_mpt = buscar_janela(endpoint, CLASSES_MPT, MOV_MPT, dt_ini, dt_fim, existentes, trib_key) todos.extend(r_mpt) mpt_total += len(r_mpt) time.sleep(0.3) _log(f" {trib_key} MPT: {mpt_total}") time.sleep(1) return todos def main(): todos = [] if OUTPUT.exists(): todos = json.loads(OUTPUT.read_text()) _log(f"Retomando: {len(todos):,} registros existentes") existentes = {r["numero"] for r in todos} # 1. TST (fonte principal — decisões mais fundamentadas) todos += coletar_tst(existentes) # 2. TRTs + MPT todos += coletar_trts(existentes) # Deduplicar seen = set() uniq = [] for d in todos: if d["numero"] not in seen: seen.add(d["numero"]) uniq.append(d) with open(OUTPUT, "w") as f: json.dump(uniq, f, ensure_ascii=False, indent=2) # Resumo cats = Counter(d["categoria"] for d in uniq) tribs = Counter(d["tribunal"] for d in uniq) _log(f"\n{'='*50}") _log(f"TOTAL ÚNICO: {len(uniq):,} decisões → {OUTPUT}") _log("Por categoria:") for cat, n in cats.most_common(): _log(f" {cat:<14} {n:>6,}") _log("Por tribunal:") for t, n in tribs.most_common(): _log(f" {t:<8} {n:>6,}") if __name__ == "__main__": main()