Merge branch 'rodrigoneal-correios'

brazilcep/client.py

@@ -11,9 +11,8 @@
 
 import enum
 import re
-import warnings
 
-from . import apicep, viacep
+from . import apicep, viacep, correios
 
 NUMBERS = re.compile(r"[^0-9]")
 DEFAULT_TIMEOUT = 5  # in seconds
@@ -32,7 +31,7 @@ class WebService(enum.Enum):
 
 
 services = {
-    WebService.CORREIOS: None,
+    WebService.CORREIOS: correios.fetch_address,
     WebService.VIACEP: viacep.fetch_address,
     WebService.APICEP: apicep.fetch_address,
 }
@@ -65,17 +64,6 @@ def get_address_from_cep(cep, webservice=WebService.APICEP, timeout=None, proxie
     """
         )
 
-    if webservice == WebService.CORREIOS:
-        warnings.warn(
-            "CORREIOS support has been deprecated, and we intend to remove it"
-            " in a future release of BrazilCEP. Please use the WebService.VIACEP, WebService.APICEP"
-            " instead, as described in the documentation.",
-            DeprecationWarning,
-        )
-
-        # override deprecated option
-        webservice = WebService.APICEP
-
     kwargs = {}
 
     if timeout and isinstance(timeout, int):

brazilcep/correios.py

@@ -0,0 +1,84 @@
+"""
+brazilcep.correios
+~~~~~~~~~~~~~~~~
+
+This module implements the BrazilCEP correios adapter.
+
+:copyright: (c) 2023 by Michell Stuttgart.
+:license: MIT, see LICENSE for more details.
+"""
+
+import requests
+from bs4 import BeautifulSoup
+
+from . import exceptions
+
+
+URL = "https://www2.correios.com.br/sistemas/buscacep/resultadoBuscaEndereco.cfm"
+
+# Lista de novas chaves
+address_keys = ["street", "district", "cep", "complement", "city", "uf"]
+
+
+def transform_address(address: dict) -> dict:
+    """Transforms an address dictionary to a new dictionary
+    with standardized keys."""
+
+    # Separar rua e complemento, se existir
+    street, *complement = address["logradouro"].split("-", maxsplit=1)
+    address["logradouro"] = street.strip()
+    address["complemento"] = complement[0].strip() if complement else ""
+
+    # Separar cidade e estado
+    city, state = address.pop("localidade").split("/")
+    address["municipio"] = city.strip()
+    address["estado"] = state.strip()
+
+    # Mapear valores para as novas chaves
+    return {k: v.strip() for k, v in zip(address_keys, address.values())}
+
+
+def extract_and_transform_cep(html: bytes) -> dict:
+    """Extracts address data from HTML and transforms it into a dictionary.
+    with standardized keys."""
+
+    # Analisar o HTML
+    soup = BeautifulSoup(html, "html.parser")
+    table = soup.find("table")
+    if not table:
+        return None
+
+    # Extrair dados da tabela
+    data = {}
+    for th_data, td_data in zip(table.find_all("th"), table.find_all("td")):
+        # Processar chaves e valores
+        key = th_data.get_text(strip=True).replace(":", "").split("/")[0].lower()
+        value = td_data.get_text(strip=True)
+        data[key] = value
+
+    # Transformar e retornar o endereço
+    return transform_address(data)
+
+
+def fetch_address(cep, **kwargs):
+    """Fetch VIACEP webservice for CEP address. VIACEP provide
+    a REST API to query CEP requests.
+
+    Args:
+        cep (str):CEP to be searched.
+        timeout (int): How many seconds to wait for the server to return data before giving up.
+        proxies (dict):  Dictionary mapping protocol to the URL of the proxy.
+
+    Returns:
+        address (dict): respective address data from CEP.
+    """
+    data = {
+        "CEP": cep,
+    }
+    response = requests.post(URL, data=data, **kwargs)  # pylint = missing-timeout
+
+    address_data = extract_and_transform_cep(response.content)
+
+    if address_data:
+        return address_data
+    raise exceptions.InvalidCEP()

noxfile.py

@@ -1,6 +1,6 @@
 import nox
 
-nox.options.sessions = [ "tests", "pylint"]
+nox.options.sessions = ["tests", "pylint"]
 
 
 @nox.session
@@ -20,15 +20,15 @@ def pylint(session: nox.Session) -> None:
     session.install("-e.[lint]")
     session.run("pylint", "brazilcep")
 
+
 @nox.session
 def coverage(session: nox.Session) -> None:
     """
     Run coverage.
     """
     session.install("-e.[coverage]")
-    session.run(
-        "pytest", "--cov", "brazilcep"
-    )
+    session.run("pytest", "--cov", "brazilcep")
+
 
 @nox.session
 def check_manifest(session: nox.Session) -> None:
@@ -39,11 +39,11 @@ def check_manifest(session: nox.Session) -> None:
     session.install("check-manifest")
     session.run("check-manifest", *session.posargs)
 
+
 @nox.session
 def docs(session: nox.Session) -> None:
     """
     Build the docs. Will serve unless --non-interactive
     """
     session.install("-e.[docs]")
     session.run("mkdocs", "serve" if session.interactive else "build")
-

pyproject.toml

@@ -30,7 +30,7 @@ classifiers = [
   "Natural Language :: Portuguese (Brazilian)",
 ]
 
-dependencies = ["zeep>=4.2.1", "requests>=2.28.2"]
+dependencies = ["zeep>=4.2.1", "requests>=2.28.2", "beautifulsoup4==4.12.3"]
 
 [project.optional-dependencies]
 dev = ["nox>=2022.11.21"]

tests/test_apicep.py

@@ -111,9 +111,7 @@ def test_fetch_address_blocked_by_flood(requests_mock):
 
 
 def test_fetch_address_404(requests_mock):
-    requests_mock.get(
-        "https://ws.apicep.com/cep/37503130.json", status_code=404
-    )  # noqa
+    requests_mock.get("https://ws.apicep.com/cep/37503130.json", status_code=404)  # noqa
 
     with pytest.raises(exceptions.BrazilCEPException):
         get_address_from_cep("37503-130", webservice=WebService.APICEP)

tests/test_client.py

@@ -14,25 +14,25 @@ def test_search_error():
         get_address_from_cep("37.503-130", webservice="VIACEP")
 
 
-def test_search_correios(requests_mock):
-    """Set mock get return"""
-    req_mock_text = """{
-        "status":200,
-        "ok":true,
-        "code":"37503-130",
-        "state":"MG",
-        "city":"Itajubá",
-        "district":"Santo Antônio",
-        "address":"Rua Geraldino Campista - até 214/215",
-        "statusText":"ok"
-    }"""
-
-    requests_mock.get("https://ws.apicep.com/cep/37503130.json", text=req_mock_text)
-
-    with catch_warnings(record=True) as warn:
-        get_address_from_cep("37503130", webservice=WebService.CORREIOS)
-        assert len(warn) == 1
-        assert issubclass(warn[0].category, DeprecationWarning)
+# def test_search_correios(requests_mock):
+#     """Set mock get return"""
+#     req_mock_text = """{
+#         "status":200,
+#         "ok":true,
+#         "code":"37503-130",
+#         "state":"MG",
+#         "city":"Itajubá",
+#         "district":"Santo Antônio",
+#         "address":"Rua Geraldino Campista - até 214/215",
+#         "statusText":"ok"
+#     }"""
+
+#     requests_mock.get("https://ws.apicep.com/cep/37503130.json", text=req_mock_text)
+
+#     with catch_warnings(record=True) as warn:
+#         get_address_from_cep("37503130", webservice=WebService.CORREIOS)
+#         assert len(warn) == 1
+#         assert issubclass(warn[0].category, DeprecationWarning)
 
 
 def test_format_cep_success():

tests/test_correios.py

@@ -0,0 +1,19 @@
+import pytest
+
+
+from brazilcep import WebService, exceptions, get_address_from_cep
+
+
+def test_fetch_address_success_real():
+    address = get_address_from_cep("37.503-130", webservice=WebService.CORREIOS)
+    assert address["district"] == "Santo Antônio"
+    assert address["cep"] == "37503-130"
+    assert address["city"] == "Itajubá"
+    assert address["complement"] == "até 214/215"
+    assert address["street"] == "Rua Geraldino Campista"
+    assert address["uf"] == "MG"
+
+
+def test_fetch_address_fail_real():
+    with pytest.raises(exceptions.InvalidCEP):
+        get_address_from_cep("99999-999", webservice=WebService.CORREIOS)

tests/test_viacep.py

@@ -78,9 +78,7 @@ def test_get_address_invalid_cep(requests_mock):
 
 
 def test_fetch_address_404(requests_mock):
-    requests_mock.get(
-        "http://www.viacep.com.br/ws/37503130/json", status_code=404
-    )  # noqa
+    requests_mock.get("http://www.viacep.com.br/ws/37503130/json", status_code=404)  # noqa
 
     with pytest.raises(exceptions.BrazilCEPException):
         get_address_from_cep("37503-130", webservice=WebService.VIACEP)