evandro/pagodera
1
0 Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(library): add vocation filter, normalize categories, improve grid layout

Browse Source 
Added vocation state and included it in query string builder, updated onFilter to accept a vocation parameter, improved grid layout with grid flow classes, normalized category names for consistent item icons, and refined stats mapping to handle type safety.
This commit is contained in:
Evandro Fonseca Junior committed 2025-10-30 18:44:52 -03:00
1 parent 44d7f8ce81
commit 2b790d8eb4
19 files changed
+2638 -173
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

iconScript/tibiawiki/spiders/__pycache__/selenium_spider.cpython-313.pyc
BIN
View File
Binary file not shown.
iconScript/tibiawiki/spiders/selenium_spider.py
+433 -87
View File
@@ -4,6 +4,7 @@ Spider usando Selenium para contornar bloqueios anti-bot do TibiaWiki
import time
import sqlite3
import os
from concurrent.futures import ThreadPoolExecutor
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
@@ -13,8 +14,8 @@ 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
from urllib.parse import urljoin, urlparse
from tibiawiki.utils import normalize_name, parse_int, parse_float, clean_text, normalize_column_name
import logging
logging.basicConfig(level=logging.INFO)
@@ -36,10 +37,48 @@ class SeleniumSpider:
self.created_tables = set()
self.items_scraped = 0
# Cache de otimização
self._columns_cache = {} # Cache de colunas já verificadas por tabela: {table_name: {column_names}}
self.image_queue = [] # Fila de imagens para download assíncrono
self._cached_html = None # Cache do HTML da página
# 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 validate_url(self, url):
"""
Valida se a URL é do domínio TibiaWiki válido
Args:
url: URL para validar
Returns:
bool: True se válida, False caso contrário
"""
try:
parsed = urlparse(url)
# Verifica se é HTTPS e do domínio tibiawiki.com.br
if parsed.scheme != 'https':
logger.warning(f"URL deve usar HTTPS: {url}")
return False
valid_domains = ['tibiawiki.com.br', 'www.tibiawiki.com.br']
if parsed.netloc not in valid_domains:
logger.warning(f"URL deve ser do domínio tibiawiki.com.br: {url}")
return False
# Verifica se tem /wiki/ no path
if '/wiki/' not in parsed.path:
logger.warning(f"URL deve conter /wiki/ no caminho: {url}")
return False
return True
except Exception as e:
logger.error(f"Erro ao validar URL {url}: {e}")
return False
def setup_driver(self):
"""Configura o driver do Selenium"""
chrome_options = Options()
@@ -89,6 +128,13 @@ class SeleniumSpider:
self.conn = sqlite3.connect(self.db_path)
self.conn.row_factory = sqlite3.Row
self.cursor = self.conn.cursor()
# Otimizações SQLite para performance
self.cursor.execute("PRAGMA journal_mode=WAL")
self.cursor.execute("PRAGMA synchronous=NORMAL")
self.cursor.execute("PRAGMA cache_size=10000")
self.cursor.execute("PRAGMA temp_store=MEMORY")
logger.info(f"Conectado ao banco SQLite: {self.db_path}")
def close_database(self):
@@ -111,14 +157,61 @@ class SeleniumSpider:
logger.info(f"Carregando página: {url}")
self.driver.get(url)
# Aguarda elementos principais carregarem
# Aguarda elementos principais carregarem (timeout reduzido para performance)
try:
WebDriverWait(self.driver, 10).until(
WebDriverWait(self.driver, 3).until(
EC.presence_of_element_located((By.TAG_NAME, "table"))
)
except Exception as e:
logger.warning(f"Timeout esperando elementos: {e}")
def detect_category_from_page(self, url):
"""
Detecta o nome da categoria baseado no conteúdo da página
Extrai do título H1 ou título da página
"""
try:
html = self.driver.page_source
soup = BeautifulSoup(html, 'html.parser')
# Primeiro, tenta encontrar o H1 principal
h1 = soup.find('h1')
if h1:
category_text = clean_text(h1.get_text(strip=True))
# Remove prefixos comuns
prefixes = ['Categoria:', 'Lista de', 'Lista das', 'Lista dos']
for prefix in prefixes:
if category_text.startswith(prefix):
category_text = category_text[len(prefix):].strip()
if category_text:
logger.info(f"Categoria detectada do H1: {category_text}")
return category_text
# Fallback: tenta extrair do título da página
title = soup.find('title')
if title:
title_text = clean_text(title.get_text(strip=True))
# Remove sufixos comuns
if ' - TibiaWiki' in title_text:
category_text = title_text.replace(' - TibiaWiki', '').strip()
if category_text:
logger.info(f"Categoria detectada do título: {category_text}")
return category_text
# Fallback: tenta extrair do próprio URL
if '/wiki/' in url:
category_text = url.split('/wiki/')[-1]
if category_text:
logger.info(f"Categoria detectada do URL: {category_text}")
return category_text
logger.warning("Não foi possível detectar categoria, usando 'custom'")
return "custom"
except Exception as e:
logger.error(f"Erro ao detectar categoria: {e}")
return "custom"
def get_categories(self):
"""Extrai links das categorias da página principal"""
url = "https://www.tibiawiki.com.br/wiki/Itens"
@@ -169,13 +262,35 @@ class SeleniumSpider:
soup = BeautifulSoup(html, 'html.parser')
# Procura por tabelas
tables = soup.find_all('table', class_='wikitable')
# APENAS tabelas com ID 'tabelaDPL'
tables = soup.find_all('table', id='tabelaDPL')
if tables:
logger.info(f"Tabela com ID 'tabelaDPL' encontrada ({len(tables)} tabela(s))")
else:
logger.warning(f"Nenhuma tabela com ID 'tabelaDPL' encontrada em {category_name}")
logger.info("Apenas tabelas com ID 'tabelaDPL' são processadas")
return
# Filtrar tabelas: remover tabelas com classe específica
filtered_tables = []
ignored_count = 0
for table in tables:
# Ignora tabelas com classe "tibia_client_background_light"
if table.get('class') and 'tibia_client_background_light' in table.get('class'):
logger.info(f"Ignorando tabela com classe tibia_client_background_light")
ignored_count += 1
continue
filtered_tables.append(table)
if ignored_count > 0:
logger.info(f"Tabelas ignoradas (classe tibia_client_background_light): {ignored_count}")
tables = filtered_tables
logger.info(f"Total de tabelas válidas após filtragem: {len(tables)}")
if not tables:
tables = soup.find_all('table')
if not tables:
logger.warning(f"Nenhuma tabela encontrada em {category_name}")
logger.warning(f"Nenhuma tabela válida encontrada após filtragem em {category_name}")
return
for table_idx, table in enumerate(tables):
@@ -216,7 +331,8 @@ class SeleniumSpider:
window.__scrapeStyleInjected = true;
}
document.querySelectorAll('tr.__scrapeHighlight').forEach(el=>el.classList.remove('__scrapeHighlight'));
let tables = document.querySelectorAll('table.wikitable');
let tables = document.querySelectorAll('table#tabelaDPL');
if (!tables || tables.length===0) tables = document.querySelectorAll('table.wikitable');
if (!tables || tables.length===0) tables = document.querySelectorAll('table');
const t = tables[tableIndex];
if (!t) return;
@@ -233,6 +349,10 @@ class SeleniumSpider:
except Exception:
pass
# Delay visual removido para performance (pode reativar com DEBUG_MODE=1)
# if not self.headless and os.getenv('DEBUG_MODE'):
# time.sleep(0.05)
# Processa cada célula
for cell_idx, cell in enumerate(cells):
if cell_idx >= len(headers):
@@ -290,10 +410,26 @@ class SeleniumSpider:
# 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)
# 3) Demais colunas: capturar automaticamente
# Normaliza nome da coluna para identificador SQL válido
field_name = normalize_column_name(header_clean)
# Se já existe com outro nome (prioridade para campos especiais), pula
if field_name in ['id', 'categoria', 'image_url', 'image_local_path']:
continue
# Campo "nome" sempre como texto (é a chave primária)
if field_name == 'nome':
if cell_text and cell_text.strip():
item_data['nome'] = cell_text.strip()
else:
# Detecta tipo e salva valor
value = self._detect_and_convert_value(cell_text)
# Apenas salva se houver valor
if value is not None and value != '':
item_data[field_name] = value
# 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:
@@ -311,30 +447,33 @@ class SeleniumSpider:
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()
def _detect_and_convert_value(self, text):
"""
Detecta automaticamente o tipo do valor e converte
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
Args:
text: Texto a ser convertido
Returns:
int, float ou str: Valor convertido no tipo apropriado
"""
if not text or text.strip() == '':
return None
# Tenta converter para int primeiro
int_value = parse_int(text)
if int_value is not None:
# Verifica se não perdeu informação (ex: "10.5" -> 10 perderia o .5)
if str(int_value) in text.replace(',', '').replace('.', ''):
return int_value
# Tenta converter para float
float_value = parse_float(text)
if float_value is not None:
return float_value
# Retorna como texto
return text.strip()
def save_item(self, item_data):
"""Salva item no banco de dados"""
@@ -344,6 +483,9 @@ class SeleniumSpider:
# Cria tabela se necessário
self._create_table_if_not_exists(table_name, item_data)
# SEMPRE garante que todas as colunas do item_data existem na tabela
self._ensure_columns_exist(table_name, item_data)
# Prepara valores para SQL
columns = list(item_data.keys())
values = list(item_data.values())
@@ -360,11 +502,64 @@ class SeleniumSpider:
"""
self.cursor.execute(sql, values)
self.conn.commit()
# Download da imagem
# Batch commit: commit a cada 50 itens para performance
if self.items_scraped % 50 == 0:
self.conn.commit()
# Enfileira imagem para download assíncrono
if item_data.get('image_url'):
self.download_image(item_data)
self.image_queue.append(item_data.copy())
def _ensure_columns_exist(self, table_name, item_data):
"""
Garante que todas as colunas do item_data existem na tabela
Usa cache granular para evitar verificações redundantes
"""
try:
# Inicializa cache para esta tabela se necessário
if table_name not in self._columns_cache:
self._columns_cache[table_name] = set()
# Obtém colunas existentes na tabela (apenas se necessário)
needs_to_fetch = False
unknown_columns = []
# Primeiro passo: identificar colunas que precisam verificação
for field_name in item_data.keys():
if field_name not in self._columns_cache[table_name]:
unknown_columns.append(field_name)
needs_to_fetch = True
# Se não há colunas desconhecidas, não precisa fazer nada
if not needs_to_fetch:
return
# Busca colunas existentes apenas uma vez
self.cursor.execute(f"PRAGMA table_info({table_name})")
existing_columns = [row[1] for row in self.cursor.fetchall()]
# Adiciona colunas que não existem
for field_name in unknown_columns:
if field_name not in existing_columns:
# Determina tipo SQL baseado no valor
field_value = item_data[field_name]
sql_type = "INTEGER" if isinstance(field_value, int) else "REAL" if isinstance(field_value, float) else "TEXT"
try:
self.cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {field_name} {sql_type}")
self.conn.commit()
logger.info(f"Coluna '{field_name}' adicionada à tabela '{table_name}'")
except sqlite3.OperationalError as e:
# Coluna já existe (possível race condition)
if 'duplicate column name' not in str(e).lower():
logger.warning(f"Erro ao adicionar coluna '{field_name}': {e}")
# Marca como verificada no cache (independentemente se existe ou não)
self._columns_cache[table_name].add(field_name)
except Exception as e:
logger.error(f"Erro ao verificar colunas: {e}")
def _create_table_if_not_exists(self, table_name, item_data):
"""Cria tabela se não existir"""
@@ -397,15 +592,21 @@ class SeleniumSpider:
self.created_tables.add(table_name)
def download_image(self, item_data):
"""Baixa imagem do item"""
"""
Faz APENAS o download da imagem (sem tocar no banco SQLite)
Thread-safe: retorna informações para update posterior na thread principal
Returns:
tuple: (success: bool, file_path: str, table_name: str, item_name: str) ou None se erro
"""
image_url = item_data.get('image_url')
if not image_url:
return
return None
# Nunca baixar ícones Tick/Cross
basename = image_url.split('/')[-1].lower()
if 'tick' in basename or 'cross' in basename:
return
return None
categoria = item_data.get('categoria', 'uncategorized')
nome = item_data.get('nome', 'unknown')
@@ -427,47 +628,128 @@ class SeleniumSpider:
file_path = os.path.join(category_dir, f"{nome_norm}.{ext}")
# Download
# Download (timeout reduzido para performance)
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 = requests.get(image_url, headers=headers, timeout=10)
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
# Retorna informações para update posterior (thread-safe)
return (True, file_path, categoria, nome)
except Exception as e:
logger.warning(f"Erro ao baixar imagem {image_url}: {e}")
return None
def run(self):
"""Executa o processo completo de scraping"""
def process_image_queue(self):
"""
Processa fila de imagens em paralelo usando ThreadPoolExecutor
Arquitetura híbrida: downloads HTTP paralelos + updates SQLite sequenciais na thread principal
"""
if not self.image_queue:
return
logger.info(f"Processando fila de {len(self.image_queue)} imagens em paralelo...")
# ETAPA 1: Downloads paralelos (thread-safe, sem SQLite)
results = []
with ThreadPoolExecutor(max_workers=5) as executor:
# Submete todas as imagens para download paralelo
futures = [executor.submit(self.download_image, item_data) for item_data in self.image_queue]
# Coleta resultados
for future in futures:
try:
result = future.result()
if result: # (success, file_path, categoria, nome)
results.append(result)
except Exception as e:
logger.warning(f"Erro no download paralelo: {e}")
# ETAPA 2: Updates SQLite sequenciais na thread principal
if results:
logger.info(f"Atualizando {len(results)} registros no banco...")
# Garante que coluna image_local_path existe em todas as tabelas necessárias
tables_to_check = set(r[2] for r in results) # categoria
for table in tables_to_check:
table_name = normalize_name(table)
if table_name not in self._columns_cache:
self._columns_cache[table_name] = set()
# Verifica se coluna image_local_path existe
if 'image_local_path' not in self._columns_cache[table_name]:
try:
self.cursor.execute(f"PRAGMA table_info({table_name})")
existing = [row[1] for row in self.cursor.fetchall()]
if 'image_local_path' not in existing:
self.cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN image_local_path TEXT")
self.conn.commit()
logger.info(f"Coluna 'image_local_path' adicionada à tabela '{table_name}'")
self._columns_cache[table_name].add('image_local_path')
except Exception as e:
logger.warning(f"Erro ao verificar coluna image_local_path: {e}")
# Atualiza cada registro
for success, file_path, categoria, nome in results:
if not success:
continue
table_name = normalize_name(categoria)
try:
self.cursor.execute(
f"UPDATE {table_name} SET image_local_path=? WHERE nome=?",
(file_path, nome)
)
except Exception as e:
logger.warning(f"Erro ao atualizar image_local_path para {nome}: {e}")
# Commit de todos os updates
self.conn.commit()
# Limpa fila
self.image_queue.clear()
logger.info("Fila de imagens processada")
def scrape_single_url(self, url):
"""
Faz scraping de uma URL específica
Detecta a categoria automaticamente e processa
"""
logger.info(f"Iniciando scraping de URL única: {url}")
# Valida URL
if not self.validate_url(url):
logger.error(f"URL inválida: {url}")
return
# Carrega a página
try:
self.get_page(url, wait_time=0)
except Exception as e:
logger.error(f"Erro ao carregar página {url}: {e}")
return
# Detecta categoria
category_name = self.detect_category_from_page(url)
logger.info(f"Categoria detectada: {category_name}")
# Processa a categoria
self.scrape_category(category_name, url)
def run(self, single_url=None, urls_file=None):
"""
Executa o processo completo de scraping
Args:
single_url: URL específica para fazer scraping
urls_file: Caminho para arquivo TXT com múltiplas URLs
"""
try:
logger.info("Iniciando scraping com Selenium...")
@@ -475,24 +757,85 @@ class SeleniumSpider:
self.setup_driver()
self.setup_database()
# Obtém categorias
categories = self.get_categories()
# Modo: URL única
if single_url:
logger.info("Modo: URL única")
# Valida URL antes de continuar
if not self.validate_url(single_url):
logger.error(f"URL fornecida é inválida: {single_url}")
return
self.scrape_single_url(single_url)
# Modo: Arquivo com múltiplas URLs
elif urls_file:
logger.info(f"Modo: Arquivo com URLs ({urls_file})")
if not os.path.exists(urls_file):
logger.error(f"Arquivo não encontrado: {urls_file}")
return
# Lê URLs do arquivo
urls = []
with open(urls_file, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
# Ignora linhas vazias e comentários
if not line or line.startswith('#'):
continue
# Valida cada URL antes de adicionar
if self.validate_url(line):
urls.append(line)
else:
logger.warning(f"Linha {line_num}: URL inválida ignorada: {line}")
if not urls:
logger.error("Nenhuma URL válida encontrada no arquivo!")
return
logger.info(f"Total de URLs válidas para processar: {len(urls)}")
# Processa cada URL
for url_idx, url in enumerate(urls, 1):
try:
logger.info(f"\nProcessando URL {url_idx}/{len(urls)}")
self.scrape_single_url(url)
# Delay removido para performance
except Exception as e:
logger.error(f"Erro ao processar URL {url}: {e}")
continue
if not categories:
logger.error("Nenhuma categoria encontrada!")
return
# Modo: Completo (padrão)
else:
logger.info("Modo: Completo")
# 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 removido para performance
except Exception as e:
logger.error(f"Erro ao processar {category_name}: {e}")
continue
# 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
# Commit final para garantir que tudo foi salvo
logger.info("Salvando dados finais no banco...")
self.conn.commit()
# Processa fila de imagens em paralelo
if self.image_queue:
logger.info(f"\nIniciando download paralelo de {len(self.image_queue)} imagens...")
self.process_image_queue()
logger.info(f"\n{'='*80}")
logger.info(f"Scraping concluído!")
@@ -501,6 +844,9 @@ class SeleniumSpider:
except Exception as e:
logger.error(f"Erro fatal: {e}")
# Commit de emergência
if self.conn:
self.conn.commit()
raise
finally: