Merge pull request #71 from olist/remote-freight

Remote freight calculation using Correios API

correios/client.py

@@ -15,16 +15,21 @@
 
 import os
 from datetime import datetime
-from typing import Union, Sequence, List, Dict
+from decimal import Decimal
+from typing import Union, Sequence, List, Dict, Optional
 
 from correios import xml_utils, DATADIR
 from correios.exceptions import PostingListSerializerError, TrackingCodesLimitExceededError
+from correios.models.data import EXTRA_SERVICE_MP, EXTRA_SERVICE_AR
+from correios.utils import to_decimal, to_integer
 from .models.address import ZipAddress, ZipCode
 from .models.posting import (NotFoundTrackingEvent, TrackingCode, PostingList, ShippingLabel,
-                             TrackingEvent, EventStatus)
-from .models.user import User, FederalTaxNumber, StateTaxNumber, Contract, PostingCard, Service
+                             TrackingEvent, EventStatus, Package, Freight, FreightError)
+from .models.user import User, FederalTaxNumber, StateTaxNumber, Contract, PostingCard, Service, ExtraService
 from .soap import SoapClient
 
+KG = 1000  # g
+
 
 class ModelBuilder:
     def build_service(self, service_data):
@@ -149,6 +154,38 @@ def load_tracking_events(self, tracking_codes: Dict[str, TrackingCode], response
 
         return result
 
+    def build_freights_list(self, response):
+        result = []
+        for service_data in response.Servicos.cServico:
+            service = Service.get(service_data.Codigo)
+            error_code = to_integer(service_data.Erro)
+            if error_code:
+                freight = FreightError(
+                    service=service,
+                    error_code=error_code,
+                    error_message=service_data.MsgErro,
+                )
+            else:
+                delivery_time = int(service_data.PrazoEntrega)
+                value = to_decimal(service_data.ValorSemAdicionais)
+                declared_value = to_decimal(service_data.ValorValorDeclarado)
+                ar_value = to_decimal(service_data.ValorAvisoRecebimento)
+                mp_value = to_decimal(service_data.ValorMaoPropria)
+                saturday = service_data.EntregaSabado or ""
+                home = service_data.EntregaDomiciliar or ""
+                freight = Freight(
+                    service=service,
+                    delivery_time=delivery_time,
+                    value=value,
+                    declared_value=declared_value,
+                    ar_value=ar_value,
+                    mp_value=mp_value,
+                    saturday=saturday.upper() == "S",
+                    home=home.upper() == "S",
+                )
+            result.append(freight)
+        return result
+
 
 class PostingListSerializer:
     def _get_posting_list_element(self, posting_list):
@@ -278,6 +315,7 @@ class Correios:
         'test': ("https://apphom.correios.com.br/SigepMasterJPA/AtendeClienteService/AtendeCliente?wsdl", False),
     }
     websro_url = "https://webservice.correios.com.br/service/rastro/Rastro.wsdl"
+    freight_url = "http://ws.correios.com.br/calculador/CalcPrecoPrazo.asmx?WSDL"
 
     def __init__(self, username, password, timeout=8, environment="production"):
         self.username = username
@@ -294,6 +332,9 @@ def __init__(self, username, password, timeout=8, environment="production"):
         self.websro_client = SoapClient(self.websro_url, timeout=self.timeout)
         self.websro = self.websro_client.service
 
+        self.freight_client = SoapClient(self.freight_url, timeout=self.timeout)
+        self.freight = self.freight_client.service
+
         self.model_builder = ModelBuilder()
 
     def _auth_call(self, method_name, *args, **kwargs):
@@ -380,3 +421,39 @@ def get_tracking_code_events(self, tracking_list):
         response = self.websro.buscaEventosLista(self.username, self.password, "L", "T", "101",
                                                  tuple(tracking_codes.keys()))
         return self.model_builder.load_tracking_events(tracking_codes, response)
