akira / modules /doc_analyzer.py
akra35567's picture
Upload 22 files
3b6c24d verified
import os
import io
import json
from typing import Dict, Any, Optional
from loguru import logger
try:
import google.generativeai as genai
except ImportError:
genai = 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.
"""
def __init__(self, api_key: str = ""):
self.api_key = api_key or os.getenv("GEMINI_API_KEY", "")
if genai and self.api_key:
genai.configure(api_key=self.api_key)
self.model = genai.GenerativeModel('gemini-1.5-flash')
else:
self.model = None
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:
return {"success": False, "error": "Gemini não configurado para documentos"}
try:
mime_type = self._get_mime_type(file_path)
# Para arquivos de texto simples, lemos diretamente
if mime_type == "text/plain":
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
content = f.read()
prompt = f"DOCUMENTO:\n{content}\n\nPERGUNTA/ACAO: {query}"
response = self.model.generate_content(prompt)
else:
# Para PDF e outros, usamos o sistema de arquivos do GenAI (se disponível) ou bytes
# Nota: Em ambientes restritos, pode ser necessário ler bytes
with open(file_path, "rb") as f:
doc_data = f.read()
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