Update worker.py

worker.py

@@ -1,11 +1,12 @@
 #!/usr/bin/env python3
 """
-TJ-PR - Extrator ULTRA SIMPLIFICADO (v8.4 - SINGLE WORKER OPTIMIZED)
+TJ-PR - Extrator ULTRA SIMPLIFICADO (v7.8 - Git Resiliente & Stats)
 
-Otimização:
-1. STARTUP: Carrega estado do Git para memória (Fundamental).
-2. RUNTIME: Confia na memória RAM (não relê disco a cada push).
-3. PUSH: Apenas Append das novidades nos CSVs.
+Worker com arquitetura produtor-consumidor. 
+v7.8 MODIFICAÇÕES:
+1. Git Resiliente: Commit local antes de Pull/Push. Estratégia Rebase.
+2. Logs Limpos: Apenas eventos de lote e push.
+3. Estatísticas: Monitoramento de médias (5m, 30m, 120m) salvo em JSON.
 """
 
 import re
@@ -23,19 +24,19 @@ import time
 import gzip
 import tarfile
 import io
+from collections import deque
 from concurrent.futures import ThreadPoolExecutor
 from threading import Thread, Lock
 from queue import Queue, Empty
 from git import Repo, GitCommandError
 
 # -----------------------------
-# Configuração & Caminhos
+# Configuração
 # -----------------------------
-BASE_DIR = Path(__file__).parent.resolve()
 GIT_TOKEN = os.getenv("GIT_TOKEN")
 GITHUB_BRANCH = os.getenv("GITHUB_BRANCH", "main")
-LOCAL_REPO_PATH = BASE_DIR / "repo_clone_v8"
-STATS_FILE = BASE_DIR / "monitor_stats.json"
+LOCAL_REPO_PATH = Path("./repo_clone_v7")
+STATS_FILE_PATH = LOCAL_REPO_PATH / "stats.json"
 
 github_repo_env = os.getenv("GITHUB_REPO")
 if github_repo_env: 
@@ -45,279 +46,398 @@ else:
 
 MAX_PAGINA = int(os.getenv("MAX_PAGINA") or "121792")
 MIN_PAGINA = int(os.getenv("MIN_PAGINA") or "1")
+
 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
 MAX_TENTATIVAS_EXTRACAO = int(os.getenv("MAX_TENTATIVAS_EXTRACAO") or "5")
 
-logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S")
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    datefmt="%H:%M:%S"
+)
 logger = logging.getLogger(__name__)
 
 # ===================================================================================
-# CLASSES UTILITÁRIAS
+# GERENCIADOR DE ESTATÍSTICAS
+# ===================================================================================
+class EstatisticaManager:
+    def __init__(self):
+        self.lock = Lock()
+        self.total_paginas = 0
+        self.total_processos = 0
+        # Armazena timestamps dos eventos para cálculo de média móvel
+        self.history_paginas = deque()
+        self.history_processos = deque()
+    
+    def registrar(self, qtd_paginas, qtd_processos):
+        now = time.time()
+        with self.lock:
+            self.total_paginas += qtd_paginas
+            self.total_processos += qtd_processos
+            for _ in range(qtd_paginas): self.history_paginas.append(now)
+            for _ in range(qtd_processos): self.history_processos.append(now)
+            self._limpar_antigos(now)
+
+    def _limpar_antigos(self, now):
+        # Remove eventos mais velhos que 120 minutos (7200 segundos)
+        limit = now - 7200
+        while self.history_paginas and self.history_paginas[0] < limit: self.history_paginas.popleft()
+        while self.history_processos and self.history_processos[0] < limit: self.history_processos.popleft()
+
+    def calcular_medias(self):
+        now = time.time()
+        with self.lock:
+            self._limpar_antigos(now)
+            stats = {
+                "total_paginas": self.total_paginas,
+                "total_processos": self.total_processos,
+                "medias_paginas": self._calc_rate(self.history_paginas, now),
+                "medias_processos": self._calc_rate(self.history_processos, now),
+                "timestamp_atualizacao": time.strftime("%Y-%m-%d %H:%M:%S")
+            }
+        return stats
+
+    def _calc_rate(self, history, now):
+        # Retorna contagem absoluta nos intervalos
+        def count_in_window(seconds):
+            limit = now - seconds
+            return sum(1 for t in history if t >= limit)
+        
+        # Opcional: Para virar "por minuto", dividiria pelo tempo. 
+        # Aqui retornaremos o TOTAL processado na janela de tempo.
+        return {
+            "5min": count_in_window(300),
+            "30min": count_in_window(1800),
+            "120min": count_in_window(7200)
+        }
+
+    def salvar_arquivo(self):
+        dados = self.calcular_medias()
+        try:
+            with open(STATS_FILE_PATH, 'w', encoding='utf-8') as f:
+                json.dump(dados, f, indent=2)
+        except Exception:
+            pass # Falha não crítica
+        return dados
+
+# ===================================================================================
+# CLASSE DE EXTRAÇÃO DE DESTAQUES (Integrada)
 # ===================================================================================
 class DestaqueExtractor:
     def __init__(self):
