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)]*>(.*?)“)

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() [/sourcecode] Agora, é só você se dedicar e fazer isso pros sites que te interessam. Notas?