diff --git a/roulier/carriersv2/__init__.py b/roulier/carriersv2/__init__.py index 1a84411..0db66e3 100644 --- a/roulier/carriersv2/__init__.py +++ b/roulier/carriersv2/__init__.py @@ -1 +1,2 @@ +from .ciblex import transporter from .mondialrelay import transporter diff --git a/roulier/carriersv2/ciblex/__init__.py b/roulier/carriersv2/ciblex/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/roulier/carriersv2/ciblex/schema.py b/roulier/carriersv2/ciblex/schema.py new file mode 100644 index 0000000..e8c088a --- /dev/null +++ b/roulier/carriersv2/ciblex/schema.py @@ -0,0 +1,171 @@ +# Copyright 2024 Akretion (http://www.akretion.com). +# @author Florian Mounier +# 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], + ) diff --git a/roulier/carriersv2/ciblex/transporter.py b/roulier/carriersv2/ciblex/transporter.py new file mode 100644 index 0000000..acd7ab3 --- /dev/null +++ b/roulier/carriersv2/ciblex/transporter.py @@ -0,0 +1,150 @@ +# Copyright 2024 Akretion (http://www.akretion.com). +# @author Florian Mounier +# 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)