<commit_before># Copyright 2016 <NAME>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import re
from collections import namedtuple
from decimal import Decimal
from pathlib import Path
from typing import List, Optional, Sequence, Union

from requests.exceptions import Timeout
from zeep.exceptions import Fault

from .exceptions import (
    AuthenticationError,
    CanceledPostingCardError,
    ClientError,
    ClosePostingListError,
    ConnectTimeoutError,
    NonexistentPostingCardError,
    TrackingCodesLimitExceededError,
)
from .models.address import ZipAddress, ZipCode
from .models.builders import ModelBuilder
from .models.data import EXTRA_SERVICE_AR, EXTRA_SERVICE_MP
from .models.posting import FreightResponse, Package, PostingList, TrackingCode
from .models.user import ExtraService, PostingCard, Service, User
from .serializers import PostingListSerializer
from .soap import SoapClient
from .utils import get_resource_path

KG = 1000  # g
# environ servico url filename
CORREIOS_WEBSERVICES = {
    'sigep-production': (
        'https://apps.correios.com.br/SigepMasterJPA/AtendeClienteService/AtendeCliente?wsdl',
        'AtendeCliente-production.wsdl',
    ),
    'sigep-test': (
        'https://apphom.correios.com.br/SigepMasterJPA/AtendeClienteService/AtendeCliente?wsdl',
        'AtendeCliente-test.wsdl',
    ),
    'websro': (
        'https://webservice.correios.com.br/service/rastro/Rastro.wsdl',
        'Rastro.wsdl',
    ),
    'freight': (
        'http://ws.correios.com.br/calculador/CalcPrecoPrazo.asmx?WSDL',
        'CalcPrecoPrazo.asmx',
    ),
}

CorreiosAPIError = namedtuple('CorreiosAPIError', ['regex', 'exception'])

ERRORS = (
    CorreiosAPIError(
        regex=re.compile(r"autenticacao"),
        exception=AuthenticationError,
    ),
    CorreiosAPIError(
        regex=re.compile(r"^O Cartão de Postagem.*Cancelado.$"),
        exception=CanceledPostingCardError,
    ),
    CorreiosAPIError(
        regex=re.compile(r"^Cartao de Postagem inexistente"),
        exception=NonexistentPostingCardError,
    ),
)


class Correios:
    PRODUCTION = "production"
    TEST = "test"
    MAX_TRACKING_CODES_PER_REQUEST = 50

    def __init__(
        self,
        username: str,
        password: str,
        timeout: int = 8,
        environment: str = "production",
        local_wsdl_path: Optional[Path] = None
    ) -> None:

        if local_wsdl_path is None:
            local_wsdl_path = get_resource_path("wsdls")
        self.local_wsdl_path = local_wsdl_path

        ssl_check = environment == "production"
        sigep_key = "sigep-{}".format(environment)
        if not self.local_wsdl_path:
            sigep_url = (CORREIOS_WEBSERVICES[sigep_key][0], ssl_check)
            websro_url = CORREIOS_WEBSERVICES["websro"][0]
            freight_url = CORREIOS_WEBSERVICES["freight"][0]
        else:
            sigep_url = (str(self.local_wsdl_path / CORREIOS_WEBSERVICES[sigep_key][1]), ssl_check)
            websro_url = str(self.local_wsdl_path / CORREIOS_WEBSERVICES["websro"][1])
            freight_url = str(self.local_wsdl_path / CORREIOS_WEBSERVICES["freight"][1])

        self.username = username
        self.password = password
        self.timeout = timeout

        self.sigep_url = sigep_url[0]
        self.sigep_verify = sigep_url[1]

        self.sigep_client = SoapClient(self.sigep_url, verify=self.sigep_verify, timeout=self.timeout)
        self.sigep = self.sigep_client.service

        self.websro_client = SoapClient(websro_url, timeout=self.timeout)
        self.websro = self.websro_client.service

        self.freight_client = SoapClient(freight_url, timeout=self.timeout)
        self.freight = self.freight_client.service

        self.model_builder = ModelBuilder()

    def _parse_exception(self, exception):
        message = str(exception)

        for regex, exception in ERRORS:
            if regex.search(message):
                raise exception(message)

        raise ClientError(message)

    def _auth_call(self, method_name, *args, **kwargs):
        kwargs.update({
            "usuario": self.username,
            "senha": self.password,
        })
        return self._call(method_name, *args, **kwargs)

    def _call(self, method_name, *args, **kwargs):
        method = getattr(self.sigep, method_name)
        try:
            return method(*args, **kwargs)

        except Timeout:
            raise ConnectTimeoutError("Timeout connection error ({} seconds)".format(self.timeout))

        except Fault as exc:
            self._parse_exception(exc)

    def get_user(self, contract_number: Union[int, str], posting_card_number: Union[int, str]) -> User:
        contract_number = str(contract_number)<commit_msg>Rename _parse_exception method to _handle_exception<commit_after># Copyright 2016 <NAME>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import re
from collections import namedtuple
from decimal import Decimal
from pathlib import Path
from typing import List, Optional, Sequence, Union

