diff --git a/roulier/carriers/__init__.py b/roulier/carriers/__init__.py index de3a5fd..7326d72 100755 --- a/roulier/carriers/__init__.py +++ b/roulier/carriers/__init__.py @@ -1,4 +1,5 @@ from . import laposte_fr +from . import colissimo_fr from .gls_fr import rest as gls_fr_rest from .gls_fr import glsbox as gls_fr_glsbox from . import chronopost_fr diff --git a/roulier/carriers/colissimo_fr/__init__.py b/roulier/carriers/colissimo_fr/__init__.py new file mode 100644 index 0000000..a0c5503 --- /dev/null +++ b/roulier/carriers/colissimo_fr/__init__.py @@ -0,0 +1 @@ +from . import carrier diff --git a/roulier/carriers/colissimo_fr/carrier.py b/roulier/carriers/colissimo_fr/carrier.py new file mode 100644 index 0000000..5c21be3 --- /dev/null +++ b/roulier/carriers/colissimo_fr/carrier.py @@ -0,0 +1,230 @@ +# Copyright 2024 Akretion (http://www.akretion.com). +# @author Florian Mounier +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). +import logging +import base64 +import json +import requests +from requests_toolbelt import MultipartDecoder +from ...carrier import Carrier, action +from ...exception import CarrierError +from .schema import ( + ColissimoFrLabelInput, + ColissimoFrLabelOutput, + ColissimoFrPackingSlipInput, + ColissimoFrPackingSlipOutput, + ColissimoFrGetDocumentsInput, + ColissimoFrGetDocumentInput, + ColissimoFrCreateUpdateDocumentInput, + ColissimoFrDocumentOutput, + ColissimoFrDocumentsOutput, + ColissimoFrCreateUpdateDocumentOutput, +) + + +_logger = logging.getLogger(__name__) + + +def int_maybe(value): + try: + return int(value) + except ValueError: + return value + + +class ColissimoFr(Carrier): + __key__ = "colissimo_fr" + __url__ = "https://ws.colissimo.fr/sls-ws/SlsServiceWSRest/2.0" + __doc_url__ = "https://ws.colissimo.fr/api-document/rest" + __auth_url__ = "https://ws.colissimo.fr/widget-colissimo/rest" + __ref__ = "https://www.colissimo.fr/doc-colissimo/redoc-sls/en" + + def _raise_for_status(self, response): + try: + response.raise_for_status() + except requests.exceptions.HTTPError as e: + try: + json = response.json() + if "error" in json: + msg = [{"id": 0, "message": json["error"]}] + elif "errors" in json: + msg = [ + { + "id": int_maybe(error.get("code", 0)), + "message": error["message"], + } + for error in json["errors"] + ] + elif "messages" in json: + msg = [ + { + "id": int_maybe(error["id"]), + "message": error["messageContent"], + } + for error in json["messages"] + if error["type"] == "ERROR" + ] + elif "errorCode" in json and json["errorCode"] != "000": + msg = [ + { + "id": int_maybe(json["errorCode"]), + "message": json["errorLabel"], + } + ] + else: + raise + except Exception: + msg = response.text + + raise CarrierError(response, msg) from e + + return response + + def _raise_for_error_code(self, response): + if "errorCode" in response and response["errorCode"] != "000": + raise CarrierError( + response, + [ + { + "id": int_maybe(response["errorCode"]), + "message": response["errorLabel"], + } + ], + ) + + def request(self, method, json, url=None): + headers = {} + if json and "apiKey" in json: + json = json.copy() + headers["apiKey"] = json.pop("apiKey") + + response = requests.post( + f"{url or self.__url__}/{method}", json=json, headers=headers + ) + self._raise_for_status(response) + return response + + def doc_request(self, method, json, files=None): + kwargs = {} + if files: + kwargs["headers"] = json.pop("credential") + kwargs["data"] = json + kwargs["files"] = files + else: + kwargs["json"] = json + + response = requests.post(f"{self.__doc_url__}/{method}", **kwargs) + self._raise_for_status(response) + return response + + def validate(self, params): + response = self.request("checkGenerateLabel", params) + _logger.debug("Validation response: %s", response.text) + return response.text + + def _parse_response(self, response): + parsed = {} + decoder = MultipartDecoder.from_response(response) + for part in decoder.parts: + content_id = part.headers.get(b"Content-ID", b"").decode("utf-8") + content_type = part.headers.get(b"Content-Type", b"").decode("utf-8") + + content = part.content + + # Process each part based on its content type + if "application/json" in content_type: + parsed[content_id] = json.loads(content.decode("utf-8")) + elif ( + "application/pdf" in content_type + or "application/octet-stream" in content_type + ): + parsed[content_id] = content + else: + _logger.warning( + "Unknown content type: %s for id : %s", content_type, content_id + ) + return parsed + + @action + def get_label(self, input: ColissimoFrLabelInput) -> ColissimoFrLabelOutput: + params = input.params() + self.validate(params) + response = self.request("generateLabel", params) + + result = self._parse_response(response) + return ColissimoFrLabelOutput.from_params(result, input) + + @action + def get_packing_slip( + self, input: ColissimoFrPackingSlipInput + ) -> ColissimoFrPackingSlipOutput: + + params = input.params() + + if input.packing_slip_number: + raise NotImplementedError( + "Fetching packing slip by number does not seem to " + "be supported by the REST API" + ) + # Getting the auth token + params["login"] = params.pop("contractNumber") + response = self.request("authenticate.rest", params, url=self.__auth_url__) + self._raise_for_status(response) + response = response.json() + headers = {"token": response["token"]} + # partnerClientCode + response = requests.get( + f"{self.__url__}/SlsInternalService/getBordereauByNumber/{input.packing_slip_number}", + headers=headers, + ) + self._raise_for_status(response) + result = self._parse_response(response) + else: + response = self.request("generateBordereauByParcelsNumbers", params) + result = self._parse_response(response) + + return ColissimoFrPackingSlipOutput.from_params(result) + + @action + def get_documents( + self, input: ColissimoFrGetDocumentsInput + ) -> ColissimoFrDocumentsOutput: + params = input.params() + response = self.doc_request("documents", params) + result = response.json() + self._raise_for_error_code(result) + return ColissimoFrDocumentsOutput.from_params(result) + + @action + def get_document( + self, input: ColissimoFrGetDocumentInput + ) -> ColissimoFrDocumentOutput: + params = input.params() + response = self.doc_request("document", params) + return ColissimoFrDocumentOutput.from_params(response.content) + + @action + def create_document( + self, input: ColissimoFrCreateUpdateDocumentInput + ) -> ColissimoFrCreateUpdateDocumentOutput: + params = input.params() + + with open(input.service.document_path, "rb") as file: + files = {"file": (params["filename"], file.read())} + + response = self.doc_request("storedocument", params, files) + result = response.json() + return ColissimoFrCreateUpdateDocumentOutput.from_params(result) + + @action + def update_document( + self, input: ColissimoFrCreateUpdateDocumentInput + ) -> ColissimoFrCreateUpdateDocumentOutput: + params = input.params() + + with open(input.service.document_path, "rb") as file: + files = {"file": (params["filename"], file.read())} + + response = self.doc_request("updatedocument", params, files) + result = response.json() + return ColissimoFrCreateUpdateDocumentOutput.from_params(result) diff --git a/roulier/carriers/colissimo_fr/schema.py b/roulier/carriers/colissimo_fr/schema.py new file mode 100644 index 0000000..7dddb46 --- /dev/null +++ b/roulier/carriers/colissimo_fr/schema.py @@ -0,0 +1,678 @@ +# Copyright 2024 Akretion (http://www.akretion.com). +# @author Florian Mounier +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). +from base64 import b64encode +from datetime import date, datetime +from enum import Enum +from pathlib import Path +from pydantic import BaseModel, model_validator + +from ...helpers import merge, unaccent +from ...schema import ( + LabelInput, + Address, + LabelOutput, + Auth, + Service, + Parcel, + ParcelLabel, + Label, + Tracking, + PackingSlipInput, + PackingSlipOutput, + PackingSlip, + PackingSlipClient, + PackingSlipSupportSite, + DocumentService, + GetDocumentService, + GetDocumentInput, + GetDocumentsInput, + CreateUpdateDocumentService, + CreateUpdateDocumentInput, + DocumentOutput, +) + + +class Format(Enum): + # Shortcuts + PDF = "PDF" + ZPL = "ZPL" + DPL = "DPL" + # Formats + ZPL_10x15_203dpi = "ZPL_10x15_203dpi" + ZPL_10x15_300dpi = "ZPL_10x15_300dpi" + DPL_10x15_203dpi = "DPL_10x15_203dpi" + DPL_10x15_300dpi = "DPL_10x15_300dpi" + PDF_10x15_300dpi = "PDF_10x15_300dpi" + PDF_A4_300dpi = "PDF_A4_300dpi" + ZPL_10x10_203dpi = "ZPL_10x10_203dpi" + ZPL_10x10_300dpi = "ZPL_10x10_300dpi" + DPL_10x10_203dpi = "DPL_10x10_203dpi" + DPL_10x10_300dpi = "DPL_10x10_300dpi" + PDF_10x10_300dpi = "PDF_10x10_300dpi" + ZPL_10x15_203dpi_UL = "ZPL_10x15_203dpi_UL" + ZPL_10x15_300dpi_UL = "ZPL_10x15_300dpi_UL" + DPL_10x15_203dpi_UL = "DPL_10x15_203dpi_UL" + DPL_10x15_300dpi_UL = "DPL_10x15_300dpi_UL" + PDF_10x15_300dpi_UL = "PDF_10x15_300dpi_UL" + PDF_A4_300dpi_UL = "PDF_A4_300dpi_UL" + + @property + def final_value(self): + return ( + { + Format.PDF: Format.PDF_10x15_300dpi, + Format.ZPL: Format.ZPL_10x15_300dpi, + Format.DPL: Format.DPL_10x15_300dpi, + } + .get(self, self) + .value + ) + + +class ColissimoFrAuth(Auth): + login: str | None = None + password: str | None = None + apiKey: str | None = None + + def params(self): + if self.apiKey: + return {"apiKey": self.apiKey} + return { + "contractNumber": self.login, + "password": self.password, + } + + @model_validator(mode="after") + def check_login_pass_or_apikey(self): + if self.apiKey: + if self.login or self.password: + raise ValueError("Only one of login/password or apiKey is allowed") + else: + if not self.login or not self.password: + raise ValueError("Without apiKey, login and password are required") + + return self + + +class ColissimoFrAddress(Address): + country: str + firstName: str | None = None + zip: str + city: str + street0: str | None = None + street1: str + street2: str | None = None + street3: str | None = None + door1: str | None = None + door2: str | None = None + intercom: str | None = None + language: str = "FR" + landlinePhone: str | None = None + stateOrProvinceCode: str | None = None + + def params(self): + return { + "companyName": self.company, + "lastName": self.name, + "firstName": self.firstName, + "line0": self.street2, + "line1": self.street0, + "line2": self.street1, + "line3": self.street3, + "countryCode": self.country, + "city": self.city, + "zipCode": self.zip, + "phoneNumber": self.landlinePhone, + "mobileNumber": self.phone, + "doorCode1": self.door1, + "doorCode2": self.door2, + "intercom": self.intercom, + "email": self.email, + "language": self.language, + "stateOrProvinceCode": self.stateOrProvinceCode, + } + + +class ColissimoFrService(Service): + labelFormat_x: int = 0 + labelFormat_y: int = 0 + labelFormat: Format | None = Format.PDF + dematerialized: bool = False + returnType: str | None = None + printCoDDocument: bool = False + + product: str + pickupLocationId: str | None = None + + mailBoxPicking: bool = False + mailBoxPickingDate: date | None = None + vatCode: int | None = None + vatPercentage: int | None = None + vatAmount: int | None = None + transportationAmount: int | None = None + totalAmount: int | None = None + commercialName: str | None = None + returnTypeChoice: int | None = None + reseauPostal: str | None = None + + codeBarForReference: bool | None = None + serviceInfo: str | None = None + promotionCode: str | None = None + + codSenderAddress: ColissimoFrAddress | None = None + + @model_validator(mode="after") + def check_format(self): + if self.labelFormat is None: + self.labelFormat = Format.PDF + return self + + def params(self): + return { + "outputFormat": { + "x": self.labelFormat_x, + "y": self.labelFormat_y, + "outputPrintingType": self.labelFormat.final_value, + "dematerialized": self.dematerialized, + "returnType": self.returnType, + "printCoDDocument": self.printCoDDocument, + }, + "letter": { + "service": { + "productCode": self.product, + "depositDate": self.shippingDate.isoformat(), + "mailBoxPicking": self.mailBoxPicking, + "mailBoxPickingDate": ( + self.mailBoxPickingDate.isoformat() + if self.mailBoxPickingDate + else None + ), + "vatCode": self.vatCode, + "vatPercentage": self.vatPercentage, + "vatAmount": self.vatAmount, + "transportationAmount": self.transportationAmount, + "totalAmount": self.totalAmount, + "orderNumber": self.reference1, + "commercialName": self.commercialName, + "returnTypeChoice": self.returnTypeChoice, + "reseauPostal": self.reseauPostal, + }, + "parcel": { + "pickupLocationId": self.pickupLocationId, + }, + "sender": { + "senderParcelRef": self.reference1, + }, + "addressee": { + "addresseeParcelRef": self.reference2, + "codeBarForReference": self.codeBarForReference, + "serviceInfo": self.serviceInfo, + "promotionCode": self.promotionCode, + }, + "codSenderAddress": ( + self.codSenderAddress.params() if self.codSenderAddress else None + ), + }, + } + + +class ColissimoFrArticle(BaseModel): + description: str | None = None + quantity: int | None = None + weight: float | None = None + value: float | None = None + hsCode: str | None = None + originCountry: str | None = None + originCountryLabel: str | None = None + currency: str | None = "EUR" + artref: str | None = None + originalIdent: str | None = None + vatAmount: float | None = None + customsFees: float | None = None + + def params(self): + return { + "description": self.description, + "quantity": self.quantity, + "weight": self.weight, + "value": self.value, + "hsCode": self.hsCode, + "originCountry": self.originCountry, + "originCountryLabel": self.originCountryLabel, + "currency": self.currency, + "artref": self.artref, + "originalIdent": self.originalIdent, + "vatAmount": self.vatAmount, + "customsFees": self.customsFees, + } + + +class ColissimoFrOriginal(BaseModel): + originalIdent: str | None = None + originalInvoiceNumber: str | None = None + originalInvoiceDate: str | None = None + originalParcelNumber: str | None = None + + def params(self): + return { + "originalIdent": self.originalIdent, + "originalInvoiceNumber": self.originalInvoiceNumber, + "originalInvoiceDate": self.originalInvoiceDate, + "originalParcelNumber": self.originalParcelNumber, + } + + +class ColissimoFrCustoms(BaseModel): + includesCustomsDeclarations: bool = False + numberOfCopies: int | None = None + + # contents + articles: list[ColissimoFrArticle] = [] + category: int + original: list[ColissimoFrOriginal] = [] + explanations: str | None = None + + importersReference: str | None = None + importersContact: str | None = None + officeOrigin: str | None = None + comments: str | None = None + description: str | None = None + invoiceNumber: str | None = None + licenseNumber: str | None = None + certificatNumber: str | None = None + importerAddress: ColissimoFrAddress | None = None + + def params(self): + return { + "includesCustomsDeclarations": True, + "numberOfCopies": self.numberOfCopies, + "contents": { + "article": [article.params() for article in self.articles], + "category": { + "value": self.category, + }, + "original": [orig.params() for orig in self.original], + "explanations": self.explanations, + }, + "importersReference": self.importersReference, + "importersContact": self.importersContact, + "officeOrigin": self.officeOrigin, + "comments": self.comments, + "description": self.description, + "invoiceNumber": self.invoiceNumber, + "licenseNumber": self.licenseNumber, + "certificatNumber": self.certificatNumber, + "importerAddress": ( + self.importerAddress.params() if self.importerAddress else None + ), + } + + +class ColissimoFrParcel(Parcel): + parcelNumber: str | None = None + insuranceAmount: int | None = None + insuranceValue: int | None = None + recommendationLevel: str | None = None + nonMachinable: bool = False + returnReceipt: bool = False + instructions: str | None = None + pickupLocationId: str | None = None + ftd: bool = False + ddp: bool = False + disabledDeliveryBlockingCode: str | None = None + cod: bool = False + codamount: int | None = None + codcurrency: str | None = None + + customs: ColissimoFrCustoms | None = None + + length: int | None = None + width: int | None = None + height: int | None = None + + def params(self): + return { + "letter": { + "parcel": { + "parcelNumber": self.parcelNumber, + "insuranceAmount": self.insuranceAmount, + "insuranceValue": self.insuranceValue, + "recommendationLevel": self.recommendationLevel, + "weight": self.weight, + "nonMachinable": self.nonMachinable, + "returnReceipt": self.returnReceipt, + "instructions": self.instructions, + "pickupLocationId": self.pickupLocationId, # TODO + "ftd": self.ftd, + "ddp": self.ddp, + "disabledDeliveryBlockingCode": self.disabledDeliveryBlockingCode, + "cod": self.cod, + "codamount": self.codamount, + "codcurrency": self.codcurrency, + }, + "customsDeclarations": self.customs.params() if self.customs else None, + }, + "fields": ( + { + "field": [ + {"key": key.upper(), "value": getattr(self, key)} + for key in ["length", "width", "height"] + if getattr(self, key) + ] + } + if self.length or self.width or self.height + else None + ), + } + + +class ColissimoFrLabelInput(LabelInput): + auth: ColissimoFrAuth + service: ColissimoFrService + parcels: list[ColissimoFrParcel] + to_address: ColissimoFrAddress + from_address: ColissimoFrAddress + + def params(self): + return unaccent( + merge( + self.auth.params(), + self.service.params(), + self.parcels[0].params(), + { + "letter": { + "sender": {"address": self.from_address.params()}, + "addressee": {"address": self.to_address.params()}, + } + }, + ) + ) + + +class ColissimoFrTracking(Tracking): + @classmethod + def from_params(cls, result): + return cls.model_construct( + number=result["parcelNumber"], + url=result["pdfUrl"], + partner=result["parcelNumberPartner"], + ) + + +class ColissimoFrLabel(Label): + @classmethod + def from_params(cls, result, name, format): + return cls.model_construct( + data=b64encode(result).decode("utf-8"), + name=name, + type=format, + ) + + +class ColissimoFrParcelLabel(ParcelLabel): + label: ColissimoFrLabel | None = None + tracking: ColissimoFrTracking | None = None + + @classmethod + def from_params(cls, result, input): + return cls.model_construct( + id=1, + reference=input.parcels[0].reference, + label=( + ColissimoFrLabel.from_params( + result["