modules/computervision.py

# type: ignore
"""
modules/computervision.py
================================================================================
VISION AI MÓDULO - MULTIMODAL GEMINI + QR CODE + fallback OCR
================================================================================
Versão 3.0 - AKIRA "The Seer"

Este módulo evoluiu de detecção de bordas para entendimento semântico.
Pipeline de Processamento:
   1. Gemini Vision (Multimodal): Descrição de cena, objetos, cores e contexto.
      Fallbacks automáticos se Gemini falhar:
       1a. Groq Vision (Llama 3.2 - gratuito)
       1b. ToRouter Vision (gpt-5.4-nano / gpt-4o-mini - $1 free/account)
       1c. OpenRouter Vision (Llama 3.2 / Qwen-VL - free tier)
       1d. Pollinations.ai (OpenAI Vision - gratuito)
   2. QR Code Scanner: Extração de dados de códigos QR.
   3. OCR (Tesseract): Extração de texto (fallback para técnica/precisão).
   4. CV2 Analytics: Contagem de formas e objetos (Haar Cascades).
   5. RAG Visual: Armazena hashes de imagens conhecidas para lembrança rápida.

Diferente da V2, este módulo não apenas "vê" pixels, ele "entende" a imagem.
================================================================================
"""

import os
import io
import json
import time
import base64
import hashlib
from datetime import datetime
from typing import Dict, Any, List, Optional, Tuple, Union
from dataclasses import dataclass
from loguru import logger

try:
    from .config import DB_PATH, GROQ_API_KEY, OPENROUTER_API_KEY, TOROUTER_MODEL, TOROUTER_VISION_MODEL, TOROUTER_BASE_URL
except (ImportError, ValueError):
    try:
        from modules.config import DB_PATH, GROQ_API_KEY, OPENROUTER_API_KEY, TOROUTER_MODEL, TOROUTER_VISION_MODEL, TOROUTER_BASE_URL
    except ImportError:
        DB_PATH = "data/kiami.db"
        GROQ_API_KEY = ""
        OPENROUTER_API_KEY = ""
        TOROUTER_MODEL = "openai/gpt-5.5"
        TOROUTER_VISION_MODEL = "openai/gpt-5.4-nano"
        TOROUTER_BASE_URL = "https://torouter.ai/v1"


# ============================================================
# Imports Lazy para Performance
# ============================================================
_cv2 = None
_np = None
_pytesseract = None
_PIL_Image = None
_genai = None
_groq_client = None
_openai_client = None

def _check_core_deps():
    global _cv2, _np, _pytesseract, _PIL_Image, _genai, _groq_client, _openai_client
    try:
        import cv2 as cv
        import numpy as np
        import pytesseract as pt
        from PIL import Image as PILImg
        _cv2, _np, _pytesseract, _PIL_Image = cv, np, pt, PILImg
        
        # Google GenAI (nova API)
        try:
            import google.genai as genai_new
            _genai = genai_new
        except ImportError:
            try:
                import google.generativeai as genai_old
                _genai = genai_old
            except ImportError:
                _genai = None
                
        return True
    except Exception as e:
        logger.warning(f"Visão parcial: {e}")
        return False

_DEPS_OK = _check_core_deps()

# ============================================================
# CONFIGURAÇÕES
# ============================================================

@dataclass
class VisionConfig:
    ocr_lang: str = "por+eng"
    similarity_threshold: float = 0.88
    max_image_res: int = 1200
    enable_gemini: bool = True
    enable_qr: bool = True
    db_path: str = DB_PATH

# ============================================================
# CLASSE PRINCIPAL
# ============================================================

