""" 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 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, urlparse from tibiawiki.utils import normalize_name, parse_int, parse_float, clean_text, normalize_column_name 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 # 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() 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() # 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): """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 (timeout reduzido para performance) try: 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" 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 # 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: logger.warning(f"Nenhuma tabela válida encontrada após filtragem 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#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; 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 # 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): 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: 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: 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 _detect_and_convert_value(self, text): """ Detecta automaticamente o tipo do valor e converte 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""" 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) # 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()) 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) # 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.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""" 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): """ 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 None # Nunca baixar ícones Tick/Cross basename = image_url.split('/')[-1].lower() if 'tick' in basename or 'cross' in basename: return None 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 (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=10) response.raise_for_status() with open(file_path, 'wb') as f: f.write(response.content) # 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 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...") # Setup self.setup_driver() self.setup_database() # 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 # 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 # 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!") 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}") # Commit de emergência if self.conn: self.conn.commit() raise finally: self.close_database() self.close_driver()