Bon voici une MàJ du programme, je vais détailler un peu ce que j'ai fait. Je suis reparti de mon script précédent mais le but c'est d'avoir un truc qui fonctionne en arrière plan donc on vire prettytable et on exécute selenium en mode headless (sans interface).
options = Options()
options.headless = True
driver = webdriver.Firefox(options=options)
driver.get(url)
article_body = driver.find_element_by_class_name('feed-grid')
articles = article_body.find_elements_by_class_name("feed-grid__item")
Ensuite, je créé une liste qui me servira à recevoir tous mes items et qui seront représentés via des dictionnaires (facile à écrire dans un fichier csv avec dictWriter. Au passage, j'en profite pour ajouter le nom du fichier, le timestamp et surtout l'ID qui nous servira de "clé primaire" au moment de récupérer tous les IDs de tous les items pour savoir si on a déjà stocké le fichier dans le csv. L'ID est facile à récupérer car il est présent dans l'url.
list_of_items = []
for article in articles:
try:
user, price, size, brand = [elem.text for elem in article.find_elements_by_class_name('Text_text__QBn4-')]
link = article.find_element_by_class_name('c-box__overlay').get_attribute('href')
item = dict()
user = user.strip()
price = price.strip()
size = size.strip()
brand = brand.strip()
link = link.strip()
name = " ".join(link.split("/")[-1].split("-")[1:]).title()
id_ = link.split('/')[-1].split('-')[0]
item['user'] = user
item['price'] = price
item['size'] = size
item['name'] = name
item['brand'] = brand
item['link'] = link
item['id'] = id_
item['datetime'] = datetime.now().isoformat()
list_of_items.append(item)
Ensuite, on écrit quelques fonctions (je ne mets pas l'intégralité du code de chaque fonction pour que ça reste lisible). Une fonction qui va récupérer tous les ids du fichier csv (on renvoie un set car plus rapide qu'on liste pour tester si un élément est présent ou non) :
def create_csv_file(filename: str, fieldnames: List[str]) -> None:
ids = set()
with open(filename, mode="r") as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
ids.add(row['id'])
return ids
Une autre fonction qui écrit l'item dans le fichier (pas très efficace car ça implique d'ouvrir, écrire, fermer le fichier pour chaque item mais on aura peu d'I/O dans notre programme donc on peut se le permettre et le programme sera plus lisible) :
def add_item_to_file(item: Dict, filename: str, fieldnames: List[Dict]) -> None:
with open(filename, mode="a", encoding="utf-8") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writerow(item)
À ce stade on a toutes nos fonctions, on n'a plus qu'à assembler tout ça dans une fonction main et de l'appeler par la suite :
def main():
FILENAME = "vinted.csv"
FIELDNAMES = ['id', 'user', 'price', 'size', 'name', 'brand', 'link', 'datetime']
URL = "https://www.vinted.fr/vetements?size_id[]=207&catalog[]=79&brand_id[]=304&price_to=20&status[]=1&status[]=6&status[]=2&order=newest_first"
# Création du fichier dans lequel on stockera nos articles
if not os.path.exists(FILENAME):
create_csv_file(FILENAME, FIELDNAMES)
logger.info(f"Creating the file `{FILENAME}`.")
# Récupération de tous les ID
set_of_ids = get_all_ids(FILENAME)
logger.info(f"Recovery of all the IDs of the `{FILENAME}` file.")
new_items = []
for item in get_all_items(URL):
if item['id'] not in set_of_ids: # Si l'ID de l'article n'est pas dans le fichier, on ne l'a pas encore scrapé.
add_item_to_file(item, FILENAME, FIELDNAMES)
logger.info(f"New item, added to the file `{FILENAME}` : `{item}`.")
new_items.append(item)
else:
logger.info(f"Item already saved in the file `{FILENAME}` : `{item}`.")
if new_items: # S'il y a des nouveaux items, on envoie une notification.
logger.info("New items saved, preparation of the telegram notification.")
# Ici, on place la fonction qui envoie une notification via telegram.
# send_telegram_notification(new_items, user, password, channel)
# On vide la liste des nouveaux items
new_items.clear()
Forcément, on ne fait plus de print étant donné que le programme tournera en arrière plan. On utilise logger configuré au début :
logging.basicConfig(format="%(asctime)s %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
level=logging.INFO,
filename="vinted.log")
logger = logging.getLogger("vinted_scraper")
Ensuite, on doit exécuter le script tous les jours pour scraper les derniers articles. Le module sched de Python est assez complexe à utiliser donc j'utilise schedule de Dan Brader (le fondateur du site RealPython). Le module s'utilise relativement facilement, il suffit d'appeler la fonction qu'on souhaite exécuter et de fixer la périodicité :
if __name__ == "__main__":
# On exécute le script directement (test et debugging)
# main()
# On exécute le script tous les jours à 18H00.
schedule.every().day.at("18:00").do(main)
# On exécute le script toutes les heures.
# schedule.every(60).minutes.do(main)
while True:
schedule.run_pending()
time.sleep(1)
Voilà, c'est terminé, tu n'as plus qu'à intégrer ta notification telegram dans la fonction main. Ensuite il faudra lancer le programme en arrière plan et le détacher du terminal. Sur Linux, on fait comme ça, sous windows je sais pas trop mais bon j'imagine que c'est faisable facilement.
nohup python3 vinted.py &
Le code complet :
# https://www.jeuxvideo.com/forums/42-47-64620468-1-0-1-0-bot-mon-bot-est-il-realisable.htm
# vinted.py
import csv
import logging
import os
import time
from datetime import datetime
from typing import Dict, List, Set
import schedule
from selenium import webdriver
from selenium.webdriver.firefox.options import Options
logging.basicConfig(format="%(asctime)s %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
level=logging.INFO,
filename="vinted.log")
logger = logging.getLogger("vinted_scraper")
def get_all_items(url: str) -> List[Dict]:
"""Function that returns a dictionary list of all items on the url page.
Args:
url (str): URL of the page to scrap.
Returns:
List: return a list of all items
"""
options = Options()
options.headless = True
driver = webdriver.Firefox(options=options)
driver.get(url)
article_body = driver.find_element_by_class_name('feed-grid')
articles = article_body.find_elements_by_class_name("feed-grid__item")
list_of_items = []
for article in articles:
try:
user, price, size, brand = [elem.text for elem in article.find_elements_by_class_name('Text_text__QBn4-')]
link = article.find_element_by_class_name('c-box__overlay').get_attribute('href')
item = dict()
user = user.strip()
price = price.strip()
size = size.strip()
brand = brand.strip()
link = link.strip()
name = " ".join(link.split("/")[-1].split("-")[1:]).title()
id_ = link.split('/')[-1].split('-')[0]
item['user'] = user
item['price'] = price
item['size'] = size
item['name'] = name
item['brand'] = brand
item['link'] = link
item['id'] = id_
item['datetime'] = datetime.now().isoformat()
list_of_items.append(item)
except ValueError:
pass
driver.quit()
return list_of_items
def get_all_ids(filename: str) -> Set:
"""Function that returns the set of all ids of all items saved in the file.
Args:
filename (str): File where items are saved.
Returns:
Set: Set of ids.
"""
ids = set()
with open(filename, mode="r") as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
ids.add(row['id'])
return ids
def create_csv_file(filename: str, fieldnames: List[str]) -> None:
"""Function that creates the backup file.
Args:
filename (str): File where items are saved.
fieldnames (List): List containing item headers.
"""
with open(filename, mode="w", encoding="utf-8") as csvfile:
writer = csv.writer(csvfile)
writer.writerow(fieldnames)
def add_item_to_file(item: Dict, filename: str, fieldnames: List[Dict]) -> None:
"""Function that adds an item to the file.
Args:
item (Dict): Dictionary representing an item.
filename (str): File where items are saved.
fieldnames (List): List containing item headers.
"""
with open(filename, mode="a", encoding="utf-8") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writerow(item)
def main():
FILENAME = "vinted.csv"
FIELDNAMES = ['id', 'user', 'price', 'size', 'name', 'brand', 'link', 'datetime']
URL = "https://www.vinted.fr/vetements?size_id[]=207&catalog[]=79&brand_id[]=304&price_to=20&status[]=1&status[]=6&status[]=2&order=newest_first"
# Création du fichier dans lequel on stockera nos articles
if not os.path.exists(FILENAME):
create_csv_file(FILENAME, FIELDNAMES)
logger.info(f"Creating the file `{FILENAME}`.")
# Récupération de tous les ID
set_of_ids = get_all_ids(FILENAME)
logger.info(f"Recovery of all the IDs of the `{FILENAME}` file.")
new_items = []
for item in get_all_items(URL):
if item['id'] not in set_of_ids: # Si l'ID de l'article n'est pas dans le fichier, on ne l'a pas encore scrapé.
add_item_to_file(item, FILENAME, FIELDNAMES)
logger.info(f"New item, added to the file `{FILENAME}` : `{item}`.")
new_items.append(item)
else:
logger.info(f"Item already saved in the file `{FILENAME}` : `{item}`.")
if new_items: # S'il y a des nouveaux items, on envoie une notification.
logger.info("New items saved, preparation of the telegram notification.")
# Ici, on place la fonction qui envoie une notification via telegram.
# send_telegram_notification(new_items, user, password, channel)
# On vide la liste des nouveaux items
new_items.clear()
if __name__ == "__main__":
# On exécute le script directement (test et debugging)
# main()
# On exécute le script tous les jours à 18H00.
schedule.every().day.at("18:00").do(main)
# On exécute le script toutes les heures.
# schedule.every(60).minutes.do(main)
while True:
schedule.run_pending()
time.sleep(1)
Voici une capture des logs, j'ai lancé le programme à 12h04 (premier scraping) puis je l'ai relancé manuellement à 12h25 et il a trouvé 3 nouveaux articles de silvamika (ajoutés 11 minutes avant).
