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 )