-        self.MIN = 3
-        self.STOP = {'OCULTARACORDAO', 'ATENCAO', 'O TEXTO ABAIXO REPRESENTA A TRANSCRICAO DE ACORDAO', 'EVENTUAIS IMAGENS SERAO SUPRIMIDAS', 'TRANSCRICAO DE ACORDAO', 'ESTADO DO PARANA', 'PODER JUDICIARIO', 'TRIBUNAL DE JUSTICA', 'RELATOR DESEMBARGADOR', 'VISTOS'}
-    def _norm(self, t): return re.sub(r'[^a-zA-Z0-9.\s]', '.', "".join([c for c in unicodedata.normalize('NFKD', t or "") if not unicodedata.combining(c)]))
-    def extrair_destaques(self, reg, pid):
-        txt = self._norm(f"{reg.get('ementa', '')} {reg.get('integra_do_acordao', '')}")
-        chunks = set()
-        for c in txt.split('.'):
-            c = re.sub(r'\s+', ' ', c).strip()
-            if len(c) >= self.MIN and c.isupper() and not any(x.islower() for x in c) and c not in self.STOP and len(c.split())<=12: chunks.add(c)
-        return [{"id": f"{pid}-{i}", "id_processo": pid, "texto": t} for i, t in enumerate(chunks, 1)]
+        self.MIN_CARACTERES_TAG = 3
+        self.STOP_PHRASES = {
+            'OCULTARACORDAO', 'ATENCAO', 'O TEXTO ABAIXO REPRESENTA A TRANSCRICAO DE ACORDAO',
+            'EVENTUAIS IMAGENS SERAO SUPRIMIDAS', 'TRANSCRICAO DE ACORDAO', 'ESTADO DO PARANA',
+            'PODER JUDICIARIO', 'TRIBUNAL DE JUSTICA', 'RELATOR DESEMBARGADOR', 'VISTOS'
+        }
+
+    def _normalizar(self, texto: str) -> str:
+        if not texto: return ""
+        nfkd_form = unicodedata.normalize('NFKD', texto)
+        texto_limpo = "".join([c for c in nfkd_form if not unicodedata.combining(c)])
+        return re.sub(r'[^a-zA-Z0-9.\s]', '.', texto_limpo)
+
+    def _extrair_chunks(self, texto_processado: str) -> list[str]:
+        chunks = texto_processado.split('.')
+        etiquetas = set()
+        for chunk in chunks:
+            chunk = re.sub(r'\s+', ' ', chunk).strip()
+            if len(chunk) >= self.MIN_CARACTERES_TAG:
+                if chunk.isupper() and not any(c.islower() for c in chunk):
+                    if chunk not in self.STOP_PHRASES and len(chunk.split()) <= 12:
+                        etiquetas.add(chunk)
+        return list(etiquetas)
 
+    def extrair_destaques(self, registro: dict, id_processo: int) -> list[dict]:
+        conteudo = f"{registro.get('ementa', '')} {registro.get('integra_do_acordao', '')}"
+        texto_normalizado = self._normalizar(conteudo)
+        destaques_encontrados = self._extrair_chunks(texto_normalizado)
+        
+        tags_para_saida = []
+        for i, texto_destaque in enumerate(destaques_encontrados, 1):
+            tag_id = f"{id_processo}-{i}" 
+            tags_para_saida.append({ "id": tag_id, "id_processo": id_processo, "texto": texto_destaque })
+        return tags_para_saida
+
+# ===================================================================================
+# CLASSE DE EXTRAÇÃO WEB
+# ===================================================================================
 class ExtratorUltraSimples:
-    def __init__(self): self.base = "https://portal.tjpr.jus.br"
-    def _limpa(self, t): return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", t.replace("&nbsp;", " ").replace("&quot;", '"').replace("&#39;", "'"))).strip()
-    def extrair_acordaos(self, html):
-        res = []
-        for tbl in re.findall(r'<table[^>]*class=["\']?[^"\']*resultTable[^"\']*["\']?[^>]*>(.*?)</table>', html, re.DOTALL | re.IGNORECASE):
-            ac = {}
-            for tr in re.findall(r"<tr[^>]*>(.*?)</tr>", tbl, re.DOTALL | re.IGNORECASE):
-                if m := re.search(r"tjpr\.url\.crypto=([a-f0-9]+)", tr): ac["url_documento"] = f"{self.base}/jurisprudencia/publico/visualizacao.do?tjpr.url.crypto={m.group(1)}"
-                if m := re.search(r"<b>([^<]+)</b>", tr, re.IGNORECASE):
-                    k = re.sub(r"[^\w]", "", unicodedata.normalize('NFKD', m.group(1).strip()).encode('ascii','ignore').decode('ascii').lower())
-                    if v := re.search(r"</b>\s*(.*?)</td>", tr, re.DOTALL | re.IGNORECASE): ac[k] = self._limpa(v.group(1))
-            if ac: res.append(ac)
-        return res
+    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); return f"{self.base_url}/jurisprudencia/publico/visualizacao.do?tjpr.url.crypto={match.group(1)}" if match else 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 = {};
+        for linha_html in self.extrair_linhas_tr(tabela_html):
+            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: return [acordao for tabela in self.extrair_tabelas(html) if (acordao := self.extrair_acordao(tabela))]
 
 # ===================================================================================
