Le portail francophone de la géomatique

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é")