Spaces:
Running
Running
| import os | |
| import io | |
| import json | |
| from typing import Dict, Any, Optional | |
| from loguru import logger | |
| # Suporta AMBAS as APIs: nova (google.genai) e antiga (google.generativeai) | |
| _genai = None | |
| _api_style = None | |
| try: | |
| import google.genai as genai_new | |
| _genai = genai_new | |
| _api_style = 'new' | |
| except ImportError: | |
| try: | |
| import google.generativeai as genai_old | |
| _genai = genai_old | |
| _api_style = 'old' | |
| except ImportError: | |
| _genai = None | |
| _api_style = None | |
| class DocumentAnalyzer: | |
| """ | |
| Módulo para análise inteligente de documentos via Gemini. | |
| Suporta extração de texto, resumo e resposta a perguntas sobre arquivos. | |
| Compatível com API nova (google.genai) e antiga (google.generativeai). | |
| """ | |
| def __init__(self, api_key: str = ""): | |
| self.api_key = api_key or os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY") or "" | |
| self.model = None | |
| self.client = None | |
| if not self.api_key: | |
| logger.warning("⚠️ [DOC ANALYZER] Nenhuma API key configurada") | |
| return | |
| if _api_style == 'new' and _genai: | |
| try: | |
| self.client = _genai.Client(api_key=self.api_key) | |
| self.model = True | |
| logger.info("✅ [DOC ANALYZER] Google GenAI (nova API) configurado") | |
| except Exception as e: | |
| logger.error(f"❌ [DOC ANALYZER] Erro ao inicializar nova API: {e}") | |
| elif _api_style == 'old' and _genai: | |
| try: | |
| _genai.configure(api_key=self.api_key) | |
| self.model = _genai.GenerativeModel('gemini-1.5-flash') | |
| logger.info("✅ [DOC ANALYZER] Google GenAI (API antiga) configurado") | |
| except Exception as e: | |
| logger.error(f"❌ [DOC ANALYZER] Erro ao inicializar API antiga: {e}") | |
| def analyze_base64(self, base64_data: str, mime_type: str = "application/pdf", file_name: str = "documento", query: str = "Resuma este documento") -> Dict[str, Any]: | |
| """Analisa documento a partir de base64 string.""" | |
| if not self.model: | |
| logger.error(f"❌ [DOC ANALYZER] Modelo não disponível. API style: {_api_style}, client: {self.client}") | |
| return {"success": False, "error": f"Gemini não configurado para documentos (api={_api_style})"} | |
| try: | |
| import base64 as b64 | |
| doc_data = b64.b64decode(base64_data) | |
| if _api_style == 'new' and self.client: | |
| model_id = os.getenv("GEMINI_MODEL") or "gemini-2.0-flash" | |
| response = self.client.models.generate_content( | |
| model=model_id, | |
| contents=[ | |
| _genai.types.Part.from_bytes(data=doc_data, mime_type=mime_type), | |
| query | |
| ] | |
| ) | |
| else: | |
| response = self.model.generate_content([ | |
| {"mime_type": mime_type, "data": doc_data}, | |
| query | |
| ]) | |
| return { | |
| "success": True, | |
| "analysis": response.text, | |
| "file_name": file_name | |
| } | |
| except Exception as e: | |
| logger.exception(f"Erro ao analisar documento base64 {file_name}: {e}") | |
| return {"success": False, "error": str(e)} | |
| def analyze_file(self, file_path: str, query: str = "Resuma este documento") -> Dict[str, Any]: | |
| """Lê um arquivo local e envia para o Gemini analisar.""" | |
| if not os.path.exists(file_path): | |
| return {"success": False, "error": "Arquivo não encontrado"} | |
| if not self.model: | |
| logger.error(f"❌ [DOC ANALYZER] Modelo não disponível para analyze_file. API: {_api_style}") | |
| return {"success": False, "error": f"Gemini não configurado para documentos (api={_api_style})"} | |
| try: | |
| mime_type = self._get_mime_type(file_path) | |
| if mime_type == "text/plain": | |
| with open(file_path, "r", encoding="utf-8", errors="ignore") as f: | |
| content = f.read() | |
| if _api_style == 'new' and self.client: | |
| model_id = os.getenv("GEMINI_MODEL") or "gemini-2.0-flash" | |
| prompt = f"DOCUMENTO:\n{content}\n\nPERGUNTA/ACAO: {query}" | |
| response = self.client.models.generate_content(model=model_id, contents=prompt) | |
| else: | |
| prompt = f"DOCUMENTO:\n{content}\n\nPERGUNTA/ACAO: {query}" | |
| response = self.model.generate_content(prompt) | |
| else: | |
| with open(file_path, "rb") as f: | |
| doc_data = f.read() | |
| if _api_style == 'new' and self.client: | |
| model_id = os.getenv("GEMINI_MODEL") or "gemini-2.0-flash" | |
| response = self.client.models.generate_content( | |
| model=model_id, | |
| contents=[ | |
| _genai.types.Part.from_bytes(data=doc_data, mime_type=mime_type), | |
| query | |
| ] | |
| ) | |
| else: | |
| response = self.model.generate_content([ | |
| {"mime_type": mime_type, "data": doc_data}, | |
| query | |
| ]) | |
| return { | |
| "success": True, | |
| "analysis": response.text, | |
| "file_name": os.path.basename(file_path) | |
| } | |
| except Exception as e: | |
| logger.exception(f"Erro ao analisar documento {file_path}: {e}") | |
| return {"success": False, "error": str(e)} | |
| def _get_mime_type(self, file_path: str) -> str: | |
| ext = os.path.splitext(file_path)[1].lower() | |
| mapping = { | |
| ".pdf": "application/pdf", | |
| ".txt": "text/plain", | |
| ".py": "text/plain", | |
| ".js": "text/plain", | |
| ".md": "text/plain", | |
| ".json": "application/json" | |
| } | |
| return mapping.get(ext, "application/octet-stream") | |
| _analyzer = None | |
| def get_document_analyzer(api_key: str = "") -> DocumentAnalyzer: | |
| global _analyzer | |
| if not _analyzer: | |
| _analyzer = DocumentAnalyzer(api_key) | |
| return _analyzer | |