#!/usr/bin/env python3 """ scraper_mte_trf.py — SAVE Company Coleta decisões onde EMPREGADOR VENCEU autuações do Ministério do Trabalho nos TRFs. Usa DataJud CNJ (API pública). Multas administrativas do MTE são cobradas via execução fiscal na Justiça Federal. """ import json, time, requests from pathlib import Path OUTPUT = Path("/home/cristiano/meus-projetos/pesquisa-juridica/mte_empregador_venceu.json") LOG = Path("/home/cristiano/meus-projetos/pesquisa-juridica/mte_run.log") DATAJUD_KEY = "cDZHYzlZa0JadVREZDJCendQbXY6SkJlTzNjLV9TRENyQk1RdnFKZGRQdw==" BASE = "https://api-publica.datajud.cnj.jus.br" TRFS = { "TRF1": "api_publica_trf1", "TRF2": "api_publica_trf2", "TRF3": "api_publica_trf3", "TRF4": "api_publica_trf4", "TRF5": "api_publica_trf5", } # Classes processuais para contestar multas MTE na Justiça Federal CLASSES = [ 1116, # Execução Fiscal 1028, # Embargos à Execução Fiscal 1120, # Mandado de Segurança Cível 109, # Ação Declaratória 7655, # Ação Anulatória ] # Assuntos CNJ relacionados a infrações trabalhistas administrativas ASSUNTOS_MTE = [ 10820, # Auto de Infração / Autuação 10823, # Multa Administrativa 10826, # Nulidade de Auto de Infração 10680, # NR — Normas Regulamentadoras 10660, # Registro em CTPS 10661, # Jornada de Trabalho 10662, # Horas Extras 10670, # Insalubridade 10671, # Periculosidade 10672, # EPI — Equipamento de Proteção Individual 10673, # CIPA 10675, # PCMSO / PPRA 10676, # Acidente de Trabalho (administrativo) 10677, # Trabalho Infantil 10678, # Trabalho Escravo / Análogo 10630, # FGTS (administrativo) 10631, # Lei de Cotas PCD 10632, # Menor Aprendiz 10640, # Terceirização (administrativa) ] # Movimentos de vitória do empregador MOV_EMPREGADOR_GANHA = [ 219, # Procedência do pedido 221, # Procedência em parte 238, # Provimento (recurso do empregador provido) 240, # Provimento em Parte 460, # Extinção do crédito 236, # Não Provimento do recurso do MTE 198, # Improcedência do pedido do MTE ] MOTIVO_MAPA = { "auto de infra": "Auto de infração anulado — vício formal ou mérito", "autuac": "Autuação do MTE anulada pelo Judiciário", "multa": "Multa administrativa trabalhista cancelada", "nulidade": "Nulidade formal do auto de infração reconhecida", "epi": "EPI fornecido e documentado — auto cancelado", "cipa": "CIPA dispensada por dimensionamento do estabelecimento", "pcmso": "PCMSO/PPRA existentes — ausência não comprovada pelo fiscal", "ppra": "PPRA existente — autuação anulada", "insalubrid": "Insalubridade afastada por laudo técnico (LTCAT)", "jornada": "Jornada regular comprovada — banco de horas ou acordo coletivo", "horas extra": "Horas extras: banco de horas por ACT afastou autuação", "ctps": "Registro em CTPS: autônomo comprovado — vínculo não configurado", "trabalho infant": "Trabalho infantil: menor aprendiz regularmente contratado", "trabalho escrav": "Trabalho análogo à escravidão: caracterização afastada", "cotas": "Cotas PCD: cumprimento por estabelecimento comprovado", "aprendiz": "Menor aprendiz com contrato regular — auto cancelado", "terceiriz": "Terceirização lícita (pós-Reforma) — fraude não configurada", "fgts": "FGTS recolhido regularmente — execução extinta", "decad": "Decadência da ação punitiva (Lei 9.873/1999) reconhecida", "prescri": "Prescrição intercorrente — processo paralisado pelo MTE", "acidente": "Acidente de trabalho: culpa exclusiva da vítima reconhecida", "execu": "Execução fiscal de multa MTE: dívida extinta ou prescrita", } def motivo_legivel(assunto_str): a = assunto_str.lower() for chave, descricao in MOTIVO_MAPA.items(): if chave in a: return descricao return "Autuação trabalhista administrativa anulada" def buscar(tribunal_key, endpoint, existentes: set): url = f"{BASE}/{endpoint}/_search" # CORREÇÃO: era "ApiKey" (errado) → "APIKey" (correto) headers = {"Authorization": f"APIKey {DATAJUD_KEY}", "Content-Type": "application/json"} resultados = [] # Query unificada: classe + movimento (sem filtro de assunto — MTE usa # assuntos genéricos de direito administrativo/fiscal que variam por TRF) body = { "size": 100, "from": 0, "query": {"bool": {"must": [ {"terms": {"classe.codigo": CLASSES}}, {"terms": {"movimentos.codigo": MOV_EMPREGADOR_GANHA}}, {"range": {"dataHoraUltimaAtualizacao": {"gte": "2023-01-01"}}}, ]}}, "_source": [ "numeroProcesso", "classe", "assuntos", "movimentos", "orgaoJulgador", "dataAjuizamento", "dataHoraUltimaAtualizacao", ], "sort": [{"dataHoraUltimaAtualizacao": {"order": "desc"}}], } offset = 0 while offset < 500: body["from"] = offset try: r = requests.post(url, json=body, headers=headers, timeout=30) if r.status_code != 200: LOG.open("a").write(f"ERRO {tribunal_key}: HTTP {r.status_code}\n") break hits = r.json().get("hits", {}).get("hits", []) if not hits: break for h in hits: s = h["_source"] num = s.get("numeroProcesso", "") if not num or num in existentes: continue existentes.add(num) assunto_str = " | ".join( a.get("nome", "") for a in s.get("assuntos", []) if isinstance(a, dict) ) mov_codigos = {m.get("codigo", 0) for m in s.get("movimentos", []) if isinstance(m, dict)} tipo = "" for m in s.get("movimentos", []): if isinstance(m, dict) and m.get("codigo", 0) in MOV_EMPREGADOR_GANHA: tipo = m.get("nome", "") break resultados.append({ "numero": num, "tribunal": tribunal_key, "classe": s.get("classe", {}).get("nome", "") if isinstance(s.get("classe"), dict) else "", "orgao_julgador": s.get("orgaoJulgador", {}).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_legivel(assunto_str), "resultado": "empregador_venceu", "fonte": "DataJud", }) if len(hits) < 100: break offset += 100 time.sleep(0.5) except Exception as e: LOG.open("a").write(f"ERRO {tribunal_key}: {e}\n") break return resultados def main(): todos = [] if OUTPUT.exists(): todos = json.loads(OUTPUT.read_text()) print(f"Retomando: {len(todos)} existentes") existentes = {r["numero"] for r in todos} for trib, endpoint in TRFS.items(): print(f"Coletando {trib}...", flush=True) r = buscar(trib, endpoint, existentes) todos.extend(r) print(f" {len(r)} decisões encontradas", flush=True) with open(LOG, "a") as f: f.write(f"{trib}: {len(r)} decisões\n") # deduplicar por número de processo 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) print(f"Total único: {len(uniq)} → {OUTPUT}") with open(LOG, "a") as f: f.write(f"TOTAL: {len(uniq)} decisões únicas\n") if __name__ == "__main__": main()