#!/usr/bin/env python3 """ ParaAI - Queue Director API (v5.0 — batch) ============================================ MUDANÇAS v5.0 — Comunicação em lote para reduzir 429: - [NEW] GET /queue/work/batch?worker_id=X&batch_size=N Entrega até N tasks em 1 única request (register_id + chunk_id incluídos). - [NEW] POST /queue/result/batch Recebe um lote de resultados em 1 única request. - [CHANGED] KEEPALIVE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_TIMEOUT_SEC: 900 → 550 s Worker considerado morto após 550 s sem keepalive. - [CHANGED] _requeue_timed_out(): requeue de tasks expiradas agora é por LOTE (agrupa por worker_id), não task por task. Herda de v4.0 (HTTP-only, sem clone Git): - GET CSV de bordo no startup - Topup/Reaproveitamento via HTTP - Upload HTTP direto (sem git push) """ import io as _io import json import logging import os import re import tarfile as _tarfile import threading import time from collections import deque as _deque from datetime import datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Optional import pandas as pd from fastapi import Depends, FastAPI, Header, HTTPException, Query from fastapi.responses import HTMLResponse, JSONResponse from pydantic import BaseModel logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)-5s] %(name)s: %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger("queue_director") from workers_tools import StateManager, ResultCollector from http_director_manager import HttpDirectorManager # ══════════════════════════════════════════════════════════════════════ # CONFIGURAÇÃO # ══════════════════════════════════════════════════════════════════════ def _env_int(name: str, default: int) -> int: try: return int(os.getenv(name, str(default))) except (TypeError, ValueError): return default def _env_float(name: str, default: float) -> float: try: return float(os.getenv(name, str(default))) except (TypeError, ValueError): return default QUEUE_API_TOKEN = os.getenv("QUEUE_API_TOKEN", "paraai-secret-token") GIT_TOKEN = os.getenv("GIT_TOKEN", "") GITHUB_REPO_SLUG = os.getenv("GITHUB_REPO_SLUG", "") GITHUB_BRANCH = os.getenv("GITHUB_BRANCH", "main") WORK_DIR = Path(os.getenv("WORK_DIR", "/data")) PROTOCOL_FILE = Path(os.getenv("PROTOCOL_FILE", "system_role.md")) STATE_CSV_NAME = os.getenv("STATE_CSV_NAME", "new_tracking_processamento_up.csv") SYNC_INTERVAL = _env_int("SYNC_INTERVAL_SEC", 320) PUSH_INTERVAL = _env_int("PUSH_INTERVAL_SEC", 30) TASK_TIMEOUT_MIN = _env_int("TASK_TIMEOUT_MIN", 320) ID_MINIMO_CHUNK = _env_int("ID_MINIMO_CHUNK", 1) MIN_CHUNKS_ATIVOS = _env_int("MIN_CHUNKS_ATIVOS", 40) MAX_CHUNKS_ATIVOS = _env_int("MAX_CHUNKS_ATIVOS", 45) BUFFER_MONITOR_INTERVAL = _env_int("BUFFER_MONITOR_INTERVAL", 20) MAX_RETRIES_PER_TASK = _env_int("MAX_RETRIES_PER_TASK", 3) UPDATE_ONLY_MODE = bool(int(os.getenv("UPDATE_ONLY_MODE", "1"))) # ← modo cirúrgico: só atualiza LLMs existentes KEEPALIVE_TIMEOUT_SEC = 320 # ← v5.0: era 900 — worker morto após 550 s sem KA # Registros cujo orgao_julgador bater neste padrão são ignorados (skip) silenciosamente. _ORGAO_IGNORADO_RE = re.compile(r".*?Vice.*?Presidência.*?", re.IGNORECASE | re.DOTALL) WORK_DIR.mkdir(parents=True, exist_ok=True) (WORK_DIR / "SILVER").mkdir(exist_ok=True) # ══════════════════════════════════════════════════════════════════════ # ESTADO GLOBAL (thread-safe) # ══════════════════════════════════════════════════════════════════════ _lock = threading.Lock() _task_queue: List[dict] = [] _in_progress: Dict[str, dict] = {} _chunk_collectors: Dict[str, object] = {} _chunk_counters: Dict[str, dict] = {} _pending_push: List[dict] = [] _chunk_payload_logs: Dict[str, List[dict]] = {} _chunk_error_logs: Dict[str, List[dict]] = {} _http_manager: Optional[HttpDirectorManager] = None _system_prompt: str = "" _initialized: bool = False _init_error: str = "" _start_time: Optional[datetime] = None _stats = { "tasks_entregues": 0, "tasks_ok": 0, "tasks_erro": 0, "tasks_skip": 0, "tasks_aceitas_sem_reserva": 0, "chunks_concluidos": 0, "tokens_total": 0, } _pressure_429: int = 0 _pressure_498: int = 0 _rpm_window: "_deque[float]" = _deque() _tpm_window: "_deque[tuple]" = _deque() _WINDOW_SEC = 60 _topup_trigger = threading.Event() # ══════════════════════════════════════════════════════════════════════ # MÉTRICAS DESLIZANTES # ══════════════════════════════════════════════════════════════════════ def _record_task_ok(tokens: int): now = datetime.utcnow().timestamp() _rpm_window.append(now) _tpm_window.append((now, tokens)) def _sliding_metrics() -> tuple: now = datetime.utcnow().timestamp() cutoff = now - _WINDOW_SEC while _rpm_window and _rpm_window[0] < cutoff: _rpm_window.popleft() while _tpm_window and _tpm_window[0][0] < cutoff: _tpm_window.popleft() rpm = round(len(_rpm_window) / (_WINDOW_SEC / 60), 1) tpm = sum(t for _, t in _tpm_window) return rpm, tpm # ══════════════════════════════════════════════════════════════════════ # HELPERS INTERNOS # ══════════════════════════════════════════════════════════════════════ def _load_system_prompt() -> str: if PROTOCOL_FILE.exists(): return PROTOCOL_FILE.read_text(encoding="utf-8") logger.warning("⚠️ system_role.md não encontrado — usando prompt padrão.") return "Você é um assistente jurídico especializado em análise de decisões." def _publish_chunk_llm(chunk_id: str, gz_path: Path, msg: str) -> bool: try: return _http_manager.upload_chunk_llm(chunk_id, gz_path.read_bytes(), msg) except Exception as e: logger.error(f"❌ [publish] chunk_llm {chunk_id}: {e}") return False def _publish_diario(msg: str = "update tracking") -> bool: csv_path = WORK_DIR / STATE_CSV_NAME if not csv_path.exists(): logger.error("❌ [publish] CSV não existe ainda") return False df = pd.read_csv(csv_path) return _http_manager.salvar_diario(df, msg) def _ler_jsonl_de_tar(tar_path: Path) -> list: registros = [] linhas_validas = [] # texto cru das linhas OK (para reescrita se necessário) n_corrompidas = 0 if not tar_path.exists(): return registros try: with _tarfile.open(tar_path, "r:gz") as tar: member = next( (m for m in tar.getmembers() if m.name.endswith("jurisprudencias.jsonl")), None ) if not member: return registros for raw_line in tar.extractfile(member): line = raw_line.decode("utf-8").strip() if not line: continue try: r = json.loads(line) except Exception: n_corrompidas += 1 logger.warning(f"⚠️ [tar] {tar_path.name}: JSON inválido — linha removida.") continue if not isinstance(r, dict): n_corrompidas += 1 logger.warning( f"⚠️ [tar] {tar_path.name}: tipo inesperado {type(r).__name__} " f"(esperado dict) — linha removida." ) continue registros.append(r) linhas_validas.append(line) except Exception as e: logger.warning(f"⚠️ [tar] {tar_path.name}: {e}") return registros if n_corrompidas > 0: logger.warning( f"🧹 [tar] {tar_path.name}: {n_corrompidas} linha(s) corrompida(s) removida(s). " f"Reescrevendo tar local com {len(linhas_validas)} linhas válidas." ) _reescrever_tar_jsonl(tar_path, linhas_validas) return registros def _reescrever_tar_jsonl(tar_path: Path, linhas_validas: list): """Reconstrói o tar.gz preservando todos os membros exceto jurisprudencias.jsonl, que é substituído pelas linhas saudáveis.""" try: outros_membros: dict = {} with _tarfile.open(tar_path, "r:gz") as tar: for m in tar.getmembers(): if not m.name.endswith("jurisprudencias.jsonl"): f = tar.extractfile(m) if f: outros_membros[m.name] = f.read() novo_jsonl = "\n".join(linhas_validas).encode("utf-8") tmp_path = tar_path.with_suffix(".tmp.gz") with _tarfile.open(tmp_path, "w:gz") as tar: info = _tarfile.TarInfo(name="jurisprudencias.jsonl") info.size = len(novo_jsonl) tar.addfile(info, _io.BytesIO(novo_jsonl)) for name, data in outros_membros.items(): info = _tarfile.TarInfo(name=name) info.size = len(data) tar.addfile(info, _io.BytesIO(data)) tmp_path.replace(tar_path) logger.info(f"✅ [tar] {tar_path.name}: reescrito sem linhas corrompidas.") except Exception as e: logger.error(f"❌ [tar] Falha ao reescrever {tar_path.name}: {e}") def _ler_jsonl_de_bytes(gz_bytes: bytes) -> tuple: """Retorna (registros: list[dict], linhas_validas: list[str], n_corrompidas: int). Linhas que não são dict válido são descartadas e contadas.""" registros = [] linhas_validas = [] n_corrompidas = 0 try: with _tarfile.open(fileobj=_io.BytesIO(gz_bytes), mode="r:gz") as tar: member = next( (m for m in tar.getmembers() if m.name.endswith("jurisprudencias.jsonl")), None ) if not member: return registros, linhas_validas, n_corrompidas for raw_line in tar.extractfile(member): line = raw_line.decode("utf-8").strip() if not line: continue try: r = json.loads(line) except Exception: n_corrompidas += 1 logger.warning("⚠️ [tar/mem]: JSON inválido — linha removida.") continue if not isinstance(r, dict): n_corrompidas += 1 logger.warning( f"⚠️ [tar/mem]: tipo inesperado {type(r).__name__} " f"(esperado dict) — linha removida." ) continue registros.append(r) linhas_validas.append(line) except Exception as e: logger.warning(f"⚠️ [tar/mem] {e}") return registros, linhas_validas, n_corrompidas def _reuplodar_chunk_llm_limpo(chunk_id: str, gz_bytes_original: bytes, linhas_validas: list): """Reconstrói o tar.gz do chunk_llm sem linhas corrompidas e faz re-upload para o GitHub.""" try: outros_membros: dict = {} with _tarfile.open(fileobj=_io.BytesIO(gz_bytes_original), mode="r:gz") as tar: for m in tar.getmembers(): if not m.name.endswith("jurisprudencias.jsonl"): f = tar.extractfile(m) if f: outros_membros[m.name] = f.read() novo_jsonl = "\n".join(linhas_validas).encode("utf-8") buf = _io.BytesIO() with _tarfile.open(fileobj=buf, mode="w:gz") as tar: info = _tarfile.TarInfo(name="jurisprudencias.jsonl") info.size = len(novo_jsonl) tar.addfile(info, _io.BytesIO(novo_jsonl)) for name, data in outros_membros.items(): info = _tarfile.TarInfo(name=name) info.size = len(data) tar.addfile(info, _io.BytesIO(data)) gz_limpo = buf.getvalue() ok = _http_manager.upload_chunk_llm( chunk_id, gz_limpo, f"🧹 limpeza linhas corrompidas chunk {chunk_id}" ) if ok: logger.info(f"✅ [limpeza] chunk_llm {chunk_id}: re-upload limpo OK.") else: logger.error(f"❌ [limpeza] Falha no re-upload do chunk_llm {chunk_id}.") except Exception as e: logger.error(f"❌ [limpeza] Erro ao reconstruir chunk_llm {chunk_id}: {e}") def _ler_raw_md_de_tar(tar_path: Path) -> dict: resultado = {} if not tar_path.exists(): return resultado try: with _tarfile.open(tar_path, "r:gz") as tar: for member in tar.getmembers(): if member.name.startswith("raw/") and member.name.endswith(".md"): f = tar.extractfile(member) if f: resultado[member.name[4:-3]] = f.read().decode("utf-8") except Exception as e: logger.warning(f"⚠️ [raw/md] {tar_path.name}: {e}") return resultado def _ler_raw_md_de_bytes(gz_bytes: bytes) -> dict: resultado = {} try: with _tarfile.open(fileobj=_io.BytesIO(gz_bytes), mode="r:gz") as tar: for member in tar.getmembers(): if member.name.startswith("raw/") and member.name.endswith(".md"): f = tar.extractfile(member) if f: resultado[member.name[4:-3]] = f.read().decode("utf-8") except Exception as e: logger.warning(f"⚠️ [raw/md/mem] {e}") return resultado def _carregar_registros_existentes_llm(chunk_id: str): work_tar = WORK_DIR / f"dados_llm_v3/chunk_llm_{chunk_id}.tar.gz" registros = _ler_jsonl_de_tar(work_tar) # já limpa e reescreve local se necessário if not registros: gz_bytes = _http_manager.baixar_chunk_llm(chunk_id) if gz_bytes: registros, linhas_validas, n_corrompidas = _ler_jsonl_de_bytes(gz_bytes) if registros: logger.info(f"📂 [reaprov] Chunk {chunk_id}: {len(registros)} registros do GitHub.") if n_corrompidas > 0: logger.warning( f"🧹 [reaprov] Chunk {chunk_id}: {n_corrompidas} linha(s) corrompida(s) " f"no GitHub — iniciando re-upload limpo." ) _reuplodar_chunk_llm_limpo(chunk_id, gz_bytes, linhas_validas) ids_existentes: set = set() por_id: dict = {} for r in registros: mj = r.get("manifestacao_juridica", {}) mid = None if isinstance(mj, dict): mid = mj.get("id_manifestacao") if mid is None: # fallback: id_manifestacao no nível raiz mid = r.get("id_manifestacao") if mid is not None: key = str(mid) ids_existentes.add(key) por_id[key] = r return registros, ids_existentes, por_id def _tem_referencias_pedidos(registro: dict) -> bool: """Retorna True somente se o registro está completo e sem campos placeholder '...'. Verifica três caminhos críticos dentro de manifestacao_juridica: 1. referencias_pedidos_canonicos[0].identificador_unico_pedido_atomico 2. fundamentacao.teses_relator[0].identificador_unico_fundamento_atomico 3. fatos_pedidos_nexos_causa.pedidos[0].causalidade_futura_positiva_pedido.cordas[0] Se qualquer um desses campos existir e for "..." (equivalente a null/vazio), o registro é considerado incompleto e será recolocado na fila para atualização. """ _PLACEHOLDER = "..." mj = registro.get("manifestacao_juridica") # ── campo raiz deve existir ────────────────────────────────────────────── if not isinstance(mj, dict): return False # ── 1. referencias_pedios_canonicos ──────────────────────────────────── metadados = mj.get("metadados") if isinstance(metadados, dict): relator_ou_juiz = metadados.get("relator_ou_juiz") if relator_ou_juiz is not None and relator_ou_juiz != _PLACEHOLDER: pass else: logger.debug("[tem_refs] registro incompleto — metadados.relator_ou_juiz é None ou placeholder") return False else: logger.debug("[tem_refs] metadados não é um dicionário") return False # ── 2. fundamentacao.teses_relator ────────────────────────────────────── fundamentacao = mj.get("fundamentacao") if isinstance(fundamentacao, dict): teses = fundamentacao.get("teses_relator") if isinstance(teses, list) and len(teses) > 0: id_fund = teses[0].get("identificador_unico_fundamento_atomico") if isinstance(teses[0], dict) else None if id_fund == _PLACEHOLDER: logger.debug( f"[tem_refs] registro incompleto — teses_relator[0] é placeholder" ) return False # ── 3. fatos_pedidos_nexos_causa.pedidos[0].causalidade_futura_positiva_pedido.cordas[0] ── fpnc = mj.get("fatos_pedidos_nexos_causa") if isinstance(fpnc, dict): pedidos = fpnc.get("pedidos") if isinstance(pedidos, list) and len(pedidos) > 0: pedido0 = pedidos[0] if isinstance(pedidos[0], dict) else {} cfpp = pedido0.get("causalidade_futura_positiva_pedido") if isinstance(cfpp, dict): cordas = cfpp.get("cordas") if isinstance(cordas, list) and len(cordas) > 0: if cordas[0] == _PLACEHOLDER: logger.debug( f"[tem_refs] registro incompleto — cordas[0] é placeholder" ) return False return True def _tem_referencias_pedidos1(registro: dict) -> bool: """Retorna True somente se o registro está completo e sem campos placeholder '...'. Verifica três caminhos críticos dentro de manifestacao_juridica: 1. referencias_pedidos_canonicos[0].identificador_unico_pedido_atomico 2. fundamentacao.teses_relator[0].identificador_unico_fundamento_atomico 3. fatos_pedidos_nexos_causa.pedidos[0].causalidade_futura_positiva_pedido.cordas[0] Se qualquer um desses campos existir e for "..." (equivalente a null/vazio), o registro é considerado incompleto e será recolocado na fila para atualização. """ _PLACEHOLDER = "..." mj = registro.get("manifestacao_juridica") # ── campo raiz deve existir ────────────────────────────────────────────── if not isinstance(mj, dict): return False # ── 3. fatos_pedidos_nexos_causa.pedidos[0].causalidade_futura_positiva_pedido.cordas[0] ── fpnc = mj.get("fatos_pedidos_nexos_causa") if isinstance(fpnc, dict): pedidos = fpnc.get("pedidos") if isinstance(pedidos, list) and len(pedidos) > 0: pedido0 = pedidos[0] if isinstance(pedidos[0], dict) else {} cfpp = pedido0.get("causalidade_futura_positiva_pedido") if isinstance(cfpp, dict): cordas = cfpp.get("cordas") if isinstance(cordas, list) and len(cordas) > 0: if cordas[0] == _PLACEHOLDER: logger.debug( f"[tem_refs] registro incompleto — cordas[0] é placeholder" ) return False return True # ══════════════════════════════════════════════════════════════════════ # CARREGAMENTO DE CHUNKS # ══════════════════════════════════════════════════════════════════════ def _carregar_chunk_na_fila(chunk_id: str) -> int: chunk_id = str(int(chunk_id)).zfill(6) dados_bytes = _http_manager.baixar_chunk_dados(chunk_id) if not dados_bytes: logger.warning(f"⚠️ [carregar] Chunk {chunk_id}: não encontrado no GitHub.") return 0 tarefas = _http_manager.extrair_tarefas_de_chunk_dados(chunk_id, dados_bytes) if not tarefas: logger.warning(f"⚠️ [carregar] Chunk {chunk_id}: sem tarefas válidas.") return 0 _, ids_existentes, existentes_por_id = _carregar_registros_existentes_llm(chunk_id) # ── Filtro: orgao_julgador bloqueado ───────────────────────────────────── tarefas_ignoradas: List[dict] = [] tarefas_permitidas: List[dict] = [] for t in tarefas: orgao = str(t.get("dados_originais", {}).get("orgao_julgador") or "") if _ORGAO_IGNORADO_RE.fullmatch(orgao): tarefas_ignoradas.append(t) else: tarefas_permitidas.append(t) if tarefas_ignoradas: logger.info( f"🚫 [filtro] Chunk {chunk_id}: {len(tarefas_ignoradas)} registro(s) ignorado(s) " f"por orgao_julgador bloqueado (Vice-Presidência)." ) tarefas = tarefas_permitidas # ───────────────────────────────────────────────────────────────────────── tarefas_faltantes: List[dict] = [] tarefas_pre_existentes: List[dict] = [] tarefas_novas_ignoradas: List[dict] = [] # UPDATE_ONLY_MODE: novos sem LLM existente for t in tarefas: if str(t["id"]) in ids_existentes: registro = existentes_por_id[str(t["id"])] if _tem_referencias_pedidos(registro): tarefas_pre_existentes.append(t) else: t_upd = dict(t) t_upd["dados_llm_existente"] = registro tarefas_faltantes.append(t_upd) else: if UPDATE_ONLY_MODE: tarefas_novas_ignoradas.append(t) else: tarefas_faltantes.append(t) n_novas_ignoradas = len(tarefas_novas_ignoradas) # UPDATE_ONLY_MODE total_original = len(tarefas) + len(tarefas_ignoradas) # mantém contagem real do chunk n_ignoradas = len(tarefas_ignoradas) n_existentes = len(tarefas_pre_existentes) n_faltantes = len(tarefas_faltantes) _modo_label = " [UPDATE_ONLY]" if UPDATE_ONLY_MODE else "" if n_existentes: logger.info( f"♻️ [reaprov{_modo_label}] Chunk {chunk_id}: " f"{n_existentes}/{total_original} pré-existentes | " f"{n_ignoradas} ignoradas | {n_faltantes} para atualizar" + (f" | {n_novas_ignoradas} novos pulados (update_only)" if UPDATE_ONLY_MODE and n_novas_ignoradas else "") + "." ) else: logger.info( f"🆕 [carregar{_modo_label}] Chunk {chunk_id}: {total_original} tarefas " f"({n_ignoradas} ignoradas" + (f", {n_novas_ignoradas} novos pulados" if UPDATE_ONLY_MODE and n_novas_ignoradas else "") + ")." ) if not tarefas_faltantes: logger.info(f"✅ [reaprov] Chunk {chunk_id}: completo. Marcando finalizado.") sm = StateManager(WORK_DIR, WORK_DIR) sm.marcar_chunk_como_finalizado(chunk_id, {"total_ok": n_existentes, "erros": []}) _publish_diario(f"✅ Chunk {chunk_id} já completo (reaproveitado)") return 0 collector = ResultCollector(WORK_DIR) # Registros ignorados por orgao_julgador bloqueado → contabilizados como skip for t in tarefas_ignoradas: mid_str = str(t["id"]) collector.adicionar_resultado( t, { "status": "skip", "resultado": {}, "tokens": 0, "motivo": "orgao_julgador_bloqueado", "_raw_markdown": "", "_id_manifestacao": mid_str, }, ) for t in tarefas_pre_existentes: mid_str = str(t["id"]) record = existentes_por_id.get(mid_str, {}) eh_erro = ( isinstance(record.get("manifestacao_juridica"), dict) and record["manifestacao_juridica"].get("error") ) collector.adicionar_resultado( t, { "status": "falha" if eh_erro else "sucesso", "resultado": record, "tokens": 0, "motivo": "preexistente_reaproveitado", "_raw_markdown": "", #_build_raw_markdown(t.get("dados_originais", {}), mid_str), "_id_manifestacao": mid_str, }, ) # UPDATE_ONLY_MODE: registros sem LLM existente → skip (não cria novos) if UPDATE_ONLY_MODE and tarefas_novas_ignoradas: logger.info( f"🔒 [update-only] Chunk {chunk_id}: " f"{n_novas_ignoradas} registro(s) sem LLM existente ignorados (UPDATE_ONLY_MODE)." ) for t in tarefas_novas_ignoradas: mid_str = str(t["id"]) collector.adicionar_resultado( t, { "status": "skip", "resultado": {}, "tokens": 0, "motivo": "update_only_sem_existente", "_raw_markdown": "", "_id_manifestacao": mid_str, }, ) with _lock: _task_queue.extend(tarefas_faltantes) _chunk_collectors[chunk_id] = collector _chunk_counters[chunk_id] = { "total": total_original, "done": n_existentes + n_ignoradas + n_novas_ignoradas, # ignoradas + update_only já contam como concluídas "tokens": 0, } _stats["tasks_skip"] += n_ignoradas + n_novas_ignoradas logger.info( f"📥 Chunk {chunk_id}: {n_faltantes} novas | " f"{n_existentes} reaproveitadas | {n_ignoradas} ignoradas | {total_original} total" ) return n_faltantes def _topup_chunks_se_necessario(): with _lock: ativos = len(_chunk_counters) if ativos >= MAX_CHUNKS_ATIVOS: return 0 faltam = MIN_CHUNKS_ATIVOS - ativos if faltam <= 0: return 0 logger.info(f"📊 [buffer] {ativos} ativos — carregando {faltam} chunk(s)...") sm = StateManager(WORK_DIR, WORK_DIR) sm.id_minimo = ID_MINIMO_CHUNK carregados = 0 for _ in range(faltam): chunk_id = sm.selecionar_proximo_chunk() if not chunk_id: logger.info("⏸️ [buffer] Sem chunks pendentes para topup.") break sm.marcar_chunk_como_iniciado(chunk_id) n = _carregar_chunk_na_fila(chunk_id) if n > 0: carregados += 1 _publish_diario(f"🔒 LOCK chunk {chunk_id}") logger.info(f"✅ [topup] Chunk {chunk_id}: {n} tasks enfileiradas.") else: sm.marcar_chunk_como_finalizado(chunk_id, {"total_ok": 0, "erros": ["sem_tarefas"]}) _publish_diario(f"⚠️ Chunk {chunk_id} sem tarefas") return carregados def _sync_e_carregar_proximos_chunks(): with _lock: if len(_chunk_counters) >= MAX_CHUNKS_ATIVOS: logger.info(f"⏸️ [sync] {len(_chunk_counters)} chunks ativos — máximo atingido.") return logger.info("📥 [sync] Baixando CSV de bordo do GitHub...") df = _http_manager.baixar_diario() if df is not None: df.to_csv(WORK_DIR / STATE_CSV_NAME, index=False) logger.info(f"✅ [sync] CSV atualizado: {len(df)} registros.") else: logger.error("❌ [sync] Falha ao baixar CSV — usando local se disponível.") sm = StateManager(WORK_DIR, WORK_DIR) sm.id_minimo = ID_MINIMO_CHUNK with _lock: faltam = max(MIN_CHUNKS_ATIVOS - len(_chunk_counters), 1) carregados = 0 for _ in range(faltam): chunk_id = sm.selecionar_proximo_chunk() if not chunk_id: break sm.marcar_chunk_como_iniciado(chunk_id) n = _carregar_chunk_na_fila(chunk_id) if n > 0: carregados += 1 _publish_diario(f"🔒 LOCK chunk {chunk_id}") else: sm.marcar_chunk_como_finalizado(chunk_id, {"total_ok": 0, "erros": ["sem_tarefas"]}) if carregados: logger.info(f"✅ [sync] {carregados} chunk(s) carregados via HTTP.") else: logger.info("⏸️ [sync] Nenhum chunk novo disponível.") # ══════════════════════════════════════════════════════════════════════ # FINALIZAÇÃO DE CHUNK # ══════════════════════════════════════════════════════════════════════ def _finalizar_chunk_se_completo(chunk_id: str): with _lock: counter = _chunk_counters.get(chunk_id) if not counter or counter["done"] < counter["total"]: return collector = _chunk_collectors.pop(chunk_id, None) del _chunk_counters[chunk_id] if not collector: return logger.info(f"🎉 Chunk {chunk_id} completo ({counter['done']}/{counter['total']})!") collector.finalizar_chunk_parcial(chunk_id) _pos_processar_jsonl_chunk(chunk_id) stats = collector.get_chunk_stats_and_reset() sm = StateManager(WORK_DIR, WORK_DIR) sm.marcar_chunk_como_finalizado(chunk_id, stats) msg = f"✅ Chunk {chunk_id} | {stats['total_ok']} OK | {len(stats['erros'])} erros" with _lock: _stats["chunks_concluidos"] += 1 # ── Push síncrono em thread própria — não confia no disco efêmero ── # Se falhar, cai no _pending_push como fallback pro bg-push. def _tentar_push(): ok = _push_chunk_agora(chunk_id, msg) if not ok: with _lock: _pending_push.append({"chunk_id": chunk_id, "msg": msg}) logger.warning(f"⚠️ [push] Chunk {chunk_id} falhou — delegado ao BG-push.") threading.Thread(target=_tentar_push, daemon=True, name=f"push-{chunk_id}").start() logger.info(f"🚀 Chunk {chunk_id} disparando push síncrono.") _topup_trigger.set() def _pos_processar_jsonl_chunk(chunk_id: str): work_tar = WORK_DIR / f"dados_llm_v3/chunk_llm_{chunk_id}.tar.gz" if not work_tar.exists(): logger.warning(f"⚠️ [merge] {work_tar} não encontrado.") return def _sort_key(obj): mid = obj.get("manifestacao_juridica", {}).get("id_manifestacao") \ if isinstance(obj.get("manifestacao_juridica"), dict) else None try: return int(mid) if mid is not None else 0 except (TypeError, ValueError): return 0 try: existentes_bytes = _http_manager.baixar_chunk_llm(chunk_id) existentes, _, _n = _ler_jsonl_de_bytes(existentes_bytes) if existentes_bytes else ([], [], 0) raw_md_existentes = _ler_raw_md_de_bytes(existentes_bytes) if existentes_bytes else {} existentes_por_id = { str(r["manifestacao_juridica"]["id_manifestacao"]): r for r in existentes if isinstance(r.get("manifestacao_juridica"), dict) and r["manifestacao_juridica"].get("id_manifestacao") } novos = _ler_jsonl_de_tar(work_tar) novos_por_id = { str(r["manifestacao_juridica"]["id_manifestacao"]): r for r in novos if isinstance(r.get("manifestacao_juridica"), dict) and r["manifestacao_juridica"].get("id_manifestacao") } merged = {**existentes_por_id, **novos_por_id} linhas = sorted(merged.values(), key=_sort_key) # raw_md desativado — não puxa do GitHub nem escreve no tar # (evita tar.gz > 100 MB e falha permanente na GitHub API) logger.info( f"🔀 [merge] Chunk {chunk_id}: " f"{len(existentes_por_id)} existentes + {len(novos_por_id)} novos " f"→ {len(linhas)} registros" ) dados_bytes = "\n".join(json.dumps(r, ensure_ascii=False) for r in linhas).encode("utf-8") with _lock: payload_logs = _chunk_payload_logs.pop(chunk_id, []) error_logs = _chunk_error_logs.pop(chunk_id, []) payloads_bytes = ( "\n".join(json.dumps(e, ensure_ascii=False) for e in payload_logs).encode("utf-8") if payload_logs else b"" ) erros_bytes = ( "\n".join(json.dumps(e, ensure_ascii=False) for e in error_logs).encode("utf-8") if error_logs else b"" ) # ── Escreve em arquivo temporário e só substitui se tiver ok ── # Evita perda de dados se algo explodir durante a escrita. tmp_tar = work_tar.with_suffix(".tmp.gz") with _tarfile.open(tmp_tar, "w:gz") as tar: for name, data in [ ("jurisprudencias.jsonl", dados_bytes), ("payloads.jsonl", payloads_bytes), ("erros.jsonl", erros_bytes), ]: if data: info = _tarfile.TarInfo(name=name) info.size = len(data) tar.addfile(info, _io.BytesIO(data)) # Atomic replace — só deleta o original depois que o tmp está pronto tmp_tar.replace(work_tar) logger.info( f"📦 [merge] chunk_llm_{chunk_id}.tar.gz: {len(linhas)} registros " f"| {len(payload_logs)} payloads | {len(error_logs)} erros" ) except Exception as e: logger.error(f"❌ [merge] Chunk {chunk_id}: {e}", exc_info=True) # ══════════════════════════════════════════════════════════════════════ # TIMEOUT / REQUEUE — v5.0: requeue em LOTE por worker_id # ══════════════════════════════════════════════════════════════════════ def _requeue_timed_out(): """ Verifica tasks em _in_progress que ultrapassaram KEEPALIVE_TIMEOUT_SEC (550 s). v5.0 — comportamento em lote: • Agrupa tasks expiradas por worker_id. • Workers com tentativas restantes: devolve o LOTE inteiro ao início da fila de uma vez (não task por task), mantendo a ordem interna. • Workers que atingiram MAX_RETRIES: falha o lote inteiro de uma vez. """ now = datetime.utcnow() deadline = now - timedelta(seconds=KEEPALIVE_TIMEOUT_SEC) # Agrupa por worker_id by_worker_requeue: Dict[str, List] = {} by_worker_fail: Dict[str, List] = {} with _lock: for tid, info in list(_in_progress.items()): referencia = info.get("last_keepalive") or info["claimed_at"] if referencia >= deadline: continue retries = info.get("retries", 0) elapsed = int((now - referencia).total_seconds()) wid = info["worker_id"] if retries + 1 >= MAX_RETRIES_PER_TASK: logger.error( f"⏱️❌ [timeout-batch] Task {tid} | worker={wid} | " f"{elapsed}s sem KA | {retries+1}/{MAX_RETRIES_PER_TASK} → FALHA DEFINITIVA" ) by_worker_fail.setdefault(wid, []).append((tid, info)) else: new_r = retries + 1 logger.warning( f"⏱️♻️ [timeout-batch] Task {tid} | worker={wid} | " f"{elapsed}s sem KA | tentativa {new_r}/{MAX_RETRIES_PER_TASK} → requeue" ) info["retries"] = new_r by_worker_requeue.setdefault(wid, []).append((tid, info)) # Remove todas as expiradas de _in_progress for wid, items in {**by_worker_requeue, **by_worker_fail}.items(): for tid, _ in items: _in_progress.pop(tid, None) # ── Requeue em lote: insere o grupo inteiro no início da fila ── for wid, items in by_worker_requeue.items(): lote = [] for tid, info in items: task = info["task"] task["_retries"] = info["retries"] lote.append(task) for t in reversed(lote): _task_queue.insert(0, t) logger.info( f"♻️ [timeout-batch] worker={wid}: {len(lote)} task(s) " f"devolvidas à fila como lote." ) # ── Falhas definitivas (fora do lock) ──────────────────────────── chunks_afetados = set() for wid, items in by_worker_fail.items(): for tid, info in items: chunk_id = info["chunk_id"] task = info["task"] chunks_afetados.add(chunk_id) with _lock: collector = _chunk_collectors.get(chunk_id) if collector: collector.adicionar_resultado( task, { "status": "falha", "resultado": _normalizar_resultado(task, "falha", {}), "tokens": 0, "motivo": "timeout_keepalive_batch_max_retries", }, ) with _lock: _stats["tasks_erro"] += 1 if chunk_id in _chunk_counters: _chunk_counters[chunk_id]["done"] += 1 logger.error( f"💀 [timeout-batch] worker={wid}: {len(items)} task(s) " f"falharam definitivamente (max retries)." ) for cid in chunks_afetados: _finalizar_chunk_se_completo(cid) # ══════════════════════════════════════════════════════════════════════ # NORMALIZAÇÃO / SANITIZAÇÃO # ══════════════════════════════════════════════════════════════════════ def _limpar_resultado_llm(resultado: dict) -> dict: _SCHEMA_KEYS = {"type", "required", "$schema", "$defs", "definitions", "additionalProperties"} def _is_schema_wrapper(obj): return isinstance(obj, dict) and "properties" in obj and set(obj.keys()) <= _SCHEMA_KEYS | {"properties"} def _strip(obj): if isinstance(obj, dict): return {k: _strip(v) for k, v in obj.items() if k not in _SCHEMA_KEYS} if isinstance(obj, list): return [_strip(i) for i in obj] return obj resultado.pop("$schema", None) if _is_schema_wrapper(resultado): resultado = resultado.get("properties", resultado) if "properties" in resultado and "manifestacao_juridica" not in resultado: inner = resultado["properties"] if isinstance(inner, dict) and "manifestacao_juridica" in inner: resultado = inner mj = resultado.get("manifestacao_juridica") if isinstance(mj, dict) and list(mj.keys()) == ["properties"] and isinstance(mj.get("properties"), dict): resultado["manifestacao_juridica"] = mj["properties"] return _strip(resultado) def _normalizar_resultado(task: dict, status: str, resultado: dict) -> dict: id_manifestacao = ( task.get("dados_originais", {}) .get("manifestacao_juridica", {}) .get("id_manifestacao") ) if status in ("falha", "skip"): return {"manifestacao_juridica": {"id_manifestacao": id_manifestacao, ("error" if status == "falha" else "skip"): True}} if id_manifestacao: mj = resultado.get("manifestacao_juridica") if isinstance(mj, dict): if not mj.get("id_manifestacao"): mj["id_manifestacao"] = id_manifestacao else: resultado.setdefault("manifestacao_juridica", {"id_manifestacao": id_manifestacao}) return resultado def _build_raw_markdown(dados_originais: dict, id_manifestacao) -> str: d = dados_originais or {} num_proc = d.get("numero_processo") or d.get("Id") or str(id_manifestacao) linhas = [ f"# Acórdão TJPR — id_manifestacao: {id_manifestacao}", "", f"**Número do processo:** {num_proc}", ] for label, keys in [ ("Data do julgamento", ["data_julgamento", "data"]), ("Relator", ["relator", "relator_ou_juiz"]), ("Órgão julgador", ["orgao_julgador", "tribunal"]), ]: v = next((d.get(k) for k in keys if d.get(k)), None) if v: linhas.append(f"**{label}:** {v}") linhas.append("") for section, keys in [ ("Ementa", ["ementa"]), ("Decisão", ["decisao"]), ("Íntegra do Acórdão", ["integra_do_acordao", "integra"]), ]: v = next((d.get(k) for k in keys if d.get(k)), None) if v: linhas += [f"## {section}", "", v.strip(), ""] return "\n".join(linhas) def _contar_chunks_pendentes() -> int: try: sm = StateManager(WORK_DIR, WORK_DIR) sm.id_minimo = ID_MINIMO_CHUNK return sm.contar_chunks_pendentes() if hasattr(sm, "contar_chunks_pendentes") else 0 except Exception: return 0 # ══════════════════════════════════════════════════════════════════════ # BACKGROUND THREADS # ══════════════════════════════════════════════════════════════════════ def _bg_sync(): while True: time.sleep(SYNC_INTERVAL) logger.info("🔄 [BG-sync] Verificando novos chunks via HTTP...") try: _sync_e_carregar_proximos_chunks() except Exception as e: logger.error(f"❌ [BG-sync] {e}", exc_info=True) def _bg_buffer_monitor(): while True: triggered = _topup_trigger.wait(timeout=BUFFER_MONITOR_INTERVAL) _topup_trigger.clear() try: with _lock: ativos = len(_chunk_counters) if ativos < MIN_CHUNKS_ATIVOS: motivo = "sinalizado" if triggered else "periódico" logger.info(f"🔔 [buffer] Topup ({motivo}): {ativos}/{MIN_CHUNKS_ATIVOS} ativos.") _topup_chunks_se_necessario() except Exception as e: logger.error(f"❌ [buffer-monitor] {e}", exc_info=True) def _bg_requeue(): while True: time.sleep(30) try: _requeue_timed_out() except Exception as e: logger.error(f"❌ [BG-requeue] {e}", exc_info=True) def _push_chunk_agora(chunk_id: str, msg: str) -> bool: """ Publica um chunk_llm no GitHub imediatamente. Retorna True se ok_llm + ok_csv. Thread-safe (pode ser chamado de qualquer lugar). """ gz_path = WORK_DIR / f"dados_llm_v3/chunk_llm_{chunk_id}.tar.gz" if not gz_path.exists(): logger.error(f"❌ [push] {gz_path} não existe — chunk {chunk_id} perdido.") return False try: ok_llm = _publish_chunk_llm(chunk_id, gz_path, msg) ok_csv = _publish_diario(msg) if ok_llm else False if ok_llm and ok_csv: logger.info(f"✅ [push] Chunk {chunk_id} publicado no GitHub.") try: gz_path.unlink(missing_ok=True) # limpa disco após push confirmado except Exception: pass return True else: logger.error(f"❌ [push] Chunk {chunk_id} falhou (llm={ok_llm} csv={ok_csv}).") return False except Exception as e: logger.error(f"❌ [push] Exceção no chunk {chunk_id}: {e}", exc_info=True) return False def _bg_push(): while True: try: time.sleep(PUSH_INTERVAL) with _lock: pending = list(_pending_push) _pending_push.clear() if not pending: continue logger.info(f"🚀 [BG-push] Publicando {len(pending)} chunk(s) via HTTP...") for item in pending: try: ok = _push_chunk_agora(item["chunk_id"], item["msg"]) if not ok: with _lock: _pending_push.append(item) # re-enfileira se falhou except Exception as e_item: logger.error(f"❌ [BG-push] chunk {item.get('chunk_id','?')}: {e_item}", exc_info=True) with _lock: _pending_push.append(item) # nunca perde except Exception as e_outer: logger.error(f"❌ [BG-push] loop externo: {e_outer}", exc_info=True) # NÃO deixa a thread morrer — volta no próximo ciclo # ══════════════════════════════════════════════════════════════════════ # APP FASTAPI # ══════════════════════════════════════════════════════════════════════ app = FastAPI(title="ParaAI Queue Director", version="5.0.0") @app.on_event("startup") async def startup(): global _http_manager, _system_prompt, _initialized, _init_error, _start_time logger.info("=" * 60) logger.info("🚀 ParaAI Queue Director v5.0 (batch) iniciando...") logger.info("=" * 60) try: if not GIT_TOKEN or not GITHUB_REPO_SLUG: raise ValueError("GIT_TOKEN e GITHUB_REPO_SLUG são obrigatórios.") _http_manager = HttpDirectorManager(GIT_TOKEN, GITHUB_REPO_SLUG, GITHUB_BRANCH) _system_prompt = _load_system_prompt() _sync_e_carregar_proximos_chunks() for target, name in [ (_bg_sync, "bg-sync"), (_bg_push, "bg-push"), (_bg_requeue, "bg-requeue"), (_bg_buffer_monitor, "bg-buffer"), ]: threading.Thread(target=target, daemon=True, name=name).start() _initialized = True _start_time = datetime.utcnow() with _lock: fila = len(_task_queue) logger.info(f"✅ Queue Director v5.0 pronto. {fila} tasks na fila inicial.") except Exception as e: _init_error = str(e) logger.critical(f"❌ Falha na inicialização: {e}", exc_info=True) # ══════════════════════════════════════════════════════════════════════ # AUTH # ══════════════════════════════════════════════════════════════════════ def _require_auth(authorization: str = Header(None)): if not authorization or authorization != f"Bearer {QUEUE_API_TOKEN}": raise HTTPException(status_code=401, detail="Não autorizado.") # ══════════════════════════════════════════════════════════════════════ # MODELS # ══════════════════════════════════════════════════════════════════════ class TaskResult(BaseModel): worker_id: str = "unknown" chunk_id: Optional[str] = None status: str = "falha" resultado: Optional[Any] = None tokens: Optional[Any] = 0 motivo: Optional[str] = "" class Config: extra = "allow" @property def resultado_safe(self) -> dict: return self.resultado if isinstance(self.resultado, dict) else {} @property def tokens_safe(self) -> int: try: return int(self.tokens or 0) except (TypeError, ValueError): return 0 class BatchTaskResult(BaseModel): """Um item dentro de um lote de resultados.""" task_id: str chunk_id: Optional[str] = None register_id: Optional[str] = None status: str = "falha" resultado: Optional[Any] = None tokens: Optional[Any] = 0 motivo: Optional[str] = "" worker_id: str = "unknown" class Config: extra = "allow" @property def resultado_safe(self) -> dict: return self.resultado if isinstance(self.resultado, dict) else {} @property def tokens_safe(self) -> int: try: return int(self.tokens or 0) except (TypeError, ValueError): return 0 class BatchResultPayload(BaseModel): """Lote completo de resultados enviado pelo worker.""" worker_id: str = "unknown" results: List[BatchTaskResult] # ══════════════════════════════════════════════════════════════════════ # ROTAS # ══════════════════════════════════════════════════════════════════════ @app.get("/health") def health(): with _lock: fila = len(_task_queue) em_proc = len(_in_progress) return { "status": "ok" if _initialized else "initializing", "error": _init_error or None, "queue_size": fila, "in_progress": em_proc, "timestamp": datetime.utcnow().isoformat(), } @app.get("/queue/status", dependencies=[Depends(_require_auth)]) def queue_status(): with _lock: p429 = _pressure_429 p498 = _pressure_498 return { "queue_size": len(_task_queue), "in_progress": len(_in_progress), "chunks_ativos": { cid: {"total": c["total"], "done": c["done"], "tokens": c.get("tokens", 0)} for cid, c in _chunk_counters.items() }, "stats": dict(_stats), "pending_push": len(_pending_push), "pressure": {"429": p429, "498": p498, "total": p429 * 2 + p498}, "timestamp": datetime.utcnow().isoformat(), } # ── GET individual (mantido para compatibilidade) ───────────────────── @app.get("/queue/work", dependencies=[Depends(_require_auth)]) def get_work(worker_id: str = "unknown"): if not _initialized: raise HTTPException(status_code=503, detail="Queue Director inicializando.") with _lock: tarefas_pendentes = [ (tid, info) for tid, info in list(_in_progress.items()) if info["worker_id"] == worker_id ] chunks_afetados: set = set() requeue_tasks: List[dict] = [] for tid, info in tarefas_pendentes: retries = info.get("retries", 0) + 1 chunks_afetados.add(info["chunk_id"]) if retries >= MAX_RETRIES_PER_TASK: logger.error( f"⏱️❌ [get_work] Task {tid} | worker={worker_id} | " f"{retries}/{MAX_RETRIES_PER_TASK} tentativas → FALHA." ) collector = _chunk_collectors.get(info["chunk_id"]) if collector: collector.adicionar_resultado( info["task"], { "status": "falha", "resultado": _normalizar_resultado(info["task"], "falha", {}), "tokens": 0, "motivo": "worker_abandonou_max_retries", }, ) _stats["tasks_erro"] += 1 if info["chunk_id"] in _chunk_counters: _chunk_counters[info["chunk_id"]]["done"] += 1 else: old_task = info["task"] old_task["_retries"] = retries requeue_tasks.append(old_task) logger.warning(f"♻️ [get_work] Worker {worker_id} pediu nova task sem devolver {tid}. Requeue.") del _in_progress[tid] if _task_queue: task = _task_queue.pop(0) for t in reversed(requeue_tasks): _task_queue.insert(0, t) elif requeue_tasks: task = requeue_tasks.pop(0) for t in reversed(requeue_tasks): _task_queue.insert(0, t) else: return JSONResponse({"task_id": None, "message": "Fila vazia."}) task_id = task["id"] chunk_id = task["chunk_id"] now = datetime.utcnow() _in_progress[task_id] = { "claimed_at": now, "last_keepalive": now, "worker_id": worker_id, "chunk_id": chunk_id, "task": task, "retries": task.pop("_retries", 0), } _stats["tasks_entregues"] += 1 logger.info(f"❇️ [get_work] Task {task_id} (chunk {chunk_id}) → worker {worker_id}") for cid in chunks_afetados: _finalizar_chunk_se_completo(cid) return { "task_id": task_id, "chunk_id": chunk_id, "dados_originais": task["dados_originais"], "dados_llm_existente": task.get("dados_llm_existente"), # null se task nova "system_prompt": _system_prompt, } # ── GET em lote — novo v5.0 ─────────────────────────────────────────── @app.get("/queue/work/batch", dependencies=[Depends(_require_auth)]) def get_work_batch( worker_id: str = "unknown", batch_size: int = Query(default=10, ge=1, le=50), ): """ Entrega até `batch_size` tasks em UMA única request. Cada item do retorno inclui: • task_id, chunk_id, register_id (id_manifestacao), dados_originais Comportamento de requeue idêntico ao endpoint individual: • Se o mesmo worker_id já tinha tasks reservadas sem devolução, faz requeue delas (ou falha definitivamente se MAX_RETRIES atingido) antes de entregar o novo lote. """ if not _initialized: raise HTTPException(status_code=503, detail="Queue Director inicializando.") tasks_out: List[dict] = [] chunks_afetados: set = set() with _lock: # Requeue das tasks antigas deste worker tarefas_pendentes = [ (tid, info) for tid, info in list(_in_progress.items()) if info["worker_id"] == worker_id ] requeue_tasks: List[dict] = [] for tid, info in tarefas_pendentes: retries = info.get("retries", 0) + 1 chunks_afetados.add(info["chunk_id"]) if retries >= MAX_RETRIES_PER_TASK: logger.error( f"⏱️❌ [batch/get] Task {tid} | worker={worker_id} | " f"{retries}/{MAX_RETRIES_PER_TASK} → FALHA DEFINITIVA" ) collector = _chunk_collectors.get(info["chunk_id"]) if collector: collector.adicionar_resultado( info["task"], { "status": "falha", "resultado": _normalizar_resultado(info["task"], "falha", {}), "tokens": 0, "motivo": "worker_abandonou_max_retries_batch", }, ) _stats["tasks_erro"] += 1 if info["chunk_id"] in _chunk_counters: _chunk_counters[info["chunk_id"]]["done"] += 1 else: old_task = info["task"] old_task["_retries"] = retries requeue_tasks.append(old_task) logger.warning( f"♻️ [batch/get] Worker {worker_id} pediu novo lote sem devolver " f"{tid}. Requeue (tentativa {retries})." ) del _in_progress[tid] # Reinsere no início da fila (preserva ordem interna) for t in reversed(requeue_tasks): _task_queue.insert(0, t) # Pega até batch_size tasks da fila now = datetime.utcnow() count = min(batch_size, len(_task_queue)) for _ in range(count): if not _task_queue: break task = _task_queue.pop(0) task_id = task["id"] chunk_id = task["chunk_id"] dados_orig = task.get("dados_originais") or {} mj = dados_orig.get("manifestacao_juridica") or {} register_id = str( mj.get("id_manifestacao") or dados_orig.get("Id") or dados_orig.get("id_manifestacao") or task_id ) _in_progress[task_id] = { "claimed_at": now, "last_keepalive": now, "worker_id": worker_id, "chunk_id": chunk_id, "task": task, "retries": task.pop("_retries", 0), "batch": True, } _stats["tasks_entregues"] += 1 item_out = { "task_id": task_id, "chunk_id": chunk_id, "register_id": register_id, "dados_originais": dados_orig, "dados_llm_existente": task.get("dados_llm_existente"), # null se nova } tasks_out.append(item_out) logger.info( f"❇️ [batch/get] {len(tasks_out)} task(s) → worker={worker_id} " f"(pediu={batch_size}, fila tinha={count})" ) for cid in chunks_afetados: _finalizar_chunk_se_completo(cid) return {"tasks": tasks_out, "count": len(tasks_out)} # ── POST resultado individual (mantido para compatibilidade) ────────── # # ATENÇÃO: a função post_result é definida SEM decorator aqui para que # post_result_batch (que a chama) possa ser REGISTRADA primeiro no router. # A rota paramétrica /queue/result/{task_id} é registrada via # app.add_api_route() DEPOIS de /queue/result/batch, garantindo que o # FastAPI nunca capture "batch" como task_id. def post_result(task_id: str, payload: TaskResult): if not _initialized: raise HTTPException(status_code=503, detail="Queue Director inicializando.") wid = payload.worker_id or "unknown" resultado_normalizado = payload.resultado_safe if isinstance(resultado_normalizado, dict): resultado_normalizado["instance_id"] = wid with _lock: info = _in_progress.pop(task_id, None) sem_reserva = (info is None) if sem_reserva: chunk_id = payload.chunk_id if chunk_id: try: chunk_id = str(int(chunk_id)).zfill(6) except (ValueError, TypeError): pass if not chunk_id: logger.warning(f"⚠️ [result] Task {task_id} sem reserva e sem chunk_id — descartado.") with _lock: _stats["tasks_aceitas_sem_reserva"] += 1 return {"status": "aceito_sem_reserva", "task_id": task_id, "aviso": "chunk_id ausente"} logger.warning(f"⚠️ [result] Task {task_id} sem reserva (chunk={chunk_id}) — aceitando.") task_stub = {"id": task_id, "chunk_id": chunk_id, "dados_originais": {}} with _lock: _stats["tasks_aceitas_sem_reserva"] += 1 else: chunk_id = info["chunk_id"] task_stub = info["task"] if payload.status == "sucesso" and isinstance(resultado_normalizado, dict): resultado_normalizado = _limpar_resultado_llm(resultado_normalizado) resultado_final = _normalizar_resultado(task_stub, payload.status, resultado_normalizado) if isinstance(resultado_final, dict): resultado_final["instance_id"] = wid motivo_lower = (payload.motivo or "").lower() with _lock: global _pressure_429, _pressure_498 if "429" in motivo_lower or "rate" in motivo_lower or "too many" in motivo_lower: _pressure_429 += 1 if "498" in motivo_lower: _pressure_498 += 1 extra = payload.model_extra or {} if chunk_id: wpl = extra.get("_log_payloads") or [] wel = extra.get("_log_erros") or [] with _lock: _chunk_payload_logs.setdefault(chunk_id, []).extend(wpl if isinstance(wpl, list) else []) _chunk_error_logs.setdefault(chunk_id, []).extend(wel if isinstance(wel, list) else []) with _lock: collector = _chunk_collectors.get(chunk_id) if collector: dados_orig = task_stub.get("dados_originais") or {} id_man_raw = ( dados_orig.get("manifestacao_juridica", {}).get("id_manifestacao") if isinstance(dados_orig.get("manifestacao_juridica"), dict) else dados_orig.get("Id") or dados_orig.get("id_manifestacao") ) or task_id worker_raw_md = "" #extra.get("_raw_markdown") or "" worker_id_man = extra.get("_id_manifestacao") if worker_id_man: id_man_raw = str(worker_id_man) raw_md = "" #worker_raw_md or (_build_raw_markdown(dados_orig, id_man_raw) if dados_orig else None) collector.adicionar_resultado( task_stub, { "status": payload.status, "resultado": resultado_final, "tokens": payload.tokens_safe, "motivo": payload.motivo, "_raw_markdown": raw_md, "_id_manifestacao": str(id_man_raw), }, ) else: logger.warning(f"⚠️ [result] Collector não encontrado para chunk {chunk_id}.") with _lock: if payload.status == "sucesso": _stats["tasks_ok"] += 1 _record_task_ok(payload.tokens_safe) elif payload.status == "skip": _stats["tasks_skip"] += 1 else: _stats["tasks_erro"] += 1 _stats["tokens_total"] += payload.tokens_safe if chunk_id in _chunk_counters: cnt = _chunk_counters[chunk_id] if not sem_reserva or cnt["done"] < cnt["total"]: cnt["done"] += 1 cnt["tokens"] += payload.tokens_safe done = _chunk_counters.get(chunk_id, {}).get("done", "?") total = _chunk_counters.get(chunk_id, {}).get("total", "?") tokens = _chunk_counters.get(chunk_id, {}).get("tokens", 0) tag = " [sem-reserva]" if sem_reserva else "" logger.info( f"📥 [result{tag}] Task {task_id} | {payload.status} | " f"chunk {chunk_id} [{done}/{total}] | tokens={payload.tokens_safe} " f"(chunk={tokens:,}) | worker={wid}" ) _finalizar_chunk_se_completo(chunk_id) return {"status": "aceito", "task_id": task_id} # ── POST resultado em lote — novo v5.0 ─────────────────────────────── # # DEVE ser registrada ANTES de /queue/result/{task_id} para que o # FastAPI não capture POST /queue/result/batch como task_id="batch". @app.post("/queue/result/batch", dependencies=[Depends(_require_auth)]) def post_result_batch(payload: BatchResultPayload): """ Recebe um lote de resultados do worker em UMA única request. Internamente chama post_result() para cada item, reutilizando toda a lógica existente de merge, collector e finalização de chunk. """ if not _initialized: raise HTTPException(status_code=503, detail="Queue Director inicializando.") accepted = 0 errors = 0 for item in payload.results: try: task_result = TaskResult( worker_id = item.worker_id or payload.worker_id, chunk_id = item.chunk_id, status = item.status, resultado = item.resultado, tokens = item.tokens, motivo = item.motivo, ) # Preserva campos extras (_raw_markdown, _id_manifestacao) se presentes extra = item.model_extra or {} if extra: task_result.__pydantic_extra__.update(extra) post_result(item.task_id, task_result) accepted += 1 except Exception as e: logger.error(f"❌ [batch/result] Task {item.task_id}: {e}") errors += 1 logger.info( f"📥 [batch/result] worker={payload.worker_id} | " f"{accepted} aceitos | {errors} erros | lote={len(payload.results)}" ) return {"accepted": accepted, "errors": errors, "total": len(payload.results)} # Registro tardio da rota paramétrica — DEVE vir DEPOIS de /queue/result/batch app.add_api_route( "/queue/result/{task_id}", post_result, methods=["POST"], dependencies=[Depends(_require_auth)], ) # ── Admin / keepalive ───────────────────────────────────────────────── @app.post("/queue/keepalive/{task_id}", dependencies=[Depends(_require_auth)]) def keepalive(task_id: str, worker_id: str = "unknown"): with _lock: info = _in_progress.get(task_id) if not info: return JSONResponse( {"status": "not_found", "task_id": task_id, "message": "Task não reservada."}, status_code=404, ) with _lock: info["last_keepalive"] = datetime.utcnow() logger.debug(f"💓 [keepalive] Task {task_id} | worker={worker_id}") return {"status": "ok", "task_id": task_id} @app.post("/queue/admin/reload", dependencies=[Depends(_require_auth)]) def admin_reload(): threading.Thread(target=_sync_e_carregar_proximos_chunks, daemon=True).start() return {"status": "reload iniciado"} @app.post("/queue/admin/force_push", dependencies=[Depends(_require_auth)]) def admin_force_push(chunk_id: Optional[str] = None): """ Força push imediato. - sem chunk_id : drena _pending_push + escaneia disco por tar.gz órfãos - com chunk_id : força push daquele chunk específico do disco """ if chunk_id: cid = str(int(chunk_id)).zfill(6) msg = f"🔧 force_push manual chunk {cid}" ok = _push_chunk_agora(cid, msg) if not ok: gz = WORK_DIR / f"dados_llm_v3/chunk_llm_{cid}.tar.gz" raise HTTPException( status_code=404 if not gz.exists() else 502, detail=f"Push falhou para chunk {cid}." ) return {"chunk_id": cid, "ok": True} # Escaneia disco por órfãos e injeta em _pending_push output_dir = WORK_DIR / "dados_llm_v3" with _lock: ja_pendentes = {item["chunk_id"] for item in _pending_push} recuperados = 0 if output_dir.exists(): for gz_path in sorted(output_dir.glob("chunk_llm_*.tar.gz")): cid = gz_path.stem.replace("chunk_llm_", "").replace(".tar", "").zfill(6) if cid not in ja_pendentes: with _lock: _pending_push.append({"chunk_id": cid, "msg": f"♻️ force_push disco {cid}"}) recuperados += 1 with _lock: pending = list(_pending_push) _pending_push.clear() results = [] for item in pending: ok = _push_chunk_agora(item["chunk_id"], item["msg"]) results.append({"chunk_id": item["chunk_id"], "ok": ok}) if not ok: with _lock: _pending_push.append(item) ok_n = sum(1 for r in results if r["ok"]) fail_n = len(results) - ok_n return { "total": len(results), "ok": ok_n, "falha": fail_n, "recuperados_disco": recuperados, "detalhes": results } @app.get("/queue/admin/push_status", dependencies=[Depends(_require_auth)]) def admin_push_status(): """Quantos chunks estão esperando push + quais tar.gz existem no disco.""" output_dir = WORK_DIR / "dados_llm_v3" disco = [] if output_dir.exists(): for gz in sorted(output_dir.glob("chunk_llm_*.tar.gz")): disco.append({"file": gz.name, "size_kb": round(gz.stat().st_size / 1024, 1)}) with _lock: pendentes = list(_pending_push) return { "pending_push_ram": len(pendentes), "chunks_em_disco": len(disco), "disco": disco, "ram": pendentes, } @app.get("/queue/admin/disco", dependencies=[Depends(_require_auth)]) def admin_disco(): """Lista todos os chunk_llm_*.tar.gz presentes no disco efêmero.""" output_dir = WORK_DIR / "dados_llm_v3" arquivos = [] if output_dir.exists(): for gz in sorted(output_dir.glob("chunk_llm_*.tar.gz")): arquivos.append({ "chunk_id": gz.stem.replace("chunk_llm_", "").replace(".tar", "").zfill(6), "file": gz.name, "size_kb": round(gz.stat().st_size / 1024, 1), }) return {"total": len(arquivos), "chunks": arquivos} from fastapi.responses import FileResponse @app.get("/queue/admin/disco/{chunk_id}", dependencies=[Depends(_require_auth)]) def admin_disco_download(chunk_id: str): """ Baixa o chunk_llm_{chunk_id}.tar.gz direto do disco efêmero. Útil para resgatar mel antes que o Space morra de novo 🍯 """ cid = str(int(chunk_id)).zfill(6) gz_path = WORK_DIR / f"dados_llm_v3/chunk_llm_{cid}.tar.gz" if not gz_path.exists(): raise HTTPException(status_code=404, detail=f"chunk_llm_{cid}.tar.gz não encontrado no disco.") return FileResponse( path = gz_path, filename = gz_path.name, media_type = "application/gzip", ) @app.post("/queue/admin/requeue/{task_id}", dependencies=[Depends(_require_auth)]) def admin_requeue_task(task_id: str): with _lock: info = _in_progress.pop(task_id, None) if not info: raise HTTPException(status_code=404, detail=f"Task {task_id} não está em processamento.") task = info["task"] retries = info.get("retries", 0) + 1 task["retries"] = retries _task_queue.insert(0, task) logger.info(f"♻️ [admin] Task {task_id} devolvida à fila (retries={retries})") return {"status": "requeued", "task_id": task_id, "retries": retries} @app.post("/queue/admin/mark_error/{task_id}", dependencies=[Depends(_require_auth)]) def admin_mark_error(task_id: str): with _lock: info = _in_progress.pop(task_id, None) if not info: raise HTTPException(status_code=404, detail=f"Task {task_id} não está em processamento.") chunk_id = info["chunk_id"] task_stub = info["task"] collector = _chunk_collectors.get(chunk_id) _stats["tasks_erro"] += 1 if collector: collector.adicionar_resultado( task_stub, {"status": "erro", "resultado": {}, "tokens": 0, "motivo": "marcado_erro_manual"} ) logger.warning(f"🚫 [admin] Task {task_id} marcada como erro (chunk={chunk_id})") _finalizar_chunk_se_completo(chunk_id) return {"status": "marked_error", "task_id": task_id, "chunk_id": chunk_id} # ══════════════════════════════════════════════════════════════════════ # DASHBOARD HTML # ══════════════════════════════════════════════════════════════════════ @app.get("/", response_class=HTMLResponse) def root(): now = datetime.utcnow() with _lock: fila = len(_task_queue) em_proc = len(_in_progress) chunks_a = len(_chunk_counters) cnt_snap = dict(_chunk_counters) s = dict(_stats) in_prog_snap = dict(_in_progress) p429 = _pressure_429 p498 = _pressure_498 buffer_ok = chunks_a >= MIN_CHUNKS_ATIVOS buffer_cor = "#22c55e" if buffer_ok else "#f59e0b" buffer_label = f"{'✅' if buffer_ok else '⚠️'} {chunks_a}/{MIN_CHUNKS_ATIVOS} mín." chunks_pend = _contar_chunks_pendentes() rows = "" for cid, cnt in sorted(cnt_snap.items()): pct = int(cnt["done"] / cnt["total"] * 100) if cnt["total"] else 0 bar_cor = "#f59e0b" if (100 - pct) <= 15 else "#38bdf8" alert = " ⚠️" if (100 - pct) <= 15 else "" rows += ( f"{cid}{alert}" f"{cnt['done']} / {cnt['total']} ({pct}%)" f"
" f"{cnt.get('tokens',0):,}" f"" ) if not rows: rows = "Nenhum chunk ativo" tasks_rows = "" for tid, info in in_prog_snap.items(): claimed_at = info.get("claimed_at", now) last_kp = info.get("last_keepalive") or claimed_at elapsed_total = int((now - claimed_at).total_seconds()) elapsed_kp = int((now - last_kp).total_seconds()) retries = info.get("retries", 0) pct_kp = max(0, min(100, int(elapsed_kp / KEEPALIVE_TIMEOUT_SEC * 100))) color = "#22c55e" if retries == 0 else "#eab308" if retries == 1 else "#f97316" if retries == 2 else "#ef4444" bar_color = "#22c55e" if pct_kp < 60 else "#eab308" if pct_kp < 90 else "#ef4444" batch_tag = " 📦" if info.get("batch") else "" tasks_rows += ( "" f"{tid} / {info.get('chunk_id','?')}" f"{info.get('worker_id','?')}{batch_tag}" f"{retries}" f"{elapsed_total}s total | {elapsed_kp}s sem KA" "
" f"
" "
" f"" f"" "" ) if not tasks_rows: tasks_rows = "Nenhuma tarefa em processamento" uptime_sec = (now - _start_time).total_seconds() if _start_time else 0 uptime_str = f"{int(uptime_sec//3600):02d}h {int((uptime_sec%3600)//60):02d}m {int(uptime_sec%60):02d}s" rpm, tpm = _sliding_metrics() pt = p429 * 2 + p498 pt_color = "#22c55e" if pt == 0 else "#eab308" if pt < 20 else "#ef4444" html = f""" ParaAI Queue Director v5

📋 ParaAI Queue Director v5.0 batch

Status: {'✅ Pronto' if _initialized else '⏳ Inicializando'} · HTTP-only · KA timeout={KEEPALIVE_TIMEOUT_SEC}s · auto-refresh 10s

Fila
{fila}
Em processamento
{em_proc}
Chunks ativos
{buffer_label}
Chunks pendentes
{chunks_pend}
Tasks OK
{s['tasks_ok']}
Tasks erro
{s['tasks_erro']}
Tokens total
{s['tokens_total']:,}
Reg/min
{rpm}
Tok/min
{int(tpm):,}
{'
Pressão 429
'+str(p429)+'
' if p429 else ''} {'
Pressão 498
'+str(p498)+'
' if p498 else ''}
Uptime
{uptime_str}

📦 Chunks ativos

{rows}
Chunk IDProgressoBarraTokens

⚙️ Tarefas em processamento

{tasks_rows}
Task / ChunkWorkerRetriesTempoKA timer ({KEEPALIVE_TIMEOUT_SEC}s)Ação

🕐 {now.strftime('%Y-%m-%d %H:%M:%S')} UTC

""" return HTMLResponse(html)