Um parser das notas de corretagem da RICO no formato ir_investidor

parser_rico.py

import subprocess
from xml.etree import ElementTree as ET
import pandas as pd
import re
import logging
import sys
import os.path

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

TIPOS_DE_MERCADO = [ 
    "VISTA", 
    "FRACIONÁRIO",
    # "LEILÃO",
    # "EXERCÍCIO DE OPÇÕES DE COMPRA",
    # "EXERCÍCIO DE OPÇÕES DE VENDA",
    # "TERMO",
    # "FUTURO COM RETENÇÃO DE GANHO",
    # "FUTURO COM MOVIMENTAÇÃO CONTÍNUA",
    # "OPÇÕES DE COMPRA",
    # "OPÇÕES DE VENDA"
]

def get_array_data_pregao(tree):
    return get_proximo_elemento(tree, "Data pregão")

# In RICO's PDF, date comes after the operations
# Get all dates found in pages
# Useful for parsing consolidated files provided by Rico
def get_proximo_elemento(tree, tag, next=True):
    array_tags = []
    previous = []

    try:
        elements = tree.findall(".//text")

        tag_found = False
        for el in elements:            
            if el.text == tag:
                tag_found = True
            elif tag_found: # no xpath support
                if next:
                    array_tags.append(el.text)
                else:
                    array_tags.append(previous[0])
                tag_found = False
            
            previous.append(el.text)
            
            if len(previous) > 2:
                previous.pop(0)

    except Exception as e:
        logger.error("Erro ao processar '%s'.", tag, e)
    
    return array_tags


def contains_obs(buffer):
    return not buffer[5].isdigit()


def add_negocios_realizados(negocios_realizados, buffer, data_pregao):
    # print(buffer)
    
    if len(buffer) < 5:
        logger.warning("Ignorando linha.")
        return 
    # 2 - Corretora ou pessoa vinculada atuou na contra parte. 
    # # - Negócio direto
    # 8 - Liquidação Institucional
    # D - Day Trade
    # F - Cobertura B - Debêntures
    # A - Posição futuro
    # C - Clubes e fundos de Ações P - Carteira Própria
    # H - Home Broker
    # X - Box
    # Y - Desmanche de Box
    # L - Precatório
    # T - Liquidação pelo Bruto I - POP
    # regex = r"([\d\w-]+) (C|V|S) (FRACION.RIO|VISTA) ([\w ]+) (CI ER|CI ES|CI|ON N1|PNA N1|ON NM|PN N2|PN|ON|PNA|PNB) ?([#|D|F|A|C|H|X|Y|L|T|I]*) (\d+) ([\d,]+) ([\d,\.]+) (D|C)"
    regex = r"([\d\w-]+) (C|V|S) (FRACION.RIO|VISTA) ([\w ]+) (CI ER|CI ES|CI|ON N1|PNA N1|ON NM|PN N2|PN|ON|PNA|PNB|PN N1|UNT N2) ?([#|D|F|A|C|H|X|Y|L|T|I]*) (\d+) ([\d,]+) ([\d,\.]+) (D|C)"
    
    pattern = re.compile(regex)

    line = " ".join(buffer)
    logger.debug(line)
    match = pattern.match(line)
    
    if match:
        try:
            negociacao = match.group(1)
            cv = match.group(2)
            tipo_mercado = match.group(3)
            papel = match.group(4)
            espec_titulo = match.group(5)
            obs = match.group(6)
            quantidade = match.group(7)
            preco_ajuste = match.group(8)
            valor_operacao = match.group(9)
            dc = match.group(10)

            ativo = "FII" if "FII" in papel else "ACAO"

            negocios_realizados.get("data_operacao").append(data_pregao)
            negocios_realizados.get("data_liquidacao").append(data_pregao)
            negocios_realizados.get("ativo").append(ativo)
            negocios_realizados.get("papel").append(papel)
            negocios_realizados.get("operacao").append('COMPRA' if 'C' in cv else 'VENDA')
            negocios_realizados.get("qtd").append(quantidade)
            negocios_realizados.get("preco").append(preco_ajuste)
            negocios_realizados.get("valor").append(valor_operacao)
            negocios_realizados.get("day_trade").append('S' if 'D' in obs else 'N')
            negocios_realizados.get("corretagem").append("0,00")
            negocios_realizados.get("iss").append("0,00")
            negocios_realizados.get("taxas_b3").append("0,00")
            negocios_realizados.get("obs").append(obs)
            
            # ir_investidor
            negocios_realizados.get("IRRF").append("0,00")
            negocios_realizados.get("NOVO PAPEL").append("")
            negocios_realizados.get("QTD ANTIGA").append(0)
            negocios_realizados.get("QTD NOVA").append(0)
            negocios_realizados.get("COMUNS A COMPENSAR").append("0,00")
            negocios_realizados.get("DAYTRADE A COMPENSAR").append("0,00")
            negocios_realizados.get("FIIS A COMPENSAR").append("0,00")
            negocios_realizados.get("IRRF A COMPENSAR").append("0,00")
            negocios_realizados.get("ALUGUEL").append("0,00")
            negocios_realizados.get("CUSTOS").append("0,00")
                
            logger.debug("Registro processado com sucesso. '{}'".format(line))
        except Exception as e:
            logger.error("Erro ao processar registro. '{}'".format(line), e)
    else:
        logger.warn("Erro ao processar linha. '{}'".format(line))

    return negocios_realizados

def is_same_line(previous_line, curr_line):
    return previous_line == curr_line


