NotaControl ConsultarNfse Python

anapolis.py

import os
import re
import requests
import dateutil.parser
from dateutil.relativedelta import relativedelta
from dateutil.parser import ParserError
from datetime import timedelta


# Date Parse
from openpyxl.utils.datetime import days_to_time


def date(date_iso8601):
    try:
        tmp = dateutil.parser.parse(date_iso8601)
        tmp = tmp.replace(tzinfo=None)
    except ParserError as err:
        tmp = None
    return tmp


# Consulta XMLs
def consulta(cnpj, data, im):
    xml = "<?xml version=\"1.0\" encoding=\"utf-8\"?>\n " \
          "<ConsultarNfseEnvio " \
          " xmlns=\"http://www.issnetonline.com.br/webserviceabrasf/vsd/servico_consultar_nfse_envio.xsd\" " \
          " xmlns:tc=\"http://www.issnetonline.com.br/webserviceabrasf/vsd/tipos_complexos.xsd\"> \n" \
          "<Prestador>\n" \
          "   <tc:CpfCnpj>\n" \
          f"      <tc:Cnpj>{cnpj}</tc:Cnpj>\n" \
          "   </tc:CpfCnpj>\n" \
          f"  <tc:InscricaoMunicipal>{im}</tc:InscricaoMunicipal>\n" \
          "</Prestador>\n" \
          "<PeriodoEmissao>\n" \
          f"   <DataInicial>{data}</DataInicial>\n" \
          f"   <DataFinal>{data}</DataFinal>\n" \
          "</PeriodoEmissao>\n" \
          "</ConsultarNfseEnvio>"

    xml = xml.replace("<", "&lt;").replace(">", "&gt;")

    payload = "<?xml version=\"1.0\" encoding=\"utf-8\"?> \n" \
              "<soap:Envelope xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\" \n" \
              "                 xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" \n" \
              "                 xmlns:xsd=\"http://www.w3.org/2001/XMLSchema\">\n" \
              "  <soap:Body>\n" \
              "    <ConsultarNfse xmlns =\"http://www.issnetonline.com.br/webservice/nfd\">\n" \
              f"      <xml>{xml}</xml>\n" \
              "    </ConsultarNfse>\n" \
              "  </soap:Body>\n" \
              "</soap:Envelope>\n"

    headers = {"Accept": "*/*",
               "User-Agent": "Mozilla/4.0 (compatible; MSIE 6.0; MS Web Services Client Protocol 2.0.50727.8806)",
               "Content-type": "text/xml;charset=UTF-8",
               "Content-length": str(len(payload))
               }

    cert_file_path = "cert.pem"
    key_file_path = "key.pem"
    cert = (cert_file_path, key_file_path)
    url = "http://www.issnetonline.com.br/webserviceabrasf/anapolis/servicos.asmx"

    response = requests.post(url, headers=headers, data=payload, cert=cert)
    return {"response": response.text, "status": response.status_code}


def main():
    inicio = date("2022-10-01")
    # fim = inicio + timedelta(days=30)
    fim = date("2022-10-31")
    cnpj = "12345678000189"
    inscricao = "12345"

    while inicio <= fim:
        data = inicio.strftime("%Y-%m-%d")
        print(f"processando {data}...")
        result = consulta(cnpj=cnpj, data=data, im=inscricao)

        item = re.findall("<ConsultarNfseResult>(.*?)</ConsultarNfseResult>", result["response"], re.DOTALL)
        if len(item):
            content = item[0]
        else:
            item = re.findall("<soap:Fault>(.*?)</soap:Fault>", result["response"], re.DOTALL)
            if len(item):
                content = item[0]
            else:
                content = ""

        content = content.replace("&lt;", "<").replace("&gt;", ">").replace("\n", "").replace("\r\r", "\r")

        if "MensagemRetorno" in content:
            print("Erro na consulta!")
            inicio = inicio + timedelta(days=1)
            continue

        ano = inicio.year
        mes = inicio.month
        if not os.path.isdir(f"{ano}"):
            os.mkdir(f"{ano}")
        if not os.path.isdir(f"{ano}/{mes}"):
            os.mkdir(f"{ano}/{mes}")

        filename = cnpj + "_" + data.replace("-", "") + ".xml"
        with open(f"{ano}/{mes}/{filename}", 'w', encoding="utf-8") as f:
            f.write(content)

        inicio = inicio + timedelta(days=1)


if __name__ == '__main__':
    main()