Un script qui dort en attendant qu'un fichier tombe dans un dossier, c'est exactement ce que fait Watchdog. Pas de polling toutes les deux secondes, pas de cron qui scanne un répertoire à l'aveugle : la bibliothèque s'appuie directement sur le système de fichiers pour te prévenir à l'instant où un fichier est créé, modifié ou supprimé. Tes automatisations réagissent en quelques millisecondes, sans gaspiller de CPU.
La méthode naïve, c'est la boucle : toutes les deux secondes, lister les fichiers d'un dossier, comparer avec la liste précédente, et agir si quelque chose a bougé. Ça marche, mais ça consomme du CPU en permanence, ça rate les fichiers créés puis supprimés entre deux passages, et ça complique le code pour rien. Watchdog délègue ce travail au noyau : inotify sur Linux, FSEvents sur macOS, ReadDirectoryChangesW sur Windows. Ton script ne se réveille que quand il y a réellement du nouveau.
Watchdog est un standard du monde Python. On le retrouve dans les chaînes de facturation, les pipelines de traitement d'images ou les serveurs qui reconstruisent un site dès qu'un article est modifié. Le principe se lit en une minute et le code tient en une vingtaine de lignes.
1. Installe Watchdog
pip install watchdogRien d'autre à configurer : Watchdog est une bibliothèque autonome, sans service à lancer ni dépendance exotique. Un environnement virtuel reste recommandé, comme pour tout projet Python.
2. Le squelette : voir les événements défiler
On commence par le minimum vital. On crée un gestionnaire qui hérite de FileSystemEventHandler, on surcharge la méthode on_created, puis on branche un Observer sur le dossier de ton choix.
import time
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class GestionnaireFichiers(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
print(f"[+] Nouveau fichier : {Path(event.src_path).name}")
def main():
dossier = Path.home() / "inbox"
dossier.mkdir(exist_ok=True)
observateur = Observer()
observateur.schedule(GestionnaireFichiers(), str(dossier), recursive=False)
observateur.start()
print(f"Surveillance de {dossier} ... Ctrl+C pour arreter")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observateur.stop()
observateur.join()
if __name__ == "__main__":
main()Lance ce script, ouvre un second terminal et dépose un fichier dans inbox : son nom s'affiche immédiatement. L'Observer tourne dans son propre thread, ton programme principal reste libre. Ctrl+C pour arrêter proprement.
3. Les quatre événements qui comptent
on_created n'est que le début. FileSystemEventHandler expose quatre méthodes que tu peux surcharger selon le cas :
class GestionnaireFichiers(FileSystemEventHandler):
def on_created(self, event): ... # un fichier apparait
def on_modified(self, event): ... # un fichier change
def on_deleted(self, event): ... # un fichier disparait
def on_moved(self, event): ... # un fichier est renomme ou deplaceChaque event expose src_path (le chemin concerné) et is_directory (vrai si c'est un dossier). Pour on_moved, tu récupères aussi dest_path.
4. Une vraie réaction : trier les fichiers par extension
Afficher des noms, c'est bien. Déclencher une action, c'est mieux. Le script suivant déplace chaque nouveau fichier vers un sous-dossier selon son extension : images, tableaux, documents.
import time
import shutil
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
DOSSIERS = {
".jpg": "images", ".jpeg": "images", ".png": "images",
".csv": "tableaux", ".xlsx": "tableaux",
".pdf": "documents", ".txt": "documents",
}
def attendre_fichier_complet(chemin, patience=5.0, intervalle=0.3):
taille_prec = -1
debut = time.time()
while time.time() - debut < patience:
if not chemin.exists():
return False
taille = chemin.stat().st_size
if taille == taille_prec:
return True
taille_prec = taille
time.sleep(intervalle)
return False
class TrieurFichiers(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
chemin = Path(event.src_path)
cible = DOSSIERS.get(chemin.suffix.lower())
if cible is None:
return
attendre_fichier_complet(chemin)
dossier = chemin.parent / cible
dossier.mkdir(exist_ok=True)
shutil.move(str(chemin), str(dossier / chemin.name))
print(f"Deplace : {chemin.name} -> {cible}/")
def main():
dossier = Path.home() / "inbox"
dossier.mkdir(exist_ok=True)
observateur = Observer()
observateur.schedule(TrieurFichiers(), str(dossier), recursive=False)
observateur.start()
print(f"Surveillance de {dossier} ...")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observateur.stop()
observateur.join()
if __name__ == "__main__":
main()Dépose un CSV, un PNG et un PDF dans inbox : chacun atterrit dans le bon sous-dossier sans que tu lèves le petit doigt.
5. Le piège du fichier à moitié écrit
Sur un gros fichier, on_created peut se déclencher avant la fin de l'écriture. Lis-le trop tôt et tu récupères une version tronquée. La parade : attendre que sa taille se stabilise. C'est le rôle de la fonction attendre_fichier_complet glissée dans le script précédent : elle compare la taille deux fois de suite et rend la main quand plus rien ne bouge. La patience (5 secondes par défaut) évite de bloquer éternellement sur un transfert interrompu.
6. Filtrer avant de réagir
Inutile de traiter tous les fichiers qui passent. PatternMatchingEventHandler restreint la surveillance à certains motifs, et RegexMatchingEventHandler va encore plus loin avec des expressions régulières. Tu gagnes en clarté et tu évites de déclencher tout un pipeline sur un fichier temporaire.
from watchdog.events import PatternMatchingEventHandler
class TrieurFichiers(PatternMatchingEventHandler):
patterns = ["*.csv", "*.xlsx", "*.jpg", "*.png", "*.pdf"]
def on_created(self, event):
chemin = Path(event.src_path)
# ... même logique de tri que dans l'étape 4 ...Remplace FileSystemEventHandler par PatternMatchingEventHandler, définis patterns, et ton handler n'est appelé que si le nom du fichier correspond. Les extensions non listées sont simplement ignorées.
Pour surveiller aussi les sous-dossiers, passe recursive=True à observateur.schedule. Prudence : si ton handler déplace un fichier vers un sous-dossier lui-même surveillé, tu risques de déclencher on_created en boucle. Garde un dossier d'entrée plat tant que tu n'as pas besoin du contraire.
Enfin, protège ta réaction : enveloppe la logique de on_created dans un try/except. Si le traitement d'un fichier plante, l'Observer continue de tourner et les événements suivants sont traités normalement. Une exception non attrapée ne fait pas tomber la boucle de surveillance, mais mieux vaut logger l'erreur plutôt que de la laisser passer en silence.
7. Des idées pour brancher l'IA dessus
Une fois le mécanisme en place, la réaction devient un simple appel de fonction. Quelques pistes concrètes : lancer une reconnaissance de texte sur chaque image qui arrive (pytesseract), nourrir un modèle sur chaque nouveau CSV (Pandas), indexer chaque PDF dans une base vectorielle (ChromaDB), ou notifier un canal via un bot Telegram. Watchdog ne fait qu'une chose, mais il la fait au bon moment : te prévenir pile quand un fichier est prêt.
Watchdog remplace avantageusement la boucle while True qui scanne un dossier en continu : moins de CPU, une latence quasi nulle et un code plus lisible. Le squelette tient en vingt lignes et se déploie partout où un fichier qui tombe doit déclencher un travail. Branche ton traitement dedans, et ton dossier devient une vraie boîte de réception automatique.
Auteur
Krugz
Développeur fullstack
Développeur fullstack qui tente, tant bien que mal, de suivre l'actualité de l'IA au milieu d'une marée d'informations imbuvables et des milliers d'outils et de modèles qui sortent chaque jour.