if __name__ == "__main__":
    file = ""
    
    if len(sys.argv) < 2:
        logger.info("Missing file argument.")
        logger.info(f"{sys.argv[0]} <name-of-brokerage-file>")
        sys.exit(1)
    else:
        file = sys.argv[1]
        
        if not os.path.exists(file):
            logger.info(f"File not found: {file}")
            sys.exit(2)
        else:
            logger.info(f"Parsing file: {file}")
    
    result = subprocess.check_output(["pdftohtml", "{}".format(file), "-stdout", "-xml", "-i"]) 
    tree = ET.fromstring(result)
    child = tree.getchildren()

    # ET.dump(tree)

    # Used to store operations and later convert to pandas df
    negocios_realizados = {
        "data_operacao": [],
        "data_liquidacao": [],
        "ativo": [],
        "operacao": [],
        "day_trade": [],
        "papel": [],
        "qtd": [],
        "preco": [],
        "valor": [],
        "corretagem": [],
        "iss": [],
        "taxas_b3": [],
        "obs": [],
        
        # Campos necessários para o ir_investidor
        "IRRF": [],
        "NOVO PAPEL": [], 
        "QTD ANTIGA": [], 
        "QTD NOVA": [],
        "COMUNS A COMPENSAR": [],
        "DAYTRADE A COMPENSAR": [], 
        "FIIS A COMPENSAR": [], 
        "IRRF A COMPENSAR": [],
        "ALUGUEL": [],
        "CUSTOS": []
    }

    # Get dates if using multiple sheets. For example, end of year files published by RICO
    array_data_pregao = get_array_data_pregao(tree)

    page = 0
    for t in tree.iter("page"):
        logger.info("Processando página '%s' - data pregão '%s'...", page, array_data_pregao[page])

        curr_line_top = -1
        data_pregao_is_next = False
        buffer = []
        
        for tx in t.iter('text'):
            if tx.text == "1-BOVESPA":
                if buffer:
                    add_negocios_realizados(negocios_realizados, buffer, array_data_pregao[page])
                    buffer = []
                    curr_line_top = -1
                
                curr_line_top = tx.attrib.get("top")

            if is_same_line(curr_line_top, tx.attrib.get("top")):
                buffer.append(tx.text.strip())

         # Add last record
        negocios_realizados = add_negocios_realizados(negocios_realizados, buffer, array_data_pregao[page])            
        buffer = []
        curr_line_top = -1
    
        page += 1

    array_taxa_liquidacao = get_proximo_elemento(tree, "Taxa de liquidação", False)
    array_taxa_operacional = get_proximo_elemento(tree, "Taxa Operacional", False)
    array_emolumentos = get_proximo_elemento(tree, "Emolumentos", False)

    logger.info("Taxa de liquidação: '%s'", array_taxa_liquidacao)
    logger.info("  Taxa Operacional: '%s'", array_taxa_operacional)
    logger.info("       Emolumentos: '%s'", array_emolumentos)

            # negocios_realizados.get("data_operacao").append(data_pregao)
            # negocios_realizados.get("data_liquidacao").append(data_pregao)
            # negocios_realizados.get("").append(ativo)
            # negocios_realizados.get("").append(papel)
            # negocios_realizados.get("operacao").append(cv)
            # negocios_realizados.get("qtd").append(quantidade)
            # negocios_realizados.get("preco").append(preco_ajuste)
            # negocios_realizados.get("valor").append(valor_operacao)
            # negocios_realizados.get("day_trade").append('S' if 'D' in obs else 'N')
            # negocios_realizados.get("corretagem").append("0,00")
            # negocios_realizados.get("iss").append("0,00")
            # negocios_realizados.get("taxas_b3").append("0,00")
            # negocios_realizados.get("obs").append(obs)
            
    print("#############")            
    print(negocios_realizados)

    df = pd.DataFrame(negocios_realizados) 
    df.fillna(0)
    
    # DATA OPERACAO,DATA LIQUIDACAO,ATIVO,OPERACAO,DAYTRADE?,PAPEL,QTD,PRECO,VALOR,CORRETAGEM,ISS (5%),
    # TAXAS B3,ALUGUEL,CUSTOS,IRRF,NOVO PAPEL,QTD ANTIGA,QTD NOVA,
    # COMUNS A COMPENSAR,DAYTRADE A COMPENSAR,FIIS A COMPENSAR,IRRF A COMPENSAR
    df.rename(columns={"data_operacao": "DATA OPERACAO",
                        "data_liquidacao": "DATA LIQUIDACAO",
                        "ativo": "ATIVO",
                        "operacao": "OPERACAO",
                        "papel": "PAPEL",
                        "qtd": "QTD",
                        "valor": "VALOR",
                        "day_trade": "DAYTRADE?",
                        "preco": "PRECO",
                        "custos": "CUSTOS",
                        "corretagem": "CORRETAGEM",
                        "iss": "ISS (5%)",
                        "taxas_b3": "TAXAS B3"
                        },
                 inplace=True)

    #print(df)

    df.to_csv('file_{}.csv'.format(array_data_pregao[0].replace("/","-")), 
        index=False, 
        columns=["DATA OPERACAO","DATA LIQUIDACAO","ATIVO","OPERACAO","DAYTRADE?","PAPEL","QTD","PRECO","VALOR","CORRETAGEM","ISS (5%)",
            "TAXAS B3","ALUGUEL","CUSTOS","IRRF","NOVO PAPEL","QTD ANTIGA","QTD NOVA",
            "COMUNS A COMPENSAR","DAYTRADE A COMPENSAR","FIIS A COMPENSAR","IRRF A COMPENSAR"])

    logger.info("A planilha de operações foi gerada e salva em: '%s'", 'file_{}.csv'.format(array_data_pregao[0].replace("/","-")))