Add ciblex transporter

roulier/carriersv2/__init__.py

@@ -1 +1,2 @@
+from .ciblex import transporter
 from .mondialrelay import transporter

roulier/carriersv2/ciblex/schema.py

@@ -0,0 +1,171 @@
+# Copyright 2024 Akretion (http://www.akretion.com).
+# @author Florian Mounier <[email protected]>
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
+from ..helpers import prefix, suffix, none_as_empty, REMOVED
+from ..schema import (
+    LabelInput,
+    Address,
+    LabelOutput,
+    Auth,
+    Service,
+    Parcel,
+    ParcelLabel,
+    Label,
+    Tracking,
+)
+
+
+class CiblexAuth(Auth):
+    login: str
+    password: str
+
+    def params(self):
+        return none_as_empty(
+            {
+                "USER_COMPTE": self.login,
+                "USER_PASSWORD": self.password,
+                "lang": "fr",
+                "LOGIN": "Connexion sécurisée",
+            }
+        )
+
+
+class CiblexService(Service):
+    customerId: str
+    product: str
+    imperative_time: str | None = None  # 08:00, 09:00
+    opt_ssm: bool | None = None
+
+    def params(self):
+        return none_as_empty(
+            {
+                "expediteur": self.customerId,
+                "prestation": self.product,
+                "date_cmd": self.shippingDate.strftime("%d/%m/%Y"),
+                "imperatif": self.imperative_time,
+                "opt_ssm": self.opt_ssm,
+            }
+        )
+
+
+class CiblexParcel(Parcel):
+    reference2: str | None = None
+    reference3: str | None = None
+    delivery_versus: float | None = None
+    check_payable_to: str | None = None
+    # ad_valorem_types: 1 : standand, 2 : sensible, 4 : international
+    ad_valorem_type: int | None = None
+    ad_valorem: float | None = None
+    ad_valorem_agreed: bool | None = None
+
+    def params(self):
+        return none_as_empty(
+            {
+                "poids": self.weight,
+                "ref1": self.reference,
+                "ref2": self.reference2,
+                "ref3": self.reference3,
+                "cpa": self.delivery_versus,
+                "ordre_chq": self.check_payable_to,
+                "opt_adv": self.ad_valorem_type,
+                "adv": self.ad_valorem,
+                "adv_cond": self.ad_valorem_agreed,
+            }
+        )
+
+
+class CiblexAddress(Address):
+    zip: str
+    city: str
+    country: str  # FR ou MC, enum?
+    street3: str | None = None
+    street4: str | None = None
+
+    def params(self):
+        return none_as_empty(
+            {
+                "raison": ", ".join(
+                    [part for part in (self.name, self.company) if part]
+                ),
+                "adr1": self.street1,
+                "adr2": self.street2,
+                "adr3": self.street3,
+                "adr4": self.street4,
+                "cp": self.zip,
+                "ville": self.city,
+                "pays": self.country,
+                "tel": self.phone,
+                "email": self.email,
+            }
+        )
+
+
+class CiblexLabelInput(LabelInput):
+    auth: CiblexAuth
+    service: CiblexService
+    parcels: list[CiblexParcel]
+    to_address: CiblexAddress
+    from_address: CiblexAddress
+
+    def params(self):
+        return none_as_empty(
+            {
+                "module": "cmdsai",
+                "commande": None,
+                **self.service.params(),
+                **prefix(self.from_address.params(), "exp_"),
+                **prefix(self.to_address.params(), "dest_"),
+                "nb_colis": len(self.parcels),
+                **{
+                    k: v
+                    for i, parcel in enumerate(self.parcels)
+                    for k, v in suffix(parcel.params(), f"_{i+1}").items()
+                },
+            }
+        )
+
+
+class CiblexTracking(Tracking):
+    @classmethod
+    def from_params(cls, result):
+        return cls.model_construct(
+            number=result["tracking"],
+            url=(
+                "https://secure.extranet.ciblex.fr/extranet/client/"
+                "corps.php?module=colis&colis=%s" % result["tracking"]
+            ),
+        )
+
+
+class CiblexLabel(Label):
+    @classmethod
+    def from_params(cls, result):
+        return cls.model_construct(
+            data=result["label"],
+            name="label",
+            type=result["format"],
+        )
+
+
+class CiblexParcelLabel(ParcelLabel):
+    label: CiblexLabel | None = None
+    tracking: CiblexTracking | None = None
+
+    @classmethod
+    def from_params(cls, result):
+        return cls.model_construct(
+            id=result["id"],
+            reference=result["reference"],
+            label=CiblexLabel.from_params(result),
+            tracking=CiblexTracking.from_params(result),
+        )
+
+
+class CiblexLabelOutput(LabelOutput):
+    parcels: list[CiblexParcelLabel]
+
+    @classmethod
+    def from_params(cls, results):
+        return cls.model_construct(
+            parcels=[CiblexParcelLabel.from_params(result) for result in results],
+        )

roulier/carriersv2/ciblex/transporter.py

@@ -0,0 +1,150 @@
+# Copyright 2024 Akretion (http://www.akretion.com).
+# @author Florian Mounier <[email protected]>
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
+import base64
+import logging
+import requests
+from lxml.html import fromstring
+from ..api import Transporter, action
+from ...exception import CarrierError
+from .schema import CiblexLabelInput, CiblexLabelOutput
+
+_logger = logging.getLogger(__name__)
+
+
+class Ciblex(Transporter):
+    __key__ = "ciblex"
+    __url__ = "https://secure.extranet.ciblex.fr/extranet/client"
+
+    def _xpath(self, response, xpath):
+        root = fromstring(response.text)
+        return root.xpath(xpath)
+
+    def _xpath_to_text(self, response, xpath):
+        nodes = self._xpath(response, xpath)
+        if nodes:
+            return "\n".join([e.text_content() for e in nodes])
+
+    def _encode_all_params_to_latin_1(self, params):
+        for key, value in params.items():
+            if isinstance(value, str):
+                params[key] = value.encode("latin-1")
+        return params
+
+    def _auth(self, auth):
+        response = requests.post(f"{self.__url__}/index.php", data=auth.params())
+        error = self._xpath_to_text(response, '//td[@class="f_erreur_small"]')
+        if error:
+            raise CarrierError(response, error)
+
+        return response.cookies
+
+    def _validate(self, auth, params):
+        # 1) Validate
+        response = requests.get(
+            f"{self.__url__}/corps.php",
+            params={"action": "Valider", **params},
+            cookies=auth,
+        )
+
+        # Handle approximative city
+        cp_dest = self._xpath(response, '//select[@name="cp_dest"]')
+        if cp_dest:
+            good_city = cp_dest[0].getchildren()[0].text.split(" ", 1)[1]
+            if params["dest_ville"] == good_city:
+                raise CarrierError(response, "City not found")
+            _logger.warning(f"Replacing {params['dest_ville']} by {good_city}")
+            params["dest_ville"] = good_city.encode("latin-1")
+            return self._validate(auth, params)
+
+        error = self._xpath_to_text(response, '//p[@class="f_erreur"]')
+        if error:
+            raise CarrierError(response, error)
+
+    def _print(self, auth, params, format="PDF"):
+        # 2) Print
+        response = requests.get(
+            f"{self.__url__}/corps.php",
+            params={
+                "action": (
+                    "Imprimer(PDF)" if format == "PDF" else "Imprimer(Thermique)"
+                ),
+                **params,
+            },
+            cookies=auth,
+        )
+
+        labels = self._xpath(response, '//input[@name="liste_cmd"]')
+        if not labels:
+            raise CarrierError(response, "No label found")
+        if len(labels) > 1:
+            raise CarrierError(response, "Multiple labels found")
+        label = labels[0]
+        order = label.attrib["value"]
+        return {
+            "order": order,
+            "format": format,
+        }
+
+    def _download(self, auth, order):
+        # 3) Get label
+        response = requests.get(
+            f"{self.__url__}/label_ool.php",
+            params={
+                "origine": "OOL",
+                "output": order["format"],
+                "url_retour": f"{self.__url__}/corps.php?module=cmdjou",
+                "liste_cmd": order["order"],
+            },
+            cookies=auth,
+        )
+        return base64.b64encode(response.content)
+
+    def _get_tracking(self, auth, order, label, input):
+        # 4) Get tracking
+        response = requests.get(
+            f"{self.__url__}/corps.php?module=cmdjou",
+            cookies=auth,
+        )
+        # Order format is like "04282,17,1,1" : customerId, order, parcel count, ?
+        customer_id, order_id, count, _ = order["order"].split(",")
+
+        count = int(count)
+        assert count == len(input.parcels), "Parcel count mismatch"
+
+        order_ref = f"{customer_id}-{order_id.zfill(6)}"
+        orders = self._xpath(response, '//tr[@class="t_liste_ligne"]')
+        order = next(
+            filter(lambda o: o.getchildren()[0].text == order_ref, orders), None
+        )
+        if not order:
+            raise CarrierError(response, f"Order {order_ref} not found")
+
+        trackings = [a.text for a in order.getchildren()[3].findall("a")]
+        return [
+            {
+                "id": f"{order_ref}_{i+1}",
+                "reference": input.parcels[i].reference,
+                "format": "PDF",
+                "label": label,  # TODO: Label contain all parcels, split it?
+                "tracking": trackings[i],
+            }
+            for i in range(count)
+        ]
+
+    @action
+    def get_label(self, input: CiblexLabelInput) -> CiblexLabelOutput:
+        auth = self._auth(input.auth)
+        format = input.service.labelFormat or "PDF"
+        if format != "PDF":
+            # Website also use "PRINTER" but this can't work here
+            raise CarrierError(None, "Only PDF format is supported")
+
+        # requests send all params as utf-8, but Ciblex expect latin-1
+        params = self._encode_all_params_to_latin_1(input.params())
+        self._validate(auth, params)
+        order = self._print(auth, params, format)
+        label = self._download(auth, order)
+        results = self._get_tracking(auth, order, label, input)
+
+        return CiblexLabelOutput.from_params(results)