From eb1d2578e73b0039bb5d8d7d3ea188ae4703e81e Mon Sep 17 00:00:00 2001 From: zep Date: Sat, 12 Jul 2025 20:52:57 +0200 Subject: [PATCH] Add mondialRelay.py --- mondialRelay.py | 212 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 212 insertions(+) create mode 100644 mondialRelay.py diff --git a/mondialRelay.py b/mondialRelay.py new file mode 100644 index 0000000..85a5f9e --- /dev/null +++ b/mondialRelay.py @@ -0,0 +1,212 @@ +import os +import pickle +import time +import csv +import sys +import configparser +from selenium import webdriver +from selenium.webdriver.common.by import By +from selenium.webdriver.support.ui import Select, WebDriverWait +from selenium.webdriver.support import expected_conditions as EC +from selenium.common.exceptions import TimeoutException +import argparse + + +class MondialRelaySession: + SUBMIT_URL = "https://www.mondialrelay.fr/envoi-de-colis/" + COOKIE_FILE = "cookies.pkl" + CONFIG_FILE = "config.ini" + + def __init__(self, driver=None, login_email=None, login_password=None, config_path=None): + # Load credentials from parameters or config file + self.login_email, self.login_password = self._load_credentials( + login_email, login_password, config_path or self.CONFIG_FILE + ) + + if driver: + self.driver = driver + else: + # If running as a PyInstaller bundle, drivers may live in _MEIPASS + base_path = getattr(sys, '_MEIPASS', os.path.abspath('.')) + driver_exe = os.path.join(base_path, 'chromedriver.exe') + if os.path.exists(driver_exe): + self.driver = webdriver.Chrome(executable_path=driver_exe) + else: + self.driver = webdriver.Chrome() + + self.wait = WebDriverWait(self.driver, 5) + self._prepare_session() + + def _save_cookies(self): + with open(self.COOKIE_FILE, "wb") as f: + pickle.dump(self.driver.get_cookies(), f) + + def _load_cookies(self): + with open(self.COOKIE_FILE, "rb") as f: + cookies = pickle.load(f) + for c in cookies: + # Selenium expects int expiry + if isinstance(c.get("expiry"), float): + c["expiry"] = int(c["expiry"]) + self.driver.add_cookie(c) + + def _load_credentials(self, email, password, path): + if email and password: + return email, password + config = configparser.ConfigParser(interpolation=None) + if not os.path.exists(path): + raise RuntimeError(f"Config file not found: {path}") + config.read(path) + try: + creds = config['credentials'] + return creds.get('email'), creds.get('password') + except KeyError: + raise RuntimeError("'credentials' section missing in config file") + + + def _is_logged_in(self): + try: + self.wait.until(EC.visibility_of_element_located((By.ID, "MenuConnecte"))) + return True + except TimeoutException: + return False + + def _prepare_session(self): + # Go to page to set domain context + self.driver.get(self.SUBMIT_URL) + time.sleep(1) + # Try load cookies + if os.path.exists(self.COOKIE_FILE): + self._load_cookies() + self.driver.refresh() + # If still not logged, perform inline login + if not self._is_logged_in(): + # open login modal + print("Not logged in") + self.driver.execute_script("callConnexion();") + form = self.wait.until(EC.visibility_of_element_located((By.ID, "LogOn_Form"))) + form.find_element(By.ID, "LogOn_LoginJson").send_keys(self.login_email) + form.find_element(By.ID, "LogOn_MotDePasse").send_keys(self.login_password) + form.find_element(By.CSS_SELECTOR, "button[type='submit']").click() + # wait for login + self.wait.until(EC.visibility_of_element_located((By.ID, "MenuConnecte"))) + # save cookies + self._save_cookies() + + def send_parcel(self, country: str, weight_kg: float = 1.0, + email: str = "test@test.com", comment: str = "Coucou"): # -> bool + # Navigate back to form page each time + self.driver.get(self.SUBMIT_URL) + self.wait.until(EC.presence_of_element_located((By.TAG_NAME, "body"))) + # Weight + w = self.wait.until(EC.element_to_be_clickable((By.ID, "poids"))) + w.clear(); w.send_keys(str(weight_kg)) + # Mode relais + lbl_mode = self.wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, "label[for='mode-relais']"))) + self.driver.execute_script("arguments[0].scrollIntoView(true);", lbl_mode) + lbl_mode.click() + # Laisser destinataire decider + lbl_dest = self.wait.until(EC.element_to_be_clickable((By.ID, "label-email-destinataire"))) + self.driver.execute_script("arguments[0].scrollIntoView(true);", lbl_dest) + lbl_dest.click() + # Email + e = self.wait.until(EC.element_to_be_clickable((By.ID, "destinataire-email"))) + e.clear(); e.send_keys(email) + # Country + sel = Select(self.wait.until(EC.element_to_be_clickable((By.ID, "CreerEnvoi_Destinataire_Email_Pays")))) + sel.select_by_visible_text(country) + # Comment + c = self.wait.until(EC.element_to_be_clickable((By.ID, "Destinataire_Commentaire"))) + c.clear(); c.send_keys(comment) + # Submit + submit_btn = self.wait.until( + EC.element_to_be_clickable((By.ID, "ECommerce_CreerEnvoi_Form_Submit")) + ) + submit_btn.click() + # Check confirmation overlay + try: + overlay = self.wait.until( + EC.visibility_of_element_located((By.ID, "ECommerce_CreerEnvoi_OverlayConfirm")) + ) + return overlay.is_displayed() + except TimeoutException: + return False + + def quit(self): + self.driver.quit() + +def process_csv(input_csv="MondialRelay.csv", + sent_csv="sent.csv", + not_sent_csv="not_sent.csv", + config_path=None): + """ + Reads rows from input_csv, sends parcel for valid countries, and writes + sent/not_sent outputs with a Status column. + """ + # Valid destination list + valid_countries = { + "Allemagne", "Autriche", "Belgique", "Espagne", + "France", "Italie", "Luxembourg", "Pays Bas", + "Pologne", "Portugal" + } + + # Initialize session once + session = MondialRelaySession(config_path=config_path) + + with open(input_csv, newline='', encoding='utf-8') as fin, \ + open(sent_csv, 'w', newline='', encoding='utf-8') as f_sent, \ + open(not_sent_csv, 'w', newline='', encoding='utf-8') as f_not: + + reader = csv.DictReader(fin) + fieldnames = reader.fieldnames + ["Status"] + sent_writer = csv.DictWriter(f_sent, fieldnames=fieldnames) + not_sent_writer = csv.DictWriter(f_not, fieldnames=fieldnames) + + sent_writer.writeheader() + not_sent_writer.writeheader() + + for row in reader: + country = row.get("Pays", "").strip() + email = row.get("E-mail", row.get("Email", "")) + weight = float(row.get("poids", row.get("Poids", 1))) + comment = row.get("Commentaire", row.get("commentaire", "")) + + if country in valid_countries: + try: + ok = session.send_parcel( + country=country, + weight_kg=weight, + email=email, + comment=comment + ) + row["Status"] = "sent" if ok else "not sent" + if ok: + print(f"Email sent to {email}") + sent_writer.writerow(row) + else: + print(f"Email FAILED TO SENT to {email} for country {country}") + not_sent_writer.writerow(row) + except Exception as e: + row["Status"] = f"error: {e}" + not_sent_writer.writerow(row) + else: + row["Status"] = "not sent" + not_sent_writer.writerow(row) + + session.quit() + print(f"Processing complete. Sent: {sent_csv}, Not sent: {not_sent_csv}") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Process Mondial Relay shipments from a CSV file") + parser.add_argument('input_csv', help='Path to the input CSV file') + parser.add_argument('--sent_csv', default='sent.csv', help='Path for output file of sent parcels') + parser.add_argument('--not_sent_csv', default='not_sent.csv', help='Path for output file of not sent parcels') + parser.add_argument('--config', default=MondialRelaySession.CONFIG_FILE, help='Path to config.ini for credentials') + args = parser.parse_args() + process_csv( + input_csv=args.input_csv, + sent_csv=args.sent_csv, + not_sent_csv=args.not_sent_csv, + config_path=args.config + )