Add ciblex carrier with api v2

roulier/carriers/__init__.py

@@ -6,3 +6,4 @@
 from . import geodis_fr
 from . import mondialrelay
 from . import mondialrelay_fr
+from . import ciblex

roulier/carriers/ciblex/__init__.py

@@ -0,0 +1 @@
+from . import carrier

roulier/carriers/ciblex/carrier.py

@@ -0,0 +1,159 @@
+# 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 ...carrier import Carrier, action
+from ...exception import CarrierError
+from .schema import CiblexLabelInput, CiblexLabelOutput
+
+_logger = logging.getLogger(__name__)
+
+
+class Ciblex(Carrier):
+    __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 _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)",  # This is only to get the liste_cmd
+                **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, format="PDF"):
+        # 3) Get label
+        response = requests.get(
+            f"{self.__url__}/label_ool.php",
+            params={
+                "origine": "OOL",
+                "output": order["format"] if format == "PDF" else "PRINTER",
+                "url_retour": f"{self.__url__}/corps.php?module=cmdjou",
+                "liste_cmd": order["order"],
+            },
+            cookies=auth,
+        )
+        if format == "EPL":
+            # We need to get the file name
+            button = self._xpath(response, '//input[@id="btn_imp"]')
+            if not button:
+                raise CarrierError(response, "No generated EPL found")
+            epl_fn = button[0].attrib["onclick"].split("'")[3]
+            response = requests.get(f"{self.__url__}/tmp/{epl_fn}", cookies=auth)
+
+        return base64.b64encode(response.content)
+
+    def _get_tracking(self, auth, order, label, input, format="PDF"):
+        # 4) Get tracking
+        response = requests.get(
+            f"{self.__url__}/corps.php",
+            params={
+                "codecli": "tous",
+                "date1": input.service.shippingDate.strftime("%d/%m/%Y"),
+                "date2": input.service.shippingDate.strftime("%d/%m/%Y"),
+                "etat": 0,
+                "cmdsui": "Rechercher",
+                "module": "cmdsui",
+                "action": "rechercher",
+            },
+            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 order is None or not len(order):
+            raise CarrierError(response, f"Order {order_ref} not found")
+
+        trackings = [a.text for a in order.getchildren()[4].findall("a")]
+        return [
+            {
+                "id": f"{order_ref}_{i+1}",
+                "reference": input.parcels[i].reference,
+                "format": format,
+                "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 not in ["PDF", "EPL"]:
+            # Website also use "PRINTER" but this can't work here
+            raise CarrierError(None, "Only PDF and EPL format are supported")
+
+        # requests send all params as utf-8, but Ciblex expect latin-1
+        params = input.params()
+        self._validate(auth, params)
+        order = self._print(auth, params, format)
+        label = self._download(auth, order, format)
+        results = self._get_tracking(auth, order, label, input, format)
+
+        return CiblexLabelOutput.from_params(results)

roulier/carriers/ciblex/schema.py

@@ -0,0 +1,164 @@
+# 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, unaccent
+from ...schema import (
+    LabelInput,
+    Address,
+    LabelOutput,
+    Auth,
+    Service,
+    Parcel,
+    ParcelLabel,
+    Label,
+    Tracking,
+)
+
+
+class CiblexAuth(Auth):
+    login: str
+    password: str
+
+    def params(self):
+        return {
+            "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 {
+            "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 {
+            "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 {
+            "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 unaccent(
+            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"].decode("utf-8"),
+            name="label",
+            type=result["format"],
+        )
+
+
+class CiblexParcelLabel(ParcelLabel):
+    id: str
+    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],
+        )