Scraping do site do governo, Agrofit, com programação síncrona e assíncrona utilizando corrotinas e asyncio.queue no python 3.6.

_ingredientes-ativos-agrofit-scraping.md

python 3.6

Raspagem de dados Agrofit

Foram feitos 2 scripts para a raspagem de dados. Um síncrono e outro assíncrono.

• Síncrono: foram utilizadas as bibliotecas requests e BeautifulSoup;
• Assíncrono: foram utilizadas as bibliotecas aiohttp, BeautifulSoup e asyncio (corrotinas e filas).

Dados raspados

517 cadastros sendo os campos:

• ID
• Nome Comum
• Nome Comum Português
• Grupo Químico
• Nome Químico (IUPAC)
• Classes

Tempo de conclusão

No meu computador os tempos foram:

• Síncrono: 136.2 segundos (2m16s)
• Assíncrono: 75.9 segundos (1m16s)

Links Agrofit

• Página de pesquisa
• Resultado de pesquisa (POST)
• Detalhes do ingrediente ativo

async.py

import asyncio
import csv
import logging
import re

from bs4 import BeautifulSoup
import aiohttp

logging.basicConfig(level=logging.ERROR)
log = logging.getLogger("agrofit")

# # # # # # # # # # # # # # # # # # # # # #
#                  Classes                #
# # # # # # # # # # # # # # # # # # # # # #

class IngredienteAtivo():
    __slots__ = ("id", "nome_comum", "nome_comum_pt", "grupo_quimico", "iupac", "classes")

    def __init__(self, id, nome_comum, nome_comum_pt, grupo_quimico, iupac, classes):
        self.id = id
        self.nome_comum = nome_comum
        self.nome_comum_pt = nome_comum_pt
        self.grupo_quimico = grupo_quimico
        self.iupac = iupac
        self.classes = classes

    def __str__(self):
        return """
         ================== Ingrediente Ativo =======================
        | ID: {0.id}
        | Nome Comum: {0.nome_comum}
        | Nome Comum Português: {0.nome_comum_pt}
        | Grupo Químico: {0.grupo_quimico}
        | IUPAC: {0.iupac}
        | Classes: {0.classes}
         ============================================================""".format(self)

    def __repr__(self):
        return "IngredienteAtivo({0.id!r}, {0.nome_comum!r}, {0.nome_comum_pt!r}, {0.grupo_quimico!r}, {0.iupac!r}, {0.classes!r})".format(self)

    def as_csv_row(self):
        return (self.id, self.nome_comum, self.nome_comum_pt, self.grupo_quimico, self.iupac, ", ".join(self.classes))


# # # # # # # # # # # # # # # # # # # # # #
#                  Utils                  #
# # # # # # # # # # # # # # # # # # # # # #

URL_BASE = "http://agrofit.agricultura.gov.br/agrofit_cons/"
URL_CONSULTA = URL_BASE + "!ap_ing_ativo_lista_cons"
URL_DETALHE = URL_BASE + "!ap_ing_ativo_detalhe_cons"

REGEX_LINK = re.compile("^!ap_ing_ativo_detalhe_cons\?p_id_ingrediente_ativo=\d+")
REGEX_CLASSES = re.compile(".*incluirClasse\('\d*' *, *'(.*)'\).*")

AGROFIT_ENCODING = "ISO-8859-1"

def _sanitize_string(string):
    return re.sub("\s+", " ", string).strip()


# # # # # # # # # # # # # # # # # # # # # #
#                  Script                 #
# # # # # # # # # # # # # # # # # # # # # #

async def produce_url(session, queue):
    post_data = {
        "p_nm_comum_port": "",              # Pesquisa: Nome
        "p_id_grupo_quimico": "Selecione",  # Pesquisa: Grupo Químico
        "p_id_classe": "",                  # Pesquisa: Classe
        "p_id_cultura": "",                 # Pesquisa: Cultura
        "p_nm_sort": "nm_comum_portugues",  # Ordenação da pesquisa
        "p_linha_inicial": 0,               # Página inicial (10/pag)
    }
    while True:
        async with session.post(URL_CONSULTA, data=post_data, timeout=10) as response:
            assert response.status == 200
            html_str = await response.text(encoding=AGROFIT_ENCODING)

            soup = BeautifulSoup(html_str, "html.parser")
            btn_next = soup.find("input", attrs={"class": "botao", "type": "BUTTON", "name": "proximo"})
            links = soup.findAll("a", href=REGEX_LINK)

            if len(links) == 0:
                break

            for url in set( (link["href"] for link in links) ):
                await queue.put(url)

            post_data['p_linha_inicial'] += 10

            # if post_data['p_linha_inicial'] > 40:  # limita em 5 páginas para testes
            #     break


