213 lines
8.5 KiB
Python
213 lines
8.5 KiB
Python
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
|
|
)
|