-# WORKER v8.4 (SINGLE WORKER OPTIMIZED)
+# CLASSE PRINCIPAL DO WORKER
 # ===================================================================================
 class AbelhaAtomica:
     def __init__(self):
-        if not GIT_TOKEN or not GITHUB_REPO: raise ValueError("GIT_TOKEN/REPO required")
+        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 = ExtratorUltraSimples()
-        self.destaques = DestaqueExtractor()
+        self.extrator_html = ExtratorUltraSimples()
+        self.destaque_extractor = DestaqueExtractor()
+        self.stats = EstatisticaManager()
         self.session = requests.Session(); self.session.headers.update({"User-Agent": f"Mozilla/5.0 ({self.worker_id})"})
         
-        self.fila, self.res = Queue(), Queue()
-        self.lk_git, self.lk_state, self.lk_stats = Lock(), Lock(), Lock()
-        self.th_push = None
+        self.fila_de_tarefas, self.fila_de_resultados = Queue(), Queue()
+        self.lock_git, self.lock_estado = Lock(), Lock()
+        self.thread_de_push = None
         
         self.paginas_status, self.processos_vistos = {}, set()
-        self.buf_regs, self.buf_tags = [], []
-        # Buf de páginas novas para append no CSV
-        self.buf_paginas_novas = {} 
-        
-        self.lote_n, self.prox_id = 1, 1
-        self.st_pg, self.st_proc, self.tot_pg = [], [], 0
-        
-        logger.info(f"--- WORKER {self.worker_id} (v8.4 - Single Worker) ---")
-        self._init_repo()
+        self.registros_acumulados, self.tags_destaques_acumuladas, self.html_bruto_acumulado = [], [], {}
+        self.num_lote_saida, self.proximo_id_registro = 1, 1
+
+        logger.info(f"Inicializando Abelha {self.worker_id} (v7.8 - Resiliente)...")
+        self._setup_git_repo()
+        logger.info(f"Abelha pronta. Estado: {len(self.paginas_status)} págs, {len(self.processos_vistos)} processos.")
 
-    def _init_repo(self):
-        url = f"https://oauth2:{GIT_TOKEN}@github.com/{GITHUB_REPO}.git"
-        if not LOCAL_REPO_PATH.exists(): Repo.clone_from(url, LOCAL_REPO_PATH, branch=GITHUB_BRANCH)
+    def _setup_git_repo(self):
+        remote_url = f"https://oauth2:{GIT_TOKEN}@github.com/{GITHUB_REPO}.git"
+        if not LOCAL_REPO_PATH.exists(): 
+            self.repo = Repo.clone_from(remote_url, LOCAL_REPO_PATH, branch=GITHUB_BRANCH)
         else: 
             self.repo = Repo(LOCAL_REPO_PATH)
-            try: os.remove(LOCAL_REPO_PATH/".git"/"index.lock")
-            except: pass
-            self.repo.git.remote("set-url", "origin", url)
+            self.repo.remotes.origin.set_url(remote_url)
         
-        with self.repo.config_writer() as c: c.set_value("user", "name", self.worker_id); c.set_value("user", "email", "bot@local")
-        
-        # CARGA INICIAL (FUNDAMENTAL)
-        with self.lk_git:
-            logger.info("Sincronizando estado inicial...")
-            self.repo.git.fetch("origin")
-            self.repo.git.reset("--hard", f"origin/{GITHUB_BRANCH}")
-            
-            # Lê CSVs para memória
-            f_pg = LOCAL_REPO_PATH / "paginas_status.csv"
-            if not f_pg.exists(): open(f_pg, 'w').write("pagina,status\n")
-            with open(f_pg, 'r') as f:
-                r = csv.reader(f); next(r, None)
-                self.paginas_status = {int(row[0]): row[1] for row in r if row}
+        with self.repo.config_writer() as config:
+            config.set_value("pull", "rebase", "true")
+            config.set_value("user", "email", "worker@tjpr.bot")
+            config.set_value("user", "name", "TJPR Worker")
+
+        self._carregar_estado_do_repo()
+
+    def _carregar_estado_do_repo(self):
+        with self.lock_git:
+            logger.info("Sincronizando (Pull inicial)...")
+            try:
+                self.repo.remotes.origin.pull()
+            except Exception as e:
+                logger.warning(f"Pull inicial falhou (continuando com estado local se existir): {e}")
+
+            path_paginas = LOCAL_REPO_PATH / "paginas_status.csv"
+            if not path_paginas.exists():
+                with open(path_paginas, 'w', newline='', encoding='utf-8') as f: csv.writer(f).writerow(['pagina', 'status'])
+            with open(path_paginas, 'r', encoding='utf-8') as f: reader = csv.reader(f); next(reader, None); self.paginas_status = {int(row[0]): row[1] for row in reader if row}
             
