Evandro Fonseca Junior
44d7f8ce81
Introduces optional `mode` prop ("market"/"catalog") to conditionally render components like rarity badges. Adds defensive checks for missing price data and priceHistory properties to prevent runtime errors. New icons `Badge` and `Boxes` imported from lucide-react.
510 lines
20 KiB
Python
510 lines
20 KiB
Python
"""
|
|
Spider usando Selenium para contornar bloqueios anti-bot do TibiaWiki
|
|
"""
|
|
import time
|
|
import sqlite3
|
|
import os
|
|
from selenium import webdriver
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.webdriver.chrome.service import Service
|
|
from selenium.webdriver.chrome.options import Options
|
|
from webdriver_manager.chrome import ChromeDriverManager
|
|
from bs4 import BeautifulSoup
|
|
import requests
|
|
from urllib.parse import urljoin
|
|
from tibiawiki.utils import normalize_name, parse_int, parse_float, clean_text
|
|
import logging
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SeleniumSpider:
|
|
"""
|
|
Spider que usa Selenium para extrair dados do TibiaWiki
|
|
"""
|
|
|
|
def __init__(self, db_path='data/tibiawiki.db', images_path='data/images', headless=True):
|
|
self.db_path = db_path
|
|
self.images_path = images_path
|
|
self.headless = headless
|
|
self.driver = None
|
|
self.conn = None
|
|
self.cursor = None
|
|
self.created_tables = set()
|
|
self.items_scraped = 0
|
|
|
|
# Cria diretórios necessários
|
|
os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
|
|
os.makedirs(self.images_path, exist_ok=True)
|
|
|
|
def setup_driver(self):
|
|
"""Configura o driver do Selenium"""
|
|
chrome_options = Options()
|
|
|
|
if self.headless:
|
|
chrome_options.add_argument('--headless=new')
|
|
|
|
# Opções para evitar detecção
|
|
chrome_options.add_argument('--disable-blink-features=AutomationControlled')
|
|
chrome_options.add_argument('--disable-extensions')
|
|
chrome_options.add_argument('--no-sandbox')
|
|
chrome_options.add_argument('--disable-dev-shm-usage')
|
|
chrome_options.add_argument('--disable-gpu')
|
|
chrome_options.add_argument('--window-size=1920,1080')
|
|
|
|
# Desabilita imagens para acelerar carregamento
|
|
chrome_prefs = {
|
|
'profile.managed_default_content_settings.images': 2
|
|
}
|
|
chrome_options.add_experimental_option('prefs', chrome_prefs)
|
|
|
|
# Estratégia de carregamento mais rápida
|
|
try:
|
|
chrome_options.page_load_strategy = 'eager'
|
|
except Exception:
|
|
pass
|
|
|
|
# User agent
|
|
chrome_options.add_argument('user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36')
|
|
|
|
# Exclude do webdriver
|
|
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
|
chrome_options.add_experimental_option('useAutomationExtension', False)
|
|
|
|
# Auto-download driver
|
|
service = Service(ChromeDriverManager().install())
|
|
|
|
self.driver = webdriver.Chrome(service=service, options=chrome_options)
|
|
|
|
# Executa script para remover webdriver property
|
|
self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
|
|
|
|
logger.info("Driver do Chrome configurado com sucesso")
|
|
|
|
def setup_database(self):
|
|
"""Configura conexão com banco de dados"""
|
|
self.conn = sqlite3.connect(self.db_path)
|
|
self.conn.row_factory = sqlite3.Row
|
|
self.cursor = self.conn.cursor()
|
|
logger.info(f"Conectado ao banco SQLite: {self.db_path}")
|
|
|
|
def close_database(self):
|
|
"""Fecha conexão com banco"""
|
|
if self.conn:
|
|
self.conn.commit()
|
|
self.conn.close()
|
|
logger.info("Conexão com SQLite fechada")
|
|
|
|
def close_driver(self):
|
|
"""Fecha o navegador"""
|
|
if self.driver:
|
|
self.driver.quit()
|
|
logger.info("Driver fechado")
|
|
|
|
def get_page(self, url, wait_time=0):
|
|
"""
|
|
Carrega uma página e aguarda carregamento
|
|
"""
|
|
logger.info(f"Carregando página: {url}")
|
|
self.driver.get(url)
|
|
|
|
# Aguarda elementos principais carregarem
|
|
try:
|
|
WebDriverWait(self.driver, 10).until(
|
|
EC.presence_of_element_located((By.TAG_NAME, "table"))
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Timeout esperando elementos: {e}")
|
|
|
|
def get_categories(self):
|
|
"""Extrai links das categorias da página principal"""
|
|
url = "https://www.tibiawiki.com.br/wiki/Itens"
|
|
self.get_page(url, wait_time=0)
|
|
|
|
# Obtém HTML após renderização
|
|
html = self.driver.page_source
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
|
|
categories = {}
|
|
|
|
# Procura por links de categorias
|
|
all_links = soup.find_all('a', href=True)
|
|
|
|
category_keywords = [
|
|
'Capacetes', 'Armaduras', 'Escudos', 'Calças', 'Spellbooks',
|
|
'Botas', 'Aljavas', 'Machados', 'Clavas', 'Espadas', 'Rods',
|
|
'Wands', 'Distância', 'Munição', 'Punhos', 'Livros', 'Decorações',
|
|
'Troféus', 'Comidas', 'Líquidos', 'Plantas', 'Amuletos', 'Colares',
|
|
'Anéis', 'Chaves', 'Ferramentas', 'Recipientes', 'Runas', 'Pergaminhos'
|
|
]
|
|
|
|
for link in all_links:
|
|
link_text = link.get_text(strip=True)
|
|
href = link.get('href', '')
|
|
|
|
for keyword in category_keywords:
|
|
if keyword in link_text and '/wiki/' in href:
|
|
full_url = urljoin(url, href)
|
|
if keyword not in categories:
|
|
categories[keyword] = full_url
|
|
logger.info(f"Categoria encontrada: {keyword} -> {full_url}")
|
|
break
|
|
|
|
logger.info(f"Total de categorias encontradas: {len(categories)}")
|
|
return categories
|
|
|
|
def scrape_category(self, category_name, url):
|
|
"""Extrai itens de uma categoria"""
|
|
logger.info(f"\n{'='*80}")
|
|
logger.info(f"Processando categoria: {category_name}")
|
|
logger.info(f"{'='*80}")
|
|
|
|
self.get_page(url, wait_time=0)
|
|
|
|
# Obtém HTML após renderização
|
|
html = self.driver.page_source
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
|
|
# Procura por tabelas
|
|
tables = soup.find_all('table', class_='wikitable')
|
|
|
|
if not tables:
|
|
tables = soup.find_all('table')
|
|
|
|
if not tables:
|
|
logger.warning(f"Nenhuma tabela encontrada em {category_name}")
|
|
return
|
|
|
|
for table_idx, table in enumerate(tables):
|
|
logger.info(f"Processando tabela {table_idx + 1}/{len(tables)}")
|
|
|
|
# Extrai cabeçalhos
|
|
headers = []
|
|
header_row = table.find('tr')
|
|
|
|
if header_row:
|
|
ths = header_row.find_all(['th', 'td'])
|
|
for th in ths:
|
|
header_text = clean_text(th.get_text(strip=True))
|
|
headers.append(header_text)
|
|
|
|
logger.debug(f"Colunas: {headers}")
|
|
|
|
# Extrai linhas
|
|
rows = table.find_all('tr')[1:] # Pula cabeçalho
|
|
|
|
for row_idx, row in enumerate(rows):
|
|
item_data = {'categoria': category_name}
|
|
cells = row.find_all('td')
|
|
|
|
if not cells:
|
|
continue
|
|
|
|
# Destacar visualmente a linha atual no navegador
|
|
try:
|
|
self.driver.execute_script(
|
|
"""
|
|
(function(tableIndex, rowIndex){
|
|
try {
|
|
if (!window.__scrapeStyleInjected) {
|
|
const style = document.createElement('style');
|
|
style.textContent = '.__scrapeHighlight{outline:2px solid #e11d48;background:rgba(225,29,72,0.08);}';
|
|
document.head.appendChild(style);
|
|
window.__scrapeStyleInjected = true;
|
|
}
|
|
document.querySelectorAll('tr.__scrapeHighlight').forEach(el=>el.classList.remove('__scrapeHighlight'));
|
|
let tables = document.querySelectorAll('table.wikitable');
|
|
if (!tables || tables.length===0) tables = document.querySelectorAll('table');
|
|
const t = tables[tableIndex];
|
|
if (!t) return;
|
|
const trs = t.querySelectorAll('tr');
|
|
const r = trs[rowIndex+1]; // +1 para ignorar cabeçalho
|
|
if (!r) return;
|
|
r.classList.add('__scrapeHighlight');
|
|
r.scrollIntoView({block:'center'});
|
|
} catch(e) {}
|
|
})(arguments[0], arguments[1]);
|
|
""",
|
|
table_idx, row_idx
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# Processa cada célula
|
|
for cell_idx, cell in enumerate(cells):
|
|
if cell_idx >= len(headers):
|
|
break
|
|
|
|
header = headers[cell_idx]
|
|
header_clean = clean_text(header)
|
|
header_key = header_clean.casefold()
|
|
cell_text = clean_text(cell.get_text(strip=True))
|
|
|
|
# Extrai possíveis imagens da célula
|
|
imgs = cell.find_all('img')
|
|
img_srcs = []
|
|
for im in imgs:
|
|
src = im.get('src') if im else None
|
|
if not src:
|
|
continue
|
|
if not src.startswith('http'):
|
|
src = urljoin(url, src)
|
|
img_srcs.append(src)
|
|
|
|
# 1) Coluna sem título: considerar como ícone do item; ignorar Tick/Cross
|
|
if not header_clean:
|
|
for img_src in img_srcs:
|
|
filename = img_src.split('/')[-1].lower()
|
|
if 'tick' in filename or 'cross' in filename:
|
|
continue
|
|
if 'image_url' not in item_data:
|
|
item_data['image_url'] = img_src
|
|
break
|
|
# nada mais a fazer nesta célula
|
|
continue
|
|
|
|
# 2) Coluna "agrupável": mapear Tick/Cross para 1/0 e não baixar imagem
|
|
if 'agrup' in header_key:
|
|
assigned = False
|
|
for img_src in img_srcs:
|
|
filename = img_src.split('/')[-1].lower()
|
|
if 'tick' in filename:
|
|
item_data['agrupavel'] = 1
|
|
assigned = True
|
|
break
|
|
if 'cross' in filename:
|
|
item_data['agrupavel'] = 0
|
|
assigned = True
|
|
break
|
|
if not assigned:
|
|
text_key = cell_text.casefold()
|
|
if text_key == 'sim':
|
|
item_data['agrupavel'] = 1
|
|
assigned = True
|
|
elif text_key in ('não', 'nao'):
|
|
item_data['agrupavel'] = 0
|
|
assigned = True
|
|
# nunca definir image_url a partir desta coluna
|
|
continue
|
|
|
|
# 3) Demais colunas: mapear campos; só usar imagem se não for Tick/Cross
|
|
# Preferir texto para campos conhecidos
|
|
self._map_column_to_field(item_data, header_clean, cell_text)
|
|
|
|
# Considerar imagem apenas se ainda não temos image_url e não for Tick/Cross
|
|
if 'image_url' not in item_data and img_srcs:
|
|
for img_src in img_srcs:
|
|
filename = img_src.split('/')[-1].lower()
|
|
if 'tick' in filename or 'cross' in filename:
|
|
continue
|
|
item_data['image_url'] = img_src
|
|
break
|
|
|
|
# Salva item se tiver nome
|
|
if item_data.get('nome'):
|
|
self.save_item(item_data)
|
|
self.items_scraped += 1
|
|
|
|
if self.items_scraped % 10 == 0:
|
|
logger.info(f"Total de itens extraídos: {self.items_scraped}")
|
|
|
|
def _map_column_to_field(self, item_data, column_name, value):
|
|
"""Mapeia colunas para campos"""
|
|
column_lower = column_name.lower()
|
|
|
|
if 'nome' in column_lower or 'item' in column_lower:
|
|
item_data['nome'] = value
|
|
elif 'nível' in column_lower or 'lvl' in column_lower:
|
|
item_data['nivel'] = parse_int(value)
|
|
elif 'peso' in column_lower:
|
|
item_data['peso'] = parse_float(value)
|
|
elif 'defesa' in column_lower or 'def' in column_lower:
|
|
item_data['defesa'] = parse_int(value)
|
|
elif 'ataque' in column_lower or 'atk' in column_lower:
|
|
item_data['ataque'] = parse_int(value)
|
|
elif 'alcance' in column_lower:
|
|
item_data['alcance'] = parse_int(value)
|
|
elif 'slots' in column_lower:
|
|
item_data['slots'] = parse_int(value)
|
|
elif 'skill' in column_lower:
|
|
item_data['skill_para_usar'] = value
|
|
elif 'valor' in column_lower:
|
|
item_data['valor_npc'] = parse_int(value)
|
|
elif 'descrição' in column_lower:
|
|
item_data['descricao'] = value
|
|
|
|
def save_item(self, item_data):
|
|
"""Salva item no banco de dados"""
|
|
category = item_data.get('categoria')
|
|
table_name = normalize_name(category)
|
|
|
|
# Cria tabela se necessário
|
|
self._create_table_if_not_exists(table_name, item_data)
|
|
|
|
# Prepara valores para SQL
|
|
columns = list(item_data.keys())
|
|
values = list(item_data.values())
|
|
placeholders = ','.join(['?'] * len(values))
|
|
|
|
# Query UPSERT
|
|
columns_str = ','.join(columns)
|
|
update_str = ','.join([f"{col}=excluded.{col}" for col in columns if col != 'id'])
|
|
|
|
sql = f"""
|
|
INSERT INTO {table_name} ({columns_str})
|
|
VALUES ({placeholders})
|
|
ON CONFLICT(nome) DO UPDATE SET {update_str}
|
|
"""
|
|
|
|
self.cursor.execute(sql, values)
|
|
self.conn.commit()
|
|
|
|
# Download da imagem
|
|
if item_data.get('image_url'):
|
|
self.download_image(item_data)
|
|
|
|
def _create_table_if_not_exists(self, table_name, item_data):
|
|
"""Cria tabela se não existir"""
|
|
if table_name in self.created_tables:
|
|
return
|
|
|
|
columns_def = []
|
|
added_fields = set()
|
|
|
|
for field_name, field_value in item_data.items():
|
|
sql_type = "INTEGER" if isinstance(field_value, int) else "REAL" if isinstance(field_value, float) else "TEXT"
|
|
constraint = " UNIQUE NOT NULL" if field_name == 'nome' else ""
|
|
columns_def.append(f"{field_name} {sql_type}{constraint}")
|
|
added_fields.add(field_name)
|
|
|
|
# Sempre adiciona campos de imagem
|
|
if 'image_url' not in added_fields:
|
|
columns_def.append("image_url TEXT")
|
|
if 'image_local_path' not in added_fields:
|
|
columns_def.append("image_local_path TEXT")
|
|
|
|
sql = f"""
|
|
CREATE TABLE IF NOT EXISTS {table_name} (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
{','.join(columns_def)}
|
|
)
|
|
"""
|
|
|
|
self.cursor.execute(sql)
|
|
self.created_tables.add(table_name)
|
|
|
|
def download_image(self, item_data):
|
|
"""Baixa imagem do item"""
|
|
image_url = item_data.get('image_url')
|
|
if not image_url:
|
|
return
|
|
|
|
# Nunca baixar ícones Tick/Cross
|
|
basename = image_url.split('/')[-1].lower()
|
|
if 'tick' in basename or 'cross' in basename:
|
|
return
|
|
|
|
categoria = item_data.get('categoria', 'uncategorized')
|
|
nome = item_data.get('nome', 'unknown')
|
|
|
|
# Normaliza nomes
|
|
categoria_norm = normalize_name(categoria)
|
|
nome_norm = normalize_name(nome)
|
|
|
|
# Cria diretório
|
|
category_dir = os.path.join(self.images_path, categoria_norm)
|
|
os.makedirs(category_dir, exist_ok=True)
|
|
|
|
# Define caminho do arquivo
|
|
ext = 'jpg'
|
|
if '.' in image_url:
|
|
ext = image_url.split('.')[-1].lower()
|
|
if ext in ['jpeg', 'jpg']:
|
|
ext = 'jpg'
|
|
|
|
file_path = os.path.join(category_dir, f"{nome_norm}.{ext}")
|
|
|
|
# Download
|
|
try:
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
|
|
}
|
|
response = requests.get(image_url, headers=headers, timeout=30)
|
|
response.raise_for_status()
|
|
|
|
with open(file_path, 'wb') as f:
|
|
f.write(response.content)
|
|
|
|
# Atualiza banco com caminho local
|
|
table_name = normalize_name(categoria)
|
|
try:
|
|
self.cursor.execute(
|
|
f"UPDATE {table_name} SET image_local_path=? WHERE nome=?",
|
|
(file_path, item_data.get('nome'))
|
|
)
|
|
self.conn.commit()
|
|
except sqlite3.OperationalError as e:
|
|
# Se coluna não existe, adiciona ela
|
|
if 'no such column' in str(e):
|
|
try:
|
|
self.cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN image_local_path TEXT")
|
|
self.conn.commit()
|
|
# Tenta atualizar novamente
|
|
self.cursor.execute(
|
|
f"UPDATE {table_name} SET image_local_path=? WHERE nome=?",
|
|
(file_path, item_data.get('nome'))
|
|
)
|
|
self.conn.commit()
|
|
except Exception:
|
|
logger.warning(f"Não foi possível adicionar coluna image_local_path: {e}")
|
|
else:
|
|
raise
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Erro ao baixar imagem {image_url}: {e}")
|
|
|
|
def run(self):
|
|
"""Executa o processo completo de scraping"""
|
|
try:
|
|
logger.info("Iniciando scraping com Selenium...")
|
|
|
|
# Setup
|
|
self.setup_driver()
|
|
self.setup_database()
|
|
|
|
# Obtém categorias
|
|
categories = self.get_categories()
|
|
|
|
if not categories:
|
|
logger.error("Nenhuma categoria encontrada!")
|
|
return
|
|
|
|
# Processa cada categoria
|
|
for category_name, url in categories.items():
|
|
try:
|
|
self.scrape_category(category_name, url)
|
|
|
|
# Delay mínimo entre categorias (reduzido)
|
|
time.sleep(0.2)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erro ao processar {category_name}: {e}")
|
|
continue
|
|
|
|
logger.info(f"\n{'='*80}")
|
|
logger.info(f"Scraping concluído!")
|
|
logger.info(f"Total de itens extraídos: {self.items_scraped}")
|
|
logger.info(f"{'='*80}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erro fatal: {e}")
|
|
raise
|
|
|
|
finally:
|
|
self.close_database()
|
|
self.close_driver()
|
|
|