Ajustes leitura do certificado e templates ginfes

generate-pdfs.py

@@ -0,0 +1,135 @@
+import os
+import requests
+import certifi
+import pandas as pd
+import json
+from lxml import html
+import xml.etree.ElementTree as ET
+from pytrustnfe.certificado import Certificado
+from pytrustnfe.nfse.ginfes import consultar_nfse_por_rps
+import re
+import time
+import unicodedata
+
+def sanitize_filename(name):
+    name = name.strip()
+    name = unicodedata.normalize('NFKD', name).encode('ascii', 'ignore').decode('ascii')
+    name = re.sub(r'[^\w\-]', '_', name)
+    name = name.lower()
+    return name
+
+def download_note_pdf(verification_id, note_number, file_name, output_dir="output"):
+    session = requests.Session()
+    session.verify = certifi.where()
+    session.headers.update({
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36'
+    })
+
+    consulta_url = "http://visualizar.ginfes.com.br/report/consultarNota"
+    params = {
+        "__report": "nfs_ver15",
+        "cdVerificacao": verification_id,
+        "numNota": str(note_number),
+        "cnpjPrestador": "null"
+    }
+
+    formulario_url = "http://visualizar.ginfes.com.br/report/exportacao"
+
+    os.makedirs(output_dir, exist_ok=True)
+
+    try:
+        response = session.get(consulta_url, params=params)
+        response.raise_for_status()
+
+        tree = html.fromstring(response.content)
+        form = tree.xpath("//form[@name='exportar']")
+        if not form:
+            raise ValueError("Export form not found.")
+
+        form_data = {input_tag.attrib.get('name'): input_tag.attrib.get('value', '')
+                     for input_tag in form[0].xpath(".//input")}
+        form_data.update({'imprime': '0', 'tipo': 'pdf'})
+
+        post_response = session.post(formulario_url, data=form_data, stream=True)
+        post_response.raise_for_status()
+
+        if 'application/pdf' not in post_response.headers.get('Content-Type', ''):
+            raise ValueError("The response is not a PDF file.")
+
+        output_path = os.path.join(output_dir, file_name)
+        with open(output_path, 'wb') as file:
+            for chunk in post_response.iter_content(chunk_size=1024):
+                if chunk:
+                    file.write(chunk)
+
+        print(f"PDF saved at: {output_path}")
+
+    except requests.exceptions.SSLError as e:
+        print(f"SSL Error: {e}")
+    except requests.exceptions.RequestException as e:
+        print(f"Request Error: {e}")
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+
+if __name__ == "__main__":
+    # listing notes to download
+    filename = '/Users/addo/Downloads/teste_emissao_nf_pdf.xlsx'
+    df = pd.read_excel(filename, skiprows=1)
+
+
+    certificado = open("/Users/addo/jobs/addodelgrossi/go-nfs-e/resources/35143637000111.pfx", "rb").read()
+    certificado = Certificado(certificado, b'k4ex2018')
+
+    previous_serie = None
+    index = 1
+    for _, row in df.iterrows():
+        current_serie = row['SERIE']
+
+        if current_serie != previous_serie:
+            index = 1  # Reset index when SERIE changes
+            previous_serie = current_serie
+        else:
+            index += 1
+
+        obj = {
+            'cnpj_prestador': row['CNPJ'],
+            'inscricao_municipal': row['INSCRICAO'],
+            'numero': str(index),
+            'serie': str(current_serie),
+            'tipo': '1'
+        }
+
+        resposta = consultar_nfse_por_rps(certificado, consulta=obj, ambiente='producao')
+        try:
+            root = ET.fromstring(resposta['received_xml'])
+            namespaces = {
+                'ns3': 'http://www.ginfes.com.br/servico_consultar_nfse_rps_resposta_v03.xsd',
+                'ns4': 'http://www.ginfes.com.br/tipos_v03.xsd'
+            }
+
+            codigo_verificacao = root.find('.//ns4:CodigoVerificacao', namespaces).text
+            numero = root.find('.//ns4:Numero', namespaces).text
+            data_emissao = root.find('.//ns4:DataEmissao', namespaces).text.split('T')[0]  # Extract date part
+
+            data_emissao_formatted = data_emissao.replace('-', '')
+            fantasia_sanitized = sanitize_filename(row['FANTASIA'])
+            file_name = f"{data_emissao_formatted}_{row['ID']}_{fantasia_sanitized}_nf.pdf"
+
+            # Call the download_note_pdf function
+            download_note_pdf(
+                verification_id=codigo_verificacao,
+                note_number=numero,
+                file_name=file_name
+            )
+
+            print('waiting 3 seconds...', fantasia_sanitized)
+            time.sleep(3)
+
+
+        except ET.ParseError as e:
+            print(f"XML Parse Error for row ID {row['ID']}: {e}")
+        except AttributeError as e:
+            print(f"Missing XML elements for row ID {row['ID']}: {e}")
+        except Exception as e:
+            print(f"An error occurred while processing row ID {row['ID']}: {e}")