-            f_pr = LOCAL_REPO_PATH / "processos_vistos.csv"
-            if not f_pr.exists(): open(f_pr, 'w').write("id,processo_25\n")
-            with open(f_pr, 'r') as f:
-                r = csv.reader(f); next(r, None)
-                ids = []
-                for row in r:
-                    if len(row)>1: 
-                        if row[0].isdigit(): ids.append(int(row[0]))
-                        self.processos_vistos.add(row[1])
-                if ids: self.prox_id = max(ids)+1
-
-            # Lote
-            new_doce = LOCAL_REPO_PATH / "new_doce"; new_doce.mkdir(exist_ok=True)
-            nums = [int(re.search(r'(\d+)', f.name).group(1)) for f in new_doce.glob("lote_*.tar.gz") if re.search(r'(\d+)', f.name)]
-            if nums: self.lote_n = max(nums) + 1
+            path_processos = LOCAL_REPO_PATH / "processos_vistos.csv"
+            if not path_processos.exists():
+                with open(path_processos, 'w', newline='', encoding='utf-8') as f: csv.writer(f).writerow(['id', 'processo_25'])
+            with open(path_processos, 'r', encoding='utf-8') as f:
+                reader = csv.reader(f); next(reader, None); ids = [int(row[0]) for row in reader if len(row) > 1 and row[0].isdigit() and self.processos_vistos.add(row[1]) is None]
+                if ids: self.proximo_id_registro = max(ids) + 1
             
-            logger.info(f"Memória carregada: {len(self.paginas_status)} pgs, {len(self.processos_vistos)} procs.")
+            path_saida = LOCAL_REPO_PATH / "new_doce"; path_saida.mkdir(exist_ok=True)
+            arquivos = list(path_saida.glob("lote_*.tar.gz"))
+            if arquivos:
+                numeros = [int(re.search(r'(\d+)', f.name).group(1)) for f in arquivos if re.search(r'(\d+)', f.name)]
+                if numeros: self.num_lote_saida = max(numeros) + 1
 
-    # --- Stats ---
-    def _metric(self, t, n=1):
-        now = time.time()
-        with self.lk_stats:
-            if t=='pg': self.st_pg.append(now); self.tot_pg += 1
-            else: self.st_proc.extend([now]*n)
+    def _selecionar_lote_de_paginas(self):
+        paginas_para_trabalhar = [p for p in range(MIN_PAGINA, MAX_PAGINA + 1) if p not in self.paginas_status]
+        if not paginas_para_trabalhar: return False
+        lote = paginas_para_trabalhar[:TAMANHO_LOTE_PAGINAS]
+        for pagina in lote: self.fila_de_tarefas.put(pagina)
+        logger.info(f"➕ Lote de {len(lote)} páginas adicionado à fila.")
+        return True
 
