Na implementação de Business Intelligence (BI) em empresas, é comum a modelagem e visualização de dados internos. No entanto, a preparação de dados externos tem ganhado destaque com o crescimento do web scraping em Python, permitindo a criação de novas fontes de dados para análises mais completas.
Para ilustrar, utilizaremos o site Yimutian, uma plataforma chinesa de produtos agrícolas, como exemplo. O objetiov é extrair dadoss de preços e categorias usando o framework Scrapy, demonstrando uma abordagem prática para coleta automatizada de informações.
Análise da Estrutura do Site Yimutian
O site Yimutian oferece dados de mercado agrícola, com categorias como frutas, hortaliças e variedades específicas. A análise envolve navegar por hierarquias de categorias (principle, média e pequena) e extrair detalhes de preços por região.
Código dos Spiders com Scrapy
Os exemplos a seguir mostram spiders configurados para extrair diferentes aspectos dos dados, com nomes de variáveis e métodos adaptados para clareza e redução de similaridade.
1. Spider para Dados de Mercado
Este spider coleta informações de categorias e variedades de produtos.
# -*- coding: utf-8 -*-
import scrapy
from copy import deepcopy
from myapp.items import ItemMercadoProduto
class SpiderMercadoProduto(scrapy.Spider):
nome = "mercado_produto"
dominios_permitidos = ["hangqing.ymt.com"]
urls_iniciais = ['http://hangqing.ymt.com/']
def analisar(self, resposta):
lista_links = resposta.xpath("//div[@id='purchase_wrapper']/div//a[@class='hide']")
for link in lista_links:
item = ItemMercadoProduto()
item["url_categoria_grande"] = link.xpath("./@href").extract_first()
item["id_categoria_grande"] = item["url_categoria_grande"].replace("http://hangqing.ymt.com/common/nav_chandi_", "")
item["nome_categoria_grande"] = link.xpath("./text()").extract_first()
yield scrapy.Request(
item["url_categoria_grande"],
callback=self.detalhes_categoria_media,
meta={"item": deepcopy(item)}
)
def detalhes_categoria_media(self, resposta):
item = resposta.meta["item"]
lista_itens = resposta.xpath("//div[@class='cate_nav_wrap']//a")
for elemento in lista_itens:
item["id_categoria_media"] = elemento.xpath("./@data-id").extract_first()
item["nome_categoria_media"] = elemento.xpath("./text()").extract_first()
yield scrapy.Request(
item["url_categoria_grande"],
callback=self.detalhes_categoria_pequena,
meta={"item": deepcopy(item)},
dont_filter=True
)
def detalhes_categoria_pequena(self, resposta):
item = resposta.meta["item"]
id_categoria_media = item["id_categoria_media"]
if int(id_categoria_media) > 0:
nav_produto_id = "nav-product-" + id_categoria_media
lista_links = resposta.xpath(f"//div[@class='cate_content_1']//div[contains(@class,'{nav_produto_id}')]//ul//a")
for link in lista_links:
item["id_categoria_pequena"] = link.xpath("./@data-id").extract_first()
item["url_categoria_pequena"] = link.xpath("./@href").extract_first()
item["nome_categoria_pequena"] = link.xpath("./text()").extract_first()
yield scrapy.Request(
item["url_categoria_pequena"],
callback=self.detalhes_variedade,
meta={"item": deepcopy(item)}
)
def detalhes_variedade(self, resposta):
item = resposta.meta["item"]
lista_variedades = resposta.xpath("//ul[@class='all_cate clearfix']//li")
if len(lista_variedades) > 0:
for variedade in lista_variedades:
item["url_variedade"] = variedade.xpath("./a/@href").extract_first()
item["nome_variedade"] = variedade.xpath("./a/text()").extract_first()
item["id_variedade"] = item["url_variedade"].split("_")[2]
yield item
else:
item["url_variedade"] = ""
item["nome_variedade"] = ""
item["id_variedade"] = -1
yield item
2. Spider para Dados de Regiões de Origem
Este spider coleta informações geográficas, como províncias, cidades e condados.
# -*- coding: utf-8 -*-
import scrapy
from copy import deepcopy
from myapp.items import ItemRegiaoOrigem
class SpiderRegiaoOrigem(scrapy.Spider):
nome = "regiao_origem"
dominios_permitidos = ['hangqing.ymt.com']
urls_iniciais = ['http://hangqing.ymt.com/chandi_8031_0_0']
def analisar(self, resposta):
lista_provincias = resposta.xpath("//div[@class='fl sku_name']/ul//li")
for provincia in lista_provincias:
item = ItemRegiaoOrigem()
item["url_provincia"] = provincia.xpath("./a/@href").extract_first()
item["id_provincia"] = item["url_provincia"].split("_")[-1]
item["nome_provincia"] = provincia.xpath("./a/text()").extract_first()
yield scrapy.Request(
item["url_provincia"],
callback=self.detalhes_cidade,
meta={"item": deepcopy(item)}
)
def detalhes_cidade(self, resposta):
item = resposta.meta["item"]
opcoes = resposta.xpath("//select[@class='location_select'][1]//option")
if len(opcoes) > 0:
for opcao in opcoes:
nome = opcao.xpath("./text()").extract_first()
if nome != "全部":
item["nome_cidade"] = nome
item["url_cidade"] = opcao.xpath("./@data-url").extract_first()
item["id_cidade"] = item["url_cidade"].split("_")[-1]
yield scrapy.Request(
item["url_cidade"],
callback=self.detalhes_area,
meta={"item": deepcopy(item)}
)
else:
item["nome_cidade"] = ""
item["url_cidade"] = ""
item["id_cidade"] = 0
yield scrapy.Request(
item["url_cidade"],
callback=self.detalhes_area,
meta={"item": deepcopy(item)}
)
def detalhes_area(self, resposta):
item = resposta.meta["item"]
lista_areas = resposta.xpath("//select[@class='location_select'][2]//option")
if len(lista_areas) > 0:
for area in lista_areas:
nome = area.xpath("./text()").extract_first()
if nome != "全部":
item["nome_area"] = nome
item["url_area"] = area.xpath("./@data-url").extract_first()
item["id_area"] = item["url_area"].split("_")[-1]
yield item
else:
item["nome_area"] = ""
item["url_area"] = ""
item["id_area"] = 0
yield item
3. Spider para Distribuição de Preços por Região
Este spider analisa preços em diferentes níveis geográficos, integrando dados de bancos de dados.
# -*- coding: utf-8 -*-
import scrapy
import pymysql
import json
from copy import deepcopy
from myapp.items import ItemPrecoRegiao
import datetime
class SpiderDistribuicaoPreco(scrapy.Spider):
nome = 'distribuicao_preco'
dominios_permitidos = ['hangqing.ymt.com']
urls_iniciais = ['http://hangqing.ymt.com/']
data_atual = datetime.datetime.now()
chave_data = str(data_atual.year) + str(data_atual.month) + str(data_atual.day)
def __init__(self):
self.conn = pymysql.connect(
host="127.0.0.1", port=3306,
user='root', password='senha',
db='banco_dados', charset='utf8'
)
def analisar(self, resposta):
cursor = self.conn.cursor()
sql = f"select id_pequeno from tabela_precos where chave_data = {self.chave_data} and preco_medio > 0"
cursor.execute(sql)
registros = cursor.fetchall()
for registro in registros:
item = ItemPrecoRegiao()
item["id_pequeno"] = registro[0]
url_api = "http://hangqing.ymt.com/chandi/location_charts"
id_pequeno_str = str(item["id_pequeno"])
dados_formulario = {
"locationId": "0",
"productId": id_pequeno_str,
"breedId": "0"
}
yield scrapy.FormRequest(
url_api,
formdata=dados_formulario,
callback=self.processar_resposta,
meta={"item": deepcopy(item)}
)
def processar_resposta(self, resposta):
item = resposta.meta["item"]
dados = json.loads(resposta.text)
if dados["status"] == 0:
item["unidade"] = dados["data"]["unit"]
item["chave_data"] = self.chave_data
lista_dados = dados["data"]["dataList"]
for entrada in lista_dados:
if isinstance(entrada, list):
item["nome_provincia"] = entrada[0]
item["preco_provincia"] = entrada[1]
elif isinstance(entrada, dict):
item["nome_provincia"] = entrada["name"]
item["preco_provincia"] = entrada["y"]
# Lógica adicional para buscar IDs de província e cidades, conforme necessário
yield scrapy.FormRequest(
"http://hangqing.ymt.com/chandi/location_charts",
formdata={"locationId": "0", "productId": str(item["id_pequeno"]), "breedId": "0"},
callback=self.processar_nivel_provincia,
meta={"item": deepcopy(item)}
)
def processar_nivel_provincia(self, resposta):
# Implementação similar para extração de preços em nível de província
pass
# Métodos adicionais para processar níveis de cidade e condado, seguindo a mesma estrutura
4. Spider para Tendências de Preços
Este spider coleta dados de preços médios por dia para análise de tendências.
# -*- coding: utf-8 -*-
import scrapy
import pymysql.cursors
from copy import deepcopy
from myapp.items import ItemTendenciaPreco
import datetime
import json
class SpiderTendenciaPreco(scrapy.Spider):
nome = 'tendencia_preco'
dominios_permitidos = ['hangqing.ymt.com']
urls_iniciais = ['http://hangqing.ymt.com/chandi_8031_0_0']
data_atual = datetime.datetime.now()
def analisar(self, resposta):
conn = pymysql.connect(
host="127.0.0.1", port=3306,
user='root', password='senha',
db='banco_dados', charset='utf8'
)
cursor = conn.cursor()
sql = "select distinct id_pequeno, nome_pequeno, url_pequeno from tabela_produtos"
cursor.execute(sql)
produtos = cursor.fetchall()
for produto in produtos:
item = ItemTendenciaPreco()
item["url_pequeno"] = produto[2]
item["id_pequeno"] = produto[0]
yield scrapy.Request(
item["url_pequeno"],
callback=self.extrair_preco,
meta={"item": deepcopy(item)}
)
def extrair_preco(self, resposta):
item = resposta.meta["item"]
item["preco_medio_dia"] = resposta.xpath("//dd[@class='c_origin_price']/p[2]//span[1]/text()").extract_first()
item["unidade"] = resposta.xpath("//dd[@class='c_origin_price']/p[2]//span[2]/text()").extract_first()
item["chave_data"] = str(self.data_atual.year) + str(self.data_atual.month) + str(self.data_atual.day)
if item["preco_medio_dia"] is None:
item["preco_medio_dia"] = 0
item["unidade"] = ""
yield item
Definição de Itens (Items)
Os itens definem a estrutura dos dados coletados, adaptados para diferentes conjuntos de informações.
# -*- coding: utf-8 -*-
import scrapy
class ItemMercadoProduto(scrapy.Item):
url_categoria_grande = scrapy.Field()
id_categoria_grande = scrapy.Field()
nome_categoria_grande = scrapy.Field()
id_categoria_media = scrapy.Field()
nome_categoria_media = scrapy.Field()
id_categoria_pequena = scrapy.Field()
url_categoria_pequena = scrapy.Field()
nome_categoria_pequena = scrapy.Field()
id_variedade = scrapy.Field()
nome_variedade = scrapy.Field()
url_variedade = scrapy.Field()
class ItemRegiaoOrigem(scrapy.Item):
id_provincia = scrapy.Field()
nome_provincia = scrapy.Field()
url_provincia = scrapy.Field()
id_cidade = scrapy.Field()
nome_cidade = scrapy.Field()
url_cidade = scrapy.Field()
id_area = scrapy.Field()
nome_area = scrapy.Field()
url_area = scrapy.Field()
class ItemTendenciaPreco(scrapy.Item):
url_pequeno = scrapy.Field()
id_pequeno = scrapy.Field()
preco_medio_dia = scrapy.Field()
unidade = scrapy.Field()
chave_data = scrapy.Field()
class ItemPrecoRegiao(scrapy.Item):
id_pequeno = scrapy.Field()
unidade = scrapy.Field()
nome_provincia = scrapy.Field()
preco_provincia = scrapy.Field()
id_provincia = scrapy.Field()
nome_cidade = scrapy.Field()
preco_cidade = scrapy.Field()
id_cidade = scrapy.Field()
nome_area = scrapy.Field()
preco_area = scrapy.Field()
id_area = scrapy.Field()
nome_cidade_variedade = scrapy.Field()
preco_cidade_variedade = scrapy.Field()
id_variedade = scrapy.Field()
chave_data = scrapy.Field()
Pipeline para Armazenamento de Dados
O pipeline gerencia a inserção dos dados coletados em um banco de dados MySQL, organizando por tipo de spider.
# -*- coding: utf-8 -*-
import pymysql
class PipelineArmazenamento(object):
def abrir_spider(self, spider):
self.conexao = pymysql.connect(
host="127.0.0.1", port=3306,
user='root', password='senha',
db='banco_dados', charset='utf8'
)
def processar_item(self, item, spider):
cursor = self.conexao.cursor()
if spider.nome == "mercado_produto":
sql = f"INSERT INTO tabela_categorias (id_grande, nome_grande, url_grande, id_media, nome_media, id_pequeno, nome_pequeno, url_pequeno, id_variedade, nome_variedade, url_variedade) VALUES ('{item['id_categoria_grande']}', '{item['nome_categoria_grande']}', '{item['url_categoria_grande']}', '{item['id_categoria_media']}', '{item['nome_categoria_media']}', '{item['id_categoria_pequena']}', '{item['nome_categoria_pequena']}', '{item['url_categoria_pequena']}', '{item['id_variedade']}', '{item['nome_variedade']}', '{item['url_variedade']}')"
cursor.execute(sql)
self.conexao.commit()
elif spider.nome == "regiao_origem":
sql = f"INSERT INTO tabela_regioes (id_provincia, nome_provincia, url_provincia, id_cidade, nome_cidade, url_cidade, id_area, nome_area, url_area) VALUES ('{item['id_provincia']}', '{item['nome_provincia']}', '{item['url_provincia']}', '{item['id_cidade']}', '{item['nome_cidade']}', '{item['url_cidade']}', '{item['id_area']}', '{item['nome_area']}', '{item['url_area']}')"
cursor.execute(sql)
self.conexao.commit()
# Lógica similar para outros spiders
cursor.close()
return item
def fechar_spider(self, spider):
self.conexao.close()