Create worker.py

worker.py

@@ -0,0 +1,339 @@
+#!/usr/bin/env python3
+"""
+TJ-PR - Extrator ULTRA SIMPLIFICADO (v7.0 - State-Driven Batch Worker)
+
+Worker com arquitetura produtor-consumidor. Esta versão foi refatorada para:
+1. Usar arquivos de estado (CSVs) no repositório Git para rastrear o progresso.
+2. Acumular registros válidos e salvá-los em lotes.
+3. Garantir a continuidade do trabalho entre reinicializações lendo o estado.
+"""
+
+import re
+import json
+import requests
+import csv
+from datetime import datetime
+from pathlib import Path
+import logging
+import random
+import uuid
+import os
+import time
+import gzip
+from concurrent.futures import ThreadPoolExecutor
+from threading import Thread, Lock, current_thread
+from queue import Queue, Empty
+from git import Repo, GitCommandError, Git
+
+# -----------------------------
+# Configuração
+# -----------------------------
+GIT_TOKEN = os.getenv("GIT_TOKEN")
+GITHUB_BRANCH = os.getenv("GITHUB_BRANCH", "main")
+LOCAL_REPO_PATH = Path("./repo_clone_v7")
+github_repo_env = os.getenv("GITHUB_REPO")
+if github_repo_env: GITHUB_REPO = github_repo_env.replace("https://github.com/", "").replace(".git", "")
+else: GITHUB_REPO = None
+
+MAX_PAGINA = int(os.getenv("MAX_PAGINA") or "121792")
+MIN_PAGINA = int(os.getenv("MIN_PAGINA") or "30001")
+
+NUM_WORKERS_EXTRACao = int(os.getenv("NUM_WORKERS_EXTRACAO") or "10")
+TAMANHO_LOTE_PAGINAS = int(os.getenv("TAMANHO_LOTE_PAGINAS") or "100")
+TAMANHO_LOTE_REGISTROS = 500 # Gatilho para criar um novo arquivo de lote
+MAX_TENTATIVAS_EXTRACAO = int(os.getenv("MAX_TENTATIVAS_EXTRACAO") or "5")
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%M:%S")
+logger = logging.getLogger(__name__)
+
+# ===================================================================================
+# CLASSE DE EXTRAÇÃO (INALTERADA)
+# ===================================================================================
+class ExtratorUltraSimples:
+    def __init__(self):
+        self.base_url = "https://portal.tjpr.jus.br"
+    def limpar_texto(self, texto: str) -> str:
+        texto = re.sub(r"<br\s*/?>", " ", texto, flags=re.IGNORECASE); texto = re.sub(r"<[^>]+>", "", texto); texto = texto.replace("&nbsp;", " ").replace("&quot;", '"').replace("&amp;", "&"); texto = texto.replace("&lt;", "<").replace("&gt;", ">").replace("&#39;", "'"); texto = texto.replace("\r", "").replace("\n", "").replace("\t", ""); texto = re.sub(r"\s+", " ", texto); return texto.strip()
+    def extrair_chave_valor_da_linha(self, tr_html: str) -> tuple:
+        match_chave = re.search(r"<b>([^<]+)</b>", tr_html, re.IGNORECASE);
+        if not match_chave: return None, None
+        chave_raw = match_chave.group(1).strip(); chave = chave_raw.replace(":", "").strip(); chave = re.sub(r"[áàâãä]", "a", chave, flags=re.IGNORECASE); chave = re.sub(r"[éèêë]", "e", chave, flags=re.IGNORECASE); chave = re.sub(r"[íìîï]", "i", chave, flags=re.IGNORECASE); chave = re.sub(r"[óòôõö]", "o", chave, flags=re.IGNORECASE); chave = re.sub(r"[úùûü]", "u", chave, flags=re.IGNORECASE); chave = re.sub(r"[ç]", "c", chave, flags=re.IGNORECASE); chave = re.sub(r"\s+", "_", chave); chave = re.sub(r"[^\w]", "", chave); chave = chave.lower()
+        match_valor = re.search(r"</b>\s*(.*?)</td>", tr_html, re.DOTALL | re.IGNORECASE)
+        if not match_valor: return chave, None
+        valor_raw = match_valor.group(1); valor = self.limpar_texto(valor_raw); return chave, valor
+    def extrair_url_documento(self, tr_html: str) -> str:
+        match = re.search(r"visualizacao\.do\?tjpr\.url\.crypto=([a-f0-9]+)", tr_html)
+        if match: return f"{self.base_url}/jurisprudencia/publico/visualizacao.do?tjpr.url.crypto={match.group(1)}"
+        return None
+    def extrair_tabelas(self, html: str) -> list:
+        return re.findall(r'<table[^>]*class=["\']?[^"\']*resultTable[^"\']*["\']?[^>]*>(.*?)</table>', html, re.DOTALL | re.IGNORECASE)
+    def extrair_linhas_tr(self, tabela_html: str) -> list:
+        return re.findall(r"<tr[^>]*>(.*?)</tr>", tabela_html, re.DOTALL | re.IGNORECASE)
+    def extrair_acordao(self, tabela_html: str) -> dict:
+        acordao = {}; linhas = self.extrair_linhas_tr(tabela_html)
+        for linha_html in linhas:
+            if url := self.extrair_url_documento(linha_html): acordao["url_documento"] = url
+            if (chave := self.extrair_chave_valor_da_linha(linha_html)[0]) and (valor := self.extrair_chave_valor_da_linha(linha_html)[1]): acordao[chave] = valor
+        return acordao
+    def extrair_todos_acordaos(self, html: str) -> list:
+        tabelas = self.extrair_tabelas(html); return [acordao for tabela in tabelas if (acordao := self.extrair_acordao(tabela))]
+
+# ===================================================================================
+# CLASSE PRINCIPAL DO WORKER (REARQUITETADA v7.0)
+# ===================================================================================
+class AbelhaAtomica:
+    def __init__(self):
+        if not GIT_TOKEN or not GITHUB_REPO: raise ValueError("GIT_TOKEN e GITHUB_REPO s��o obrigatórios.")
+        self.worker_id = f"abelha-{uuid.uuid4().hex[:6]}"
+        self.extrator_html = ExtratorUltraSimples()
+        self.session = requests.Session()
+        self.session.headers.update({"User-Agent": f"Mozilla/5.0 ({self.worker_id})"})
+        
+        self.fila_de_tarefas = Queue()
+        self.fila_de_resultados = Queue() # Irá conter registros individuais e marcadores de status de página
+        
+        # Locks
+        self.lock_git = Lock()
+        self.lock_estado = Lock() # Protege o acesso aos dicionários de estado e acumulador
+        
+        self.thread_de_push = None
+        
+        # Gerenciamento de Estado
+        self.paginas_status = {}  # {123: 'sucesso', 456: 'falha'}
+        self.processos_vistos = set() # { 'processo_25_char_1', ... }
+        self.registros_acumulados = [] # Acumula registros válidos para o próximo lote
+        self.num_lote_saida = 0
+
+        self._setup_git_repo()
+        logger.info(f"Abelha {self.worker_id} (v7.0) iniciando. Estado carregado: {len(self.paginas_status)} páginas, {len(self.processos_vistos)} processos.")
+
+    def _setup_git_repo(self):
+        remote_url = f"https://oauth2:{GIT_TOKEN}@github.com/{GITHUB_REPO}.git"
+        if LOCAL_REPO_PATH.exists():
+            self.repo = Repo(LOCAL_REPO_PATH)
+        else:
+            self.repo = Repo.clone_from(remote_url, LOCAL_REPO_PATH, branch=GITHUB_BRANCH)
+        
+        with self.repo.config_writer() as config:
+            config.set_value("pull", "rebase", "true")
+
+        self._carregar_estado_do_repo()
+
+    def _carregar_estado_do_repo(self):
+        with self.lock_git:
+            logger.info("Sincronizando com o remoto para carregar estado...")
+            self.repo.git.reset("--hard")
+            self.repo.remotes.origin.pull()
+
+            # Carregar status das páginas
+            path_paginas = LOCAL_REPO_PATH / "paginas_status.csv"
+            if path_paginas.exists():
+                with open(path_paginas, 'r', encoding='utf-8') as f:
+                    reader = csv.reader(f)
+                    next(reader, None) # Pula cabeçalho
+                    self.paginas_status = {int(row[0]): row[1] for row in reader}
+            
+            # Carregar processos já vistos
+            path_processos = LOCAL_REPO_PATH / "processos_vistos.csv"
+            if path_processos.exists():
+                with open(path_processos, 'r', encoding='utf-8') as f:
+                    reader = csv.reader(f)
+                    next(reader, None) # Pula cabeçalho
+                    self.processos_vistos = {row[1] for row in reader}
+            
+            # Determinar o próximo número de lote
+            path_saida = LOCAL_REPO_PATH / "new_doce"
+            path_saida.mkdir(exist_ok=True)
+            arquivos_existentes = list(path_saida.glob("*.jsonl.gz"))
+            if arquivos_existentes:
+                numeros = [int(re.search(r'(\d+)', f.name).group(1)) for f in arquivos_existentes if re.search(r'(\d+)', f.name)]
+                self.num_lote_saida = max(numeros) + 1 if numeros else 1
+
+    def _selecionar_lote_de_paginas(self):
+        paginas_para_trabalhar = []
+        for pagina in range(MIN_PAGINA, MAX_PAGINA + 1):
+            status = self.paginas_status.get(pagina)
+            if status not in ['sucesso', 'falha_permanente', 'vazio']:
+                paginas_para_trabalhar.append(pagina)
+                if len(paginas_para_trabalhar) >= TAMANHO_LOTE_PAGINAS:
+                    break
+        
+        if not paginas_para_trabalhar:
+            logger.info("Nenhuma página nova para processar. Trabalho concluído ou aguardando.")
+            return False
+
+        for pagina in paginas_para_trabalhar:
+            self.fila_de_tarefas.put(pagina)
+        logger.info(f"Lote de {len(paginas_para_trabalhar)} páginas adicionado à fila de tarefas.")
+        return True
+
+    def extrair_processo_25(self, processo_str):
+        if not processo_str: return ""
+        return re.sub(r'[^a-zA-Z0-9\-.]', '', str(processo_str))[:25]
+
+    def validar_registro(self, reg):
+        processo_completo = reg.get("processo", "")
+        processo_25 = self.extrair_processo_25(processo_completo)
+        
+        if not processo_25 or processo_25 in self.processos_vistos:
+            return False
+        
+        url = reg.get("url_documento", "")
+        ementa = reg.get("ementa", "")
+        integra = reg.get("integra_do_acordao", "")
+
+        if not any([url, ementa, integra]): return False
+        
+        ementa_upper = ementa.upper().strip()
+        integra_upper = integra.upper().strip()
+        
+        if len(processo_completo.strip()) < 25 and processo_completo.upper() not in ementa_upper: return False
+        if len(processo_completo.strip()) == 25 and (processo_completo.upper() not in ementa_upper and processo_completo.upper() not in integra_upper): return False
+        if ementa and integra and ementa_upper[:60] not in integra_upper: reg["ementa"] = ""
+
+        return True
+
+    def _produtor_worker(self):
+        delay = random.uniform(0.5, 1.5)
+        time.sleep(delay)
+        
+        while True:
+            try:
+                pagina = self.fila_de_tarefas.get(timeout=1)
+                self._processar_pagina_extracao(pagina)
+                self.fila_de_tarefas.task_done()
+            except Empty:
+                break
+            except Exception as e:
+                logger.error(f"[{current_thread().name}] Erro fatal no worker da página {pagina}: {e}")
+                self.fila_de_tarefas.task_done()
+
+    def extrair_dados_pagina(self, pagina: int) -> list:
+        form_data = { "actionType": "pesquisar", "criterioPesquisa": "", "idLocalPesquisa": "99", "pageSize": "50", "pageNumber": str(pagina), "sortColumn": "processos.dataJulgamento", "sortOrder": "asc", "segredoJustica": "pesquisar sem", "mostrarCompleto": "true" }
+        for _ in range(MAX_TENTATIVAS_EXTRACAO):
+            try:
+                response = self.session.post("https://portal.tjpr.jus.br/jurisprudencia/publico/pesquisa.do", data=form_data, timeout=45)
+                response.raise_for_status()
+                return self.extrator_html.extrair_todos_acordaos(response.text)
+            except requests.RequestException:
+                time.sleep(5)
+        raise ConnectionError(f"Não foi possível buscar a página {pagina} após {MAX_TENTATIVAS_EXTRACAO} tentativas.")
+        
+    def _processar_pagina_extracao(self, pagina: int):
+        logger.info(f"[{current_thread().name}] 🐝 Iniciando extração da página {pagina}...")
+        try:
+            acordaos_brutos = self.extrair_dados_pagina(pagina)
+            num_validos = 0
+            
+            if not acordaos_brutos:
+                self.fila_de_resultados.put({'_tipo': 'status_pagina', 'pagina': pagina, 'status': 'vazio'})
+                return
+
+            for acordao_raw in acordaos_brutos:
+                acordao_enriquecido = self._enriquecer_acordao(acordao_raw)
+                if self.validar_registro(acordao_enriquecido):
+                    self.fila_de_resultados.put(acordao_enriquecido)
+                    num_validos += 1
+            
+            logger.info(f"[{current_thread().name}] ✅ Página {pagina}: {num_validos}/{len(acordaos_brutos)} registros válidos.")
+            self.fila_de_resultados.put({'_tipo': 'status_pagina', 'pagina': pagina, 'status': 'sucesso'})
+        except Exception as e:
+            logger.error(f"[{current_thread().name}] ❌ FALHA na extração da página {pagina}. Erro: {e}")
+            self.fila_de_resultados.put({'_tipo': 'status_pagina', 'pagina': pagina, 'status': 'falha'})
+
+    def _enriquecer_acordao(self, acordao: dict) -> dict:
+        if "processo" in acordao:
+            match = re.search(r"^(.*?)\s*\((.*?)\)", acordao["processo"])
+            if match: acordao["processo"], acordao["tipo_mov"] = match.group(1).strip(), match.group(2).strip()
+        return acordao
+
+    def _consumidor_push_para_git(self, lote_registros, paginas_atualizadas, processos_novos):
+        with self.lock_git:
+            logger.info(f"📤 [Push Thread] Iniciando PUSH de {len(lote_registros)} registros...")
+            try:
+                self.repo.git.reset("--hard"); self.repo.remotes.origin.pull()
+
+                # 1. Escrever o novo lote de dados
+                nome_arquivo = f"acordaos_{self.num_lote_saida:05d}.jsonl.gz"
+                caminho_saida = LOCAL_REPO_PATH / "new_doce" / nome_arquivo
+                with gzip.open(caminho_saida, 'wt', encoding='utf-8') as f:
+                    for reg in lote_registros: f.write(json.dumps(reg, ensure_ascii=False) + "\n")
+                
+                # 2. Atualizar arquivos de estado
+                path_paginas = LOCAL_REPO_PATH / "paginas_status.csv"
+                with open(path_paginas, 'w', newline='', encoding='utf-8') as f:
+                    writer = csv.writer(f); writer.writerow(['pagina', 'status'])
+                    for p, s in sorted(paginas_atualizadas.items()): writer.writerow([p, s])
+
+                path_processos = LOCAL_REPO_PATH / "processos_vistos.csv"
+                # Usamos 'a' (append) para adicionar apenas os novos processos
+                with open(path_processos, 'a', newline='', encoding='utf-8') as f:
+                    writer = csv.writer(f)
+                    if path_processos.stat().st_size == 0: writer.writerow(['id', 'processo_25'])
+                    for idx, proc in enumerate(processos_novos, start=len(self.processos_vistos) - len(processos_novos)):
+                        writer.writerow([idx, proc])
+
+                # 3. Commit e Push
+                self.repo.git.add(["new_doce/", "paginas_status.csv", "processos_vistos.csv"])
+                commit_msg = f"DATA: {self.worker_id} adiciona lote {self.num_lote_saida} com {len(lote_registros)} registros"
+                self.repo.index.commit(commit_msg)
+                self.repo.remotes.origin.push()
+
+                logger.info(f"✅ [Push Thread] Push do lote {self.num_lote_saida} concluído.")
+                self.num_lote_saida += 1
+                
+                # Limpar acumulador local SOMENTE após o push bem-sucedido
+                with self.lock_estado: self.registros_acumulados.clear()
+            except Exception as e:
+                logger.critical(f"❌ [Push Thread] FALHA CRÍTICA no push para o Git: {e}. Os dados não foram limpos e serão reenviados.")
+        self.thread_de_push = None
+
+    def loop_infinito(self):
+        pool_produtores = ThreadPoolExecutor(max_workers=NUM_WORKERS_EXTRACao, thread_name_prefix='ProdutorWorker')
+        
+        while True:
+            try:
+                # Alimenta a fila de tarefas se estiver vazia
+                if self.fila_de_tarefas.empty():
+                    if not self._selecionar_lote_de_paginas():
+                        time.sleep(60) # Dorme se não houver trabalho
+                        continue
+                    for _ in range(NUM_WORKERS_EXTRACao):
+                        pool_produtores.submit(self._produtor_worker)
+                
+                # Processa resultados da fila
+                processos_novos_neste_ciclo = set()
+                while not self.fila_de_resultados.empty():
+                    item = self.fila_de_resultados.get_nowait()
+                    with self.lock_estado:
+                        if '_tipo' in item and item['_tipo'] == 'status_pagina':
+                            self.paginas_status[item['pagina']] = item['status']
+                        else: # É um registro de acórdão
+                            processo_25 = self.extrair_processo_25(item.get("processo"))
+                            if processo_25 not in self.processos_vistos:
+                                self.registros_acumulados.append(item)
+                                self.processos_vistos.add(processo_25)
+                                processos_novos_neste_ciclo.add(processo_25)
+
+                # Dispara o push se o gatilho for atingido
+                if len(self.registros_acumulados) >= TAMANHO_LOTE_REGISTROS and self.thread_de_push is None:
+                    lote_para_push = list(self.registros_acumulados) # Faz uma cópia
+                    paginas_para_push = dict(self.paginas_status) # Faz uma cópia
+                    
+                    self.thread_de_push = Thread(target=self._consumidor_push_para_git, args=(lote_para_push, paginas_para_push, processos_novos_neste_ciclo), name="PushThread")
+                    self.thread_de_push.start()
+                
+                print(f"[Orquestrador] Tarefas: {self.fila_de_tarefas.qsize()}, Acumulados: {len(self.registros_acumulados)}/{TAMANHO_LOTE_REGISTROS}, Push: {'Sim' if self.thread_de_push else 'Não'}", flush=True)
+                time.sleep(5)
+
+            except Exception as e:
+                logger.critical(f"ERRO CRÍTICO no loop principal: {e}. Reiniciando...")
+                time.sleep(15)
+
+def main():
+    logger.info("Iniciando Abelha com Pipeline Assíncrono (v7.0 - State-Driven)")
+    abelha = AbelhaAtomica()
+    abelha.loop_infinito()
+
+if __name__ == "__main__":
+    main()