[Qgis_fr] QGIS: Automatiser l Import de plusieurs CSV

Message posté par : Perl21 ---------------------------------------- Tu peux le faire assez simplement avec PyQGIS ou en utilisant les macros de projet dans QGIS. Il n’existe pas de plugin “clé en main” qui gère directement l’import et la mise à jour automatique de CSV, mais deux solutions pratiques s’offrent à toi : 1. Ajouter tes CSV comme couches “texte délimité” QGIS lit les fichiers CSV directement comme une couche. Si tes fichiers portent toujours le même nom et sont écrasés chaque jour (exemple : donnees.csv), alors il suffit de l’ajouter une fois au projet. À chaque mise à jour du fichier, la couche est rafraîchie automatiquement : tu n’as donc plus besoin de réimporter manuellement. 2. Automatiser via un script PyQGIS Si tu reçois plusieurs CSV avec des noms qui changent (par exemple datés), tu peux écrire un petit script en Python. Ce script va : parcourir un dossier, importer tous les CSV trouvés, mettre à jour les couches existantes si elles ont changé. Tu peux ensuite le lancer automatiquement à l’ouverture de ton projet en l’ajoutant dans les macros du projet. Voici un script PyQGIS prêt à l’emploi pour importer et mettre à jour automatiquement tous les CSV d’un dossier dans ton projet QGIS. Colle-le dans Projet → Propriétés → Macros → Macros Python du projet, puis accepte les macros à l’ouverture. # les hashtags sont les paramètres à adapter CSV_DIR = r"C:/chemin/vers/tes_csv" # Dossier des CSV RECURSIVE = True # Inclure les sous-dossiers REFRESH_MINUTES = 5 # Rechargement auto (minutes) DEFAULT_EPSG = 2154 # 4326 (WGS84), 2154 (Lambert-93), etc. DELIMITER = "," # "," ou ";" (si ; mettre DELIMITER = "%3B") ENCODING = "UTF-8" # "UTF-8" ou "ISO-8859-1" selon tes fichiers POSSIBLE_X = ["x", "lon", "longitude", "easting"] POSSIBLE_Y = ["y", "lat", "latitude", "northing"] # === Rien à modifier en dessous (sauf besoin avancé) === from qgis.PyQt.QtCore import QTimer from qgis.PyQt.QtWidgets import QAction from qgis.core import QgsProject, QgsVectorLayer import os, io, csv _timer = None _action = None _seen_mtime = {} # path -> mtime suivi def _iter_csv_files(): if not os.path.isdir(CSV_DIR): return if RECURSIVE: for root, _, files in os.walk(CSV_DIR): for f in files: if f.lower().endswith(".csv"): yield os.path.join(root, f) else: for f in os.listdir(CSV_DIR): if f.lower().endswith(".csv"): yield os.path.join(CSV_DIR, f) def _read_header(path): # Tente lecture robuste (UTF-8-sig), retombe sur ENC0DING si besoin try: with io.open(path, "r", encoding="utf-8-sig", errors="ignore") as f: r = csv.reader(f, delimiter="," if DELIMITER=="," else ";") return next(r, None) except Exception: try: with io.open(path, "r", encoding=ENCODING, errors="ignore") as f: r = csv.reader(f, delimiter="," if DELIMITER=="," else ";") return next(r, None) except Exception: return None def _find_xy_cols(header): if not header: return (None, None) cols = [c.strip().lower() for c in header] x = next((c for c in POSSIBLE_X if c in cols), None) y = next((c for c in POSSIBLE_Y if c in cols), None) return x, y def _layer_name(path): return os.path.splitext(os.path.basename(path))[0] def _get_layer_by_name(name): for lyr in QgsProject.instance().mapLayers().values(): if lyr.name() == name: return lyr return None def _csv_uri(path, header): base = f"file:///{path.replace(os.sep,'/')}" params = [ f"encoding={ENCODING}", f"delimiter={DELIMITER if DELIMITER != ';' else '%3B'}", 'quote="', "escape=\\", "decimal=." ] x, y = _find_xy_cols(header) if x and y: params += [f"xField={x}", f"yField={y}", "geomType=point", f"crs=EPSG:{DEFAULT_EPSG}", "detectTypes=yes"] else: params += ["geomType=none", "detectTypes=yes"] return base + "?" + "&".join(params) def _add_or_update(path): header = _read_header(path) if not header: print(f"[CSV] Entête illisible: {path}") return False uri = _csv_uri(path, header) name = _layer_name(path) lyr = _get_layer_by_name(name) if lyr and isinstance(lyr, QgsVectorLayer): ok = lyr.setDataSource(uri, name, "delimitedtext") if ok: lyr.triggerRepaint() print(f"[CSV] MAJ: {name}") return True else: QgsProject.instance().removeMapLayer(lyr.id()) new_lyr = QgsVectorLayer(uri, name, "delimitedtext") if new_lyr.isValid(): QgsProject.instance().addMapLayer(new_lyr) print(f"[CSV] Ajout: {name}") return True else: print(f"[CSV] Échec chargement: {path}") return False def _sync_all(): changed = False for path in _iter_csv_files() or []: try: mtime = os.path.getmtime(path) if path not in _seen_mtime or _seen_mtime[path] < mtime: if _add_or_update(path): _seen_mtime[path] = mtime changed = True except Exception as e: print(f"[CSV] Erreur {path}: {e}") if not changed: print("[CSV] Rien à mettre à jour") def _start_timer(): global _timer if _timer is None: _timer = QTimer() _timer.timeout.connect(_sync_all) _timer.start(max(1, int(REFRESH_MINUTES)) * 60 * 1000) def _stop_timer(): if _timer: _timer.stop() def _add_toolbar_button(): global _action if _action is None: _action = QAction("Recharger CSV", None) _action.setToolTip("Forcer l'import/maj des CSV du dossier") _action.triggered.connect(_sync_all) iface.addToolBarIcon(_action) # ====== Macros QGIS ====== def openProject(): if not os.path.isdir(CSV_DIR): print(f"[CSV] Dossier introuvable: {CSV_DIR}") return _add_toolbar_button() _sync_all() _start_timer() print(f"[CSV] Auto-sync actif sur {CSV_DIR} (toutes les {REFRESH_MINUTES} min)") def saveProject(): pass def closeProject(): _stop_timer() print("[CSV] Auto-sync arrêté") ---------------------------------------- Le message est situé https://georezo.net/forum/viewtopic.php?pid=374564#p374564 Pour y répondre : qgis_fr(a)ml.georezo.net ou reply de votre messagerie Pour vous désabonner connectez-vous sur le forum puis Profil / Abonnement -- Association GeoRezo - le portail géomatique https://georezo.net