from requests.exceptions import Timeout
from zeep.exceptions import Fault

from .exceptions import (
    AuthenticationError,
    CanceledPostingCardError,
    ClientError,
    ClosePostingListError,
    ConnectTimeoutError,
    NonexistentPostingCardError,
    TrackingCodesLimitExceededError,
)
from .models.address import ZipAddress, ZipCode
from .models.builders import ModelBuilder
from .models.data import EXTRA_SERVICE_AR, EXTRA_SERVICE_MP
from .models.posting import FreightResponse, Package, PostingList, TrackingCode
from .models.user import ExtraService, PostingCard, Service, User
from .serializers import PostingListSerializer
from .soap import SoapClient
from .utils import get_resource_path

KG = 1000  # g
# environ servico url filename
CORREIOS_WEBSERVICES = {
    'sigep-production': (
        'https://apps.correios.com.br/SigepMasterJPA/AtendeClienteService/AtendeCliente?wsdl',
        'AtendeCliente-production.wsdl',
    ),
    'sigep-test': (
        'https://apphom.correios.com.br/SigepMasterJPA/AtendeClienteService/AtendeCliente?wsdl',
        'AtendeCliente-test.wsdl',
    ),
    'websro': (
        'https://webservice.correios.com.br/service/rastro/Rastro.wsdl',
        'Rastro.wsdl',
    ),
    'freight': (
        'http://ws.correios.com.br/calculador/CalcPrecoPrazo.asmx?WSDL',
        'CalcPrecoPrazo.asmx',
    ),
}

CorreiosAPIError = namedtuple('CorreiosAPIError', ['regex', 'exception'])

ERRORS = (
    CorreiosAPIError(
        regex=re.compile(r"autenticacao"),
        exception=AuthenticationError,
    ),
    CorreiosAPIError(
        regex=re.compile(r"^O Cartão de Postagem.*Cancelado.$"),
        exception=CanceledPostingCardError,
    ),
    CorreiosAPIError(
        regex=re.compile(r"^Cartao de Postagem inexistente"),
        exception=NonexistentPostingCardError,
    ),
)


class Correios:
    PRODUCTION = "production"
    TEST = "test"
    MAX_TRACKING_CODES_PER_REQUEST = 50

    def __init__(
        self,
        username: str,
        password: str,
        timeout: int = 8,
        environment: str = "production",
        local_wsdl_path: Optional[Path] = None
    ) -> None:

        if local_wsdl_path is None:
            local_wsdl_path = get_resource_path("wsdls")
        self.local_wsdl_path = local_wsdl_path

        ssl_check = environment == "production"
        sigep_key = "sigep-{}".format(environment)
        if not self.local_wsdl_path:
            sigep_url = (CORREIOS_WEBSERVICES[sigep_key][0], ssl_check)
            websro_url = CORREIOS_WEBSERVICES["websro"][0]
            freight_url = CORREIOS_WEBSERVICES["freight"][0]
        else:
            sigep_url = (str(self.local_wsdl_path / CORREIOS_WEBSERVICES[sigep_key][1]), ssl_check)
            websro_url = str(self.local_wsdl_path / CORREIOS_WEBSERVICES["websro"][1])
            freight_url = str(self.local_wsdl_path / CORREIOS_WEBSERVICES["freight"][1])

        self.username = username
        self.password = password
        self.timeout = timeout

        self.sigep_url = sigep_url[0]
        self.sigep_verify = sigep_url[1]

        self.sigep_client = SoapClient(self.sigep_url, verify=self.sigep_verify, timeout=self.timeout)
        self.sigep = self.sigep_client.service

        self.websro_client = SoapClient(websro_url, timeout=self.timeout)
        self.websro = self.websro_client.service

        self.freight_client = SoapClient(freight_url, timeout=self.timeout)
        self.freight = self.freight_client.service

        self.model_builder = ModelBuilder()

    def _handle_exception(self, exception):
        message = str(exception)

        for regex, exception in ERRORS:
            if regex.search(message):
                raise exception(message)

        raise ClientError(message)

    def _auth_call(self, method_name, *args, **kwargs):
        kwargs.update({
            "usuario": self.username,
            "senha": <PASSWORD>,
        })
        return self._call(method_name, *args, **kwargs)

    def _call(self, method_name, *args, **kwargs):
        method = getattr(self.sigep, method_name)
        try:
            return method(*args, **kwargs)

        except Timeout:
            raise ConnectTimeoutError("Timeout connection error ({} seconds)".format(self.timeout))

        except Fault as exc:
            self._handle_exception(exc)

    def get_user(self, contract_number: Union[int, str], posting_card_number: Union[int, str]) -> User:
        contract_number = str(contract_number)
{'fraction_non_alphanumeric': 0.0969276134309902, 'fraction_numerical': 0.0043755350518405785, 'mean_word_length': 3.165610142630745, 'pattern_counts': {'":': 4, '<': 6, '<?xml version=': 0, '>': 10, 'https://': 6, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3991999', 'n_tokens_mistral': 3372, 'n_tokens_neox': 3228, 'n_words': 709}