feat: add TibiaWiki scraper project and update .gitignore

Integrates a Scrapy-based TibiaWiki scraper (spider, pipelines, SQLite/Images settings, README). Updates .gitignore to exclude generated data directories (/iconScript/data, /data/images) to keep the repository clean.
This commit is contained in:
Evandro Fonseca Junior committed 2025-10-30 00:55:25 -03:00
1 parent bf12f2c6f1
commit 29ba5aece0
27 files changed
+1702 -4

No files matched your search

+46
View File
@@ -0,0 +1,46 @@
"""
Definição dos Items para scraping do TibiaWiki
"""
import scrapy
class TibiaItem(scrapy.Item):
"""Item representando um item do TibiaWiki"""
# Campos obrigatórios
nome = scrapy.Field()
categoria = scrapy.Field()
# Campos numéricos (podem estar vazios dependendo da categoria)
nivel = scrapy.Field() # Nível mínimo
peso = scrapy.Field() # Peso em oz.
# Campos de equipamento
defesa = scrapy.Field() # Armaduras, escudos, calças
ataque = scrapy.Field() # Armas
alcance = scrapy.Field() # Armas de distância
slots = scrapy.Field() # Slots necessários
skill_para_usar = scrapy.Field() # Skill necessária
# Campos de valores
valor_npc = scrapy.Field() # Valor de venda ao NPC
# Campos descritivos
descricao = scrapy.Field()
# Campos de imagem
image_url = scrapy.Field() # URL da imagem original
image_urls = scrapy.Field() # Para ImagesPipeline do Scrapy
images = scrapy.Field() # Para ImagesPipeline do Scrapy
image_local_path = scrapy.Field() # Caminho local da imagem salva
# Campos adicionais (podem aparecer em categorias específicas)
atk_velocidade = scrapy.Field() # Velocidade de ataque
durabilidade = scrapy.Field() # Durabilidade
dano = scrapy.Field() # Dano (para armas)
cura = scrapy.Field() # Cura (para poções)
mana = scrapy.Field() # Mana (para poções)
vocacao = scrapy.Field() # Vocações que podem usar
capacidade = scrapy.Field() # Capacidade de itens recipientes
tempo = scrapy.Field() # Tempo de efeito
+206
View File
@@ -0,0 +1,206 @@
"""
Pipelines para processar os itens extraídos
"""
import sqlite3
import os
import scrapy
from itemadapter import ItemAdapter
from scrapy.pipelines.images import ImagesPipeline
from scrapy.exceptions import DropItem
from tibiawiki.utils import normalize_name
class SQLitePipeline:
"""
Pipeline que salva os itens em um banco SQLite
Uma tabela por categoria de itens
"""
def __init__(self, db_path):
self.db_path = db_path
self.conn = None
self.cursor = None
self.created_tables = set()
@classmethod
def from_crawler(cls, crawler):
db_path = crawler.settings.get('SQLITE_DB_PATH', 'data/tibiawiki.db')
return cls(db_path)
def open_spider(self, spider):
"""Abre conexão com o banco quando o spider iniciar"""
# Cria diretório se não existir
os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
self.conn = sqlite3.connect(self.db_path)
self.conn.row_factory = sqlite3.Row
self.cursor = self.conn.cursor()
spider.logger.info(f"Conectado ao banco SQLite: {self.db_path}")
def close_spider(self, spider):
"""Fecha conexão com o banco quando o spider terminar"""
if self.conn:
self.conn.commit()
self.conn.close()
spider.logger.info("Conexão com SQLite fechada")
def process_item(self, item, spider):
"""Processa cada item e salva no banco"""
adapter = ItemAdapter(item)
category = adapter.get('categoria')
if not category:
raise DropItem("Item sem categoria")
# Normaliza nome da tabela
table_name = normalize_name(category)
# Cria tabela se não existir
self._create_table_if_not_exists(table_name, adapter)
# Prepara dados para inserção
data = dict(adapter)
# Remove campo 'images' que é usado apenas pelo ImagesPipeline
data.pop('images', None)
data.pop('image_urls', None)
# Prepara valores para SQL
columns = list(data.keys())
values = list(data.values())
placeholders = ','.join(['?'] * len(values))
# Query de UPSERT (inserir ou atualizar)
columns_str = ','.join(columns)
update_str = ','.join([f"{col}=excluded.{col}" for col in columns])
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()
spider.logger.debug(f"Salvo item: {data.get('nome')} em tabela {table_name}")
return item
def _create_table_if_not_exists(self, table_name, adapter):
"""Cria tabela para categoria se não existir"""
if table_name in self.created_tables:
return
# Prepara definições de colunas
columns_def = []
# Adiciona todas as colunas possíveis baseadas nos campos do item
for field_name, field_value in adapter.items():
sql_type = self._get_sql_type(field_value)
constraint = ""
# Nome é chave única
if field_name == 'nome':
constraint = " UNIQUE NOT NULL"
columns_def.append(f"{field_name} {sql_type}{constraint}")
# Cria tabela
columns_sql = ','.join(columns_def)
sql = f"""
CREATE TABLE IF NOT EXISTS {table_name} (
id INTEGER PRIMARY KEY AUTOINCREMENT,
{columns_sql}
)
"""
self.cursor.execute(sql)
self.created_tables.add(table_name)
def _get_sql_type(self, value):
"""Retorna o tipo SQL apropriado para um valor"""
if value is None:
return "TEXT"
if isinstance(value, int):
return "INTEGER"
elif isinstance(value, float):
return "REAL"
else:
return "TEXT"
class CustomImagesPipeline(ImagesPipeline):
"""
Pipeline customizado para baixar e organizar imagens
Organiza imagens por categoria em pastas separadas
"""
def file_path(self, request, response=None, info=None, *, item=None):
"""
Define o caminho onde a imagem será salva
Formato: {categoria_normalizada}/nome_do_item.jpg
"""
# Obtém nome e categoria do item
adapter = ItemAdapter(item) if item else {}
categoria = adapter.get('categoria', 'uncategorized')
nome = adapter.get('nome', 'unknown')
# Normaliza nomes
categoria_norm = normalize_name(categoria)
nome_norm = normalize_name(nome)
# Extrai extensão original
original_url = request.url
if '/' in original_url:
original_filename = original_url.split('/')[-1]
else:
original_filename = original_url
# Extrai extensão
if '.' in original_filename:
ext = original_filename.split('.')[-1].lower()
# Normaliza extensão para jpg/jpeg
if ext in ['jpeg', 'jpg']:
ext = 'jpg'
else:
ext = 'jpg'
# Cria caminho: categoria/nome.ext
path = f"{categoria_norm}/{nome_norm}.{ext}"
return path
def get_media_requests(self, item, info):
"""
Faz download das imagens baseado no campo image_urls
"""
adapter = ItemAdapter(item)
image_urls = adapter.get('image_urls', [])
if not image_urls:
return []
for image_url in image_urls:
yield scrapy.Request(image_url)
def item_completed(self, results, item, info):
"""
Atualiza campo image_local_path quando imagem é baixada
"""
if not results:
return item
# results é uma lista de tuplas (success, image_info_or_failure)
for success, result in results:
if success:
adapter = ItemAdapter(item)
adapter['image_local_path'] = result['path']
info.spider.logger.debug(f"Imagem salva: {result['path']}")
return item
+74
View File
@@ -0,0 +1,74 @@
# Scrapy settings for tibiawiki project
#
# For simplicity, this file contains only settings considered important or
# commonly used. You can find more settings consulting the documentation:
#
# https://docs.scrapy.org/en/latest/topics/settings.html
# https://docs.scrapy.org/en/latest/topics/downloader-middleware.html
# https://docs.scrapy.org/en/latest/topics/spider-middleware.html
BOT_NAME = 'tibiawiki'
SPIDER_MODULES = ['tibiawiki.spiders']
NEWSPIDER_MODULE = 'tibiawiki.spiders'
# Obey robots.txt rules
ROBOTSTXT_OBEY = True
# Configure delays for requests for more polite scraping
DOWNLOAD_DELAY = 1.5
RANDOMIZE_DOWNLOAD_DELAY = True
# The download delay setting will honor only one of:
CONCURRENT_REQUESTS_PER_DOMAIN = 2
CONCURRENT_REQUESTS = 2
# Enable autothrottle
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 1
AUTOTHROTTLE_MAX_DELAY = 3
AUTOTHROTTLE_TARGET_CONCURRENCY = 2.0
# User agent
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'
# Configure item pipelines
ITEM_PIPELINES = {
'tibiawiki.pipelines.SQLitePipeline': 300,
'tibiawiki.pipelines.CustomImagesPipeline': 400,
}
# Images settings
IMAGES_STORE = 'data/images'
IMAGES_EXPIRES = 90
# SQLite database path
SQLITE_DB_PATH = 'data/tibiawiki.db'
# Enable caching
HTTPCACHE_ENABLED = False
# Logging
LOG_LEVEL = 'INFO'
# Feeds exports (opcional - para exportar JSON/CSV)
# FEEDS = {
# 'data/items.json': {'format': 'json'},
# 'data/items.csv': {'format': 'csv'},
# }
# Disable cookies
COOKIES_ENABLED = False
# Request settings
DOWNLOAD_TIMEOUT = 30
# Retry settings
RETRY_ENABLED = True
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 522, 524, 408, 429]
# Telnet console
TELNETCONSOLE_ENABLED = False
@@ -0,0 +1,217 @@
"""
Spider principal para extrair dados de itens do TibiaWiki.br
"""
import scrapy
from scrapy.http import Request
from tibiawiki.items import TibiaItem
from tibiawiki.utils import normalize_name, parse_int, parse_float, clean_text
class ItemsSpider(scrapy.Spider):
"""
Spider que extrai informações de todas as categorias de itens do TibiaWiki
"""
name = 'tibiawiki_items'
start_urls = ['https://www.tibiawiki.com.br/wiki/Itens']
custom_settings = {
'CONCURRENT_REQUESTS': 2,
'DOWNLOAD_DELAY': 2,
'RANDOMIZE_DOWNLOAD_DELAY': True,
'AUTOTHROTTLE_ENABLED': True,
'AUTOTHROTTLE_START_DELAY': 1,
'AUTOTHROTTLE_MAX_DELAY': 3,
}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.categories_found = set()
def parse(self, response):
"""
Parseia a página principal de itens e extrai links para categorias
"""
self.logger.info(f"Parseando página principal: {response.url}")
# Procura por links de categorias
# Os links geralmente estão em <a> dentro de listas ou tabelas
category_links = []
# Método 1: Procurar links com imagens de itens (ícones das categorias)
images_with_links = response.css('a:has(img)')
for link in images_with_links:
href = link.attrib.get('href', '')
text = link.css('::text').get('').strip()
# Verifica se é um link para categoria de itens
if '/wiki/' in href and href not in category_links:
category_links.append(response.urljoin(href))
# Método 2: Procurar por palavras-chave comuns de categorias
all_links = response.css('a::attr(href)').getall()
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:
for keyword in category_keywords:
if keyword in link and link not in category_links:
category_links.append(response.urljoin(link))
break
self.logger.info(f"Encontradas {len(category_links)} categorias")
# Faz request para cada categoria
for link in category_links:
yield Request(
url=link,
callback=self.parse_category,
dont_filter=False,
meta={'category_url': link}
)
def parse_category(self, response):
"""
Parseia uma página de categoria específica e extrai dados dos itens
"""
# Extrai nome da categoria da URL
category_name = response.url.split('/')[-1].replace('_', ' ')
self.logger.info(f"Parseando categoria: {category_name}")
# Verifica se já foi processada
if category_name in self.categories_found:
return
self.categories_found.add(category_name)
# Procura por tabelas com itens
# Tables geralmente têm class 'wikitable' no MediaWiki
tables = response.css('table.wikitable')
if not tables:
# Se não encontrar wikitable, tenta qualquer table
tables = response.css('table')
for table in tables:
# Extrai cabeçalhos da tabela
headers = []
header_row = table.css('tr').get(0)
if header_row:
# Tenta pegar <th> ou primeira linha de <td>
ths = table.css('tr:first-child th, tr:first-child td')
for th in ths:
header_text = clean_text(''.join(th.css('::text').getall()))
headers.append(header_text)
self.logger.debug(f"Colunas encontradas: {headers}")
# Processa cada linha da tabela (pula o cabeçalho)
rows = table.css('tr')[1:] # Pula primeira linha (cabeçalho)
for row in rows:
item = TibiaItem()
item['categoria'] = category_name
# Extrai células
cells = row.css('td')
if not cells:
continue
# Mapeia cada célula para seu cabeçalho correspondente
for idx, cell in enumerate(cells):
if idx >= len(headers):
break
header = headers[idx]
cell_text = clean_text(''.join(cell.css('::text').getall()))
# Extrai imagem se houver
img = cell.css('img::attr(src)').get()
if img:
image_url = response.urljoin(img)
item['image_url'] = image_url
item['image_urls'] = [image_url]
# Mapeia colunas para campos do item
self._map_column_to_field(item, header, cell_text)
# Só yield item se tiver nome
if item.get('nome'):
yield item
def _map_column_to_field(self, item, column_name, value):
"""
Mapeia nomes de colunas para campos do item
Args:
item: Item scrapy a ser preenchido
column_name: Nome da coluna da tabela
value: Valor da célula
"""
# Normaliza nome da coluna para comparação
column_lower = column_name.lower()
# Mapeamento de campos comuns
if 'nome' in column_lower or 'item' in column_lower:
item['nome'] = value
elif 'nível' in column_lower or 'lvl' in column_lower:
item['nivel'] = parse_int(value)
elif 'peso' in column_lower:
item['peso'] = parse_float(value)
elif 'defesa' in column_lower or 'def' in column_lower:
item['defesa'] = parse_int(value)
elif 'ataque' in column_lower or 'atk' in column_lower:
item['ataque'] = parse_int(value)
elif 'alcance' in column_lower:
item['alcance'] = parse_int(value)
elif 'slots' in column_lower:
item['slots'] = parse_int(value)
elif 'skill' in column_lower:
item['skill_para_usar'] = value
elif 'valor' in column_lower or 'price' in column_lower:
item['valor_npc'] = parse_int(value)
elif 'descrição' in column_lower or 'descricao' in column_lower:
item['descricao'] = value
elif 'velocidade' in column_lower:
item['atk_velocidade'] = parse_int(value)
elif 'durabilidade' in column_lower:
item['durabilidade'] = parse_int(value)
elif 'dano' in column_lower:
item['dano'] = value
elif 'cura' in column_lower:
item['cura'] = parse_int(value)
elif 'mana' in column_lower:
item['mana'] = parse_int(value)
elif 'vocação' in column_lower or 'voc' in column_lower:
item['vocacao'] = value
elif 'capacidade' in column_lower:
item['capacidade'] = parse_int(value)
elif 'tempo' in column_lower:
item['tempo'] = value
@@ -0,0 +1,414 @@
"""
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')
# 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=3):
"""
Carrega uma página e aguarda carregamento
"""
logger.info(f"Carregando página: {url}")
self.driver.get(url)
# Aguarda um tempo para a página carregar
time.sleep(wait_time)
# Aguarda elementos principais carregarem
try:
WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.TAG_NAME, "body"))
)
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=4)
# 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=3)
# 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
# Processa cada célula
for cell_idx, cell in enumerate(cells):
if cell_idx >= len(headers):
break
header = headers[cell_idx]
cell_text = clean_text(cell.get_text(strip=True))
# Extrai imagem
img = cell.find('img')
if img and img.get('src'):
img_src = img.get('src')
if not img_src.startswith('http'):
img_src = urljoin(url, img_src)
item_data['image_url'] = img_src
# Mapeia coluna para campo
self._map_column_to_field(item_data, header, cell_text)
# 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
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 entre categorias
time.sleep(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()
+98
View File
@@ -0,0 +1,98 @@
"""
Funções utilitárias para normalização e formatação de dados
"""
import unicodedata
import re
def normalize_name(name):
"""
Normaliza nomes removendo acentos e substituindo espaços por underscores
Args:
name: Nome a ser normalizado
Returns:
str: Nome normalizado (lowercase, sem acentos, underscores no lugar de espaços)
Exemplos:
"Capacetes" -> "capacetes"
"Antigas Wands e Rods" -> "antigas_wands_rods"
"Calças" -> "calcas"
"""
if not name:
return ""
# Remove acentos
nfd = unicodedata.normalize('NFD', name)
name_no_accent = ''.join(c for c in nfd if unicodedata.category(c) != 'Mn')
# Converte para lowercase e substitui espaços por underscores
normalized = re.sub(r'[^a-z0-9]+', '_', name_no_accent.lower())
# Remove underscores do início e fim
return normalized.strip('_')
def parse_int(text):
"""
Extrai um inteiro de um texto, removendo caracteres não numéricos
Args:
text: Texto que pode conter números
Returns:
int ou None: Número extraído ou None se não encontrar
"""
if not text:
return None
# Remove tudo que não for dígito
numbers = re.sub(r'\D', '', str(text))
return int(numbers) if numbers else None
def parse_float(text):
"""
Extrai um float de um texto
Args:
text: Texto que pode conter números decimais
Returns:
float ou None: Número extraído ou None se não encontrar
"""
if not text:
return None
# Remove tudo exceto dígitos, ponto e vírgula
clean_text = re.sub(r'[^\d.,]', '', str(text))
# Substitui vírgula por ponto
clean_text = clean_text.replace(',', '.')
try:
return float(clean_text)
except ValueError:
return None
def clean_text(text):
"""
Limpa texto removendo espaços extras e quebras de linha
Args:
text: Texto a ser limpo
Returns:
str: Texto limpo
"""
if not text:
return ""
# Remove espaços extras e quebras de linha
cleaned = re.sub(r'\s+', ' ', str(text)).strip()
return cleaned