-    def _upd_stats(self):
-        now = time.time()
-        with self.lk_stats:
-            self.st_pg = [x for x in self.st_pg if x > now-7500]
-            self.st_proc = [x for x in self.st_proc if x > now-7500]
-            def avg(l, m): return round(sum(1 for x in l if x >= now - m*60)/m, 2)
-            s = {
-                "worker": self.worker_id, "last": time.strftime("%H:%M:%S"),
-                "sessao_pg": self.tot_pg, "db_procs": len(self.processos_vistos),
-                "fila": self.fila.qsize(), "buf": len(self.buf_regs),
-                "rate_pg": {k: avg(self.st_pg, v) for k,v in [("5m",5),("30m",30),("2h",120)]},
-                "rate_proc": {k: avg(self.st_proc, v) for k,v in [("5m",5),("30m",30),("2h",120)]}
-            }
-        try: 
-            tmp = STATS_FILE.with_suffix(".tmp"); json.dump(s, open(tmp,'w')); tmp.replace(STATS_FILE)
-        except: pass
-        return s
-
-    # --- Flow ---
-    def _fill(self):
-        p = [x for x in range(MIN_PAGINA, MAX_PAGINA+1) if x not in self.paginas_status]
-        if not p: return False
-        for x in p[:TAMANHO_LOTE_PAGINAS]: self.fila.put(x)
-        logger.info(f"Fila +{len(p[:TAMANHO_LOTE_PAGINAS])}"); return True
-
-    def _worker(self):
-        time.sleep(random.random())
+    def _gerar_hash(self, texto: str) -> str: return hashlib.sha256(texto.encode('utf-8')).hexdigest() if texto else None
+    def _extrair_crypto_url(self, url: str) -> str: match = re.search(r"tjpr\.url\.crypto=([a-f0-9]+)", url); return match.group(1) if match else None
+    def extrair_processo_25(self, processo_str: str) -> str: return re.sub(r'[^a-zA-Z0-9\-.]', '', str(processo_str))[:25]
+
+    def _transformar_e_normalizar(self, acordao: dict, pagina: int, registro_id: int) -> dict:
+        processo_completo, ementa, integra = acordao.get("processo", ""), acordao.get("ementa", ""), acordao.get("integra_do_acordao", "")
+        pacote_principal = { "Id": registro_id, "pagina": pagina, "processo": self.extrair_processo_25(processo_completo), "url_documento": self._extrair_crypto_url(acordao.get("url_documento")), "relatora": acordao.get("relatora", acordao.get("relator(a)", "")), "orgao_julgador": acordao.get("orgao_julgador", ""), "comarca": acordao.get("comarca", acordao.get("comarca_de_origem", "")), "data_do_julgamento": acordao.get("data_do_julgamento", ""), "tipo_mov": acordao.get("tipo_mov", ""), "hash_interior_teor": self._gerar_hash(integra), "hash_ementa": self._gerar_hash(ementa) }
+        pacote_ementa = {"hash_texto": pacote_principal["hash_ementa"], "texto": ementa, "id_processo": registro_id} if ementa else None
+        pacote_integra = {"hash_texto": pacote_principal["hash_interior_teor"], "texto": integra, "id_processo": registro_id} if integra else None
+        return {"principal": pacote_principal, "ementa": pacote_ementa, "integra": pacote_integra}
+
+    def validar_registro(self, reg):
+        processo_completo, processo_25 = reg.get("processo", ""), self.extrair_processo_25(reg.get("processo", ""))
+        if not processo_25 or processo_25 in self.processos_vistos: return False
+        url, ementa, integra = reg.get("url_documento", ""), reg.get("ementa", ""), reg.get("integra_do_acordao", "")
+        if not any([url, ementa, integra]): return False
+        ementa_upper, integra_upper = ementa.upper().strip(), 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):
+        time.sleep(random.uniform(0.5, 2.0))
         while True:
             try:
-                pg = self.fila.get(timeout=3)
-                if pg in self.paginas_status: 
-                    self.res.put(('ign', pg)) # Já processado (duplicado na fila ou memória)
-                else:
-                    try:
-                        data = {"actionType": "pesquisar", "idLocalPesquisa": "99", "pageSize": "50", "pageNumber": str(pg), "sortColumn": "processos.dataJulgamento", "sortOrder": "asc", "mostrarCompleto": "true"}
-                        # Retry loop simplificado
-                        acs = None
-                        for _ in range(MAX_TENTATIVAS_EXTRACAO):
-                            try:
-                                r = self.session.post("https://portal.tjpr.jus.br/jurisprudencia/publico/pesquisa.do", data=data, timeout=45)
-                                if r.status_code == 200: acs = self.extrator.extrair_acordaos(r.text); break
-                            except: time.sleep(5)
-                        
-                        if acs is None: raise Exception("Max retries")
-                        
-                        if not acs: self.res.put(('vazio', pg))
-                        else:
-                            for a in acs: self.res.put(('ok', pg, a))
-                            self.res.put(('fim_pg', pg))
-                        self._metric('pg')
-                    except: self.res.put(('err', pg))
-                self.fila.task_done()
+                pagina = self.fila_de_tarefas.get(timeout=3); self._processar_pagina_extracao(pagina); self.fila_de_tarefas.task_done()
             except Empty: break
-            except: self.fila.task_done()
+            except Exception as e: logger.error(f"Erro worker: {e}"); self.fila_de_tarefas.task_done()
+
+    def extrair_dados_pagina(self, pagina: int) -> tuple[list, str]:
+        form_data = {"actionType": "pesquisar", "criterioPesquisa": "", "idLocalPesquisa": "99", "pageSize": "50", "pageNumber": str(pagina), "sortColumn": "processos.dataJulgamento", "sortOrder": "asc", "segredoJustica": "pesquisar sem", "mostrarCompleto": "true"}
+        for tentativa 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), response.text
+            except requests.RequestException: time.sleep(5)
+        raise ConnectionError(f"Excedido o número de tentativas para a página {pagina}.")
+        
+    def _processar_pagina_extracao(self, pagina: int):
+        # Removido logger.info por página para limpar output
+        try:
+            acordaos_brutos, html_conteudo = self.extrair_dados_pagina(pagina)
+            self.fila_de_resultados.put({'tipo': 'html_bruto', 'pagina': pagina, 'conteudo': html_conteudo})
+            if not acordaos_brutos:
+                self.fila_de_resultados.put({'tipo': 'status_pagina', 'pagina': pagina, 'status': 'vazio'}); return
+            num_validos = sum(1 for ac in acordaos_brutos if self.validar_registro(self._enriquecer_acordao(ac)) and self.fila_de_resultados.put({'tipo': 'acordao', 'dados': ac, 'pagina': pagina}) is None)
+            self.fila_de_resultados.put({'tipo': 'status_pagina', 'pagina': pagina, 'status': 'sucesso'})
+        except Exception: 
+            self.fila_de_resultados.put({'tipo': 'status_pagina', 'pagina': pagina, 'status': 'falha'})
 
