twisted crawler, alvo: mercadolivre

Já pensou em fazer um programa que acessa o Mercado Livre, identifica o link de cada categoria, e extrai todos os produtos da primeira página de cada uma dessas categorias?

Pode até parecer complexo, mas não é. Esse programa existe, é simples, e está aqui, neste artigo, pronto pra você testar e modificar. :)

A algumas semanas tenho feito alguns crawlers pra esses sites que vendem produtos, como o Web Motors, Submarino, e até mesmo o eBay. A idéia é simplesmente extrair todos os produtos da primeira página de cada categoria, e montar uma base de dados com essas informações. O objetivo? Segredo de Estado.

Um dos problemas que encontrei ao fazer esse tipo de crawler, é controlar a concorrência de acesso. Se o programa acessa uma categoria por vez, demora uma eternidade pra baixar e processar cada página. Por outro lado, se ele acessa todas as categorias descontroladamente, o processo acaba criando muitos File Descriptors e isso causa diversos outros problemas pro sistema operacional – e não adianta falar em aumentar o limite usando ulimit -n, porque esses crawlers têm baixado milhões de links por dia.

A melhor maneira que encontrei pra solucionar esse problema, foi criando uma classe chamada “Controller”, que coloca as requisições em uma fila, baixa N páginas por vez, e ao invés de processá-las, simplesmente salva em um arquivo no disco.

Todos esses arquivos, com nomes únicos, são gerados usando um hash SHA1, e colocados em uma lista para serem processados offline, após o processo de download terminar.

O twisted tem um recurso muito interessante, que permite agrupar diversos deferreds em um único, e quando todos eles terminam, executa um callback. Nesse caso, quando todos os processos de baixar a página da subcategoria terminam, executa a função “offline”, que começa a processar cada uma, e extrair os produtos de lá.

Esta segunda etapa também poderia ser feita em paralelo, usando o twisted-parallels, mas decidi não colocar porque o código ficaria muito maior e você não teria tanta paciência pra entendê-lo. Usar um thread pool pra isso não valeria a pena, pois vale lembrar que o python usa o GIL, e I/O em thread só consome recurso e não aumenta em nada o desempenho.

A classe “MercadoLivre” tem 3 funções:

• __str__: retorna o link com todas as categorias do Mercado Livre
• parse_categories: um parser que retorna uma lista composta por tuples de (url, nome)
• parse_subcategory: um parser pro conteúdo de cada categoria, que retorna uma lista com o título dos produtos anunciados lá

Todos esses parsers são baseados no BeautifulSoup 3.1. É necessário tê-lo instalado pra usar o programa.

Por fim, para usar este programa em outros sites, basta substituir a classe “MercadoLivre” pela sua própria, tipo “Submarino”.

Detalhe: essa porcaria de WordPress mostra a identação do código errada no artigo, mas se você clicar em “view plain”, poderá copiar e colar o código correto. Chame o suporte!

Aqui o código (muito belo, por sinal…):

#!/usr/bin/env python
# coding: utf-8

import os, re, sys
import shutil, hashlib
from Queue import Queue
from BeautifulSoup import BeautifulSoup

# twisted
from twisted.web import client
from twisted.internet import defer, threads, reactor

class MercadoLivre:
    noscript = re.compile(r"(?is)<script[^>]*>(.*?)</script>")

    def __str__(self):
	return 'http://www.mercadolivre.com.br/jm/ml.allcategs.AllCategsServlet'

    def parse_categories(self, content):
	catlist = {}
	cleanup = lambda s: s.replace('\n', '').strip()

	# parse the document
	soup = BeautifulSoup(content)

	# find all categories and their items
	current = ''
	for item in soup.findAll('a', {'class':['categ','seglnk']}):
	    text = cleanup(item.contents[0])
	    attrs = dict(item.attrs)
	    if attrs.get(u'class') == u'categ':
		current = text
		catlist[current] = []
	    else:
		catlist[current].append((attrs.get(u'href', ''), text))

	# return the list of categories and their items
	return catlist

    def parse_subcategory(self, content):
	result = []
	soup = BeautifulSoup(self.noscript.sub('', content),
	    convertEntities=BeautifulSoup.HTML_ENTITIES)

	for item in soup.findAll('div', {'class':'col_titulo'}):
	    result.append(item.find('a').contents[0].strip())

	return result