gerar-notas.py

@@ -0,0 +1,82 @@
+import pandas as pd
+import json
+
+from datetime import datetime
+from pytrustnfe.certificado import Certificado
+from pytrustnfe.nfse.ginfes import recepcionar_lote_rps
+
+certificado = open("/Users/addo/jobs/addodelgrossi/go-nfs-e/resources/35143637000111.pfx", "rb").read()
+certificado = Certificado(certificado, b'k4ex2018')
+
+
+filename = '/Users/addo/Downloads/teste_emissao_nf.xlsx'
+df = pd.read_excel(filename, skiprows=1)
+
+lista_rps = []
+serie = '3'
+for index, row in df.iterrows():
+    item = {
+        'numero': str(index + 1),  # Número sequencial (começa em 1)
+        'serie': serie,
+        'tipo_rps': '1',
+        # 'data_emissao': '2025-01-02T14:25:00',
+        'data_emissao': datetime.now().strftime('%Y-%m-%dT%H:%M:%S'),
+        'natureza_operacao': '1',
+        'regime_tributacao': '6',
+        'optante_simples': '1',
+        'incentivador_cultural': '2',
+        'status': '1',
+        'prestador': {
+            'cnpj': str(row['CNPJ']),
+            'inscricao_municipal': str(row['INSCRICAO']),
+        },
+        'tomador': {
+            'cnpj_cpf': str(row['CLIENTE_CNPJ']),  # CNPJ
+            'razao_social': row['RAZAO SOCIAL'],  # Razão social
+            'logradouro': row['LOGRADOURO'],  # Logradouro
+            'numero': str(row['NUMERO']),  # Número
+            'bairro': row['BAIRRO'],  # Bairro
+            'cidade': row['CIDADE'],  # Cidade
+            'uf': row['UF'],  # UF
+            'cep': str(row['CEP']).replace('-', ''),  # CEP sem pontuação
+            'telefone': str(row['TELEFONE']),  # Telefone
+            'email': row['EMAIL'],  # E-mail
+        },
+        'valor_servico': f"{row['VALOR_SERVICO']:.2f}",  # Valor do serviço
+        'iss_retido': '2',
+        'base_calculo': f"{row['VALOR_SERVICO']:.2f}",  # Base de cálculo
+        'aliquota_issqn': f"{(row['ALIQUOTA_ISSQN']/100):.4f}",  # Alíquota
+        'valor_liquido_nfse': f"{row['VALOR_SERVICO']:.2f}",  # Valor líquido
+        'codigo_servico': str(row['CODIGO_SERVICO']),  # Código do serviço
+        'codigo_tributacao_municipio': str(row['CODIGO_TRIBUTACAO_MUNICIPIO']),  # Código de tributação
+        'descricao': row['DESCRICAO'],  # Descrição
+        'codigo_municipio': '3503208',  # Código do município fixo
+    }
+    lista_rps.append(item)
+
+# resultado = {'lista_rps': lista_rps}
+
+obj = {
+    'numero_lote': datetime.now().strftime('%Y%m%d%H%M%S'),
+    'cnpj_prestador': '35143637000111',
+    'inscricao_municipal': '1366453',
+    'lista_rps': lista_rps
+}
+
+print(json.dumps(obj, indent=4, ensure_ascii=False))
+
+resposta = recepcionar_lote_rps(certificado, nfse=obj, ambiente='producao')
+print(resposta)
+
+# create file from response
+with open("response.xml", "w") as f:
+    f.write(resposta['received_xml'])
+
+with open("lote.xml", "w") as f:
+    f.write(resposta['sent_xml'])
+
+
+
+# # Opcional: Salvar em um arquivo JSON
+# with open('resultado_rps.json', 'w', encoding='utf-8') as f:
+#     json.dump(resultado, f, indent=4, ensure_ascii=False)

gerar-pdf.py