-    def _norm(self, ac, pg, rid):
-        p25 = re.sub(r'[^a-zA-Z0-9\-.]', '', str(ac.get("processo","")))[:25]
-        h = lambda x: hashlib.sha256(x.encode('utf-8')).hexdigest() if x else None
-        princ = {"Id": rid, "pagina": pg, "processo": p25, "url": ac.get("url_documento"), "rel": ac.get("relatora"), "org": ac.get("orgao_julgador"), "dt": ac.get("data_do_julgamento"), "hash_int": h(ac.get("integra_do_acordao")), "hash_em": h(ac.get("ementa"))}
-        return {"principal": princ, "ementa": {"id_proc": rid, "txt": ac.get("ementa")} if princ["hash_em"] else None, "integra": {"id_proc": rid, "txt": ac.get("integra_do_acordao")} if princ["hash_int"] else None}
+    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 _push(self, lote, tags, pgs_novas, procs_novos):
-        with self.lk_git:
-            logger.info(f"🚀 Push Lote {self.lote_n} ({len(lote)} docs)...")
+    def _consumidor_push_para_git(self, lote_pacotes, lote_tags, lote_html, paginas_atualizadas, processos_novos):
+        with self.lock_git:
+            logger.info(f"📤 [PUSH] Preparando Lote {self.num_lote_saida}: {len(lote_pacotes)} registros de {len(paginas_atualizadas)} páginas...")
+            
             try:
-                bn = f"{self.lote_n:05d}"
-                with tarfile.open(LOCAL_REPO_PATH / "new_doce" / f"lote_{bn}.tar.gz", "w:gz") as tar:
-                    def w(n, d): b = io.BytesIO(json.dumps(d, ensure_ascii=False).encode('utf8')); ti = tarfile.TarInfo(n); ti.size = len(b.getvalue()); tar.addfile(ti, b)
-                    # Escreve um JSON por linha em cada arquivo dentro do tar (formato JSONL)
-                    buf_a, buf_e, buf_i, buf_t = io.StringIO(), io.StringIO(), io.StringIO(), io.StringIO()
-                    for x in lote:
-                        buf_a.write(json.dumps(x['principal'], ensure_ascii=False)+'\n')
-                        if x['ementa']: buf_e.write(json.dumps(x['ementa'], ensure_ascii=False)+'\n')
-                        if x['integra']: buf_i.write(json.dumps(x['integra'], ensure_ascii=False)+'\n')
-                    for t in tags: buf_t.write(json.dumps(t, ensure_ascii=False)+'\n')
+                # 1. Preparar Arquivo .tar.gz
+                path_saida = LOCAL_REPO_PATH / "new_doce"; path_saida.mkdir(exist_ok=True)
+                base_name = f"{self.num_lote_saida:05d}"; archive_path = path_saida / f"lote_{base_name}.tar.gz"
+                
+                with tarfile.open(archive_path, "w:gz") as tar:
+                    buffers = { 
+                        f"acordaos_{base_name}.jsonl": io.StringIO(), 
+                        f"ementa_{base_name}.jsonl": io.StringIO(), 
+                        f"integra_do_acordao_{base_name}.jsonl": io.StringIO(), 
+                        f"tags_destaques_{base_name}.jsonl": io.StringIO() 
+                    }
+                    for pacote in lote_pacotes:
+                        buffers[f"acordaos_{base_name}.jsonl"].write(json.dumps(pacote['principal'], ensure_ascii=False) + '\n')
+                        if pacote['ementa']: buffers[f"ementa_{base_name}.jsonl"].write(json.dumps(pacote['ementa'], ensure_ascii=False) + '\n')
+                        if pacote['integra']: buffers[f"integra_do_acordao_{base_name}.jsonl"].write(json.dumps(pacote['integra'], ensure_ascii=False) + '\n')
+                    for tag in lote_tags: buffers[f"tags_destaques_{base_name}.jsonl"].write(json.dumps(tag, ensure_ascii=False) + '\n')
                     