class Controller:
    def __init__(self, fetch):
	self.count = 0
	self.limit = fetch
	self.queue = Queue()
	self.tmpdir = '/tmp/mercadolivre.%d' % os.getpid()
	self.dispatch()

    def encode(self, text):
	return unicode(text).encode('utf-8', 'replace')

    def getPage(self, url, *args, **kwargs):
	d = defer.Deferred()
	self.queue.put((d, url, args, kwargs))
	return d

    def dispatch(self):
	while True:
	    try:
		assert self.count < self.limit
		d, url, args, kwargs = self.queue.get_nowait()
	    except: break
	    self.count += 1
	    deferred = client.getPage(url, *args, **kwargs)
	    deferred.addBoth(self.decrease_count)
	    deferred.chainDeferred(d)
	reactor.callLater(1, self.dispatch)

    def decrease_count(self, result):
	self.count -= 1
	return result

class main(Controller):
    def __init__(self, fetch, parser):
	# set the concurrent download limit for the crawler
	Controller.__init__(self, fetch)

	# fetch the main categories page
	self.total = 0
	self.files = []
	self.parser = parser
	d = self.getPage(str(parser), timeout=60)
	d.addCallback(self.fetch_categories)
	d.addErrback(self.error_categories)

    def error_categories(self, error):
	# hmmm... fatal error, cannot continue
	print 'cannot fetch categories from %s: %s' % (str(self.parser), str(error))
	reactor.stop()

    def error_subcategory(self, error, href, category, subcategory):
	# problem fetching subcategory contents...
	print 'error "%s / %s": [%s] %s' % (category, subcategory, href, error.value)

    def fetch_categories(self, content):
	# parse the contents in a thread
	reactor.callInThread(self.parse_categories, content)

    def parse_categories(self, content):
	try: categories = self.parser.parse_categories(content)
	except Exception, e:
	    print 'error parsing categories: %s' % str(e)
	    reactor.stop()
	    return

	print 'going to fetch %d categories...' % len(categories)
	tasks = []
	for category, contents in categories.items():
	    category = self.encode(category)
	    for href, subcategory in contents:
		href, subcategory = self.encode(href), self.encode(subcategory)
		d = self.getPage(href, timeout=60)
		d.addCallback(self.save_subcategory, href, category, subcategory)
		d.addErrback(self.error_subcategory, href, category, subcategory)
		tasks.append(d)

	# call the offline subcategory parser after downloading everything...
	d = defer.gatherResults(tasks)
	d.addCallback(self.offline)

    def save_subcategory(self, contents, href, category, subcategory):
	# create the tmpdir if it doesn't exist
	if not os.path.exists(self.tmpdir): os.mkdir(self.tmpdir)

	# create a unique hash for each category
	hash = hashlib.new('sha1', category+subcategory).hexdigest()
	filename = os.path.join(self.tmpdir, hash+'.dump')
	try:
	    fd = open(filename, 'w')
	    fd.write(contents)
	    fd.close()
	except: return
	self.files.append((href, category, subcategory, filename))
	print 'saving "%s / %s": %s' % (category, subcategory, hash)

    def offline(self, null):
	# start processing each subcategory...
	reactor.stop()
	for item in self.files:
	    href, category, subcategory, filename = item
	    sys.stdout.write('parsing "%s / %s": ' % (category, subcategory))
	    sys.stdout.flush()
	    fd = open(filename)
	    try: results = self.parser.parse_subcategory(fd.read())
	    except Exception, e:
		print 'error! %s' % str(e)
		continue

	    lr = len(results)
	    self.total += lr
	    print '%d items' % lr
	    for result in results:
		print '  ' + self.encode(result)

	print '\n%d items processed. cleaning up!' % self.total
	shutil.rmtree(self.tmpdir)

if __name__ == '__main__':
    reactor.callWhenRunning(main, 10, MercadoLivre())
    reactor.run()

Agora, é só você se dedicar e fazer isso pros sites que te interessam.

Notas?