#!/usr/bin/env python3 # -*- coding: UTF-8 -*- import os import sys import csv # Ajouter le chemin courant sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) try: import db_adapter except ImportError: print("Content-type: text/plain; charset=UTF-8\n") print("ERROR: Impossible de charger le module db_adapter.py.") sys.exit(1) # Format: [colonnes SELECT, LEFT JOIN clause, champ WHERE, type, étiquettes] base_construct_requetes = { "nomGite": ("nomGite", "LEFT JOIN nomgite on gite.nGiteId = nomgite.nGiteId", "nomgite.nGiteId", 0, "Nom du gîte:"), "typeGite": ("typeGite", "LEFT JOIN typegite ON gite.tGiteId = typegite.tGiteId", "typegite.tGiteId", 0, "Type de gîte:"), "alteration": ("alteration", "LEFT JOIN alteration ON gite.alterationId = alteration.alterationId", "alteration.alterationId", 0, "Altération:"), "mineralisation": ("mineralisation", "LEFT JOIN mineralisation ON gite.mineralisationId = mineralisation.mineralisationId", "mineralisation.mineralisationId", 0, "Type de minéralisation:"), "rocheEncaissante": ("rocheEncaissante", "LEFT JOIN rocheencaissante ON gite.rocheEncId = rocheencaissante.rocheEncId", "rocheencaissante.rocheEncId", 0, "Roche encaissante:"), "identificateur": ("identificateur", "LEFT JOIN origine on gite.origId = origine.origId", "", 0, "Origine:"), "tiroir": ("tiroir", "", "tiroir", 0, "Tiroir:"), "numeroGm": ("numeroGm", "", "numeroGm", 0, "Numéro:"), "numero": ("numeroGm,numeroGmRef,ancienNumero,autreNumero", "", "", 0, "NuméroGm:NuméroGmRef:Ancien numéro:Autre numéro:"), "structure": ("structure,categorieStructure", "LEFT JOIN relationstructure on gite.giteId = relationstructure.giteId " "LEFT JOIN structure on structure.structureId = relationstructure.structureId " "LEFT JOIN categoriestructure on categoriestructure.catStructureId = relationstructure.catStructureId", "relationstructure.structureId", 1, "Structure:Catégorie structure:"), "texture": ("texture,categorieTexture", "LEFT JOIN relationtexture on gite.giteId = relationtexture.giteId " "LEFT JOIN texture on texture.textureId = relationtexture.textureId " "LEFT JOIN categorietexture on categorietexture.catTextureId = relationtexture.catTextureId", "relationtexture.textureId", 1, "Texture:Catégorie texture:"), "mineralisation_principale": ("mineral,categorieMineral", "LEFT JOIN relationmineral on gite.giteId = relationmineral.giteId " "LEFT JOIN mineral on mineral.mineralId = relationmineral.mineralId " "LEFT JOIN categoriemineral on categoriemineral.catMineralId = relationmineral.catMineralId", "categorieMineral = 'Minéralisation Principale' AND relationmineral.mineralId", 1, "Minéralisation:Catégorie minéralisation:"), "localisation_geo": ("region,province,pays", "LEFT JOIN relationregion on gite.giteId = relationregion.giteId " "LEFT JOIN region on region.regionId = relationregion.regionId " "LEFT JOIN province on province.provinceId = relationregion.provinceId " "LEFT JOIN pays on pays.paysId = relationregion.paysId", "region.regionId", 1, "Région:Province:Pays:"), "pays": ("pays", "LEFT JOIN relationregion on gite.giteId = relationregion.giteId " "LEFT JOIN pays on pays.paysId = relationregion.paysId", "pays.paysId", 0, "Pays:"), "hyperlien": ("hyperlien", "LEFT JOIN hyperlien ON gite.giteId = hyperlien.giteId", "", 1, "Hyperlien:"), "photo": ("photoid", "LEFT JOIN photo ON gite.giteId = photo.giteId", "", 2, "Photo:") } def genere_requete_tmp(form, dict_selection): """Génère la requête SQL pour remplir la table temporaire 'tmp' (critères WHERE).""" req_tmp = "INSERT into tmp SELECT DISTINCT gite.giteId FROM gite" search_query = form.getvalue("search_query", "") if isinstance(search_query, list): search_query = search_query[0] if search_query else "" search_query = search_query.strip() # 1. Joindre les tables nécessaires pour la recherche textuelle globale if search_query: search_joins = [ "LEFT JOIN nomgite on gite.nGiteId = nomgite.nGiteId", "LEFT JOIN typegite ON gite.tGiteId = typegite.tGiteId", "LEFT JOIN relationregion on gite.giteId = relationregion.giteId", "LEFT JOIN pays on pays.paysId = relationregion.paysId" ] for join_clause in search_joins: for join in join_clause.strip().split("LEFT JOIN"): join = join.strip() if not join: continue full_join = f"LEFT JOIN {join}" table_name = join.split()[0] if f"JOIN {table_name} " not in req_tmp and f"JOIN {table_name}\n" not in req_tmp and not req_tmp.endswith(f"JOIN {table_name}"): req_tmp = req_tmp + ' ' + full_join # 2. Génération des LEFT JOIN pour les clauses WHERE (sans doublons de table) for key in dict_selection: if key in base_construct_requetes: join_clause = base_construct_requetes[key][1] if join_clause: for join in join_clause.strip().split("LEFT JOIN"): join = join.strip() if not join: continue full_join = f"LEFT JOIN {join}" table_name = join.split()[0] if f"JOIN {table_name} " not in req_tmp and f"JOIN {table_name}\n" not in req_tmp and not req_tmp.endswith(f"JOIN {table_name}"): req_tmp = req_tmp + ' ' + full_join # 3. Génération des clauses WHERE where_clauses = [] if search_query: escaped_search = search_query.replace("'", "''") or_clauses = [ f"nomgite.nomGite LIKE '%{escaped_search}%'", f"typegite.typeGite LIKE '%{escaped_search}%'", f"pays.pays LIKE '%{escaped_search}%'" ] if search_query.isdigit(): or_clauses.append(f"gite.numeroGm = {int(search_query)}") where_clauses.append("(" + " OR ".join(or_clauses) + ")") for key in dict_selection: if key in base_construct_requetes: field_name = base_construct_requetes[key][2] value = form.getvalue(key) escaped_val = value.replace("'", "''") where_clauses.append(f"{field_name} = '{escaped_val}'") if where_clauses: req_tmp += " WHERE " + " AND ".join(where_clauses) if where_clauses: req_tmp += " AND gite.giteId > 0" else: req_tmp += " WHERE gite.giteId > 0" return req_tmp def genere_requete_0(dict_cb): """Génère la requête SELECT pour les données uniques par échantillon (Type 0).""" etiq = "No échantillon:" req_0 = "SELECT gite.giteId" for key in dict_cb: if key in base_construct_requetes: if base_construct_requetes[key][3] == 0: req_0 += ',' + base_construct_requetes[key][0] etiq += base_construct_requetes[key][4] req_0 += ' ' + "FROM tmp LEFT JOIN gite on tmp.giteId = gite.giteId" for key in dict_cb: if key in base_construct_requetes: if base_construct_requetes[key][3] == 0: join_clause = base_construct_requetes[key][1] if join_clause: for join in join_clause.strip().split("LEFT JOIN"): join = join.strip() if not join: continue full_join = f"LEFT JOIN {join}" table_name = join.split()[0] if f"JOIN {table_name} " not in req_0 and f"JOIN {table_name}\n" not in req_0 and not req_0.endswith(f"JOIN {table_name}"): req_0 += ' ' + full_join return etiq, req_0 def genere_requete_m(key): """Génère la requête SELECT pour les relations multiples (Type 1 et 2).""" etiq = "No échantillon:" req_m = (f"SELECT gite.giteId, {base_construct_requetes[key][0]} " f"FROM tmp LEFT JOIN gite on tmp.giteId = gite.giteId ") if base_construct_requetes[key][1] != "": req_m += base_construct_requetes[key][1] etiq += base_construct_requetes[key][4] return etiq, req_m def run_report(form): selection_done = False cb_done = False dict_selection = {} # Critères de sélection (listes déroulantes) dict_cb = {} # Champs à afficher (checkboxes) # 1. Traitement des valeurs du formulaire search_query_present = False for key in form.keys(): if key in ("export_type", "search_query"): continue if key.endswith("_cb"): cb_done = True col_name = key[:-3] dict_cb[col_name] = 1 else: val = form.getvalue(key) if val != '*': selection_done = True dict_selection[key] = val # Vérification de la présence d'une recherche textuelle libre search_val = form.getvalue("search_query", "") if isinstance(search_val, list): search_val = search_val[0] if search_val else "" if search_val.strip(): search_query_present = True # 2. Validation if not cb_done or (not selection_done and not search_query_present): # Envoi de l'en-tête texte/csv d'erreur sys.stdout.write("Content-type: text/csv; charset=UTF-8\n\n") sys.stdout.write("ERROR,Champs ou criteres de selection insuffisants.\n") return # 3. Connexion à la base conn, cursor, is_sqlite = db_adapter.connect_database() # 4. Requête et insertion dans la table temporaire try: req_tmp = genere_requete_tmp(form, dict_selection) db_adapter.setup_temporary_table(cursor, is_sqlite, req_tmp) except Exception as err: sys.stdout.write("Content-type: text/csv; charset=UTF-8\n\n") sys.stdout.write(f"SQL_ERROR,Initialisation de la table temporaire echouee: \"{err}\"\n") cursor.close() conn.close() return # 5. Extraction des données de Type 0 (uniques par échantillon) try: etiq_0, req_0 = genere_requete_0(dict_cb) etiq_0_list = [lbl for lbl in etiq_0.split(":") if lbl] # Enlever les libellés vides cursor.execute(req_0) allRecords_0 = cursor.fetchall() except Exception as err: sys.stdout.write("Content-type: text/csv; charset=UTF-8\n\n") sys.stdout.write(f"SQL_ERROR,Requete de donnees principales echouee: \"{err}\"\n") cursor.close() conn.close() return # 6. Extraction des données de Type M (relations multiples) allRecords_multi = {} etiq_multi = {} try: for key in dict_cb: if key in base_construct_requetes: type_requete = base_construct_requetes[key][3] if type_requete in (1, 2): etiq_multi[key], req_m = genere_requete_m(key) cursor.execute(req_m) allRecords_multi[key] = cursor.fetchall() except Exception as err: sys.stdout.write("Content-type: text/csv; charset=UTF-8\n\n") sys.stdout.write(f"SQL_ERROR,Requete de donnees multiples ({key}) echouee: \"{err}\"\n") cursor.close() conn.close() return # Fermeture de la connexion à la base cursor.close() conn.close() # 7. Indexation / Groupement des résultats multiples par giteId grouped_multi = {} for key in allRecords_multi: grouped_multi[key] = {} for record in allRecords_multi[key]: giteId = record[0] if giteId not in grouped_multi[key]: grouped_multi[key][giteId] = [] grouped_multi[key][giteId].append(record) # 8. --- ENVOI DES EN-TÊTES DE TÉLÉCHARGEMENT --- sys.stdout.write("Content-type: text/csv; charset=UTF-8\n") sys.stdout.write("Content-Disposition: attachment; filename=export_gites.csv\n\n") sys.stdout.flush() # Initialisation du writer CSV de Python writer = csv.writer(sys.stdout, lineterminator='\n') # Écriture de la première ligne d'en-tête (Champs Type 0) writer.writerow(etiq_0_list) # Écriture des données for record_0 in allRecords_0: giteId = record_0[0] # Ligne de données principales (Type 0) writer.writerow(record_0) # Lignes secondaires de Type 1 (multiples) pour cet échantillon for key in dict_cb: if key in base_construct_requetes and base_construct_requetes[key][3] == 1: if key in grouped_multi and giteId in grouped_multi[key]: records = grouped_multi[key][giteId] # En-tête de la section multiple etiq_list = [lbl for lbl in etiq_multi[key].split(":") if lbl] writer.writerow(etiq_list) # Données associées for record_multi in records: writer.writerow(record_multi) # Note : Les photos (Type 2) ne sont pas exportables en format CSV brut, # on les omet donc dans le fichier tableur. sys.stdout.flush()