-                    def add_buf(nome, buf): b = buf.getvalue().encode('utf8'); ti = tarfile.TarInfo(nome); ti.size = len(b); tar.addfile(ti, io.BytesIO(b))
-                    add_buf(f"acordaos_{bn}.jsonl", buf_a); add_buf(f"ementa_{bn}.jsonl", buf_e)
-                    add_buf(f"integra_{bn}.jsonl", buf_i); add_buf(f"tags_{bn}.jsonl", buf_t)
+                    for filename, buffer in buffers.items():
+                        data = buffer.getvalue().encode('utf-8')
+                        tarinfo = tarfile.TarInfo(name=filename); tarinfo.size = len(data)
+                        tar.addfile(tarinfo, io.BytesIO(data))
+
+                # 2. Atualizar CSVs (Modo Append/Rewrite local)
+                with open(LOCAL_REPO_PATH / "paginas_status.csv", 'w', newline='', encoding='utf-8') as f:
+                    writer = csv.writer(f); writer.writerow(['pagina', 'status'])
+                    writer.writerows(sorted(self.paginas_status.items())) # Escreve todo o estado conhecido
+                
+                with open(LOCAL_REPO_PATH / "processos_vistos.csv", 'a', newline='', encoding='utf-8') as f:
+                    writer = csv.writer(f)
+                    writer.writerows(sorted(processos_novos.items(), key=lambda item: item[1]))
 
-                with self.lk_state: 
-                    self.buf_regs = []; self.buf_tags = []; self.buf_paginas_novas = {}
-                self.lote_n += 1
+                # 3. Estatísticas e Logs
+                self.stats.registrar(len(paginas_atualizadas), len(processos_novos))
+                dados_stats = self.stats.salvar_arquivo()
+                
+                # 4. Git Flow Resiliente
+                self.repo.git.add(["new_doce/", "paginas_status.csv", "processos_vistos.csv", "stats.json"])
                 
-                # GIT OPERATIONS
-                for i in range(3):
+                if self.repo.is_dirty():
+                    self.repo.index.commit(f"DATA: Lote {self.num_lote_saida} ({len(lote_pacotes)} regs)")
+                    
+                    try:
+                        self.repo.remotes.origin.pull(rebase=True)
+                        logger.info("✅ Pull (Rebase) OK.")
+                    except GitCommandError as e:
+                        logger.warning(f"⚠️ Pull falhou (conflito ou rede), mas commit local salvo: {e}")
+
                     try:
-                        try: os.remove(LOCAL_REPO_PATH/".git"/"index.lock")
-                        except: pass
-                        self.repo.git.fetch("origin")
-                        # Reset hard para garantir base limpa
-                        self.repo.git.reset("--hard", f"origin/{GITHUB_BRANCH}")
-                        
-                        # APPEND (Apenas novidades)
-                        with open(LOCAL_REPO_PATH/"paginas_status.csv", 'a', newline='', encoding='utf-8') as f:
-                            csv.writer(f).writerows(sorted(pgs_novas.items()))
-                            
-                        with open(LOCAL_REPO_PATH/"processos_vistos.csv", 'a', newline='', encoding='utf-8') as f:
-                            csv.writer(f).writerows(sorted(procs_novos.items(), key=lambda x: x[1]))
-                            
-                        self.repo.git.add(["new_doce/", "paginas_status.csv", "processos_vistos.csv"])
-                        self.repo.index.commit(f"Lote {bn}")
                         self.repo.remotes.origin.push()
+                        logger.info(f"🚀 PUSH do lote {self.num_lote_saida} concluído.")
+                        
+                        # Log Estatístico
+                        p = dados_stats['medias_paginas']
+                        pr = dados_stats['medias_processos']
+                        logger.info(f"📊 ESTATÍSTICAS: Total: {dados_stats['total_paginas']} págs | "
+                                    f"5m: {p['5min']}p/{pr['5min']}proc | "
+                                    f"30m: {p['30min']}p/{pr['30min']}proc | "
+                                    f"120m: {p['120min']}p/{pr['120min']}proc")
                         
-                        s = self._upd_stats()
-                        logger.info(f"✅ Push OK. Rate: {s['rate_pg']['5m']} pg/m")
-                        break
-                    except Exception as e: 
-                        logger.warning(f"Retry push {i}: {e}"); time.sleep(10)
-            except Exception as e: logger.critical(f"Push Fatal: {e}")
-        self.th_push = None
-
-    def run(self):
-        pool = ThreadPoolExecutor(NUM_WORKERS_EXTRACao)
-        last_s = time.time()
+                    except GitCommandError as e:
+                        logger.error(f"❌ Push falhou (será enviado na próxima rodada): {e}")
+
+                self.num_lote_saida += 1
+                with self.lock_estado: self.registros_acumulados.clear(); self.tags_destaques_acumuladas.clear(); self.html_bruto_acumulado.clear()
+
+            except Exception as e:
+                logger.critical(f"❌ FALHA CRÍTICA GERAL no push: {e}", exc_info=True)
+        
+        self.thread_de_push = None
+
+    def loop_infinito(self):
+        pool_produtores = ThreadPoolExecutor(max_workers=NUM_WORKERS_EXTRACao, thread_name_prefix='ProdutorWorker')
         while True:
             try:
