Upload 8 files

app/database.py

@@ -1,11 +1,26 @@
 import os
-from typing import List, Dict
+from typing import List, Dict, Any
 import duckdb
+import faiss
 import pandas as pd
 from huggingface_hub import hf_hub_download
+from sentence_transformers import SentenceTransformer, CrossEncoder
+import torch
+from datasets import load_dataset
+from dotenv import load_dotenv
+import pyarrow as pa
+import pyarrow.compute as pc
 
 # Initialisations
+load_dotenv()
+HF_TOKEN = os.getenv('HF_TOKEN')
+
 REPO_ID = "Loren/articles_database"
+FAISS_REPO_ID = "Loren/articles_faiss"
+FAISS_INDEX_FILE = "faiss_index.bin"
+MODEL_NAME = "intfloat/multilingual-e5-small"
+CROSS_ENCODER_NAME = "jinaai/jina-reranker-v2-base-multilingual"
+
 cache_dir = "/tmp"
 os.makedirs(cache_dir, exist_ok=True)
 # Rediriger le cache HF globalement
@@ -38,6 +53,32 @@ con.execute(f"CREATE VIEW articles AS SELECT * FROM parquet_scan('{articles_parq
 con.execute(f"CREATE VIEW tags AS SELECT * FROM parquet_scan('{tags_parquet}')")
 con.execute(f"CREATE VIEW tag_article AS SELECT * FROM parquet_scan('{tag_article_parquet}')")
 
+# Téléchargement des fichiers de la base faiss depuis le dataset Hugging Face
+hf_faiss_index = hf_hub_download(
+    repo_id=FAISS_REPO_ID,
+    filename=FAISS_INDEX_FILE,
+    repo_type="dataset",
+    token=HF_TOKEN,
+    cache_dir=cache_dir
+)
+
+# Chargement de l’index FAISS
+faiss_index = faiss.read_index(hf_faiss_index)
+
+# Téléchargement des metadatas Faiss depuis le dataset Hugging Face
+dataset = load_dataset(FAISS_REPO_ID, split="train")
+arrow_table = dataset.data
+
+# Creation du Sentence transformer model
+device = "cuda" if torch.cuda.is_available() else "cpu"
+print(f"*** Device: {device}")
+model = SentenceTransformer(MODEL_NAME, device=device)
+
+# Création du cross-encoder
+cross_encoder = CrossEncoder(CROSS_ENCODER_NAME, device=device,
+                             trust_remote_code=True)
+
+
 # Fonctions d'accès aux données
 
 def fetch_tags() -> List[str]:
@@ -108,4 +149,95 @@ def fetch_articles_by_tags(tags: List[str]) -> List[Dict]:
         result = con.execute(query, tags).fetchdf()
         return {"status": "ok", "result": result.to_dict(orient="records")}
     except Exception as e:
-        return {"status": "error", "code": type(e).__name__, "message": str(e)}
+        return {"status": "error", "code": type(e).__name__, "message": str(e)}
+
+def fetch_query_results(query: str, k_model: int = 10, k_cross: int = 5) -> Dict[str, Any]:
+    """
+    Exécute une requête de recherche sémantique avec FAISS, puis rerank avec un cross-encoder
+    et retourne les meilleurs passages enrichis avec des métadonnées provenant de DuckDB.
+
+    Paramètres
+    ----------
+    query : str
+        La requête texte fournie par l'utilisateur.
+    k_model : int, optionnel (défaut = 10)
+        Nombre de résultats les plus proches à récupérer depuis l'index FAISS.
+    k_cross : int, optionnel (défaut = 5)
+        Nombre de résultats finaux à conserver après reranking avec le cross-encoder.
+
+    Retour
+    ------
+    Dict[str, Any]
+        Un dictionnaire contenant :
+        - status : "ok" si succès, sinon "error"
+        - result : liste de résultats (si succès)
+        - code et message : informations d'erreur (si échec)
+    """
+    if not query:
+        return {"status": "error", "code": "no_query", "message": "Aucun query fourni."}
+    try:
+        query_vec = model.encode(["query: "+query], convert_to_numpy=True, normalize_embeddings=True)
+        distances, indices = faiss_index.search(query_vec, k_model)
+
+        # Résultats FAISS
+        faiss_ids_list = indices[0].tolist()
+        distances_list = distances[0].tolist()
+
+        # Filtrer Arrow sur les IDs trouvés
+        filtered_table = arrow_table.filter(
+            pc.is_in(arrow_table['faiss_id'],
+                     value_set=pa.array(faiss_ids_list))
+        )
+
+        # Convertir Arrow → pandas pour ajouter la distance
+        df = filtered_table.to_pandas()
+
+        # Ajouter la distance en gardant l'ordre faiss_ids_list
+        distance_map = dict(zip(faiss_ids_list, distances_list))
+        df["distance"] = df["faiss_id"].map(distance_map)
+
+        # Cross-encoder
+        top_passages = df["chunk_text"].tolist()
+        cross_input = [(query, p) for p in top_passages]
+        cross_scores = cross_encoder.predict(cross_input)
+
+        # Rerank
+        df["cross_score"] = cross_scores
+        df = df.sort_values(by="cross_score", ascending=False)
+
+        # Garder top k_cross
+        df_top = df.head(k_cross)
+
+        # Enregistrer dans DuckDB
+        con.register("faiss_tmp", df_top)
+
+        sql = """
+        SELECT
+            f.faiss_id,
+            f.document_id,
+            f.distance,
+            f.cross_score,
+            f.chunk_text,
+            a.article_title,
+            CASE WHEN a.article_online
+                THEN a.article_url
+                ELSE 'Article unavailable' END AS url,
+            STRING_AGG(t.tag_name, ', ') AS tags
+        FROM faiss_tmp f
+        JOIN articles a ON f.document_id = a.article_id
+        JOIN tag_article ta ON a.article_id = ta.article_id
+        JOIN tags t ON ta.tag_id = t.tag_id
+        WHERE (LENGTH(article_text) - LENGTH(REPLACE(article_text, ' ', '')) + 1) >= 100
+        GROUP BY f.faiss_id, f.document_id, f.distance, f.cross_score, f.chunk_text,
+              a.article_title, a.article_online, a.article_url
+        ORDER BY AVG(f.cross_score)
+        """
+
+        duck_res = con.execute(sql).fetchdf()
+
+        # Liste finale de dictionnaires
+        list_result = duck_res.to_dict(orient="records")
+
+        return {"status": "ok", "result": list_result}
+    except Exception as e:
+        return {"status": "error", "code": type(e).__name__, "message": str(e)}

app/main.py

@@ -1,5 +1,5 @@
 from fastapi import FastAPI, Query
-from typing import List
+from typing import List, Optional, Dict, Any
 from app import database
 from fastapi.middleware.cors import CORSMiddleware
 
@@ -92,3 +92,34 @@ def get_articles_with_tags(tags: List[str] = Query(..., description="Liste des t
             return dict_result
     except Exception as e:
         return {"status": "error", "code": type(e).__name__, "message": str(e)}
+
+
+@app.get("/get_query_results")
+def get_query_results(query: str = Query(..., description="Requête de recherche textuelle"),
+                      k_model: int = Query(10, description="Nombre de candidats retournés par FAISS"),
+                      k_cross: int = Query(5, description="Nombre de résultats conservés après reranking")
+                      ) -> Dict[str, Any]:
+    """
+    Récupère les résultats d'une requête en utilisant deux modèles de recherche.
+
+    Args:
+        query (str): La requête utilisateur pour laquelle récupérer les résultats.
+        k_model (int, optional): Nombre de résultats à retourner pour le modèle principal. Par défaut à 10.
+        k_cross (int, optional): Nombre de résultats à retourner pour le modèle croisé. Par défaut à 5.
+
+    Returns:
+        Dict[str, Any]: Un dictionnaire contenant soit les résultats de la requête, soit les informations d'erreur.
+
+    Notes:
+        - L'appel de cet endpoint utilise la fonction `fetch_query_result` pour obtenir les résultats.
+        - En cas de problème lors du traitement de la requête, un message d'erreur détaillé est retourné.
+    """
+    try:
+        dict_result = database.fetch_query_results(query, k_model, k_cross)
+        if dict_result["status"] == "ok":
+            return {"status": "ok",
+                    "results": dict_result["result"]}
+        else:
+            return dict_result
+    except Exception as e:
+        return {"status": "error", "code": type(e).__name__, "message": str(e)}

requirements.txt

@@ -1,7 +1,13 @@
-fastapi==0.109.2
-uvicorn[standard]==0.23.2
-numpy==1.26.4
-pandas==2.1.1
-pyarrow==12.0.1
-huggingface_hub==0.35.3
-duckdb==1.4.0
+fastapi==0.109.2
+uvicorn[standard]==0.23.2
+numpy==1.26.4
+pandas==2.1.1
+pyarrow==12.0.1
+huggingface_hub==0.35.3
+duckdb==1.4.0
+faiss-cpu==1.12.0
+langchain==0.3.27
+torch
+sentence-transformers
+pyarrow==22.0.0
+python-dotenv

script/1_create_dataset.py

@@ -0,0 +1,187 @@
+##############################################################################################
+### Script de création de la base de données articles à partir du fichier parquet,
+### correspondant au jeu d'essai : https://www.kaggle.com/code/fabiochiusano/medium-articles-simple-data-analysis
+### Téléchargement du csv puis conversion en Parquet avec compression snappy :
+### df = pd.read_csv("medium_articles.csv")
+### df.to_parquet("medium_articles.parquet", engine="pyarrow", compression="snappy")
+###
+### Le fichier a été uploadé dans un dataset HF : Loren/articles_db
+###
+### Ce script
+###    - crée une base SQLite articles.db constituée des 3 tables : tags, articles, et tag_article
+###    - l'upload dans le dataset HF Loren/articles_db
+###    - crée les fichiers Parquet compressés à partir des tables SQLite
+###    - l'upload dans le dataset HF Loren/articles_database
+###
+### 👉 Ils peuvent alors être utilisés par un space Hugging Face
+##############################################################################################
+
+import sqlite3
+import pandas as pd
+import os
+from dotenv import load_dotenv
+import itertools
+import ast
+import uuid
+from huggingface_hub import hf_hub_download, upload_file
+from pathlib import Path
+from collections import Counter
+
+# Initialisations
+print("Initialisations ...")
+load_dotenv()
+HF_TOKEN = os.getenv('API_HF_TOKEN')
+
+# Constantes
+MIN_COUNT = 5 # nombre minimum d'occurrences pour qu'un tag soit conservé
+DATA_DIR = Path("../../Data")   # dossier parent du script
+REPO_ID_DB = "Loren/articles_db"  # dataset HF
+REPO_ID = "Loren/articles_database"  # dataset HF
+DB_NAME = 'articles.db'
+SQLITE_FILE = DATA_DIR / DB_NAME
+LIST_TABLES = ["articles", "tags", "tag_article"]
+PARQUET_DIR = DATA_DIR / "parquet_tables"
+
+# Chargement des données
+parquet_path = hf_hub_download(repo_id=REPO_ID_DB,
+                               filename="medium_articles.parquet",
+                               repo_type="dataset")
+
+# Créer les dossiers s'ils n'existent pas
+DATA_DIR.mkdir(exist_ok=True)
+PARQUET_DIR.mkdir(exist_ok=True)
+
+# Chargement des données
+print("Chargement des données ...")
+df = pd.read_parquet(parquet_path)
+
+# Initialisations de la base SQLite
+print("Initialisations de la base SQLite ...")
+conn = sqlite3.connect(SQLITE_FILE)
+cur = conn.cursor()
+
+# Suppression des anciennes tables
+cur.execute("DROP TABLE IF EXISTS tag_article")
+cur.execute("DROP TABLE IF EXISTS tags")
+cur.execute("DROP TABLE IF EXISTS articles")
+
+# Création des tables Articles, Tags, et de la table d'association articles <-> tags
+cur.execute("""
+CREATE TABLE articles (
+    article_id TEXT PRIMARY KEY,         -- UUID
+    article_title TEXT,
+    article_text TEXT,
+    article_url TEXT,
+    article_authors TEXT,
+    article_date TEXT                    -- YYYY-MM-DD
+)""")
+
+cur.execute("""
+CREATE TABLE tags (
+    tag_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    tag_name TEXT UNIQUE
+)""")
+
+cur.execute("""
+CREATE TABLE tag_article (
+    tag_article_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    article_id TEXT,
+    tag_id INTEGER,
+    FOREIGN KEY(article_id) REFERENCES articles(article_id),
+    FOREIGN KEY(tag_id) REFERENCES tags(tag_id)
+)""")
+
+# Extraction des tags en une liste
+print("Extraction des tags en une liste  ...")
+df['list_tags'] = df['tags'].apply(lambda x: ast.literal_eval(x) if isinstance(x, str) else [])
+# Extraire tous les tags uniques
+all_tags = list(itertools.chain.from_iterable(df['list_tags']))
+# Comptage du nombre d'occurrences de chaque tag
+tag_counts = Counter(all_tags)
+# On ne va conserver que les tags avec au moins 100 occurrences
+list_tags = [tag for tag, count in tag_counts.items() if count >= MIN_COUNT]
+
+# Insertion des tags dans la table
+print("Insertion des tags dans la table ...")
+cur.executemany("INSERT INTO tags (tag_name) VALUES (?)", [(tag,) for tag in list_tags])
+
+# Récupération des correspondances tag_name -> tag_id
+print("Récupération des correspondances tag_name -> tag_id ...")
+cur.execute("SELECT tag_id, tag_name FROM tags")
+dict_tag_map = {tag_name: tag_id for tag_id, tag_name in cur.fetchall()}
+
+# Insertion des articles et table d'association dans les tables
+print("Insertion des articles et table d'association dans les tables ...")
+for _, row in df.iterrows():
+    # Détermination de l'id article
+    article_id = str(uuid.uuid4())
+
+    # Extraction de la date du timestamp
+    date_value = None
+    if pd.notna(row["timestamp"]):
+        try:
+            date_value = str(pd.to_datetime(row["timestamp"]).date())
+        except Exception:
+            date_value = None
+
+    # Insertion dans la table Articles
+    cur.execute("""
+        INSERT INTO articles (article_id, article_title, article_text, article_url, article_authors, article_date)
+        VALUES (?, ?, ?, ?, ?, ?)""",
+            (article_id, row["title"], row["text"], row["url"], row["authors"], date_value))
+
+    # Association aux tags
+    for tag_name in row['list_tags']:
+        try:
+            tag_id = dict_tag_map[tag_name]
+            cur.execute("INSERT INTO tag_article (article_id, tag_id) VALUES (?, ?)",
+                        (article_id, tag_id))
+        except:
+            pass
+
+print("-> ", len(list_tags), " tags")
+cur.execute("SELECT COUNT(*) FROM tag_article")
+nb_lignes = cur.fetchone()[0]
+print("-> ", nb_lignes, " associations articles <-> tags")
+print("-> ", len(df), " articles")
+
+# Commit
+print("Commit ...")
+conn.commit()
+
+# Upload dans le dataset hugging face
+print("Upload base Sqlite dans le dataset hugging face ...")
+upload_file(
+    path_or_fileobj=SQLITE_FILE,
+    path_in_repo=DB_NAME,
+
+    repo_id=REPO_ID_DB,
+    repo_type="dataset",
+    token=HF_TOKEN
+)
+
+# Création des fichiers Parquet compressés
+print("Création des fichiers Parquet compressés ...")
+parquet_files = []
+for table in LIST_TABLES:
+    df = pd.read_sql_query(f"SELECT * FROM {table}", conn)
+    parquet_path = PARQUET_DIR / f"{table}.parquet"
+    df.to_parquet(parquet_path, engine="pyarrow", index=False, compression="snappy")
+    parquet_files.append(parquet_path)
+
+# Upload des fichiers Parquet vers HF
+print("Upload des fichiers Parquet dans le dataset hugging face ...")
+for parquet_file in parquet_files:
+    print(f"Uploading {parquet_file.name} ...")
+    upload_file(
+        path_or_fileobj=parquet_file,
+        path_in_repo=parquet_file.name,
+        repo_id=REPO_ID,
+        repo_type="dataset",
+        token=HF_TOKEN
+    )
+
+print("Upload terminé ✅")
+
+conn.close()
+print("Traitement terminé.")

script/2_check_dataset.py

@@ -0,0 +1,174 @@
+##############################################################################################
+### Script de vérification des url des articles
+###
+### Ce script
+###    - charge la table articles depuis le dataset HF Loren/articles_database
+###    - check si les url des articles sont des pages actives (status code 200)
+###    - ajoute cette information dans une colonne article_online
+###    - crée les fichiers Parquet compressés à partir de la table créée
+###    - l'upload dans le dataset HF Loren/articles_database
+###
+### 👉 Il peut alors être utilisé par un space Hugging Face
+##############################################################################################
+
+import os
+from dotenv import load_dotenv
+from huggingface_hub import hf_hub_download, upload_file
+from pathlib import Path
+from typing import List, Any, Tuple
+import duckdb
+import asyncio
+import aiohttp
+from tqdm.asyncio import tqdm_asyncio
+
+# Fonctions
+async def check_url(session: aiohttp.ClientSession, sem: asyncio.Semaphore, url: str) -> bool:
+    """
+    Vérifie si une URL est accessible (retourne un code HTTP < 400).
+
+    Args:
+        session (aiohttp.ClientSession): Session HTTP réutilisable pour les requêtes.
+        sem (asyncio.Semaphore): Sémaphore pour limiter le nombre de requêtes simultanées.
+        url (str): L'URL à vérifier.
+
+    Returns:
+        bool: True si l’URL répond avec un code < 400, sinon False.
+    """
+    if not url:
+        return False
+
+    async with sem:  # limite de concurrence
+        try:
+            async with session.head(url, allow_redirects=True, timeout=TIMEOUT) as resp:
+                return resp.status < 400
+        except Exception:
+            return False
+
+
+async def process_batch(batch: List[Tuple[Any, ...]]) -> List[bool]:
+    """
+    Traite un batch d’URLs avec une limite de requêtes simultanées.
+
+    Args:
+        batch (List[Tuple[Any, ...]]): Liste de tuples représentant les lignes d'articles.
+            Chaque tuple doit contenir au moins une colonne d’URL à l’index 3.
+
+    Returns:
+        List[bool]: Liste de statuts (True/False) correspondant à l’accessibilité de chaque URL.
+    """
+    sem = asyncio.Semaphore(MAX_CONCURRENCY)
+    async with aiohttp.ClientSession() as session:
+        tasks = [
+            check_url(session, sem, row[3])  # row[3] = article_url
+            for row in batch
+        ]
+        return await tqdm_asyncio.gather(*tasks)
+
+
+async def main() -> None:
+    """
+    Exécute le traitement complet :
+    - Récupère les articles par batch depuis la base.
+    - Vérifie la disponibilité des URLs.
+    - Insère les résultats enrichis dans une table de sortie.
+
+    Returns:
+        None
+    """
+    total_rows = con.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
+    total_batches = (total_rows + BATCH_SIZE - 1) // BATCH_SIZE
+
+    print(f"🔍 {total_rows} lignes à traiter ({total_batches} batchs de {BATCH_SIZE})")
+
+    for batch_index in range(total_batches):
+        offset = batch_index * BATCH_SIZE
+
+        # Charger un batch depuis la table
+        batch = con.execute(f"""
+            SELECT * FROM articles
+            LIMIT {BATCH_SIZE} OFFSET {offset}
+        """).fetchall()
+
+        # Vérifier les URLs
+        online_statuses = await process_batch(batch)
+
+        # Préparer les données enrichies
+        enriched_rows = [
+            (*row, status)
+            for row, status in zip(batch, online_statuses)
+        ]
+
+        # Insérer dans la table physique
+        con.executemany(f"""
+            INSERT INTO {TABLE_OUTPUT} VALUES (?, ?, ?, ?, ?, ?, ?)
+        """, enriched_rows)
+
+        print(f"✅ Batch {batch_index + 1}/{total_batches} traité ({len(batch)} lignes)")
+
+    print("🎉 Traitement terminé !")
+    print(f"Résultat stocké dans la table '{TABLE_OUTPUT}'")
+#
+
+if __name__ == "__main__":
+    # Initialisations
+    print("Initialisations ...")
+    load_dotenv()
+    HF_TOKEN = os.getenv('API_HF_TOKEN')
+
+    # Constantes
+    DATA_DIR = Path("../../Data")   # dossier parent du script
+    REPO_ID = "Loren/articles_database"  # dataset HF
+    PARQUET_DIR = DATA_DIR / "parquet_tables"
+    REPO_ID = "Loren/articles_database"
+    CACHE_DIR = "/tmp"
+    TABLE_OUTPUT = "articles_checked"  # Table de sortie
+    BATCH_SIZE = 1000
+    MAX_CONCURRENCY = 100
+    TIMEOUT = 5  # secondes
+    parquet_path = PARQUET_DIR / f"{TABLE_OUTPUT}.parquet"
+
+    os.makedirs(CACHE_DIR, exist_ok=True)
+    # Rediriger le cache HF globalement
+    os.environ["HF_HOME"] = CACHE_DIR
+    os.environ["HF_DATASETS_CACHE"] = CACHE_DIR
+    os.environ["TRANSFORMERS_CACHE"] = CACHE_DIR
+
+    # Téléchargement des fichiers Parquet depuis Hugging Face
+    articles_parquet = hf_hub_download(
+        repo_id=REPO_ID,
+        filename="articles.parquet",
+        repo_type="dataset",
+        cache_dir=CACHE_DIR)
+
+    # Connexion DuckDB en mémoire
+    con = duckdb.connect()
+
+    # Créer des tables DuckDB directement à partir des fichiers Parquet
+    con.execute(f"CREATE VIEW articles AS SELECT * FROM parquet_scan('{articles_parquet}')")
+
+    # Créer la table cible
+    con.execute(f"""CREATE TABLE {TABLE_OUTPUT} AS
+                        SELECT *, NULL::BOOLEAN AS article_online
+                            FROM articles
+                            WHERE 1=0""")
+
+    # Traitement principal
+    asyncio.run(main())
+
+    # Sauvegarde du résultat dans un fichier Parquet
+    con.execute(f"""COPY {TABLE_OUTPUT} TO '{parquet_path}'
+                (FORMAT PARQUET, , COMPRESSION 'SNAPPY')""")
+
+    print(f"✅ Fichier Parquet créé : {parquet_path}")
+
+    # Upload des fichiers Parquet vers HF
+    print(f"Uploading {parquet_path} ...")
+    upload_file(
+        path_or_fileobj=parquet_path,
+        path_in_repo=f"{TABLE_OUTPUT}.parquet",
+        repo_id=REPO_ID,
+        repo_type="dataset",
+        token=HF_TOKEN
+    )
+
+    print("✅ Traitement terminé.")

script/3_create_faiss_database.py

@@ -0,0 +1,229 @@
+##############################################################################################
+### Script de création de la base de données FAISS des articles
+###
+### Ce script
+###    - charge la table articles depuis le dataset HF Loren/articles_database
+###    - la traite par batch :
+###        - création de chunks de texte
+###        - création des embeddings avec le modèle SentenceTransformer "intfloat/e5-small"
+###        - ajout des embeddings dans un index FAISS
+###        - sauvegarde des métadonnées des chunks dans un fichier parquet
+###        - sauvegarde de l'index FAISS dans un fichier faiss_index.bin
+###    - upload dans le dataset HF Loren/articles_faiss
+###
+### 👉 L'index Faiss peut alors être utilisé par un space Hugging Face
+##############################################################################################
+
+import os
+import torch
+import duckdb
+from huggingface_hub import hf_hub_download, upload_file
+from huggingface_hub import HfApi, HfFolder, CommitOperationAdd
+import faiss
+from sentence_transformers import SentenceTransformer
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from functools import partial
+import pyarrow as pa
+import pyarrow.parquet as pq
+from pathlib import Path
+from dotenv import load_dotenv
+
+# Fonctions
+# Batch processing function
+def batch_process(list_articles: list, faiss_id_start: int) -> int:
+    """
+    Traite un batch d'articles pour générer des embeddings et des métadonnées,
+    puis les sauvegarde de manière sécurisée pour garantir la persistance en cas de problème.
+
+    Étapes réalisées :
+    1. Découpage de chaque article en chunks via le splitter.
+    2. Création d'un dictionnaire de métadonnées pour chaque chunk contenant :
+       - faiss_id : identifiant unique aligné avec l'index FAISS
+       - document_id : identifiant de l'article
+       - chunk_text : texte du chunk
+    3. Calcul des embeddings pour tous les chunks du batch.
+    4. Ajout des embeddings au FAISS index existant (append).
+    5. Écriture immédiate de l'index FAISS sur disque pour assurer la persistance.
+    6. Sauvegarde des métadonnées batch dans un fichier Parquet distinct.
+
+    Args:
+        list_articles (list): Liste de tuples (document_id, document_text) représentant les articles du batch.
+        faiss_id_start (int): Identifiant de départ pour le premier chunk du batch,
+                              utilisé pour aligner FAISS et les métadonnées.
+
+    Returns:
+        int: Identifiant FAISS suivant, à utiliser pour le batch suivant afin de maintenir l'alignement.
+
+    Notes :
+        - Cette fonction est conçue pour être utilisée batch par batch.
+        - Les fichiers Parquet et le fichier FAISS sont mis à jour à chaque batch pour éviter toute perte de données.
+    """
+    global faiss_index
+
+    try:
+        list_chunks = []
+        list_metadata = []
+
+        for doc_id, doc_content in list_articles:
+            chunks = splitter.split_text(doc_content)
+            for chunk_text in chunks:
+                list_chunks.append(chunk_text)
+                list_metadata.append({
+                    "faiss_id": faiss_id_start,
+                    "document_id": doc_id,
+                    "chunk_text": chunk_text
+                })
+                faiss_id_start += 1
+
+        # Embeddings
+        if list_chunks:
+            passage_texts = [f"passage: {p}" for p in list_chunks]
+            embeddings = model.encode(passage_texts, convert_to_numpy=True,
+                                      normalize_embeddings=True)
+            faiss_index.add(embeddings)
+            faiss.write_index(faiss_index, str(FAISS_INDEX_FILE))
+
+        # Sauvegarde batch métadonnées en Parquet
+        if list_metadata:
+            table = pa.Table.from_pylist(list_metadata)
+            batch_file = PARQUET_DIR / f"metadata_batch_{faiss_id_start}.parquet"
+            pq.write_table(table, batch_file)
+
+        return faiss_id_start
+    except Exception as e:
+        print(f"ERROR in batch_process function : {e}")
+        return None
+##
+
+# Initialisations
+global faiss_index
+print("Initialisations ...")
+load_dotenv()
+HF_TOKEN = os.getenv('API_HF_TOKEN')
+REPO_ID = "Loren/articles_database"
+DATA_DIR = Path("../../Data")   # dossier parent du script
+CHUNK_SIZE = 250
+CHUNK_OVERLAP = 50
+BATCH_SIZE = 1000
+MODEL_NAME = "intfloat/multilingual-e5-small"
+FAISS_INDEX_FILE = DATA_DIR / "faiss_index.bin"
+PARQUET_DIR = DATA_DIR / "parquet_metadata"
+
+CACHE_DIR = "/tmp"
+os.makedirs(CACHE_DIR, exist_ok=True)
+# Rediriger le cache HF globalement
+os.environ["HF_HOME"] = CACHE_DIR
+os.environ["HF_DATASETS_CACHE"] = CACHE_DIR
+os.environ["TRANSFORMERS_CACHE"] = CACHE_DIR
+
+# Téléchargement des fichiers Parquet depuis Hugging Face
+print("Téléchargement des fichiers Parquet depuis Hugging Face ...")
+articles_parquet = hf_hub_download(
+    repo_id=REPO_ID,
+    filename="articles.parquet",
+    repo_type="dataset",
+    cache_dir=CACHE_DIR)
+
+# Connexion DuckDB en mémoire
+con = duckdb.connect()
+
+# Créer des tables DuckDB directement à partir des fichiers Parquet
+print("Création des vues DuckDB à partir des fichiers Parquet ...")
+con.execute(f"CREATE VIEW articles AS SELECT * FROM parquet_scan('{articles_parquet}')")
+
+# Creating the plitter for chunking document
+print("Initialisation du text splitter ...")
+splitter = RecursiveCharacterTextSplitter(
+    chunk_size=CHUNK_SIZE,
+    chunk_overlap=CHUNK_OVERLAP,
+    keep_separator='end',
+    separators=["\n\n", "\n", "."]
+)
+
+# Creating the Sentence transformer model
+print("Initialisation du modèle de Sentence Transformer ...")
+device = "cuda" if torch.cuda.is_available() else "cpu"
+print(f"*** Device: {device}")
+model = SentenceTransformer(MODEL_NAME, device=device)
+
+# Creating the Faiss index
+embedding_dim = model.get_sentence_embedding_dimension()
+faiss_index = faiss.IndexFlatIP(embedding_dim)
+faiss_id_counter = 0  # compteur global pour lier faiss_id et métadonnées
+
+# Traitement par batchs
+print("Création des batches et traitement ...")
+cursor = con.execute("""
+            SELECT article_id, article_text
+              FROM articles
+             WHERE (LENGTH(article_text) - LENGTH(REPLACE(article_text, ' ', '')) + 1) >= 100""")
+
+# Création d'un itérateur de batches
+fetch_batch = partial(cursor.fetchmany, BATCH_SIZE)
+
+for batch_num, batch in enumerate(iter(fetch_batch, []), start=1):
+    print("Traitement batch no ", batch_num, " ...")
+    faiss_id_counter = batch_process(batch, faiss_id_counter)
+    if not faiss_id_counter:
+        print("*** Erreur traitement batch no ", batch_num)
+
+print("\n✅ Traitement terminé")
+
+# Upload des fichiers vers HF
+# Création du dataset HF
+REPO_ID = "Loren/articles_faiss"
+api = HfApi()
+HfFolder.save_token(HF_TOKEN)
+
+# Vérifier si le dataset existe
+try:
+    repo_info = api.dataset_info(REPO_ID, token=HF_TOKEN)
+    print(f"Dataset {REPO_ID} existe déjà, suppression en cours...")
+    api.delete_repo(repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN)
+except Exception as e:
+    print(f"Dataset n'existait pas : {e}")
+
+# Créer le dataset (privé)
+api.create_repo(repo_id=REPO_ID, repo_type="dataset", exist_ok=True, private=True, token=HF_TOKEN)
+print(f"Dataset {REPO_ID} créé avec succès.")
+
+# Récupérer la liste de fichiers parquet
+print("Upload des fichiers metadatas dans le dataset hugging face ", REPO_ID, " ...")
+parquet_files = [
+    os.path.join(PARQUET_DIR, f)
+    for f in os.listdir(PARQUET_DIR)
+    if f.endswith(".parquet")
+]
+
+# Ajouter tous les fichiers
+operations = [
+    CommitOperationAdd(
+        path_in_repo=f"data/{os.path.basename(f)}",
+        path_or_fileobj=f
+    )
+    for f in parquet_files
+]
+
+api.create_commit(
+    repo_id=REPO_ID,
+    repo_type="dataset",
+    operations=operations,
+    commit_message="Upload batch metadata parquet files"
+)
+
+print("✅ Upload metadatas terminé !")
+
+
+print("Upload de l'index Faiss dans le dataset hugging face ", REPO_ID, " ...")
+upload_file(
+    path_or_fileobj=FAISS_INDEX_FILE,
+    path_in_repo=FAISS_INDEX_FILE.name,
+    repo_id=REPO_ID,
+    repo_type="dataset",
+    token=HF_TOKEN
+)
+
+print("✅ Upload faiss index terminé")
+
+con.close()
+print("✅ Traitement terminé")

script/rep.py

@@ -0,0 +1,216 @@
+##############################################################################################
+### Script de création de la base de données FAISS des articles
+###
+### Ce script
+###    - charge la table articles depuis le dataset HF Loren/articles_database
+###    - la traite par batch :
+###        - création de chunks de texte
+###        - création des embeddings avec le modèle SentenceTransformer "intfloat/e5-small"
+###        - ajout des embeddings dans un index FAISS
+###        - sauvegarde des métadonnées des chunks dans un fichier parquet
+###        - sauvegarde de l'index FAISS dans un fichier faiss_index.bin
+###    - upload dans le dataset HF Loren/articles_faiss
+###
+### 👉 L'index Faiss peut alors être utilisé par un space Hugging Face
+##############################################################################################
+
+import os
+#import torch
+import duckdb
+from huggingface_hub import hf_hub_download, upload_file
+from huggingface_hub import HfApi, HfFolder, CommitOperationAdd
+#import faiss
+#from sentence_transformers import SentenceTransformer
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from functools import partial
+import pyarrow as pa
+import pyarrow.parquet as pq
+from pathlib import Path
+from dotenv import load_dotenv
+
+# Fonctions
+# Batch processing function
+def batch_process(list_articles: list, faiss_id_start: int) -> int:
+    """
+    Traite un batch d'articles pour générer des embeddings et des métadonnées,
+    puis les sauvegarde de manière sécurisée pour garantir la persistance en cas de problème.
+
+    Étapes réalisées :
+    1. Découpage de chaque article en chunks via le splitter.
+    2. Création d'un dictionnaire de métadonnées pour chaque chunk contenant :
+       - faiss_id : identifiant unique aligné avec l'index FAISS
+       - document_id : identifiant de l'article
+       - chunk_text : texte du chunk
+    3. Calcul des embeddings pour tous les chunks du batch.
+    4. Ajout des embeddings au FAISS index existant (append).
+    5. Écriture immédiate de l'index FAISS sur disque pour assurer la persistance.
+    6. Sauvegarde des métadonnées batch dans un fichier Parquet distinct.
+
+    Args:
+        list_articles (list): Liste de tuples (document_id, document_text) représentant les articles du batch.
+        faiss_id_start (int): Identifiant de départ pour le premier chunk du batch,
+                              utilisé pour aligner FAISS et les métadonnées.
+
+    Returns:
+        int: Identifiant FAISS suivant, à utiliser pour le batch suivant afin de maintenir l'alignement.
+
+    Notes :
+        - Cette fonction est conçue pour être utilisée batch par batch.
+        - Les fichiers Parquet et le fichier FAISS sont mis à jour à chaque batch pour éviter toute perte de données.
+    """
+    global faiss_index
+
+    try:
+        list_chunks = []
+        list_metadata = []
+
+        for doc_id, doc_content in list_articles:
+            chunks = splitter.split_text(doc_content)
+            for chunk_text in chunks:
+                list_chunks.append(chunk_text)
+                list_metadata.append({
+                    "faiss_id": faiss_id_start,
+                    "document_id": doc_id,
+                    "chunk_text": chunk_text
+                })
+                faiss_id_start += 1
+
+        # Embeddings
+        #if list_chunks:
+        #    embeddings = model.encode(list_chunks, convert_to_numpy=True, normalize_embeddings=True)
+        #    faiss_index.add(embeddings)
+        #    faiss.write_index(faiss_index, FAISS_INDEX_FILE)
+
+        # Sauvegarde batch métadonnées en Parquet
+        if list_metadata:
+            table = pa.Table.from_pylist(list_metadata)
+            batch_file = PARQUET_DIR / f"metadata_batch_{faiss_id_start}.parquet"
+            pq.write_table(table, batch_file)
+
+        return faiss_id_start
+    except Exception as e:
+        print(f"ERROR in batch_process function : {e}")
+        return None
+##
+
+# Initialisations
+global faiss_index
+print("Initialisations ...")
+load_dotenv()
+HF_TOKEN = os.getenv('API_HF_TOKEN')
+REPO_ID = "Loren/articles_database"
+DATA_DIR = Path("../../Data")   # dossier parent du script
+CHUNK_SIZE = 250
+CHUNK_OVERLAP = 50
+BATCH_SIZE = 1000
+MODEL_NAME = "intfloat/e5-small"
+FAISS_INDEX_FILE = DATA_DIR / "faiss_index.bin"
+PARQUET_DIR = DATA_DIR / "parquet_metadata"
+
+CACHE_DIR = "/tmp"
+os.makedirs(CACHE_DIR, exist_ok=True)
+# Rediriger le cache HF globalement
+os.environ["HF_HOME"] = CACHE_DIR
+os.environ["HF_DATASETS_CACHE"] = CACHE_DIR
+os.environ["TRANSFORMERS_CACHE"] = CACHE_DIR
+
+# Téléchargement des fichiers Parquet depuis Hugging Face
+print("Téléchargement des fichiers Parquet depuis Hugging Face ...")
+articles_parquet = hf_hub_download(
+    repo_id=REPO_ID,
+    filename="articles.parquet",
+    repo_type="dataset",
+    cache_dir=CACHE_DIR)
+
+# Connexion DuckDB en mémoire
+con = duckdb.connect()
+
+# Créer des tables DuckDB directement à partir des fichiers Parquet
+print("Création des vues DuckDB à partir des fichiers Parquet ...")
+con.execute(f"CREATE VIEW articles AS SELECT * FROM parquet_scan('{articles_parquet}')")
+
+# Creating the plitter for chunking document
+splitter = RecursiveCharacterTextSplitter(
+    chunk_size=CHUNK_SIZE,
+    chunk_overlap=CHUNK_OVERLAP,
+    keep_separator='end',
+    separators=["\n\n", "\n", "."]
+)
+
+# Creating the Sentence transformer model
+#device = "cuda" if torch.cuda.is_available() else "cpu"
+#print(f"*** Device: {device}")
+#model = SentenceTransformer(MODEL_NAME, device=device)
+#
+## Creating the Faiss index
+#embedding_dim = model.get_sentence_embedding_dimension()
+#faiss_index = faiss.IndexFlatIP(embedding_dim)
+faiss_id_counter = 0  # compteur global pour lier faiss_id et métadonnées
+
+# Traitement par batchs
+print("Création des batches et traitement ...")
+cursor = con.execute("""
+            SELECT article_id, article_text
+              FROM articles
+             WHERE (LENGTH(article_text) - LENGTH(REPLACE(article_text, ' ', '')) + 1) >= 100""")
+
+# Création d'un itérateur de batches
+fetch_batch = partial(cursor.fetchmany, BATCH_SIZE)
+
+for batch_num, batch in enumerate(iter(fetch_batch, []), start=1):
+    print("Traitement batch no ", batch_num, " ...")
+    faiss_id_counter = batch_process(batch, faiss_id_counter)
+    if not faiss_id_counter:
+        print("*** Erreur traitement batch no ", batch_num)
+
+print("\n✅ Traitement terminé")
+
+# Upload des fichiers vers HF
+# Création du dataset HF
+REPO_ID = "Loren/articles_faiss"
+api = HfApi()
+HfFolder.save_token(HF_TOKEN)
+
+# Créer repo si besoin
+api.create_repo(repo_id=REPO_ID, repo_type="dataset", exist_ok=True, private=True)
+
+# Récupérer la liste de fichiers parquet
+print("Upload des fichiers metadatas dans le dataset hugging face ", REPO_ID, " ...")
+parquet_files = [
+    os.path.join(PARQUET_DIR, f)
+    for f in os.listdir(PARQUET_DIR)
+    if f.endswith(".parquet")
+]
+
+# Ajouter tous les fichiers
+operations = [
+    CommitOperationAdd(
+        path_in_repo=f"data/{os.path.basename(f)}",
+        path_or_fileobj=f
+    )
+    for f in parquet_files
+]
+
+api.create_commit(
+    repo_id=REPO_ID,
+    repo_type="dataset",
+    operations=operations,
+    commit_message="Upload batch metadata parquet files"
+)
+
+print("✅ Upload metadatas terminé !")
+
+
+print("Upload de l'index Faiss dans le dataset hugging face ", REPO_ID, " ...")
+upload_file(
+    path_or_fileobj=FAISS_INDEX_FILE,
+    path_in_repo=FAISS_INDEX_FILE.name,
+    repo_id=REPO_ID,
+    repo_type="dataset",
+    token=HF_TOKEN
+)
+
+print("✅ Upload faiss index terminé")
+
+con.close()
+print("✅ Traitement terminé")