diff --git a/.xsdata.xml b/.xsdata.xml index 657f8fff..f4de3d6d 100644 --- a/.xsdata.xml +++ b/.xsdata.xml @@ -10,9 +10,9 @@ false false - + @@ -21,6 +21,12 @@ + + @@ -31,6 +37,11 @@ + + + + + diff --git a/nfelib/cte/client/v4_0/__init__.py b/nfelib/cte/client/v4_0/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/nfelib/cte/client/v4_0/servers.py b/nfelib/cte/client/v4_0/servers.py new file mode 100644 index 00000000..e05c8e7e --- /dev/null +++ b/nfelib/cte/client/v4_0/servers.py @@ -0,0 +1,156 @@ +# Auto-generated file. Do not edit manually. +# ruff: noqa: D101 +# ruff: noqa: E501 + +from __future__ import annotations # Python 3.8 compat + +from enum import Enum +from typing import TypedDict + + +class Endpoint(Enum): + QRCODE = "qrcode" + CTESTATUSSERVICOV4 = "CteStatusServicoV4" + CTECONSULTAV4 = "CteConsultaV4" + CTERECEPCAOEVENTOV4 = "CteRecepcaoEventoV4" + CTERECEPCAOOSV4 = "CTeRecepcaoOSV4" + CTERECEPCAOSINCV4 = "CTeRecepcaoSincV4" + CTERECEPCAOGTVEV4 = "CTeRecepcaoGTVeV4" + CTERECEPCAOSIMPV4 = "CTeRecepcaoSimpV4" + + +class ServerConfig(TypedDict): + prod_server: str + dev_server: str + soap_version: str + endpoints: dict[Endpoint, str] + + +servers: dict[str, ServerConfig] = { + "MT": { + "prod_server": "cte.sefaz.mt.gov.br", + "dev_server": "homologacao.sefaz.mt.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.QRCODE: "/cte/qrcode", + Endpoint.CTESTATUSSERVICOV4: "/ctews2/services/CTeStatusServicoV4?wsdl", + Endpoint.CTECONSULTAV4: "/ctews2/services/CTeConsultaV4?wsdl", + Endpoint.CTERECEPCAOEVENTOV4: "/ctews2/services/CTeRecepcaoEventoV4?wsdl", + Endpoint.CTERECEPCAOOSV4: "/ctews/services/CTeRecepcaoOSV4?wsdl", + Endpoint.CTERECEPCAOSINCV4: "/ctews2/services/CTeRecepcaoSincV4?wsdl", + Endpoint.CTERECEPCAOGTVEV4: "/ctews2/services/CTeRecepcaoGTVeV4?wsdl", + Endpoint.CTERECEPCAOSIMPV4: "/cte-ws/services/CTeRecepcaoSimpV4", + }, + }, + "MS": { + "prod_server": "producao.cte.ms.gov.br", + "dev_server": "homologacao.cte.ms.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.QRCODE: "/cte/qrcode", + Endpoint.CTESTATUSSERVICOV4: "/ws/CTeStatusServicoV4", + Endpoint.CTECONSULTAV4: "/ws/CTeConsultaV4", + Endpoint.CTERECEPCAOEVENTOV4: "/ws/CTeRecepcaoEventoV4", + Endpoint.CTERECEPCAOOSV4: "/ws/CTeRecepcaoOSV4", + Endpoint.CTERECEPCAOSINCV4: "/ws/CTeRecepcaoSincV4", + Endpoint.CTERECEPCAOGTVEV4: "/ws/CTeRecepcaoGTVeV4", + Endpoint.CTERECEPCAOSIMPV4: "/ws/CTeRecepcaoSimpV4", + }, + }, + "MG": { + "prod_server": "cte.fazenda.mg.gov.br", + "dev_server": "hcte.fazenda.mg.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.QRCODE: "/portalcte/sistema/qrcode.xhtml", + Endpoint.CTESTATUSSERVICOV4: "/cte/services/CTeStatusServicoV4", + Endpoint.CTECONSULTAV4: "/cte/services/CTeConsultaV4", + Endpoint.CTERECEPCAOEVENTOV4: "/cte/services/CTeRecepcaoEventoV4", + Endpoint.CTERECEPCAOOSV4: "/cte/services/CTeRecepcaoOSV4", + Endpoint.CTERECEPCAOSINCV4: "/cte/services/CTeRecepcaoSincV4", + Endpoint.CTERECEPCAOGTVEV4: "/cte/services/CTeRecepcaoGTVeV4", + Endpoint.CTERECEPCAOSIMPV4: "/cte/services/CTeRecepcaoSimpV4", + }, + }, + "PR": { + "prod_server": "cte.fazenda.pr.gov.br", + "dev_server": "homologacao.cte.fazenda.pr.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.QRCODE: "/cte/qrcode", + Endpoint.CTESTATUSSERVICOV4: "/cte4/CTeStatusServicoV4?wsdl", + Endpoint.CTECONSULTAV4: "/cte4/CTeConsultaV4?wsdl", + Endpoint.CTERECEPCAOEVENTOV4: "/cte4/CTeRecepcaoEventoV4?wsdl", + Endpoint.CTERECEPCAOOSV4: "/cte4/CTeRecepcaoOSV4?wsdl", + Endpoint.CTERECEPCAOSINCV4: "/cte4/CTeRecepcaoSincV4?wsdl", + Endpoint.CTERECEPCAOGTVEV4: "/cte4/CTeRecepcaoGTVeV4?wsdl", + Endpoint.CTERECEPCAOSIMPV4: "/cte4/CTeRecepcaoSimpV4", + }, + }, + "RS": { + "prod_server": "cte.svrs.rs.gov.br", + "dev_server": "cte-homologacao.svrs.rs.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.QRCODE: "/cte/qrCode", + Endpoint.CTESTATUSSERVICOV4: "/ws/CTeStatusServicoV4/CTeStatusServicoV4.asmx", + Endpoint.CTECONSULTAV4: "/ws/CTeConsultaV4/CTeConsultaV4.asmx", + Endpoint.CTERECEPCAOEVENTOV4: "/ws/CTeRecepcaoEventoV4/CTeRecepcaoEventoV4.asmx", + Endpoint.CTERECEPCAOOSV4: "/ws/CTeRecepcaoOSV4/CTeRecepcaoOSV4.asmx", + Endpoint.CTERECEPCAOSINCV4: "/ws/CTeRecepcaoSincV4/CTeRecepcaoSincV4.asmx", + Endpoint.CTERECEPCAOGTVEV4: "/ws/CTeRecepcaoGTVeV4/CTeRecepcaoGTVeV4.asmx", + Endpoint.CTERECEPCAOSIMPV4: "/ws/CTeRecepcaoSimpV4/CTeRecepcaoSimpV4.asmx", + }, + }, + "SP": { + "prod_server": "nfe.fazenda.sp.gov.br", + "dev_server": "homologacao.nfe.fazenda.sp.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.QRCODE: "/CTeConsulta/qrCode", + Endpoint.CTESTATUSSERVICOV4: "/CTeWS/WS/CTeStatusServicoV4.asmx", + Endpoint.CTECONSULTAV4: "/CTeWS/WS/CTeConsultaV4.asmx", + Endpoint.CTERECEPCAOEVENTOV4: "/CTeWS/WS/CTeRecepcaoEventoV4.asmx", + Endpoint.CTERECEPCAOOSV4: "/CTeWS/WS/CTeRecepcaoOSV4.asmx", + Endpoint.CTERECEPCAOSINCV4: "/CTeWS/WS/CTeRecepcaoSincV4.asmx", + Endpoint.CTERECEPCAOGTVEV4: "/CTeWS/WS/CTeRecepcaoGTVeV4.asmx", + Endpoint.CTERECEPCAOSIMPV4: "/CTeWS/WS/CTeRecepcaoSimpV4.asmx", + }, + }, + "SVRS": { + "prod_server": "cte.svrs.rs.gov.br", + "dev_server": "cte-homologacao.svrs.rs.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.QRCODE: "/cte/qrCode", + Endpoint.CTESTATUSSERVICOV4: "/ws/CTeStatusServicoV4/CTeStatusServicoV4.asmx", + Endpoint.CTECONSULTAV4: "/ws/CTeConsultaV4/CTeConsultaV4.asmx", + Endpoint.CTERECEPCAOEVENTOV4: "/ws/CTeRecepcaoEventoV4/CTeRecepcaoEventoV4.asmx", + Endpoint.CTERECEPCAOOSV4: "/ws/CTeRecepcaoOSV4/CTeRecepcaoOSV4.asmx", + Endpoint.CTERECEPCAOSINCV4: "/ws/CTeRecepcaoSincV4/CTeRecepcaoSincV4.asmx", + Endpoint.CTERECEPCAOGTVEV4: "/ws/CTeRecepcaoGTVeV4/CTeRecepcaoGTVeV4.asmx", + Endpoint.CTERECEPCAOSIMPV4: "/ws/CTeRecepcaoSimpV4/CTeRecepcaoSimpV4.asmx", + }, + }, + "SVSP": { + "prod_server": "nfe.fazenda.sp.gov.br", + "dev_server": "homologacao.nfe.fazenda.sp.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.QRCODE: "/CTeConsulta/qrCode", + Endpoint.CTESTATUSSERVICOV4: "/CTeWS/WS/CTeStatusServicoV4.asmx", + Endpoint.CTECONSULTAV4: "/CTeWS/WS/CTeConsultaV4.asmx", + Endpoint.CTERECEPCAOEVENTOV4: "/CTeWS/WS/CTeRecepcaoEventoV4.asmx", + Endpoint.CTERECEPCAOOSV4: "/CTeWS/WS/CTeRecepcaoOSV4.asmx", + Endpoint.CTERECEPCAOSINCV4: "/CTeWS/WS/CTeRecepcaoSincV4.asmx", + Endpoint.CTERECEPCAOGTVEV4: "/CTeWS/WS/CTeRecepcaoGTVeV4.asmx", + Endpoint.CTERECEPCAOSIMPV4: "/CTeWS/WS/CTeRecepcaoSimpV4.asmx", + }, + }, + "AN": { + "prod_server": "www1.cte.fazenda.gov.br", + "dev_server": "hom1.cte.fazenda.gov.br", + "soap_version": "1.1", + "endpoints": {}, + }, +} diff --git a/nfelib/cte/client/v4_0/servers_scraper.py b/nfelib/cte/client/v4_0/servers_scraper.py new file mode 100644 index 00000000..98027a23 --- /dev/null +++ b/nfelib/cte/client/v4_0/servers_scraper.py @@ -0,0 +1,38 @@ +# Copyright (C) 2024 Raphaël Valyi - Akretion + +import sys +from pathlib import Path + +from nfelib.utils.servers_scraper import fetch_servers, save_servers +from nfelib.utils.soap_generator import generate_soap + +# Constants +PROD_URL = "https://www.cte.fazenda.gov.br/portal/webServices.aspx" +DEV_URL = "https://hom.cte.fazenda.gov.br/portal/webServices.aspx" +OUTPUT_FILE = Path("nfelib/cte/client/v4_0/servers.py") + + +def main(): + """Cli entry point.""" + download = False + if "--download" in sys.argv: + download = True + sys.argv.remove( + "--download" + ) # Remove the --download flag to avoid interfering with argparse + + generate = False + if "--generate" in sys.argv: + generate = True + sys.argv.remove( + "--generate" + ) # Remove the --generate flag to avoid interfering with argparse + + servers, endpoints = fetch_servers(PROD_URL, DEV_URL) + if servers: + save_servers(servers, endpoints, OUTPUT_FILE) + generate_soap(servers, endpoints, download, generate) + + +if __name__ == "__main__": + main() diff --git a/nfelib/mdfe/client/v3_0/__init__.py b/nfelib/mdfe/client/v3_0/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/nfelib/mdfe/client/v3_0/servers.py b/nfelib/mdfe/client/v3_0/servers.py new file mode 100644 index 00000000..6e5bb828 --- /dev/null +++ b/nfelib/mdfe/client/v3_0/servers.py @@ -0,0 +1,43 @@ +# Auto-generated file. Do not edit manually. +# ruff: noqa: D101 +# ruff: noqa: E501 + +from __future__ import annotations # Python 3.8 compat + +from enum import Enum +from typing import TypedDict + + +class Endpoint(Enum): + MDFERECEPCAOEVENTO = "MDFeRecepcaoEvento" + MDFECONSULTA = "MDFeConsulta" + MDFESTATUSSERVICO = "MDFeStatusServico" + MDFECONSNAOENC = "MDFeConsNaoEnc" + MDFEDISTRIBUICAODFE = "MDFeDistribuicaoDFe" + MDFERECEPCAOSINC = "MDFeRecepcaoSinc" + QRCODE = "QR Code" + + +class ServerConfig(TypedDict): + prod_server: str + dev_server: str + soap_version: str + endpoints: dict[Endpoint, str] + + +servers: dict[str, ServerConfig] = { + "SVRS": { + "prod_server": "mdfe.svrs.rs.gov.br", + "dev_server": "mdfe-homologacao.svrs.rs.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.MDFERECEPCAOEVENTO: "/ws/MDFeRecepcaoEvento/MDFeRecepcaoEvento.asmx", + Endpoint.MDFECONSULTA: "/ws/MDFeConsulta/MDFeConsulta.asmx", + Endpoint.MDFESTATUSSERVICO: "/ws/MDFeStatusServico/MDFeStatusServico.asmx", + Endpoint.MDFECONSNAOENC: "/ws/MDFeConsNaoEnc/MDFeConsNaoEnc.asmx", + Endpoint.MDFEDISTRIBUICAODFE: "/ws/MDFeDistribuicaoDFe/MDFeDistribuicaoDFe.asmx", + Endpoint.MDFERECEPCAOSINC: "/ws/MDFeRecepcaoSinc/MDFeRecepcaoSinc.asmx", + Endpoint.QRCODE: "/mdfe/qrCode", + }, + }, +} diff --git a/nfelib/mdfe/client/v3_0/servers_scraper.py b/nfelib/mdfe/client/v3_0/servers_scraper.py new file mode 100644 index 00000000..a385d7a9 --- /dev/null +++ b/nfelib/mdfe/client/v3_0/servers_scraper.py @@ -0,0 +1,109 @@ +# Copyright (C) 2024 Raphaël Valyi - Akretion + +from __future__ import annotations # Python 3.8 compat + +import logging +import sys +from pathlib import Path +from typing import Any + +import requests # type: ignore[import-untyped] +from bs4 import BeautifulSoup + +from nfelib.utils.servers_scraper import save_servers +from nfelib.utils.soap_generator import generate_soap + +# Constants +MDFE_SVRS_URL = "https://dfe-portal.svrs.rs.gov.br/Mdfe/Servicos" +OUTPUT_FILE = Path("nfelib/mdfe/client/v3_0/servers.py") + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +def fetch_mdfe_servers(url: str) -> dict[Any, Any]: + """Fetches the MDFe server actions for production and homologation.""" + # Fetch the page content + response = requests.get(url) + response.raise_for_status() # Raise an exception for HTTP errors + + # Parse the HTML content + soup = BeautifulSoup(response.content, "lxml") + + # Initialize dictionaries to store server actions + servers: dict[Any, Any] = { + "SVRS": { + "prod_server": "mdfe.svrs.rs.gov.br", + "dev_server": "mdfe-homologacao.svrs.rs.gov.br", + "soap_version": "1.1", + "endpoints": {}, + } + } + + # Find all tables with server information + tables = soup.find_all("table") + for table in tables: + # Determine if the table is for production or homologation + caption = table.find("caption") + if not caption: + continue + + # Extract server actions from the table + rows = table.find_all("tr")[1:] # Skip the header row + for row in rows: + cols = row.find_all("td") + if len(cols) < 4: # Ensure there are enough columns + continue + + service_name = cols[1].text.strip() + service_url = cols[3].text.strip() + + # Add the service to the servers dictionary + if service_name not in servers["SVRS"]["endpoints"]: + servers["SVRS"]["endpoints"][service_name] = {} + + servers["SVRS"]["endpoints"][service_name] = "/" + "/".join( + service_url.split("/")[3:] + ) + + logger.info("Successfully fetched MDFe servers.") + return servers + + +def main(): + """Cli entry point.""" + download = False + if "--download" in sys.argv: + download = True + sys.argv.remove( + "--download" + ) # Remove the --download flag to avoid interfering with argparse + + generate = False + if "--generate" in sys.argv: + generate = True + sys.argv.remove( + "--generate" + ) # Remove the --generate flag to avoid interfering with argparse + + # Fetch the MDFe server actions + servers = fetch_mdfe_servers(MDFE_SVRS_URL) + endpoints = { + "MDFERECEPCAOEVENTO": "MDFeRecepcaoEvento", + "MDFECONSULTA": "MDFeConsulta", + "MDFESTATUSSERVICO": "MDFeStatusServico", + "MDFECONSNAOENC": "MDFeConsNaoEnc", + "MDFEDISTRIBUICAODFE": "MDFeDistribuicaoDFe", + "MDFERECEPCAOSINC": "MDFeRecepcaoSinc", + "QRCODE": "QR Code", + } + + # Save the results to a file + if servers: + save_servers(servers, endpoints, OUTPUT_FILE) + generate_soap(servers, endpoints, download, generate) + + +if __name__ == "__main__": + main() diff --git a/nfelib/nfe/client/v4_0/__init__.py b/nfelib/nfe/client/v4_0/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/nfelib/nfe/client/v4_0/servers.py b/nfelib/nfe/client/v4_0/servers.py new file mode 100644 index 00000000..1fae4314 --- /dev/null +++ b/nfelib/nfe/client/v4_0/servers.py @@ -0,0 +1,231 @@ +# Auto-generated file. Do not edit manually. +# ruff: noqa: D101 +# ruff: noqa: E501 + +from __future__ import annotations # Python 3.8 compat + +from enum import Enum +from typing import TypedDict + + +class Endpoint(Enum): + NFEINUTILIZACAO = "NfeInutilizacao" + NFECONSULTAPROTOCOLO = "NfeConsultaProtocolo" + NFESTATUSSERVICO = "NfeStatusServico" + NFECONSULTACADASTRO = "NfeConsultaCadastro" + RECEPCAOEVENTO = "RecepcaoEvento" + NFEAUTORIZACAO = "NFeAutorizacao" + NFERETAUTORIZACAO = "NFeRetAutorizacao" + NFEDISTRIBUICAODFE = "NFeDistribuicaoDFe" + + +class ServerConfig(TypedDict): + prod_server: str + dev_server: str + soap_version: str + endpoints: dict[Endpoint, str] + + +servers: dict[str, ServerConfig] = { + "AM": { + "prod_server": "nfe.sefaz.am.gov.br", + "dev_server": "homnfe.sefaz.am.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/services2/services/NfeInutilizacao4", + Endpoint.NFECONSULTAPROTOCOLO: "/services2/services/NfeConsulta4", + Endpoint.NFESTATUSSERVICO: "/services2/services/NfeStatusServico4", + Endpoint.NFECONSULTACADASTRO: "/services2/services/CadConsultaCadastro4", + Endpoint.RECEPCAOEVENTO: "/services2/services/RecepcaoEvento4", + Endpoint.NFEAUTORIZACAO: "/services2/services/NfeAutorizacao4", + Endpoint.NFERETAUTORIZACAO: "/services2/services/NfeRetAutorizacao4", + }, + }, + "BA": { + "prod_server": "nfe.sefaz.ba.gov.br", + "dev_server": "hnfe.sefaz.ba.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/webservices/NFeInutilizacao4/NFeInutilizacao4.asmx", + Endpoint.NFECONSULTAPROTOCOLO: "/webservices/NFeConsultaProtocolo4/NFeConsultaProtocolo4.asmx", + Endpoint.NFESTATUSSERVICO: "/webservices/NFeStatusServico4/NFeStatusServico4.asmx", + Endpoint.NFECONSULTACADASTRO: "/webservices/CadConsultaCadastro4/CadConsultaCadastro4.asmx", + Endpoint.RECEPCAOEVENTO: "/webservices/NFeRecepcaoEvento4/NFeRecepcaoEvento4.asmx", + Endpoint.NFEAUTORIZACAO: "/webservices/NFeAutorizacao4/NFeAutorizacao4.asmx", + Endpoint.NFERETAUTORIZACAO: "/webservices/NFeRetAutorizacao4/NFeRetAutorizacao4.asmx", + }, + }, + "GO": { + "prod_server": "nfe.sefaz.go.gov.br", + "dev_server": "homolog.sefaz.go.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/nfe/services/NFeInutilizacao4?wsdl", + Endpoint.NFECONSULTAPROTOCOLO: "/nfe/services/NFeConsultaProtocolo4?wsdl", + Endpoint.NFESTATUSSERVICO: "/nfe/services/NFeStatusServico4?wsdl", + Endpoint.NFECONSULTACADASTRO: "/nfe/services/CadConsultaCadastro4?wsdl", + Endpoint.RECEPCAOEVENTO: "/nfe/services/NFeRecepcaoEvento4?wsdl", + Endpoint.NFEAUTORIZACAO: "/nfe/services/NFeAutorizacao4?wsdl", + Endpoint.NFERETAUTORIZACAO: "/nfe/services/NFeRetAutorizacao4?wsdl", + }, + }, + "MG": { + "prod_server": "nfe.fazenda.mg.gov.br", + "dev_server": "hnfe.fazenda.mg.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/nfe2/services/NFeInutilizacao4", + Endpoint.NFECONSULTAPROTOCOLO: "/nfe2/services/NFeConsultaProtocolo4", + Endpoint.NFESTATUSSERVICO: "/nfe2/services/NFeStatusServico4", + Endpoint.NFECONSULTACADASTRO: "/nfe2/services/CadConsultaCadastro4", + Endpoint.RECEPCAOEVENTO: "/nfe2/services/NFeRecepcaoEvento4", + Endpoint.NFEAUTORIZACAO: "/nfe2/services/NFeAutorizacao4", + Endpoint.NFERETAUTORIZACAO: "/nfe2/services/NFeRetAutorizacao4", + }, + }, + "MS": { + "prod_server": "nfe.sefaz.ms.gov.br", + "dev_server": "hom.nfe.sefaz.ms.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/ws/NFeInutilizacao4", + Endpoint.NFECONSULTAPROTOCOLO: "/ws/NFeConsultaProtocolo4", + Endpoint.NFESTATUSSERVICO: "/ws/NFeStatusServico4", + Endpoint.NFECONSULTACADASTRO: "/ws/CadConsultaCadastro4", + Endpoint.RECEPCAOEVENTO: "/ws/NFeRecepcaoEvento4", + Endpoint.NFEAUTORIZACAO: "/ws/NFeAutorizacao4", + Endpoint.NFERETAUTORIZACAO: "/ws/NFeRetAutorizacao4", + }, + }, + "MT": { + "prod_server": "nfe.sefaz.mt.gov.br", + "dev_server": "homologacao.sefaz.mt.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/nfews/v2/services/NfeInutilizacao4?wsdl", + Endpoint.NFECONSULTAPROTOCOLO: "/nfews/v2/services/NfeConsulta4?wsdl", + Endpoint.NFESTATUSSERVICO: "/nfews/v2/services/NfeStatusServico4?wsdl", + Endpoint.NFECONSULTACADASTRO: "/nfews/v2/services/CadConsultaCadastro4?wsdl", + Endpoint.RECEPCAOEVENTO: "/nfews/v2/services/RecepcaoEvento4?wsdl", + Endpoint.NFEAUTORIZACAO: "/nfews/v2/services/NfeAutorizacao4?wsdl", + Endpoint.NFERETAUTORIZACAO: "/nfews/v2/services/NfeRetAutorizacao4?wsdl", + }, + }, + "PE": { + "prod_server": "nfe.sefaz.pe.gov.br", + "dev_server": "nfehomolog.sefaz.pe.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/nfe-service/services/NFeInutilizacao4", + Endpoint.NFECONSULTAPROTOCOLO: "/nfe-service/services/NFeConsultaProtocolo4", + Endpoint.NFESTATUSSERVICO: "/nfe-service/services/NFeStatusServico4", + Endpoint.NFECONSULTACADASTRO: "/nfe-service/services/CadConsultaCadastro4?wsdl", + Endpoint.RECEPCAOEVENTO: "/nfe-service/services/NFeRecepcaoEvento4", + Endpoint.NFEAUTORIZACAO: "/nfe-service/services/NFeAutorizacao4", + Endpoint.NFERETAUTORIZACAO: "/nfe-service/services/NFeRetAutorizacao4", + }, + }, + "PR": { + "prod_server": "nfe.sefa.pr.gov.br", + "dev_server": "homologacao.nfe.sefa.pr.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/nfe/NFeInutilizacao4?wsdl", + Endpoint.NFECONSULTAPROTOCOLO: "/nfe/NFeConsultaProtocolo4?wsdl", + Endpoint.NFESTATUSSERVICO: "/nfe/NFeStatusServico4?wsdl", + Endpoint.NFECONSULTACADASTRO: "/nfe/CadConsultaCadastro4?wsdl", + Endpoint.RECEPCAOEVENTO: "/nfe/NFeRecepcaoEvento4?wsdl", + Endpoint.NFEAUTORIZACAO: "/nfe/NFeAutorizacao4?wsdl", + Endpoint.NFERETAUTORIZACAO: "/nfe/NFeRetAutorizacao4?wsdl", + }, + }, + "RS": { + "prod_server": "nfe.sefazrs.rs.gov.br", + "dev_server": "nfe-homologacao.sefazrs.rs.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/ws/nfeinutilizacao/nfeinutilizacao4.asmx", + Endpoint.NFECONSULTAPROTOCOLO: "/ws/NfeConsulta/NfeConsulta4.asmx", + Endpoint.NFESTATUSSERVICO: "/ws/NfeStatusServico/NfeStatusServico4.asmx", + Endpoint.NFECONSULTACADASTRO: "/ws/cadconsultacadastro/cadconsultacadastro4.asmx", + Endpoint.RECEPCAOEVENTO: "/ws/recepcaoevento/recepcaoevento4.asmx", + Endpoint.NFEAUTORIZACAO: "/ws/NfeAutorizacao/NFeAutorizacao4.asmx", + Endpoint.NFERETAUTORIZACAO: "/ws/NfeRetAutorizacao/NFeRetAutorizacao4.asmx", + }, + }, + "SP": { + "prod_server": "nfe.fazenda.sp.gov.br", + "dev_server": "homologacao.nfe.fazenda.sp.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/ws/nfeinutilizacao4.asmx", + Endpoint.NFECONSULTAPROTOCOLO: "/ws/nfeconsultaprotocolo4.asmx", + Endpoint.NFESTATUSSERVICO: "/ws/nfestatusservico4.asmx", + Endpoint.NFECONSULTACADASTRO: "/ws/cadconsultacadastro4.asmx", + Endpoint.RECEPCAOEVENTO: "/ws/nferecepcaoevento4.asmx", + Endpoint.NFEAUTORIZACAO: "/ws/nfeautorizacao4.asmx", + Endpoint.NFERETAUTORIZACAO: "/ws/nferetautorizacao4.asmx", + }, + }, + "SVAN": { + "prod_server": "www.sefazvirtual.fazenda.gov.br", + "dev_server": "hom.sefazvirtual.fazenda.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/NFeInutilizacao4/NFeInutilizacao4.asmx", + Endpoint.NFECONSULTAPROTOCOLO: "/NFeConsultaProtocolo4/NFeConsultaProtocolo4.asmx", + Endpoint.NFESTATUSSERVICO: "/NFeStatusServico4/NFeStatusServico4.asmx", + Endpoint.RECEPCAOEVENTO: "/NFeRecepcaoEvento4/NFeRecepcaoEvento4.asmx", + Endpoint.NFEAUTORIZACAO: "/NFeAutorizacao4/NFeAutorizacao4.asmx", + Endpoint.NFERETAUTORIZACAO: "/NFeRetAutorizacao4/NFeRetAutorizacao4.asmx", + }, + }, + "SVRS": { + "prod_server": "nfe.svrs.rs.gov.br", + "dev_server": "nfe-homologacao.svrs.rs.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/ws/nfeinutilizacao/nfeinutilizacao4.asmx", + Endpoint.NFECONSULTAPROTOCOLO: "/ws/NfeConsulta/NfeConsulta4.asmx", + Endpoint.NFESTATUSSERVICO: "/ws/NfeStatusServico/NfeStatusServico4.asmx", + Endpoint.NFECONSULTACADASTRO: "/ws/cadconsultacadastro/cadconsultacadastro4.asmx", + Endpoint.RECEPCAOEVENTO: "/ws/recepcaoevento/recepcaoevento4.asmx", + Endpoint.NFEAUTORIZACAO: "/ws/NfeAutorizacao/NFeAutorizacao4.asmx", + Endpoint.NFERETAUTORIZACAO: "/ws/NfeRetAutorizacao/NFeRetAutorizacao4.asmx", + }, + }, + "SVC-AN": { + "prod_server": "www.sefazvirtual.fazenda.gov.br", + "dev_server": "hom.sefazvirtual.fazenda.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.NFEINUTILIZACAO: "/NFeInutilizacao4/NFeInutilizacao4.asmx", + Endpoint.NFECONSULTAPROTOCOLO: "/NFeConsultaProtocolo4/NFeConsultaProtocolo4.asmx", + Endpoint.NFESTATUSSERVICO: "/NFeStatusServico4/NFeStatusServico4.asmx", + Endpoint.RECEPCAOEVENTO: "/NFeRecepcaoEvento4/NFeRecepcaoEvento4.asmx", + Endpoint.NFEAUTORIZACAO: "/NFeAutorizacao4/NFeAutorizacao4.asmx", + Endpoint.NFERETAUTORIZACAO: "/NFeRetAutorizacao4/NFeRetAutorizacao4.asmx", + }, + }, + "SVC-RS": { + "prod_server": "nfe.svrs.rs.gov.br", + "dev_server": "nfe-homologacao.svrs.rs.gov.br", + "soap_version": "1.2", + "endpoints": { + Endpoint.NFECONSULTAPROTOCOLO: "/ws/NfeConsulta/NfeConsulta4.asmx", + Endpoint.NFESTATUSSERVICO: "/ws/NfeStatusServico/NfeStatusServico4.asmx", + Endpoint.RECEPCAOEVENTO: "/ws/recepcaoevento/recepcaoevento4.asmx", + Endpoint.NFEAUTORIZACAO: "/ws/NfeAutorizacao/NFeAutorizacao4.asmx", + Endpoint.NFERETAUTORIZACAO: "/ws/NfeRetAutorizacao/NFeRetAutorizacao4.asmx", + }, + }, + "AN": { + "prod_server": "www.nfe.fazenda.gov.br", + "dev_server": "hom1.nfe.fazenda.gov.br", + "soap_version": "1.1", + "endpoints": { + Endpoint.RECEPCAOEVENTO: "/NFeRecepcaoEvento4/NFeRecepcaoEvento4.asmx", + Endpoint.NFEDISTRIBUICAODFE: "/NFeDistribuicaoDFe/NFeDistribuicaoDFe.asmx", + }, + }, +} diff --git a/nfelib/nfe/client/v4_0/servers_scraper.py b/nfelib/nfe/client/v4_0/servers_scraper.py new file mode 100644 index 00000000..2cb53800 --- /dev/null +++ b/nfelib/nfe/client/v4_0/servers_scraper.py @@ -0,0 +1,38 @@ +# Copyright (C) 2024 Raphaël Valyi - Akretion + +import sys +from pathlib import Path + +from nfelib.utils.servers_scraper import fetch_servers, save_servers +from nfelib.utils.soap_generator import generate_soap + +# Constants +PROD_URL = "https://www.nfe.fazenda.gov.br/portal/webServices.aspx" +DEV_URL = "https://hom.nfe.fazenda.gov.br/portal/webServices.aspx" +OUTPUT_FILE = Path("nfelib/nfe/client/v4_0/servers.py") + + +def main(): + """Cli entry point.""" + download = False + if "--download" in sys.argv: + download = True + sys.argv.remove( + "--download" + ) # Remove the --download flag to avoid interfering with argparse + + generate = False + if "--generate" in sys.argv: + generate = True + sys.argv.remove( + "--generate" + ) # Remove the --generate flag to avoid interfering with argparse + + servers, endpoints = fetch_servers(PROD_URL, DEV_URL) + if servers: + save_servers(servers, endpoints, OUTPUT_FILE) + generate_soap(servers, endpoints, download, generate) + + +if __name__ == "__main__": + main() diff --git a/nfelib/utils/__init__.py b/nfelib/utils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/nfelib/utils/servers_scraper.py b/nfelib/utils/servers_scraper.py new file mode 100644 index 00000000..55fa1f8d --- /dev/null +++ b/nfelib/utils/servers_scraper.py @@ -0,0 +1,263 @@ +# Copyright (C) 2024 Raphaël Valyi - Akretion + +from __future__ import annotations # Python 3.8 compat + +import logging +from io import StringIO +from os import environ +from pathlib import Path +from typing import Any + +import pandas as pd +import requests # type: ignore[import-untyped] +from brazil_fiscal_client.fiscal_client import FiscalClient, Tamb, TcodUfIbge +from bs4 import BeautifulSoup +from xsdata.formats.dataclass.serializers import PycodeSerializer + +from nfelib.cte.bindings.v4_0.cons_stat_serv_tipos_basico_v4_00 import TconsStatServ +from nfelib.cte.soap.v4_0.ctestatusservicov4 import ( + CteStatusServicoV4Soap12CteStatusServicoCt, +) +from nfelib.nfe.bindings.v4_0.cons_stat_serv_v4_00 import ConsStatServ +from nfelib.nfe.bindings.v4_0.leiaute_cons_stat_serv_v4_00 import ( + TconsStatServXServ, +) +from nfelib.nfe.soap.v4_0.nfestatusservico4 import ( + NfeStatusServico4SoapNfeStatusServicoNf, +) + +SERVICE_COLUMN = "Serviço" +URL_COLUMN = "URL" + +# here we manually maintain a list of servers requiring SOAP 1.2 headers +# when we detect new ones with SOAP 1.2 detection feature in the end +# of the file. Thus we don't always need to export a proper CERT_FILE env var +# to scrap the server URLs. +FORCE_SOAP_12_NFE = {"SVC-RS", "SVC-AN", "SP", "PR"} +FORCE_SOAP_12_CTE = {"SVSP", "RS", "SVRS", "MG", "SP", "PR"} + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +def fetch_servers(prod_url: str, dev_url: str) -> tuple[dict[str, Any], dict[str, Any]]: + """Fetches the NFe server list from the webpage using pandas and BeautifulSoup.""" + servers = {} + constants = {} # To store dynamically generated constants + force_soap_12 = FORCE_SOAP_12_NFE if "nfe" in prod_url else FORCE_SOAP_12_CTE + status_key = "NfeStatusServico" if "nfe" in prod_url else "CteStatusServicoV4" + + # Fetch production servers + prod_response = requests.get(prod_url) + prod_response.raise_for_status() + soup = BeautifulSoup(prod_response.content, "lxml") + captions = soup.find_all("caption") + + servers_list = [ + str(caption).split("(")[1].split(")")[0] + for caption in captions + if "(" in str(caption) + ] + + # Fetch development servers + dev_response = requests.get(dev_url) + dev_response.raise_for_status() + dev_html = dev_response.content.decode(dev_response.apparent_encoding) + dev_tables = pd.read_html(StringIO(dev_html)) # Wrap HTML in StringIO + + dev_servers = { + servers_list[index]: urls[-1].split("/")[2] + for index, table in enumerate(dev_tables) + if SERVICE_COLUMN in table.columns + for urls in [list(table.to_dict()[URL_COLUMN].values())] + } + + # Fetch production server details and generate constants + prod_html = prod_response.content.decode(prod_response.apparent_encoding) + prod_tables = pd.read_html(StringIO(prod_html)) # Wrap HTML in StringIO + + for index, table in enumerate(prod_tables): + if SERVICE_COLUMN not in table.columns or URL_COLUMN not in table.columns: + logger.warning(f"Skipping table {index}: missing required columns.") + continue + + actions = list(dict(table.to_dict())[SERVICE_COLUMN].values()) + urls = list(dict(table.to_dict())[URL_COLUMN].values()) + + # Generate constants from the first table + if index == 0: + for action in actions: + constant_name = action.upper().replace(" ", "_") + constants[constant_name] = action + + paths = ["/" + "/".join(url.split("/")[3:]) for url in urls] + prod_server = urls[-1].split("/")[2] + + server = servers_list[index] + action_dict = {} + + # Use dynamically generated constants as keys + for action, path in zip(actions, paths): + # if QRCODE in action.lower(): + # continue + constant_name = action.upper().replace(" ", "_") + if constant_name in constants: + action_dict[constants[constant_name]] = path + + # Handle special case for Ambiente Nacional (AN) + if server == "AN" and "NFeDistribuicaoDFe" in actions: + constant_name = "NFEDISTRIBUICAODFE" + if constant_name not in constants: + constants[constant_name] = "NFeDistribuicaoDFe" + action_dict[constants[constant_name]] = paths[ + actions.index("NFeDistribuicaoDFe") + ] + + servers[server] = { + "prod_server": prod_server, + "dev_server": dev_servers[server], + "soap_version": "1.2" if server in force_soap_12 else "1.1", + "endpoints": action_dict, + } + + logger.info("Successfully fetched servers.") + + if environ.get("CERT_FILE"): + logger.info("\nNow, let's test if some servers require SOAP 1.2 headers...") + soap_12_servers = force_soap_12 + + cert_path = Path(environ["CERT_FILE"]) + if not cert_path.is_file(): + raise FileNotFoundError(f"Certificate file not found: {cert_path}") + with open(cert_path, "rb") as pkcs12_file: + cert_data = pkcs12_file.read() + + for server, server_config in servers.items(): + if hasattr(TcodUfIbge, server): + uf = getattr(TcodUfIbge, server) + else: + uf = { + "SVRS": TcodUfIbge.SC, + "AN": TcodUfIbge.PR, + "SVAN": TcodUfIbge.MA, + "SVC-AN": TcodUfIbge.RJ, + "SVC-RS": TcodUfIbge.MA, + "SVSP": TcodUfIbge.SP, + }[server] + + if not server_config["endpoints"].get(status_key): + continue + + logger.info(f"\n\nTesting SOAP version for {server} - {uf} {uf.value}") + + client = FiscalClient( + uf=uf.value, + ambiente="2", + versao="4.00", + pkcs12_data=cert_data, + pkcs12_password=environ.get("CERT_PASSWORD"), + soap12_envelope=server_config["soap_version"] == "1.2", + ) + + try: + if "nfe" in prod_url: + client.send( + NfeStatusServico4SoapNfeStatusServicoNf, + f"https://{server_config['dev_server']}{server_config['endpoints'][status_key]}", + { + "Body": { + "nfeDadosMsg": { + "content": [ + ConsStatServ( + tpAmb=Tamb.DEV, + cUF=uf.value, + xServ=TconsStatServXServ.STATUS, + versao="4.00", + ) + ] + } + }, + }, + raise_on_soap_mismatch=True, + ) + + elif "cte" in prod_url: + client.send( + CteStatusServicoV4Soap12CteStatusServicoCt, + f"https://{server_config['dev_server']}{server_config['endpoints'][status_key]}", + { + "Body": { + "cteDadosMsg": { + "content": [ + TconsStatServ( + tpAmb=Tamb.DEV, + cUF=uf.value, + versao="4.00", + ) + ] + } + }, + }, + raise_on_soap_mismatch=True, + ) + except Exception as e: + if "Envelope" in str(e): + logger.error( + f"\nAn unexpected SOAP VERSION MISMATCH error occurred: {e}" + ) + server_config["soap_version"] = "1.2" + soap_12_servers.add(server) + else: + logger.error(f"\nAn unexpected error occurred: {e}") + + if environ.get("CERT_FILE"): + logger.info(f"sniffed 1.2 headers for {soap_12_servers}") + return servers, constants + + +def save_servers( + servers: dict[str, Any], endpoints: dict[str, str], output_file: Path +) -> None: + """Saves the extracted server data and constants as a Python file.""" + # Generate the constants section + actions = "\n".join([f' {key} = "{value}"' for key, value in endpoints.items()]) + + # Use PycodeSerializer to format the servers dictionary + serializer = PycodeSerializer() + formatted_servers = "\n".join( + serializer.render(servers, var_name="servers").replace("'", '"').split("\n")[2:] + ) + for key, value in endpoints.items(): + formatted_servers = formatted_servers.replace(f'"{value}"', f"Endpoint.{key}") + formatted_servers = formatted_servers.replace( + "servers =", "servers: dict[str, ServerConfig] =" + ) + + # Write the formatted output to the file + content = f"""# Auto-generated file. Do not edit manually. +# ruff: noqa: D101 +# ruff: noqa: E501 + +from __future__ import annotations # Python 3.8 compat + +from enum import Enum +from typing import TypedDict + + +class Endpoint(Enum): +{actions} + + +class ServerConfig(TypedDict): + prod_server: str + dev_server: str + soap_version: str + endpoints: dict[Endpoint, str] + + +{formatted_servers}""" + + output_file.parent.mkdir(parents=True, exist_ok=True) + output_file.write_text(content, encoding="utf-8") + logger.info(f"Servers and constants saved to {output_file}") diff --git a/nfelib/utils/soap_generator.py b/nfelib/utils/soap_generator.py new file mode 100644 index 00000000..525ecb60 --- /dev/null +++ b/nfelib/utils/soap_generator.py @@ -0,0 +1,147 @@ +# Copyright (C) 2024 Raphaël Valyi - Akretion + +from __future__ import annotations # Python 3.8 compat + +import logging +import os +import subprocess +from os import environ +from typing import Any + +import urllib3 +from requests import Session # type: ignore[import-untyped] +from requests_pkcs12 import Pkcs12Adapter # type: ignore[import-untyped] + +# Configure logging +_logger = logging.getLogger(__name__) +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +# Constants +SERVER = "SVRS" +WSDL_DIRS = { + "nfe": ("nfelib/nfe/wsdl/v4_0", "nfe"), + "mdfe": ("nfelib/mdfe/wsdl/v3_0", "mdfe"), + "cte": ("nfelib/cte/wsdl/v4_0", "cte"), + "bpe": ("nfelib/bpe/wsdl/v1_0", "bpe"), +} + + +def generate_soap( + servers: dict[str, Any], + endpoints: dict[str, str], + download: bool = False, + generate: bool = False, +) -> None: + """Download WSDL files for NF-e, CT-e, MDF-e, and BP-e.""" + # Access the certificate and password from environment variables + server = servers[SERVER]["prod_server"] + wsdl_urls = [ + f"https://{server}{servers[SERVER]['endpoints'].get(value, ' SKIP ' + key)}" + for key, value in endpoints.items() + ] + + # TODO make extra wsdl urls a param + wsdl_urls.append( + "https://www1.nfe.fazenda.gov.br/NFeDistribuicaoDFe/NFeDistribuicaoDFe.asmx" + ) + + doc_type = None + wsdl_dir = "" + for url in wsdl_urls: + if "SKIP" in url and "NFeDistribuicaoDFe" not in url: + _logger.error( + f"Skipping WSDL download for {url} (not found on server {server})" + ) + continue + try: + # Determine the server and mount the PKCS12 adapter + # url = f"https://{server}{path}" + + # Ensure the URL ends with ?wsdl + if not url.endswith("?wsdl"): + url += "?wsdl" + + # Determine the output file path + filename = ( + url.split("/")[-1] + .replace("?wsdl", "") + .replace(".asmx", ".wsdl") + .lower() + ) + + if doc_type is None: + # Identify the document type (NF-e, CT-e, MDF-e, BP-e) + for key, (_, _prefix) in WSDL_DIRS.items(): + if key in url.lower(): + doc_type = key + break + if not doc_type: + raise ValueError(f"Cannot determine document type for URL: {url}") + + if doc_type == "nfe": + if "cadconsultacadastro4" in url: + "the server is different in this case" + url = "https://cad.svrs.rs.gov.br/ws/cadconsultacadastro/cadconsultacadastro4.asmx" + elif "NFeDistribuicaoDFe" in url: # only for NFe + continue + + # Create the output directory if it doesn't exist + wsdl_dir, _ = WSDL_DIRS[doc_type] + os.makedirs(wsdl_dir, exist_ok=True) + + # Write the WSDL content to the file + wsdl_file = os.path.join(wsdl_dir, filename) + + if download: + cert_file = environ.get("CERT_FILE") + cert_password = environ.get("CERT_PASSWORD") + if not cert_file or not cert_password: + raise ValueError( + "Certificate file or password not provided " + "in environment variables." + ) + + session = Session() + session.verify = False # Disable SSL verification (use with caution) + session.mount( + url, + Pkcs12Adapter( + pkcs12_filename=cert_file, + pkcs12_password=cert_password, + ), + ) + # Fetch the WSDL content + response = session.get(url) + response.raise_for_status() + _logger.info(f"Writing to {wsdl_file}") + with open(wsdl_file, "w") as file: + file.write(response.text) + + if generate: + soap_dir = wsdl_dir.replace("wsdl", "soap").replace("/", ".") + command = [ + "xsdata", + "generate", + "--package", + soap_dir, + "--include-header", + wsdl_file, + ] + logging.info(" ".join(command)) + try: + subprocess.run(command, check=True) + logging.info("Successfully generated SOAP bindings for {wsdl_file}") + except subprocess.CalledProcessError as e: + logging.error( + f"Failed to generate SOAP bindings for {wsdl_file}: {e}" + ) + + except Exception as e: + _logger.error(f"Failed to download or save WSDL from {url}: {e}") + continue + + if generate: + # reset the __init__.py file to avoid arbitrary imports depending on gen order + init_file = os.path.join(wsdl_dir.replace("wsdl", "soap"), "__init__.py") + with open(init_file, "w") as file: + file.write("") diff --git a/pyproject.toml b/pyproject.toml index f0ff0658..3efa7ebd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,6 +51,11 @@ sign = [ pdf = [ "brazilfiscalreport", ] +soap = [ + "xsdata[soap]", + "erpbrasil.assinatura", + "brazil-fiscal-client", +] test = [ "pre-commit", "pytest", @@ -58,9 +63,14 @@ test = [ "pytest-cov", "xmldiff", "requests", + "requests_pkcs12", + "decorator", "beautifulsoup4", "erpbrasil.assinatura", "brazilfiscalreport", + "beautifulsoup4", + "pandas", + "brazil-fiscal-client", ] [tool.setuptools] diff --git a/tests/cte/test_servers.py b/tests/cte/test_servers.py new file mode 100644 index 00000000..c87d36ef --- /dev/null +++ b/tests/cte/test_servers.py @@ -0,0 +1,18 @@ +from pathlib import Path + +from nfelib.cte.client.v4_0.servers_scraper import main + +OUTPUT_FILE = Path("nfelib/cte/client/v4_0/servers.py") + + +def read_current_servers(): + return OUTPUT_FILE.read_text(encoding="utf-8") if OUTPUT_FILE.exists() else "" + + +def test_scraper(): + old_content = read_current_servers() + main() + new_content = read_current_servers() + assert new_content == old_content, ( + "Server list has changed. Review and commit the new file." + ) diff --git a/tests/mdfe/__init__.py b/tests/mdfe/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/mdfe/test_servers.py b/tests/mdfe/test_servers.py new file mode 100644 index 00000000..a0a1f4da --- /dev/null +++ b/tests/mdfe/test_servers.py @@ -0,0 +1,18 @@ +from pathlib import Path + +from nfelib.mdfe.client.v3_0.servers_scraper import main + +OUTPUT_FILE = Path("nfelib/mdfe/client/v3_0/servers.py") + + +def read_current_servers(): + return OUTPUT_FILE.read_text(encoding="utf-8") if OUTPUT_FILE.exists() else "" + + +def test_scraper(): + old_content = read_current_servers() + main() + new_content = read_current_servers() + assert new_content == old_content, ( + "Server list has changed. Review and commit the new file." + ) diff --git a/tests/nfe/test_servers.py b/tests/nfe/test_servers.py new file mode 100644 index 00000000..ec6c0d2e --- /dev/null +++ b/tests/nfe/test_servers.py @@ -0,0 +1,18 @@ +from pathlib import Path + +from nfelib.nfe.client.v4_0.servers_scraper import main + +OUTPUT_FILE = Path("nfelib/nfe/client/v4_0/servers.py") + + +def read_current_servers(): + return OUTPUT_FILE.read_text(encoding="utf-8") if OUTPUT_FILE.exists() else "" + + +def test_scraper(): + old_content = read_current_servers() + main() + new_content = read_current_servers() + assert new_content == old_content, ( + "Server list has changed. Review and commit the new file." + ) diff --git a/tests/test_fingerprint.py b/tests/test_fingerprint.py deleted file mode 100755 index 0f3222dd..00000000 --- a/tests/test_fingerprint.py +++ /dev/null @@ -1,75 +0,0 @@ -import hashlib -import json -import logging -from os import environ -from pathlib import Path -from unittest import TestCase - -import requests -from bs4 import BeautifulSoup - -_logger = logging.getLogger(__name__) - -HEADERS = { - "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.97 Safari/537.36" -} - -PAGES = { - "nfe": ( - "https://www.nfe.fazenda.gov.br/portal/listaConteudo.aspx?tipoConteudo=BMPFMBoln3w=", - "div", - "id", - "conteudoDinamico", - ), - # "nfe_pynfe_webservices": ("https://raw.githubusercontent.com/TadaSoftware/PyNFe/main/pynfe/utils/webservices.py",), - # "nfe_pynfe_comunicacao": ("https://raw.githubusercontent.com/TadaSoftware/PyNFe/main/pynfe/processamento/comunicacao.py"), - "cte": ( - "https://www.cte.fazenda.gov.br/portal/listaConteudo.aspx?tipoConteudo=0xlG1bdBass=", - "div", - "id", - "conteudoDinamico", - ), - "nfse": ( - "https://www.gov.br/nfse/pt-br/biblioteca/documentacao-tecnica", - "div", - "id", - "content-core", - ), - # TODO MDF-e content seems loaded with XHR - # "mdfe": ("https://portal.fazenda.sp.gov.br/servicos/mdfe/Paginas/Downloads.aspx", "div", "class", "content"), -} - - -class FingerPrintTests(TestCase): - def test_fingerprint(self) -> None: - if environ.get("SKIP_FINGERPRINT"): - _logger.info("Skipping fingerprint test") - return True - fingerprint = {} - for code, scrap_params in PAGES.items(): - url = scrap_params[0] - md5 = "ELEMENT NOT FOUND" - _logger.info(f"Fetching {url} ...") - if len(scrap_params) > 1: - page = requests.get(url, headers=HEADERS) - soup = BeautifulSoup(page.text, "html.parser") - if scrap_params[2] == "id" and soup.find( - scrap_params[1], {"id": scrap_params[3]} - ): - fragment = soup.find( - scrap_params[1], {"id": scrap_params[3]} - ).text.encode("utf-8") - md5 = hashlib.md5(fragment).hexdigest() - else: - fragment = requests.get( - url, headers=HEADERS - ).content # .decode('utf-8') - md5 = hashlib.md5(fragment).hexdigest() - fingerprint[code] = (url, md5) - - _logger.info(fingerprint) - json_string = json.dumps(fingerprint, indent=4) - target = Path("tests/fingerprints.json").read_text() - with open("tests/fingerprints.json", "w") as outfile: - outfile.write(json_string) - self.assertEqual(target.strip(), json_string.strip())