-                if self.fila.empty(): self._fill()
-                while self.fila.qsize() and len(pool._threads) < NUM_WORKERS_EXTRACao: pool.submit(self._worker)
+                if self.fila_de_tarefas.empty() and self._selecionar_lote_de_paginas():
+                    for _ in range(NUM_WORKERS_EXTRACao): pool_produtores.submit(self._produtor_worker)
                 
-                # Consumo de resultados e atualização de memória
-                novos_procs = {}
-                # Novas páginas deste ciclo para append no CSV
-                # Nota: paginas_status (memoria) acumula tudo. pgs_novas é só para o CSV.
+                processos_novos_neste_ciclo = {}
+                paginas_atualizadas_ciclo = {}
                 
-                while not self.res.empty():
-                    t, *d = self.res.get_nowait()
-                    with self.lk_state:
-                        if t == 'ign': pass
-                        elif t in ['vazio', 'fim_pg', 'err']:
-                            pg = d[0]
-                            st = 'vazio' if t=='vazio' else ('sucesso' if t=='fim_pg' else 'falha')
-                            self.paginas_status[pg] = st
-                            self.buf_paginas_novas[pg] = st # Guarda para o CSV
-                        elif t == 'ok':
-                            pg, ac = d
-                            p25 = re.sub(r'[^a-zA-Z0-9\-.]', '', str(ac.get("processo","")))[:25]
-                            if p25 and p25 not in self.processos_vistos:
-                                rid = self.prox_id
-                                self.buf_regs.append(self._norm(ac, pg, rid))
-                                self.buf_tags.extend(self.destaques.extrair_destaques(ac, rid))
-                                self.processos_vistos.add(p25)
-                                novos_procs[p25] = rid # Guarda para CSV
-                                self.prox_id += 1
-                                self._metric('proc')
-                
-                # Trigger Push
-                if len(self.buf_regs) >= TAMANHO_LOTE_REGISTROS and not self.th_push:
-                    l = list(self.buf_regs); tg = list(self.buf_tags)
-                    pn = dict(self.buf_paginas_novas) # Cópia das novas páginas
-                    # Limpa o buffer de paginas novas após capturar para envio
-                    # (Se o push falhar, perdemos o registro no CSV mas a memória segura. 
-                    # Idealmente o buffer só limparia no sucesso, mas para simplificar v8.4 está ok)
+                # Consome fila de resultados
+                while not self.fila_de_resultados.empty():
+                    item = self.fila_de_resultados.get_nowait()
+                    with self.lock_estado:
+                        if item['tipo'] == 'status_pagina': 
+                            self.paginas_status[item['pagina']] = item['status']
+                            paginas_atualizadas_ciclo[item['pagina']] = item['status']
+                        elif item['tipo'] == 'html_bruto': 
+                            self.html_bruto_acumulado[item['pagina']] = item['conteudo']
+                        elif item['tipo'] == 'acordao':
+                            acordao = item['dados']; processo_25 = self.extrair_processo_25(acordao.get("processo"))
+                            if processo_25 and processo_25 not in self.processos_vistos:
+                                reg_id = self.proximo_id_registro
+                                self.registros_acumulados.append(self._transformar_e_normalizar(acordao, item['pagina'], reg_id))
+                                self.tags_destaques_acumuladas.extend(self.destaque_extractor.extrair_destaques(acordao, reg_id))
+                                self.processos_vistos.add(processo_25); processos_novos_neste_ciclo[processo_25] = reg_id; self.proximo_id_registro += 1
+
+                # Dispara thread de persistência
+                if len(self.registros_acumulados) >= TAMANHO_LOTE_REGISTROS and not self.thread_de_push:
+                    lote = list(self.registros_acumulados)
+                    tags = list(self.tags_destaques_acumuladas)
+                    html = dict(self.html_bruto_acumulado)
+                    paginas = dict(self.paginas_status) # Cópia do estado total
                     
-                    self.th_push = Thread(target=self._push, args=(l, tg, pn, novos_procs))
-                    self.th_push.start()
+                    self.thread_de_push = Thread(
+                        target=self._consumidor_push_para_git, 
+                        args=(lote, tags, html, paginas, processos_novos_neste_ciclo), 
+                        name="PushThread"
+                    )
+                    self.thread_de_push.start()
                 
-                if time.time()-last_s > 10: self._upd_stats(); last_s = time.time()
                 time.sleep(5)
-            except Exception as e: logger.critical(f"Loop Err: {e}"); time.sleep(10)
+            except Exception as e: logger.critical(f"ERRO CRÍTICO no loop principal: {e}", exc_info=True); time.sleep(15)
+
+def main():
+    try: AbelhaAtomica().loop_infinito()
+    except Exception as e: logger.critical(f"Falha fatal na inicialização: {e}", exc_info=True); exit(1)
 
-if __name__ == "__main__": AbelhaAtomica().run()
+if __name__ == "__main__":
+    main()