Script to fetch some Brazilian stock metadata

codigos.csv

main.py

import csv
from tqdm import tqdm
from lxml import etree
from urllib.request import urlopen
from concurrent.futures import ThreadPoolExecutor


START_URL = 'http://bvmf.bmfbovespa.com.br/cias-listadas/empresas-listadas/BuscaEmpresaListada.aspx?idioma=pt-br'
PAYLOAD = b'ctl00_contentPlaceHolderConteudo_AjaxPanelBuscaPostDataValue=ctl00_contentPlaceHolderConteudo_AjaxPanelBusca%2CActiveElement%2Cctl00_contentPlaceHolderConteudo_BuscaNomeEmpresa1_btnTodas%3B&ctl00%24contentPlaceHolderConteudo%24tabMenuEmpresaListada=%7B%22State%22%3A%7B%7D%2C%22TabState%22%3A%7B%22ctl00_contentPlaceHolderConteudo_tabMenuEmpresaListada_tabNome%22%3A%7B%22Selected%22%3Atrue%7D%7D%7D&ctl00%24contentPlaceHolderConteudo%24mpgPaginas_Selected=0&httprequest=true&RadAJAXControlID=ctl00_contentPlaceHolderConteudo_AjaxPanelBusca&__EVENTTARGET=ctl00%3AcontentPlaceHolderConteudo%3ABuscaNomeEmpresa1%3AbtnTodas'


def cod_url(cod):
    return 'http://bvmf.bmfbovespa.com.br/pt-br/mercados/acoes/empresas/ExecutaAcaoConsultaInfoEmp.asp?CodCVM=' + cod


def inner_text(tag):
    return ' '.join(tag.xpath('./text()')).strip()


def bytwo(it):
    a = iter(it)
    return zip(a, a)


def parse_data(cod, tds):
    data = { 'codcvm': cod, 'codigos': [] }
    for key, val in bytwo(tds):
        text = inner_text(key).lower()
        if text.startswith('códigos de negociação'):
            data['codigos'] = list(set(val.xpath('.//a/text()')))
        elif text.startswith('cnpj'):
            data['cnpj'] = inner_text(val)
        elif text.startswith('site'):
            data['site'] = ''.join(val.itertext()).strip()
        elif text.startswith('institui'):
            data['instituicao'] = inner_text(val)
    return data


def get_data(cod):
    url = cod_url(cod)
    parser = etree.HTMLParser(encoding='utf-8')
    doc = etree.parse(urlopen(url), parser=parser)
    tds = doc.xpath('//div[@id="accordionDados"]//td')
    tds.extend(doc.xpath('//div[@id="divContatos"]//td'))
    return parse_data(cod, tds)


def cod_list():
    parser = etree.HTMLParser(encoding='utf-8')
    doc = etree.parse(urlopen(START_URL, data=PAYLOAD), parser=parser)
    table_id = "ctl00_contentPlaceHolderConteudo_BuscaNomeEmpresa1_grdEmpresa_ctl01"
    urls = doc.xpath(f'//table[@id="{table_id}"]/tbody/tr/td[1]/a/@href')
    return [x[x.find('=')+1:] for x in urls]


def main():
    with open('codigos.csv', 'w') as out, ThreadPoolExecutor() as executor:
        cods = cod_list()
        writer = csv.writer(out)
        header = ['codigo', 'codcvm', 'cnpj', 'site', 'instituicao']
        writer.writerow(header)
        for data in tqdm(executor.map(get_data, cods), total=len(cods)):
            for cod in data['codigos']:
                row = [
                    cod,
                    data.get('codcvm', None),
                    data.get('cnpj', None),
                    data.get('site', None),
                    data.get('instituicao', None),
                ]
                writer.writerow(row)


if __name__ == "__main__":
    main()