@@ -0,0 +1,83 @@
+import requests
+import ssl
+import os
+
+from lxml import html
+from requests.adapters import HTTPAdapter
+from urllib3 import PoolManager
+import certifi
+
+
+# Configura a sessão
+session = requests.Session()
+session.verify = False  # Desativa a verificação SSL (usar apenas se necessário)
+
+# Adiciona headers para simular um navegador
+session.headers.update({
+    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36'
+})
+
+# Parâmetros fornecidos
+url_consulta = "http://visualizar.ginfes.com.br/report/consultarNota"
+params = {
+    "__report": "nfs_ver15",
+    "cdVerificacao": "SJWOPKQH7",
+    "numNota": "2405",
+    "cnpjPrestador": "null"
+}
+
+# URL base para envio do formulário
+url_formulario = "http://visualizar.ginfes.com.br/report/exportacao"
+
+# Diretório para salvar o PDF
+output_dir = "output"
+os.makedirs(output_dir, exist_ok=True)
+
+try:
+    # Realiza a consulta inicial
+    response = session.get(url_consulta, params=params, verify=certifi.where())
+    response.raise_for_status()  # Levanta exceção em caso de erro HTTP
+
+    # Analisa o HTML usando lxml
+    tree = html.fromstring(response.content)
+
+    # Encontra o formulário com o nome 'exportar'
+    form = tree.xpath("//form[@name='exportar']")
+    if not form:
+        print("Formulário 'exportar' não encontrado.")
+        exit()
+
+    # Coleta os campos do formulário
+    form_data = {}
+    inputs = form[0].xpath(".//input")
+    for input_tag in inputs:
+        name = input_tag.attrib.get('name')
+        value = input_tag.attrib.get('value', '')
+        form_data[name] = value
+
+    # Altera os campos especificados
+    form_data['imprime'] = '0'
+    form_data['tipo'] = 'pdf'
+
+    # Envia o POST com os dados alterados
+    response = session.post(url_formulario, data=form_data, stream=True)
+    response.raise_for_status()  # Levanta exceção em caso de erro HTTP
+
+    # Verifica se o conteúdo é um PDF
+    content_type = response.headers.get('Content-Type')
+    if 'application/pdf' not in content_type:
+        print("O retorno não é um arquivo PDF.")
+        exit()
+
+    # Salva o PDF no disco
+    output_path = os.path.join(output_dir, f"nota_{params['numNota']}.pdf")
+    with open(output_path, 'wb') as file:
+        for chunk in response.iter_content(chunk_size=1024):
+            file.write(chunk)
+
+    print(f"PDF salvo em: {output_path}")
+
+except requests.exceptions.SSLError as e:
+    print(f"Erro SSL: {e}")
+except requests.exceptions.RequestException as e:
+    print(f"Erro na requisição: {e}")

pytrustnfe/certificado.py

@@ -3,7 +3,9 @@
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
 
 import tempfile
-from OpenSSL import crypto
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.serialization import pkcs12
+from cryptography.hazmat.backends import default_backend
 
 
 class Certificado(object):
@@ -20,13 +22,19 @@ def save_pfx(self):
 
 
 def extract_cert_and_key_from_pfx(pfx, password):
-    pfx = crypto.load_pkcs12(pfx, password.encode())
-    # PEM formatted private key
-    key = crypto.dump_privatekey(crypto.FILETYPE_PEM, pfx.get_privatekey())
-    # PEM formatted certificate
-    cert = crypto.dump_certificate(crypto.FILETYPE_PEM, pfx.get_certificate())
-    return cert.decode(), key.decode()
+    private_key, certificate, additional_certificates = pkcs12.load_key_and_certificates(
+        pfx, password, backend=default_backend()
+    )
 
+    pem_key = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption()
+    )
+
+    pem_cert = certificate.public_bytes(serialization.Encoding.PEM)    
+
+    return pem_cert.decode(), pem_key.decode()        
 
 def save_cert_key(cert, key):
     cert_temp = tempfile.mkstemp()[1]

pytrustnfe/nfse/ginfes/templates/ConsultarNfseV3.xml

@@ -5,4 +5,12 @@
         <tipos:InscricaoMunicipal>{{ consulta.inscricao_municipal }}</tipos:InscricaoMunicipal>
     </Prestador>
     <NumeroNfse>{{ consulta.numero_nfse }}</NumeroNfse>
+
+    {% if consulta.data_inicial %}
+        <PeriodoEmissao>
+            <DataInicial>{{ consulta.data_inicial }}</DataInicial>
+            <DataFinal>{{ consulta.data_final }}</DataFinal>
+        </PeriodoEmissao>
+    {% endif %}
+
 </ConsultarNfseEnvio>