class ComputerVision:
    """
    Controlador de Visão Computacional de Nova Geração.
    """

    def __init__(self, config: Optional[VisionConfig] = None):
        self.config = config or VisionConfig()
        self.db_path = self.config.db_path
        self._setup_db()
        self._init_cascades()
        
        # API Key do Gemini (preferencialmente injetada via config)
        self.api_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY") or ""
        
        # API Keys Groq + OpenRouter + ToRouter para fallback de visão
        self.groq_api_key = os.getenv("GROQ_API_KEY") or GROQ_API_KEY or ""
        self.openrouter_api_key = os.getenv("OPENROUTER_API_KEY") or OPENROUTER_API_KEY or ""
        self.torouter_api_key = os.getenv("TOROUTER_API_KEY") or os.getenv("GITAKIRA_TOROUTER_API") or ""
        self.torouter_base_url = os.getenv("TOROUTER_BASE_URL") or TOROUTER_BASE_URL or "https://torouter.ai/v1"
        self.torouter_vision_model = os.getenv("TOROUTER_VISION_MODEL") or TOROUTER_VISION_MODEL or "openai/gpt-5.4-nano"
        self.torouter_model = os.getenv("TOROUTER_MODEL") or TOROUTER_MODEL or "openai/gpt-5.5"

    def _setup_db(self):
        """Garante tabela de memória visual."""
        try:
            from .database import Database
            db = Database(self.db_path)
            db._execute_with_retry("""
                CREATE TABLE IF NOT EXISTS image_memory (
                    hash TEXT PRIMARY KEY,
                    user_id TEXT,
                    description TEXT,
                    ocr_text TEXT,
                    qr_data TEXT,
                    metadata TEXT,
                    timestamp TIMESTAMP
                )
            """, commit=True)
        except Exception as e:
            logger.error(f"Erro DB Visão: {e}")

    def _init_cascades(self):
        """Carrega modelos Haar Cascades para detecção básica."""
        if not _cv2: return
        try:
            self._face_cascade = _cv2.CascadeClassifier(_cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
        except:
            self._face_cascade = None

    # ==================================================================
    # 🎯 PIPELINE PRINCIPAL
    # ==================================================================

    # ==================================================================
    # PROCESSAMENTO
    # ==================================================================

    def analyze_image(self, input_data: Union[str, bytes], user_id: str = "anon") -> Dict[str, Any]:
        """
        Processa imagem através de todo o pipeline.
        Aceita: Caminho de arquivo (str), Base64 (str) ou Bytes brutos (bytes).
        """
        if not input_data: return {"success": False, "error": "Entrada vazia"}
        
        img_bytes = None
        
        try:
            # 1. Detecção e Normalização da Entrada
            if isinstance(input_data, bytes):
                img_bytes = input_data
            elif isinstance(input_data, str):
                # Caso A: Caminho de arquivo local
                if os.path.isfile(input_data):
                    with open(input_data, "rb") as f:
                        img_bytes = f.read()
                # Caso B: Base64
                else:
                    try:
                        b64_str = input_data
                        if "," in b64_str: b64_str = b64_str.split(",")[1]
                        img_bytes = base64.b64decode(b64_str)
                    except Exception:
                        return {"success": False, "error": "String informada não é um caminho válido nem Base64 válido"}
            
            if not img_bytes:
                return {"success": False, "error": "Falha ao extrair bytes da imagem"}

            img_hash = hashlib.md5(img_bytes).hexdigest()
            
            # 2. Check Memória Visual (Cache BD)
            cached = self._get_from_memory(img_hash)
            if cached:
                logger.info(f"🧠 Memória Visual recordada: {img_hash}")
                cached["cached"] = True
                return cached

            # 3. Preparação para OCR e CV2
            nparr = _np.frombuffer(img_bytes, _np.uint8)
            img_cv = _cv2.imdecode(nparr, _cv2.IMREAD_COLOR)
            pil_img = _PIL_Image.open(io.BytesIO(img_bytes))

            # --- EXECUÇÃO DO PIPELINE ---
            
            # A. QR Code (Rápido)
            qr_data = self._scan_qr(img_cv) if self.config.enable_qr else None
            
            # B. Gemini Vision (Semântico - O Coração)
            descricao = ""
            if self.config.enable_gemini and self.api_key:
                descricao = self._gemini_visual_analyze(img_bytes)
            
            # Cadeia de fallbacks de visão: Gemini → ToRouter (nano) → Groq → OpenRouter → Pollinations
            if not descricao:
                descricao = self._torouter_visual_analyze(img_bytes)
            if not descricao:
                descricao = self._groq_visual_analyze(img_bytes)
            if not descricao:
                descricao = self._openrouter_visual_analyze(img_bytes)
            if not descricao:
                descricao = self._pollinations_visual_analyze(img_bytes)
            
            # C. OCR (Fallback/Técnico)
            ocr_text = self._run_ocr(pil_img)
            
            # D. CV2 Analytics (Estatístico/Objetos)
            analytics = self._run_cv2_analytics(img_cv)

            # 4. Consolidação
            result = {
                "success": True,
                "hash": img_hash,
                "description": descricao or "Não foi possível descrever a imagem semanticamente.",
                "ocr": ocr_text,
                "qr": qr_data,
                "objects": analytics.get("objects", []),
                "details": {
                    "faces": analytics.get("faces", 0),
                    "resolution": f"{img_cv.shape[1]}x{img_cv.shape[0]}" if img_cv is not None else "N/A"
                },
                "timestamp": datetime.now().isoformat()
            }

            # 5. Salva na Memória
            self._save_to_memory(result, user_id)

            return result

        except Exception as e:
            logger.exception("Falha no pipeline de visão")
            return {"success": False, "error": str(e)}

    # ==================================================================
    # 👁️ MOTORES ESPECÍFICOS
    # ==================================================================

    def _gemini_visual_analyze(self, img_bytes: bytes) -> str:
        """Usa Google Gemini Multimodal para descrever a imagem."""
        if not _genai or not self.api_key: return ""
        
        try:
            # Detecta se é a API nova ou antiga
            if hasattr(_genai, 'Client'):  # Nova API google.genai
                client = _genai.Client(api_key=self.api_key)
                
                # Otimizado: Tenta os modelos mais novos (Série 3 e 3.1) primeiro
                # Prioridade: 3.1 Pro -> 3 Flash -> 2.0 Flash -> 1.5 Flash
                model_priority = [
                    "gemini-3.1-pro-preview", 
                    "gemini-3-flash-preview", 
                    "gemini-2.0-flash", 
                    "gemini-1.5-flash"
                ]
                
                # Se houver modelo configurado no ENV, coloca no topo da lista
                env_model = os.getenv("GEMINI_MODEL", "")
                if env_model and env_model not in model_priority:
                    model_priority.insert(0, env_model)
                
                # Detetar MimeType dinâmico
                mime_type = "image/png" if img_bytes.startswith(b"\x89PNG") else "image/jpeg"
                
                last_err = None
                for model_id in model_priority:
                    try:
                        logger.info(f"👁️ Tentando Gemini Vision com modelo: {model_id}")
                        response = client.models.generate_content(
                            model=model_id,
                            contents=[
                                "Analise esta imagem com extrema precisão para uma IA assistente autônoma. Descreva tudo: objetos, textos, contexto, ambiente, cores e expressões. Se houver códigos, links ou dados sensíveis, extraia-os. Seja assertivo.",
                                _genai.types.Part.from_bytes(data=img_bytes, mime_type=mime_type),
                            ]
                        )
                        if response and response.text:
                            logger.success(f"✅ Gemini Vision ({model_id}) sucesso!")
                            return response.text
                    except Exception as e:
                        last_err = e
                        if "404" in str(e) or "not found" in str(e).lower() or "permission" in str(e).lower():
                            logger.warning(f"⚠️ Modelo {model_id} indisponível ou sem permissão. Tentando próximo...")
                            continue
                        logger.error(f"❌ Erro crítico no modelo {model_id}: {e}")
                        break
                
                if last_err: raise last_err
                return ""
            else:
                # API antiga google.generativeai
                _genai.configure(api_key=self.api_key)
                # Tenta 1.5 Flash que é mais estável na API antiga
                model_name = 'gemini-1.5-flash'
                model = _genai.GenerativeModel(model_name)
                response = model.generate_content([
                    "Descreva esta imagem detalhadamente. Seja direto e informativo.",
                    _PIL_Image.open(io.BytesIO(img_bytes))
                ])
                return response.text if response else ""
        except Exception as e:
            logger.warning(f"Gemini Vision falhou: {e}")
            return ""

    def _groq_visual_analyze(self, img_bytes: bytes) -> str:
        """Fallback usando Groq (Llama 3.2 Vision - gratuito)."""
        if not self.groq_api_key:
            return ""
        try:
            from groq import Groq
            import base64
            client = Groq(api_key=self.groq_api_key)
            mime_type = "image/png" if img_bytes.startswith(b"\x89PNG") else "image/jpeg"
            b64_data = base64.b64encode(img_bytes).decode('utf-8')
            logger.info("👁️ Usando Groq Vision (Llama 3.2) como fallback...")
            for model_id in ["llama-3.2-11b-vision-preview", "llama-3.2-90b-vision-preview"]:
                try:
                    response = client.chat.completions.create(
                        model=model_id,
                        messages=[{
                            "role": "user",
                            "content": [
                                {"type": "text", "text": "Descreva esta imagem em detalhes para uma IA assistente. O que você vê?"},
                                {"type": "image_url", "image_url": {"url": f"data:{mime_type};base64,{b64_data}"}}
                            ]
                        }],
                        max_tokens=500
                    )
                    if response and response.choices:
                        text = response.choices[0].message.content or ""
                        if text.strip():
                            logger.success(f"✅ Groq Vision ({model_id}) sucesso!")
                            return text
                except Exception as e:
                    logger.warning(f"Groq Vision {model_id} falhou: {e}")
                    continue
            return ""
        except Exception as e:
            logger.warning(f"Groq Vision falhou: {e}")
            return ""

    def _openrouter_visual_analyze(self, img_bytes: bytes) -> str:
        """Fallback usando OpenRouter (multi-modelo visão - free tier)."""
        if not self.openrouter_api_key:
            return ""
        try:
            from openai import OpenAI
            import base64
            client = OpenAI(api_key=self.openrouter_api_key, base_url="https://openrouter.ai/api/v1")
            mime_type = "image/png" if img_bytes.startswith(b"\x89PNG") else "image/jpeg"
            b64_data = base64.b64encode(img_bytes).decode('utf-8')
            models = [
                "meta-llama/llama-3.2-11b-vision:free",
                "qwen/qwen2-vl-72b-instruct",
                "meta-llama/llama-3.2-90b-vision:free"
            ]
            logger.info("👁️ Usando OpenRouter Vision como fallback...")
            for model_id in models:
                try:
                    response = client.chat.completions.create(
                        model=model_id,
                        messages=[{
                            "role": "user",
                            "content": [
                                {"type": "text", "text": "Descreva esta imagem em detalhes para uma IA assistente. O que você vê?"},
                                {"type": "image_url", "image_url": {"url": f"data:{mime_type};base64,{b64_data}"}}
                            ]
                        }],
                        max_tokens=500
                    )
                    if response and response.choices:
                        text = response.choices[0].message.content or ""
                        if text.strip():
                            logger.success(f"✅ OpenRouter Vision ({model_id}) sucesso!")
                            return text
                except Exception as e:
                    logger.warning(f"OpenRouter Vision {model_id} falhou: {e}")
                    continue
            return ""
        except Exception as e:
            logger.warning(f"OpenRouter Vision falhou: {e}")
            return ""

    def _torouter_visual_analyze(self, img_bytes: bytes) -> str:
        """Fallback usando ToRouter com modelos baratos (gpt-5.4-nano / gpt-4o-mini)."""
        if not self.torouter_api_key:
            return ""
        try:
            from openai import OpenAI
            import base64
            client = OpenAI(api_key=self.torouter_api_key, base_url=self.torouter_base_url)
            mime_type = "image/png" if img_bytes.startswith(b"\x89PNG") else "image/jpeg"
            b64_data = base64.b64encode(img_bytes).decode('utf-8')
            models = [
                self.torouter_vision_model,
                "openai/gpt-4o-mini",
                "xiaomi/mimo-v2.5"
            ]
            logger.info("👁️ Usando ToRouter Vision como fallback...")
            for model_id in models:
                try:
                    response = client.chat.completions.create(
                        model=model_id,
                        messages=[{
                            "role": "user",
                            "content": [
                                {"type": "text", "text": "Descreva esta imagem em detalhes para uma IA assistente. O que você vê?"},
                                {"type": "image_url", "image_url": {"url": f"data:{mime_type};base64,{b64_data}"}}
                            ]
                        }],
                        max_tokens=500
                    )
                    if response and response.choices:
                        text = response.choices[0].message.content or ""
                        if text.strip():
                            logger.success(f"✅ ToRouter Vision ({model_id}) sucesso!")
                            return text
                except Exception as e:
                    logger.warning(f"ToRouter Vision {model_id} falhou: {e}")
                    continue
            return ""
        except Exception as e:
            logger.warning(f"ToRouter Vision falhou: {e}")
            return ""

    def _pollinations_visual_analyze(self, img_bytes: bytes) -> str:
        """Fallback Gratuito usando Pollinations.ai (Modelo OpenAI Vision)."""
        try:
            import requests
            import base64
            
            logger.info("🎙️ Usando Pollinations (Poly) para visão gratuita...")
            
            # Detetar MimeType dinâmico
            mime_type = "image/png" if img_bytes.startswith(b"\x89PNG") else "image/jpeg"
            b64_data = base64.b64encode(img_bytes).decode('utf-8')
            
            payload = {
                "model": "openai",
                "messages": [
                    {
                        "role": "user",
                        "content": [
                            {"type": "text", "text": "Descreva esta imagem em detalhes para uma IA assistente. O que você vê?"},
                            {
                                "type": "image_url",
                                "image_url": {"url": f"data:{mime_type};base64,{b64_data}"}
                            }
                        ]
                    }
                ]
            }
            
            response = requests.post(
                "https://gen.pollinations.ai/v1/chat/completions",
                json=payload,
                timeout=30
            )
            
            if response.status_code == 200:
                data = response.json()
                res_text = data['choices'][0]['message']['content']
                logger.info(f"✅ Pollinations Vision OK: {res_text[:50]}...")
                return res_text
            return ""
        except Exception as e:
            logger.warning(f"Pollinations Vision falhou: {e}")
            return ""

    def _scan_qr(self, img_cv) -> Optional[str]:
        """Detecta e decodifica QR Code."""
        if not _cv2 or img_cv is None: return None
        try:
            detector = _cv2.QRCodeDetector()
            data, _, _ = detector.detectAndDecode(img_cv)
            return data if data else None
        except:
            return None

    def _run_ocr(self, pil_img) -> str:
        """Extrai texto da imagem via Tesseract."""
        if not _pytesseract: return ""
        try:
            return _pytesseract.image_to_string(pil_img, lang=self.config.ocr_lang).strip()
        except:
            return ""

    def _run_cv2_analytics(self, img_cv) -> Dict[str, Any]:
        """Detecta faces e extrai metadados visuais básicos."""
        res = {"faces": 0, "objects": []}
        if not _cv2 or img_cv is None: return res
        
        try:
            gray = _cv2.cvtColor(img_cv, _cv2.COLOR_BGR2GRAY)
            # Faces
            if self._face_cascade:
                faces = self._face_cascade.detectMultiScale(gray, 1.1, 4)
                res["faces"] = len(faces)
                if len(faces) > 0: res["objects"].append("pessoa/rosto")
            
            # Brilho médio
            avg_color = _np.mean(img_cv, axis=(0, 1))
            res["avg_color_bgr"] = avg_color.tolist()
            
        except: pass
        return res

    # ==================================================================
    # 🗄️ PERSISTÊNCIA (MEMÓRIA VISUAL)
    # ==================================================================

    def _get_from_memory(self, img_hash: str) -> Optional[Dict]:
        try:
            from .database import Database
            db = Database(self.db_path)
            rows = db._execute_with_retry("SELECT * FROM image_memory WHERE hash = %s", (img_hash,))
            if rows:
                res = dict(rows[0])
                return {
                    "success": True,
                    "hash": res["hash"],
                    "description": res["description"],
                    "ocr": res["ocr_text"],
                    "qr": res["qr_data"],
                    "timestamp": res["timestamp"],
                    "from_memory": True
                }
        except: pass
        return None

    def _save_to_memory(self, result: Dict, user_id: str):
        try:
            from .database import Database
            db = Database(self.db_path)
            db._execute_with_retry("""
                INSERT INTO image_memory
                (hash, user_id, description, ocr_text, qr_data, metadata, timestamp)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (hash) DO UPDATE SET
                    description=EXCLUDED.description, ocr_text=EXCLUDED.ocr_text,
                    qr_data=EXCLUDED.qr_data, metadata=EXCLUDED.metadata, timestamp=EXCLUDED.timestamp
            """, (
                result["hash"],
                user_id,
                result["description"],
                result["ocr"],
                result["qr"],
                json.dumps(result.get("details", {})),
                result["timestamp"]
            ), commit=True)
        except Exception as e:
            logger.debug(f"Erro ao salvar memória visual: {e}")

# ============================================================
# SINGLETON EXPORT
# ============================================================

_vision_instance = None

def get_computer_vision(config=None) -> ComputerVision:
    global _vision_instance
    if _vision_instance is None:
        _vision_instance = ComputerVision(config)
    return _vision_instance

def analyze_image_base64(b64_str: str, user_id: str = "anon") -> Dict[str, Any]:
    return get_computer_vision().analyze_image(b64_str, user_id)

__all__ = ["ComputerVision", "get_computer_vision", "analyze_image_base64",
           "ImageFeature", "analyze_image_from_base64", "analyze_image_file"]


# ============================================================
# COMPATIBILIDADE — aliases para imports legados
# ============================================================

@dataclass
class ImageFeature:
    """Representação simplificada de features de uma imagem."""
    description: str = ""
    ocr_text: str = ""
    qr_data: Optional[str] = None
    objects: List[str] = None  # type: ignore

    def __post_init__(self):
        if self.objects is None:
            self.objects = []


def analyze_image_from_base64(b64_str: str, user_id: str = "anon") -> Dict[str, Any]:
    """Alias legado para analyze_image_base64."""
    return analyze_image_base64(b64_str, user_id)


def analyze_image_file(filepath: str, user_id: str = "anon") -> Dict[str, Any]:
    """Analisa imagem a partir de caminho de arquivo."""
    return get_computer_vision().analyze_image(filepath, user_id)