prepare_data.py

import pandas as pd
from datasets import load_dataset
import re
import os
import sys
import numpy as np
from sentence_transformers import SentenceTransformer
import faiss

def clean_text(text):
    if not isinstance(text, str):
        return ""
    # Remove extra spaces and newlines
    text = re.sub(r'\s+', ' ', text).strip()
    return text

def get_samples_from_stream(ds_name, split, token, target_count, buffer_size=2000):
    print(f"Streaming samples from {ds_name}...")
    try:
        # EhsanShahbazi datasets might have different structures, we try to handle them
        ds = load_dataset(ds_name, split=split, token=token if token else None, streaming=True)
        samples = []
        count = 0
        for item in ds:
            # Common column names for text in these datasets
            potential_cols = ['text', 'Text', 'comment', 'Comment', 'content', 'Content', 'body']
            text_val = None
            for col in potential_cols:
                if col in item:
                    text_val = item[col]
                    break

            if text_val:
                cleaned = clean_text(text_val)
                # Filter for meaningful length
                if len(cleaned) > 25:
                    samples.append({'text': cleaned})
                    count += 1

            # Stop if we have enough samples or reached buffer limit
            if count >= buffer_size or count >= target_count * 3:
                break

        if not samples:
            return None

        df = pd.DataFrame(samples)
        return df
    except Exception as e:
        print(f"Error loading {ds_name}: {e}")
        return None

def fetch_all_data(target_total=2000):
    """Fetches data from multiple Digikala comment datasets on Hugging Face."""
    hf_token = os.getenv("HUGGINGFACE_TOKEN")
    if hf_token == "": hf_token = None
    datasets_to_load = [
        ("fibonacciai/Digikala-Comments", "train"),
        ("ParsiAI/digikala-sentiment-analysis", "train"),
        ("EhsanShahbazi/digikala-comments", "train"),
        ("EhsanShahbazi/digikala-comments-with-media", "train")
    ]

    dfs = []
    per_source_target = target_total // len(datasets_to_load)

    for ds_name, split in datasets_to_load:
        df = get_samples_from_stream(ds_name, split, hf_token, per_source_target)
        if df is not None:
            dfs.append(df)

    if not dfs:
        print("Warning: No datasets could be loaded. Using mock data for safety.")
        return pd.DataFrame({'text': ["این یک نظر تستی است برای زمانی که دیتاست در دسترس نیست."] * 10})

    processed_dfs = []
    # Aim for balanced sampling
    per_source = target_total // len(dfs)

    for source_df in dfs:
        source_df = source_df.drop_duplicates(subset=['text'])
        if not source_df.empty:
            s_size = min(len(source_df), per_source)
            processed_dfs.append(source_df.sample(n=s_size, random_state=42))

    if not processed_dfs:
        return None

    df_sample = pd.concat(processed_dfs, ignore_index=True)
    # Shuffle the final dataset
    df_sample = df_sample.sample(frac=1, random_state=42).reset_index(drop=True)
    print(f"Total samples collected: {len(df_sample)}")
    return df_sample

def generate_faiss_index(df, output_dir="data"):
    print("Generating FAISS index...")
    model = SentenceTransformer('sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2')
    texts = df['text'].tolist()
    embeddings = model.encode(texts, show_progress_bar=True)
    index = faiss.IndexFlatL2(embeddings.shape[1])
    index.add(np.array(embeddings).astype('float32'))
    os.makedirs(output_dir, exist_ok=True)
    faiss.write_index(index, os.path.join(output_dir, "faiss_index.bin"))
    print(f"FAISS index saved to {output_dir}/faiss_index.bin")

def prepare_data():
    df = fetch_all_data()
    if df is not None:
        os.makedirs("data", exist_ok=True)
        df.to_csv("data/digikala_samples.csv", index=False)
        print(f"Data saved to data/digikala_samples.csv")
        generate_faiss_index(df)

if __name__ == "__main__":
    try:
        prepare_data()
        print("Data preparation completed successfully.")
        os._exit(0)
    except Exception as e:
        print(f"Fatal error during data preparation: {e}")
        os._exit(1)