OctusTech/cartalogo-api
0
1#!/usr/bin/env python32"""3API simplificada e robusta para busca de joias4Evita monkey patches complexos e foca no essencial5-- Marcador de arquivo6"""7 8import os9import sys10import uuid11from pathlib import Path12import shutil13 14# ===== CONFIGURAÇÃO INICIAL SIMPLES =====15print("🚀 Iniciando API de Busca de Joias...")16 17# Configuração de diretórios seguros18def setup_safe_directories():19 """Configura diretórios com permissão de escrita"""20 safe_dirs = {21 'THUMBNAILS_PATH': '/tmp/thumbnails',22 'DATA_PATH': '/tmp/data',23 'TEMP_PATH': '/tmp/temp_uploads',24 'MODELS_PATH': '/tmp/models',25 'CACHE_DIR': '/tmp/cache'26 }27 28 for key, path in safe_dirs.items():29 # Define variável de ambiente30 os.environ[key] = path31 32 # Cria diretório33 try:34 Path(path).mkdir(parents=True, exist_ok=True)35 print(f"📁 {key}: {path} ✅")36 except Exception as e:37 print(f"📁 {key}: {path} ❌ - {e}")38 39 return safe_dirs40 41# Configuração inicial42safe_dirs = setup_safe_directories()43 44# Importações principais45from fastapi import FastAPI, File, UploadFile, HTTPException, Form46from fastapi.responses import JSONResponse47from typing import List, Optional48 49# Configuração de diretórios principais50TEMP_DIR = Path(safe_dirs['TEMP_PATH'])51THUMBNAILS_DIR = Path(safe_dirs['THUMBNAILS_PATH'])52DATA_DIR = Path(safe_dirs['DATA_PATH'])53 54print(f"📂 Diretórios configurados:")55print(f" 🗂️ TEMP: {TEMP_DIR}")56print(f" 🖼️ THUMBNAILS: {THUMBNAILS_DIR}")57print(f" 💾 DATA: {DATA_DIR}")58 59# ===== IMPORTAÇÃO SEGURA DO SEARCHER =====60searcher = None61SimilaritySearcher = None62 63def safe_import_searcher():64 """Importa o SimilaritySearcher de forma segura"""65 global SimilaritySearcher66 67 try:68 print("📦 Tentando importar SimilaritySearcher...")69 70 # Limpa cache de módulos problemáticos primeiro71 modules_to_clean = [name for name in sys.modules.keys() 72 if any(keyword in name.lower() for keyword in ['similarity', 'image', 'jewelry'])]73 74 for module_name in modules_to_clean:75 if module_name in sys.modules:76 del sys.modules[module_name]77 print(f"🧹 Cache limpo: {module_name}")78 79 # Agora importa80 from scripts.similarity_search import SimilaritySearcher81 print("✅ SimilaritySearcher importado com sucesso!")82 return SimilaritySearcher83 84 except Exception as e:85 print(f"❌ Erro na importação do SimilaritySearcher: {e}")86 import traceback87 traceback.print_exc()88 return None89 90# Tenta importar91SimilaritySearcher = safe_import_searcher()92 93# ===== CONFIGURAÇÃO DA API =====94app = FastAPI(95 title="API de Busca de Catálogo de Joias",96 description="Uma API para buscar joias por texto, imagem ou áudio.",97 version="1.0.0"98)99 100# ===== INICIALIZAÇÃO =====101@app.on_event("startup")102async def startup_event():103 global searcher104 print("🚀 Inicializando API...")105 106 if SimilaritySearcher is None:107 print("❌ SimilaritySearcher não disponível")108 # Não falha completamente, permite que a API funcione parcialmente109 searcher = None110 return111 112 try:113 print("🔄 Inicializando SimilaritySearcher...")114 searcher = SimilaritySearcher()115 print("✅ SimilaritySearcher inicializado com sucesso!")116 except Exception as e:117 print(f"❌ Erro ao inicializar SimilaritySearcher: {e}")118 import traceback119 traceback.print_exc()120 searcher = None121 # Não falha, permite API funcionar parcialmente122 123 print("🎉 API inicializada!")124 125# ===== ENDPOINTS =====126 127@app.get("/", tags=["Status"])128def read_root():129 """Verifica se a API está online."""130 return {131 "status": "online", 132 "message": "Bem-vindo à API de Busca de Joias!",133 "searcher_status": "ready" if searcher else "not_ready",134 "environment": {135 "temp_dir": str(TEMP_DIR),136 "thumbnails_dir": str(THUMBNAILS_DIR),137 "data_dir": str(DATA_DIR)138 }139 }140 141@app.get("/health", tags=["Status"])142def health_check():143 """Health check detalhado."""144 return {145 "status": "healthy",146 "components": {147 "searcher": "ready" if searcher else "not_ready",148 "directories": {149 "temp": TEMP_DIR.exists(),150 "thumbnails": THUMBNAILS_DIR.exists(),151 "data": DATA_DIR.exists()152 }153 },154 "version": "1.0.0"155 }156 157@app.get("/debug", tags=["Debug"])158def debug_info():159 """Informações de debug da API."""160 return {161 "directories": {162 dir_name: {163 "path": str(path),164 "exists": path.exists(),165 "writable": os.access(path, os.W_OK) if path.exists() else False166 }167 for dir_name, path in [168 ("temp", TEMP_DIR),169 ("thumbnails", THUMBNAILS_DIR), 170 ("data", DATA_DIR)171 ]172 },173 "environment_vars": {174 key: value for key, value in os.environ.items() 175 if any(keyword in key.lower() for keyword in ['path', 'dir', 'cache', 'tmp'])176 },177 "searcher_status": "initialized" if searcher else "not_initialized",178 "python_path": sys.path[:5] # Primeiros 5 paths179 }180 181@app.post("/search-by-text", tags=["Busca"])182async def search_by_text(query: str = Form(...)):183 """Busca joias por uma descrição de texto."""184 if not searcher:185 raise HTTPException(186 status_code=503, 187 detail="Serviço de busca não está disponível. Verifique os logs da API."188 )189 190 try:191 print(f"🔍 Buscando por texto: '{query}'")192 results = searcher.search_by_text(text_query=query, top_k=3)193 194 # Verifica se há erro na resposta195 if isinstance(results, dict) and results.get('error'):196 print(f"❌ Erro no searcher: {results['error']}")197 return JSONResponse(198 status_code=500,199 content={"message": "Erro na busca", "error": results['error']}200 )201 202 if not results:203 return JSONResponse(content={"message": "Nenhum resultado encontrado.", "results": []})204 205 print(f"✅ Encontrados {len(results)} resultados")206 return {"results": results}207 208 except Exception as e:209 print(f"❌ Erro na busca por texto: {e}")210 import traceback211 traceback.print_exc()212 raise HTTPException(status_code=500, detail=f"Erro interno: {str(e)}")213 214@app.post("/search-by-image", tags=["Busca"])215async def search_by_image(file: UploadFile = File(...)):216 """Busca joias similares a uma imagem enviada."""217 if not searcher:218 raise HTTPException(219 status_code=503, 220 detail="Serviço de busca não está disponível. Verifique os logs da API."221 )222 223 # Validação do arquivo224 if not file.filename:225 raise HTTPException(status_code=400, detail="Nome do arquivo não fornecido")226 227 # Verifica extensão228 allowed_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}229 file_extension = Path(file.filename).suffix.lower()230 if file_extension not in allowed_extensions:231 raise HTTPException(232 status_code=400, 233 detail=f"Extensão {file_extension} não suportada. Use: {', '.join(allowed_extensions)}"234 )235 236 # Gera nome único237 temp_filename = f"{uuid.uuid4()}{file_extension}"238 temp_file_path = TEMP_DIR / temp_filename239 240 try:241 print(f"📁 Salvando arquivo: {file.filename} -> {temp_file_path}")242 243 # Salva arquivo244 with temp_file_path.open("wb") as buffer:245 shutil.copyfileobj(file.file, buffer)246 247 # Verifica se foi salvo248 if not temp_file_path.exists():249 raise HTTPException(status_code=500, detail="Falha ao salvar arquivo temporário")250 251 file_size = temp_file_path.stat().st_size252 print(f"✅ Arquivo salvo ({file_size} bytes). Iniciando busca...")253 254 # Realiza busca255 results = searcher.search_by_image(image_path=temp_file_path, top_k=3)256 257 # Verifica erro na resposta258 if isinstance(results, dict) and results.get('error'):259 print(f"❌ Erro no searcher: {results['error']}")260 return JSONResponse(261 status_code=500,262 content={"message": "Erro na busca", "error": results['error']}263 )264 265 if not results:266 return JSONResponse(content={"message": "Nenhum resultado encontrado.", "results": []})267 268 print(f"✅ Busca concluída! Encontrados {len(results)} resultados")269 return {"results": results}270 271 except HTTPException:272 # Re-raise HTTPExceptions273 raise274 except Exception as e:275 print(f"❌ Erro na busca por imagem: {e}")276 import traceback277 traceback.print_exc()278 raise HTTPException(status_code=500, detail=f"Erro interno: {str(e)}")279 finally:280 # Limpeza sempre executada281 try:282 if temp_file_path.exists():283 temp_file_path.unlink()284 print(f"🗑️ Arquivo temporário removido: {temp_filename}")285 except Exception as cleanup_error:286 print(f"⚠️ Erro ao remover arquivo temporário: {cleanup_error}")287 288# ===== ENDPOINT DE TESTE =====289@app.post("/test-upload", tags=["Teste"])290async def test_upload(file: UploadFile = File(...)):291 """Testa apenas o upload de arquivo sem busca."""292 temp_filename = f"test_{uuid.uuid4()}{Path(file.filename).suffix}"293 temp_file_path = TEMP_DIR / temp_filename294 295 try:296 with temp_file_path.open("wb") as buffer:297 shutil.copyfileobj(file.file, buffer)298 299 file_info = {300 "filename": file.filename,301 "temp_path": str(temp_file_path),302 "size": temp_file_path.stat().st_size,303 "exists": temp_file_path.exists()304 }305 306 return {"message": "Upload teste bem-sucedido", "file_info": file_info}307 308 except Exception as e:309 raise HTTPException(status_code=500, detail=f"Erro no teste de upload: {str(e)}")310 finally:311 if temp_file_path.exists():312 temp_file_path.unlink()313 314if __name__ == "__main__":315 import uvicorn316 print("🚀 Iniciando API em modo de desenvolvimento...")317 uvicorn.run(app, host="0.0.0.0", port=8000)