async def consume_url(session, queue, queue_2):

    while True:
        url = await queue.get()
        id = int(url.split("=").pop())
        
        async with session.get(URL_DETALHE, params={"p_id_ingrediente_ativo": id}, timeout=10) as response:
            log.info("+++ Fetched ID: %d", id)
            html_str = await response.text(encoding=AGROFIT_ENCODING)
            soup = BeautifulSoup(html_str, "html.parser", )

            nome_comum, nome_comum_pt, grupo_quimico, iupac = "", "", "", ""

            elem_nome_comum = soup.find("input", attrs={"name": "p_nm_comum"})
            if elem_nome_comum is not None:
                nome_comum = _sanitize_string(elem_nome_comum.attrs.get('value', ''))

            elem_nome_comum_pt = soup.find("input", attrs={"name": "p_nm_comum_portugues"})
            if elem_nome_comum_pt is not None:
                nome_comum_pt = _sanitize_string(elem_nome_comum_pt.attrs.get('value', ''))
            
            elem_grupo_quimico = soup.find("input", attrs={"name": "p_nm_grupo_quimico"})
            if elem_grupo_quimico is not None:
                grupo_quimico = _sanitize_string(elem_grupo_quimico.attrs.get('value', ''))
            
            elem_iupac = soup.find("textarea", attrs={"name": "p_nm_quimico_iupac"})
            if elem_iupac is not None:
                iupac = _sanitize_string(elem_iupac.get_text())
            
            classes_script = soup.find("table", id="classes").find('script').string.strip()
            classes = [ _sanitize_string(match.group(1)) for match in REGEX_CLASSES.finditer(classes_script) ]
            
            ia = IngredienteAtivo(id, nome_comum, nome_comum_pt, grupo_quimico, iupac, classes)
            log.debug(ia)

            queue.task_done()
            await queue_2.put(ia)


async def consume_ingrediente(queue_2):
    with open("data_queue_async.csv", "w") as csvfile:
        writer = csv.writer(csvfile)
        while True:
            ingrediente = await queue_2.get()
            log.info(">>> Wrote to CSV: %d", ingrediente.id)
            try:
                writer.writerow(ingrediente.as_csv_row())
            except UnicodeEncodeError as err:
                log.error("Ingrediente #{}: -> {}".format(ingrediente.id, err))
            finally:
                queue_2.task_done()


async def main(loop):
    async with aiohttp.ClientSession(loop=loop) as client:
        queue = asyncio.Queue(loop=loop)
        queue_2 = asyncio.Queue(loop=loop)
        
        consumer = asyncio.ensure_future(consume_url(client, queue, queue_2))
        consumer_2 = asyncio.ensure_future(consume_ingrediente(queue_2))
        
        await produce_url(client, queue)

        await queue.join()
        await queue_2.join()

        consumer.cancel()
        consumer_2.cancel()


if __name__ == '__main__':
    
    from time import perf_counter


    time_init = perf_counter()

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()

    time_final = perf_counter() - time_init

    print(">>> Tempo(segs): {:.4}".format(time_final))

resultado.csv

sync.py

import re
import csv
import logging

import requests
from bs4 import BeautifulSoup

logging.basicConfig(level=logging.ERROR)
log = logging.getLogger("agrofit")

def _sanitize_string(string):
    # string = string.encode('utf-8').decode('ISO-8859-1')
    return re.sub("\s+", " ", string).strip()


class IngredienteAtivo():
    __slots__ = ("id", "nome_comum", "nome_comum_pt", "grupo_quimico", "iupac", "classes")

    def __init__(self, id, nome_comum, nome_comum_pt, grupo_quimico, iupac, classes):
        self.id = id
        self.nome_comum = nome_comum
        self.nome_comum_pt = nome_comum_pt
        self.grupo_quimico = grupo_quimico
        self.iupac = iupac
        self.classes = classes

    def __str__(self):
        return """
         ================== Ingrediente Ativo =======================
        | ID: {0.id}
        | Nome Comum: {0.nome_comum}
        | Nome Comum Português: {0.nome_comum_pt}
        | Grupo Químico: {0.grupo_quimico}
        | IUPAC: {0.iupac}
        | Classes: {0.classes}
         ============================================================""".format(self)

    def __repr__(self):
        return "IngredienteAtivo({0.id!r}, {0.nome_comum!r}, {0.nome_comum_pt!r}, {0.grupo_quimico!r}, {0.iupac!r}, {0.classes!r})".format(self)

    def as_csv_row(self):
        return (self.id, self.nome_comum, self.nome_comum_pt, self.grupo_quimico, self.iupac, ", ".join(self.classes))