+
+    def calculate_freights(self,
+                           posting_card: PostingCard,
+                           services: List[Union[Service, int]],
+                           from_zip: Union[ZipCode, int, str], to_zip: Union[ZipCode, int, str],
+                           package: Package,
+                           value: Union[Decimal, float] = 0.00,
+                           extra_services: Optional[Sequence[Union[ExtraService, int]]] = None):
+
+        administrative_code = posting_card.administrative_code
+        services = [Service.get(s) for s in services]
+        from_zip = ZipCode.create(from_zip)
+        to_zip = ZipCode.create(to_zip)
+
+        if extra_services is None:
+            extra_services = []
+        else:
+            extra_services = [ExtraService.get(es) for es in extra_services]
+
+        response = self.freight.CalcPrecoPrazo(
+            administrative_code,
+            self.password,
+            ",".join(str(s) for s in services),
+            str(from_zip),
+            str(to_zip),
+            package.weight / KG,
+            package.package_type,
+            package.length,
+            package.height,
+            package.width,
+            package.diameter,
+            "S" if EXTRA_SERVICE_MP in extra_services else "N",
+            value,
+            "S" if EXTRA_SERVICE_AR in extra_services else "N",
+        )
+        return self.model_builder.build_freights_list(response)

correios/models/posting.py

@@ -15,14 +15,15 @@
 
 import os
 import math
-from datetime import datetime
+from datetime import datetime, timedelta
 from decimal import Decimal
-from typing import Optional, Sequence, Tuple, Union, List, Dict  # noqa: F401
+from typing import Optional, Tuple, Union, List, Dict  # noqa: F401
 
 from PIL import Image
 
 from correios import DATADIR
 from correios import exceptions
+from correios.utils import to_decimal
 from .address import Address, ZipCode
 from .data import SERVICE_PAC, TRACKING_EVENT_TYPES, TRACKING_STATUS
 from .user import Contract  # noqa: F401
@@ -255,6 +256,12 @@ class Package:
     TYPE_BOX = 2  # type: int
     TYPE_CYLINDER = 3  # type: int
 
