Delete pipeline_base (1).py

pipeline_base (1).py

@@ -1,322 +0,0 @@
-#!/usr/bin/env python3
-"""
-pipeline_base.py - Classes base e orquestrador de pipeline
-
-Fornece:
-- BaseProcessor: classe abstrata para todos os processadores
-- CPUProcessor: processadores só com CPU
-- LLMProcessor: processadores que usam Groq API
-- PipelineEngine: orquestra sequência de processadores
-- ProcessorTeam: agrupa processadores por domínio
-- PipelineBuilder: builder para pipelines complexas
-"""
-
-import json
-import logging
-from abc import ABC, abstractmethod
-from typing import Dict, List, Optional, Callable
-from datetime import datetime
-
-logger = logging.getLogger(__name__)
-
-
-def ensure_manifestacoes_autos(record: dict) -> dict:
-    """Garante que record tem chave manifestacoes_autos."""
-    if "manifestacoes_autos" not in record:
-        record["manifestacoes_autos"] = {
-            "data_extracao": datetime.now().isoformat()
-        }
-    return record
-
-
-# ============================================================================
-# CLASSES BASE
-# ============================================================================
-
-class BaseProcessor(ABC):
-    """
-    Classe abstrata para todos os processadores.
-    
-    Define interface obrigatória e tracking de stats.
-    """
-
-    def __init__(self, name: str, expertise: str):
-        """
-        Args:
-            name: Nome único do processador
-            expertise: Descrição do domínio de expertise
-        """
-        self.name = name
-        self.expertise = expertise
-        self.stats = {
-            "processados": 0,
-            "sucesso": 0,
-            "erro": 0,
-            "ignorado": 0
-        }
-
-    @abstractmethod
-    def process(self, record: dict) -> dict:
-        """
-        Processa um registro.
-        
-        Args:
-            record: Dict com dados do acórdão
-            
-        Returns:
-            Record processado (modificado ou não)
-        """
-        pass
-
-    def get_stats(self) -> dict:
-        """Retorna estatísticas do processador."""
-        return {
-            "processor": self.name,
-            "expertise": self.expertise,
-            **self.stats
-        }
-
-    def reset_stats(self):
-        """Reseta estatísticas."""
-        self.stats = {
-            "processados": 0,
-            "sucesso": 0,
-            "erro": 0,
-            "ignorado": 0
-        }
-
-
-class CPUProcessor(BaseProcessor):
-    """
-    Processador que usa apenas CPU (regex, heurística, operações em memória).
-    
-    Não faz chamadas à API externa.
-    """
-
-    def __init__(self, name: str, expertise: str):
-        super().__init__(name, expertise)
-
-
-class LLMProcessor(BaseProcessor):
-    """
-    Processador que usa chamadas a LLM (Groq API).
-    
-    Requer llm_worker_factory para funcionar.
-    Sem factory, fica no estado "ignorado".
-    """
-
-    def __init__(self, name: str, expertise: str, llm_worker_factory: Optional[Callable] = None):
-        super().__init__(name, expertise)
-        self.llm_worker_factory = llm_worker_factory
-        self.stats["llm_calls"] = 0
-        self.stats["tokens_consumed"] = 0
-
-    def _call_llm(self, tarefa: dict, system_prompt: str) -> Optional[dict]:
-        """
-        Chamada genérica a LLM via worker factory.
-        
-        Args:
-            tarefa: Dict com id, chunk_id, dados_originais
-            system_prompt: Prompt do sistema para a tarefa
-            
-        Returns:
-            Dict com resultado ou None em caso de erro
-        """
-        if not self.llm_worker_factory:
-            return None
-
-        try:
-            worker = self.llm_worker_factory()
-
-            # Prepara chamada LLM
-            user_message = json.dumps(tarefa.get("dados_originais", {}), ensure_ascii=False)
-
-            # Simula resultado (em produção seria worker.call_llm real)
-            # Esta é uma implementação stub - veja llm_worker_hybrid.py para real
-            logger.debug(f"LLM call para {self.name}: {len(user_message)} chars")
-
-            self.stats["llm_calls"] += 1
-            # self.stats["tokens_consumed"] += estimated_tokens
-
-            return {"status": "success"}  # Placeholder
-
-        except Exception as e:
-            logger.warning(f"⚠️ Erro em chamada LLM ({self.name}): {e}")
-            return None
-
-
-# ============================================================================
-# ORQUESTRADOR
-# ============================================================================
-
-class ProcessorTeam:
-    """Agrupa processadores por domínio especializado."""
-
-    def __init__(self, name: str, description: str):
-        self.name = name
-        self.description = description
-        self.processors: List[BaseProcessor] = []
-
-    def add_processor(self, processor: BaseProcessor):
-        """Adiciona processador ao time."""
-        self.processors.append(processor)
-        logger.debug(f"✓ {processor.name} adicionado a {self.name}")
-
-    def get_stats(self) -> dict:
-        """Retorna stats agregados do time."""
-        return {
-            "time": self.name,
-            "description": self.description,
-            "num_processadores": len(self.processors),
-            "processadores": [p.get_stats() for p in self.processors]
-        }
-
-
-class PipelineEngine:
-    """
-    Motor de execução de pipeline.
-    
-    Orquestra sequência de processadores e coleta stats.
-    """
-
-    def __init__(self):
-        self.processors: List[BaseProcessor] = []
-
-    def add_processor(self, processor: BaseProcessor):
-        """Adiciona processador à sequência."""
-        self.processors.append(processor)
-
-    def process_record(self, record: dict) -> dict:
-        """
-        Processa um registro através de toda a sequência.
-        
-        Args:
-            record: Dict com dados do acórdão
-            
-        Returns:
-            Record processado
-        """
-        for processor in self.processors:
-            try:
-                record = processor.process(record)
-            except Exception as e:
-                logger.error(f"❌ Erro crítico em {processor.name}: {e}")
-                raise
-
-        return record
-
-    def process_batch(self, records: List[dict]) -> List[dict]:
-        """
-        Processa lote de registros.
-        
-        Args:
-            records: Lista de dicts
-            
-        Returns:
-            Lista de records processados
-        """
-        results = []
-        for i, record in enumerate(records):
-            try:
-                processed = self.process_record(record)
-                results.append(processed)
-                if (i + 1) % 100 == 0:
-                    logger.info(f"✓ {i + 1}/{len(records)} registros processados")
-            except Exception as e:
-                logger.error(f"❌ Erro ao processar registro {i}: {e}")
-                results.append(record)  # Retorna original em caso de erro
-
-        return results
-
-    def get_full_stats(self) -> dict:
-        """Coleta stats de todos os processadores."""
-        return {
-            "pipeline": "PipelineOrchestratorMultidisciplinar",
-            "num_processadores": len(self.processors),
-            "processors": [p.get_stats() for p in self.processors]
-        }
-
-    def reset_stats(self):
-        """Reseta stats de todos os processadores."""
-        for processor in self.processors:
-            processor.reset_stats()
-
-
-class PipelineBuilder:
-    """
-    Builder para construir pipelines complexas com times.
-    
-    Permite:
-    - Criar times especializados
-    - Adicionar processadores a times
-    - Construir engine final
-    """
-
-    def __init__(self):
-        self.teams: List[ProcessorTeam] = []
-        self.engine = PipelineEngine()
-
-    def create_team(self, name: str, description: str) -> ProcessorTeam:
-        """Cria novo time."""
-        team = ProcessorTeam(name, description)
-        self.teams.append(team)
-        return team
-
-    def add_team_to_engine(self, team: ProcessorTeam):
-        """Adiciona todos os processadores de um time ao engine."""
-        for processor in team.processors:
-            self.engine.add_processor(processor)
-
-    def add_all_teams(self):
-        """Adiciona todos os times ao engine."""
-        for team in self.teams:
-            self.add_team_to_engine(team)
-
-    def build(self) -> PipelineEngine:
-        """Constrói engine final com todos os times."""
-        self.add_all_teams()
-        logger.info(f"✅ Pipeline construída com {len(self.engine.processors)} processadores")
-        return self.engine
-
-    def print_structure(self):
-        """Imprime estrutura de times e processadores."""
-        print("\n" + "=" * 100)
-        print("ESTRUTURA DA PIPELINE MULTIDISCIPLINAR")
-        print("=" * 100)
-
-        for i, team in enumerate(self.teams, 1):
-            print(f"\n{i}. 👥 {team.name}")
-            print(f"   Descrição: {team.description}")
-            print(f"   Processadores ({len(team.processors)}):")
-
-            for processor in team.processors:
-                is_llm = "LLM" if isinstance(processor, LLMProcessor) else "CPU"
-                print(f"      ✓ {processor.name:<40} [{is_llm}] - {processor.expertise}")
-
-        print("\n" + "=" * 100)
-        print(f"Total: {len(self.teams)} times | {sum(len(t.processors) for t in self.teams)} processadores")
-        print("=" * 100 + "\n")
-
-
-# ============================================================================
-# HELPER FUNCTIONS
-# ============================================================================
-
-def estimate_tokens(text: str) -> int:
-    """
-    Estima tokens em texto para cálculo de custo.
-    
-    Approximação: 1 token ≈ 4 caracteres
-    """
-    return len(text) // 4
-
-
-def log_processor_start(processor: BaseProcessor, record_id: str):
-    """Loga início de processamento."""
-    logger.debug(f"🔄 {processor.name}: iniciando para {record_id}")
-
-
-def log_processor_end(processor: BaseProcessor, record_id: str, status: str):
-    """Loga fim de processamento."""
-    emoji = "✓" if status == "sucesso" else "✗" if status == "erro" else "⊘"
-    logger.debug(f"{emoji} {processor.name}: {status} para {record_id}")