class IngredientesAtivosScrape():
    URL_BASE = "http://agrofit.agricultura.gov.br/agrofit_cons/"
    URL_CONSULTA = URL_BASE + "!ap_ing_ativo_lista_cons"
    URL_DETALHE = URL_BASE + "!ap_ing_ativo_detalhe_cons"

    REGEX_LINK = re.compile("^!ap_ing_ativo_detalhe_cons\?p_id_ingrediente_ativo=\d+")
    REGEX_CLASSES = re.compile(".*incluirClasse\('\d*' *, *'(.*)'\).*")

    @classmethod
    def get_ingrediente_ativo(cls, id):
        log.info("...Scraping {}?p_id_ingrediente_ativo={})".format(cls.URL_DETALHE, id))

        response = requests.get(cls.URL_DETALHE, { "p_id_ingrediente_ativo": id })

        if response.status_code >= 400:
            log.warning("ID: {}  Response Status: {}".format(id, response.status_code))
            return None

        soup = BeautifulSoup(response.text, "html.parser", )
        nome_comum, nome_comum_pt, grupo_quimico, iupac = "", "", "", ""

        elem_nome_comum = soup.find("input", attrs={"name": "p_nm_comum"})
        if elem_nome_comum is not None:
            nome_comum = _sanitize_string(elem_nome_comum.attrs.get('value', ''))

        elem_nome_comum_pt = soup.find("input", attrs={"name": "p_nm_comum_portugues"})
        if elem_nome_comum_pt is not None:
            nome_comum_pt = _sanitize_string(elem_nome_comum_pt.attrs.get('value', ''))
        
        elem_grupo_quimico = soup.find("input", attrs={"name": "p_nm_grupo_quimico"})
        if elem_grupo_quimico is not None:
            grupo_quimico = _sanitize_string(elem_grupo_quimico.attrs.get('value', ''))
        
        elem_iupac = soup.find("textarea", attrs={"name": "p_nm_quimico_iupac"})
        if elem_iupac is not None:
            iupac = _sanitize_string(elem_iupac.get_text())
        
        classes_script = soup.find("table", id="classes").find('script').string.strip()
        classes = [ _sanitize_string(match.group(1)) for match in cls.REGEX_CLASSES.finditer(classes_script) ]

        ia = IngredienteAtivo(id, nome_comum, nome_comum_pt, grupo_quimico, iupac, classes)
        log.debug(ia)
        return ia

    
    @classmethod
    def search(cls, nome=None, grupo_quimico=None, classe=None, cultura=None):
        post_data = {
            "p_nm_comum_port": nome or "",                      # Pesquisa: Nome
            "p_id_grupo_quimico": grupo_quimico or "Selecione", # Pesquisa: Grupo Químico
            "p_id_classe": classe or "",                        # Pesquisa: Classe
            "p_id_cultura": cultura or "",                      # Pesquisa: Cultura
            "p_nm_sort": "nm_comum_portugues",                  # Ordenação da pesquisa
            "p_linha_inicial": 0,                               # Página inicial (são 10 por página)
        }
        page = 0
        while True:
            post_data["p_linha_inicial"] = page * 10
            response = requests.post(cls.URL_CONSULTA, post_data)
            soup = BeautifulSoup(response.text, "html.parser")

            log.info("=== Ínicio da {}ª página ===".format(page + 1))
            
            btn_next = soup.find("input", attrs={"class": "botao", "type": "BUTTON", "name": "proximo"})

            links = soup.findAll("a", href=cls.REGEX_LINK)
            links_ids = ( int(link["href"].split("=").pop()) for link in links )

            for id_ingrediente in set(links_ids):
                ingrediente_ativo =  cls.get_ingrediente_ativo(id_ingrediente)

                if ingrediente_ativo:
                    yield ingrediente_ativo

            if btn_next is None:
                log.info("=== Última página terminada ===".format(page + 1))
                break

            page += 1
            

    @classmethod
    def as_csv(cls, nome_arquivo, nome=None, grupo_quimico=None, classe=None, cultura=None):
        with open(nome_arquivo, 'w') as csvfile:
            writer = csv.writer(csvfile)

            for ingrediente in cls.search(nome, grupo_quimico, classe, cultura):
                try:
                    writer.writerow(ingrediente.as_csv_row())
                except UnicodeEncodeError as err:
                    log.error("Ingrediente #{}: -> {}".format(ingrediente.id, err))


if __name__ == '__main__':
    from time import perf_counter
    time_init = perf_counter()
    IngredientesAtivosScrape.as_csv("data_sync.csv")
    time_final = perf_counter() - time_init

    print("Tempo(segs): {:.4}".format(time_final))

Olá, por acaso chegou a fazer alguma raspagem na página: https://agrofit.agricultura.gov.br/agrofit_cons/!ap_produto_form_consulta_cons