+    freight_package_types = {
+        TYPE_BOX: 1,
+        TYPE_CYLINDER: 2,
+        TYPE_ENVELOPE: 3,
+    }  # type: Dict[int, int]
+
     def __init__(self,
                  package_type: int = TYPE_BOX,
                  width: Union[float, int] = 0,  # cm
@@ -336,6 +343,19 @@ def volumetric_weight(self) -> int:
     def posting_weight(self) -> int:
         return Package.calculate_posting_weight(self.weight, self.volumetric_weight)
 
+    @property
+    def freight_package_type(self) -> int:
+        """
+        SIGEP API and Freight API different codes to identify package types:
+
+        SIGEP | Freight | Type
+        ------+---------+----------
+          1   |    3    | Envelope
+          2   |    1    | Box
+          3   |    2    | Cylinder
+        """
+        return self.freight_package_types[self.package_type]
+
     @classmethod
     def calculate_volumetric_weight(cls, width, height, length) -> int:
         return int(math.ceil((width * height * length) / IATA_COEFICIENT))
@@ -355,7 +375,7 @@ def calculate_insurance(cls,
         per_unit_value = Decimal(per_unit_value)
         if Service.get(service) == Service.get(SERVICE_PAC) and per_unit_value > INSURANCE_VALUE_THRESHOLD:
             value = (per_unit_value - INSURANCE_VALUE_THRESHOLD) * INSURANCE_PERCENTUAL_COST
-        return Decimal(value * quantity).quantize(Decimal('0.00'))
+        return to_decimal(value * quantity)
 
     @classmethod
     def validate(cls,
@@ -457,7 +477,7 @@ def __init__(self,
                  service: Union[Service, int],
                  tracking_code: Union[TrackingCode, str],
                  package: Package,
-                 extra_services: Optional[Sequence[Union[ExtraService, int]]] = None,
+                 extra_services: Optional[List[Union[ExtraService, int]]] = None,
                  logo: Optional[Union[str, Image.Image]] = None,
                  order: Optional[str] = "",
                  invoice_number: Optional[str] = "",
@@ -506,7 +526,7 @@ def __init__(self,
     def __repr__(self):
         return "<ShippingLabel tracking={!r}>".format(str(self.tracking_code))
 
-    def add_extra_services(self, extra_services: Sequence[Union["ExtraService", int]]):
+    def add_extra_services(self, extra_services: List[Union["ExtraService", int]]):
         for extra_service in extra_services:
             self.add_extra_service(extra_service)
 
@@ -639,3 +659,56 @@ def close_with_id(self, number: int):
     @property
     def closed(self):
         return self.number is not None
+
+
+class Freight:
+    def __init__(self,
+                 service: Union[Service, int],
+                 delivery_time: Union[int, timedelta],
+                 value: Union[Decimal, float, int, str],
+                 declared_value: Union[Decimal, float, int, str] = 0.00,
+                 mp_value: Union[Decimal, float, int, str] = 0.00,
+                 ar_value: Union[Decimal, float, int, str] = 0.00,
+                 saturday: bool = False,
+                 home: bool = False) -> None:
+
+        self.service = Service.get(service)
+
+        if not isinstance(delivery_time, timedelta):
+            delivery_time = timedelta(days=delivery_time)
+        self.delivery_time = delivery_time
+
+        if not isinstance(value, Decimal):
+            value = to_decimal(value)
+        self.value = value
+
+        if not isinstance(declared_value, Decimal):
+            declared_value = to_decimal(declared_value)
+        self.declared_value = declared_value
+
+        if not isinstance(mp_value, Decimal):
+            mp_value = to_decimal(mp_value)
+        self.mp_value = mp_value
+
+        if not isinstance(ar_value, Decimal):
+            ar_value = to_decimal(ar_value)
+        self.ar_value = ar_value
+
+        self.saturday = saturday
+        self.home = home
+        self.error_code = 0
+        self.error_message = ""
+
+    @property
+    def total(self):
+        return self.value + self.declared_value + self.ar_value + self.mp_value
+
+
+class FreightError(Freight):
+    def __init__(self,
+                 service: Union[Service, int],
+                 error_code: Union[str, int],
+                 error_message: str) -> None:
+        super().__init__(service=service, delivery_time=0, value=Decimal("0.00"))
+        self.error_code = int(error_code)
+        self.error_message = error_message

correios/models/user.py

@@ -13,37 +13,23 @@
 # limitations under the License.
 
 
-import os
-from datetime import datetime
+from datetime import datetime  # noqa: F401
 from decimal import Decimal
-from typing import Union, Optional, Sequence
+from typing import Union, Optional, Sequence, List  # noqa: F401
 
+import os
 from PIL import Image
 
 from correios import DATADIR
 from correios.exceptions import (InvalidFederalTaxNumberError, InvalidExtraServiceError,
                                  InvalidRegionalDirectionError, InvalidUserContractError,
                                  MaximumDeclaredValueError, MinimumDeclaredValueError)
+from correios.utils import to_integer, to_datetime
 from .data import EXTRA_SERVICES, REGIONAL_DIRECTIONS, SERVICES, EXTRA_SERVICE_VD
 
 EXTRA_SERVICE_CODE_SIZE = 2
 
 
-def to_integer(number: Union[int, str]) -> int:
-    try:
-        return int(number.strip())  # type: ignore
-    except AttributeError:
-        return int(number)
-
-
-def to_datetime(date: Union[datetime, str], fmt="%Y-%m-%d %H:%M:%S%z") -> datetime:
-    if isinstance(date, str):
-        last_colon_pos = date.rindex(":")
-        date = date[:last_colon_pos] + date[last_colon_pos + 1:]
-        return datetime.strptime(date, fmt)
-    return date
-
-
 def _to_federal_tax_number(federal_tax_number) -> "FederalTaxNumber":
     if isinstance(federal_tax_number, FederalTaxNumber):
         return federal_tax_number
@@ -155,8 +141,9 @@ def __init__(self,
         self.max_declared_value = max_declared_value
 
         if default_extra_services is None:
-            default_extra_services = []
-        self.default_extra_services = [ExtraService.get(es) for es in default_extra_services]
+            self.default_extra_services = []  # type: List
+        else:
+            self.default_extra_services = [ExtraService.get(es) for es in default_extra_services]
 
     def __str__(self):
         return str(self.code)
@@ -165,6 +152,7 @@ def __repr__(self):
         return "<Service code={!r}, name={!r}>".format(self.code, self.display_name)
 
     def __eq__(self, other):
+        other = Service.get(other)
         return (self.id, self.code) == (other.id, other.code)
 
     def validate_declared_value(self, value: Union[Decimal, float]) -> bool:
@@ -215,10 +203,12 @@ def __repr__(self):
         return "<ExtraService number={!r}, code={!r}>".format(self.number, self.code)
 
     def __eq__(self, other):
+        if isinstance(other, int):
+            return self.number == other
         return self.number == other.number
 
     def is_declared_value(self):
-        return self.number == EXTRA_SERVICE_VD
+        return self == EXTRA_SERVICE_VD
 
     @classmethod
     def get(cls, number: Union['ExtraService', int]) -> 'ExtraService':

correios/soap.py

@@ -65,5 +65,5 @@ def send(self, request):
 class SoapClient(Client):
     def __init__(self, url, cert=None, verify=True, timeout=8, *args, **kwargs):
         transport = RequestsTransport(cert=cert, verify=verify, timeout=timeout)
-        headers = {"Content-Type": "text/xml;charset=UTF-8", "SOAPAction": ""}
+        headers = {"Content-Type": "text/xml;charset=UTF-8"}
         super().__init__(url, transport=transport, headers=headers, **kwargs)

correios/utils.py

@@ -1,5 +1,9 @@
+from datetime import datetime
+from decimal import Decimal
 from itertools import chain
-from typing import Container, Iterable, Sized
+from typing import Container, Iterable, Sized, Union
+
+import re
 
 
 def capitalize_phrase(phrase: str) -> str:
@@ -51,3 +55,29 @@ def __contains__(self, elem):
 
     def __len__(self):
         return sum(len(r) for r in self.ranges)
+
+
+def to_integer(number: Union[int, str]) -> int:
+    return int(str(number).strip())
+
+
+def to_datetime(date: Union[datetime, str], fmt="%Y-%m-%d %H:%M:%S%z") -> datetime:
+    if isinstance(date, str):
+        last_colon_pos = date.rindex(":")
+        date = date[:last_colon_pos] + date[last_colon_pos + 1:]
+        return datetime.strptime(date, fmt)
+    return date
+
+
+def to_decimal(value: Union[Decimal, str, float], precision=2):
+    if not isinstance(value, Decimal):
+        value = rreplace(str(value), ",", ".", 1)
+        if "." in value:
+            real, imag = value.rsplit(".", 1)
+        else:
+            real, imag = value, "0"
+        real = re.sub("[,._]", "", real)
+        value = Decimal("{}.{}".format(real, imag))
+
+    quantize = Decimal("0." + "0" * precision)
+    return value.quantize(quantize)

tests/conftest.py

@@ -160,7 +160,7 @@ class Meta:
     package_type = Package.TYPE_BOX
     width = LazyFunction(lambda: random.randint(11, 30))
     height = LazyFunction(lambda: random.randint(2, 30))
-    length = LazyFunction(lambda: random.randint(16, 30))
+    length = LazyFunction(lambda: random.randint(18, 30))
     weight = LazyFunction(lambda: random.randint(1, 100) * 100)
     service = LazyFunction(lambda: random.choice(_services))
     sequence = Sequence(lambda n: (n, n + 1))