import numpy as np from scipy import sparse from scipy.cluster.hierarchy import linkage, dendrogram from sklearn.decomposition import TruncatedSVD import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import io import base64 import re import json # Stopwords français de base FRENCH_STOPWORDS = set([ 'le', 'la', 'les', 'un', 'une', 'des', 'de', 'du', 'au', 'aux', 'ce', 'cette', 'ces', 'mon', 'ma', 'mes', 'ton', 'ta', 'tes', 'son', 'sa', 'ses', 'notre', 'nos', 'votre', 'vos', 'leur', 'leurs', 'je', 'tu', 'il', 'elle', 'on', 'nous', 'vous', 'ils', 'elles', 'me', 'te', 'se', 'lui', 'y', 'en', 'qui', 'que', 'quoi', 'dont', 'ou', 'où', 'et', 'ou', 'mais', 'donc', 'car', 'ni', 'or', 'ne', 'pas', 'plus', 'moins', 'très', 'trop', 'peu', 'bien', 'mal', 'dans', 'sur', 'sous', 'avec', 'sans', 'pour', 'par', 'entre', 'vers', 'chez', 'contre', 'depuis', 'pendant', 'avant', 'après', 'devant', 'derrière', 'être', 'avoir', 'faire', 'dire', 'pouvoir', 'vouloir', 'devoir', 'falloir', 'est', 'sont', 'était', 'été', 'sera', 'serait', 'ai', 'as', 'avons', 'avez', 'ont', 'fait', 'dit', 'peut', 'veut', 'doit', 'faut', 'tout', 'tous', 'toute', 'toutes', 'autre', 'autres', 'même', 'mêmes', 'quel', 'quelle', 'quels', 'quelles', 'quelque', 'quelques', 'si', 'aussi', 'comme', 'comment', 'quand', 'pourquoi', 'encore', 'toujours', 'jamais', 'souvent', 'parfois', 'cela', 'ceci', 'ça', 'celui', 'celle', 'ceux', 'celles', 'puis', 'alors', 'ainsi', 'donc', 'enfin', 'ensuite', 'ici', 'là', 'là-bas', 'voici', 'voilà', 'oui', 'non', 'peut-être', 'vraiment', 'seulement', ]) def tokenize(text): """Tokenisation simple du texte""" text = text.lower() text = re.sub(r'[^a-zàâäéèêëïîôùûüœæç\s-]', ' ', text) tokens = text.split() tokens = [t for t in tokens if len(t) >= 3 and t not in FRENCH_STOPWORDS] return tokens def parse_iramuteq(text): """Parse un corpus au format Iramuteq""" segments = [] metadata = [] # Séparer par les marqueurs UCI parts = re.split(r'\*\*\*\*\s*', text) for part in parts: if not part.strip(): continue lines = part.strip().split('\n') if not lines: continue # Première ligne contient les variables étoilées first_line = lines[0] vars_match = re.findall(r'\*([a-zA-Z0-9_]+)', first_line) # Le reste est le texte if len(lines) > 1: segment_text = ' '.join(lines[1:]).strip() else: # Texte sur la même ligne après les variables segment_text = re.sub(r'\*[a-zA-Z0-9_]+', '', first_line).strip() if segment_text: segments.append(segment_text) metadata.append({'vars': vars_match}) return segments, metadata def segment_text(text, min_size): """Segmente un texte brut en unités de contexte""" # Découper par phrases sentences = re.split(r'[.!?]+', text) sentences = [s.strip() for s in sentences if s.strip()] segments = [] current_segment = [] current_word_count = 0 for sentence in sentences: words = sentence.split() current_segment.append(sentence) current_word_count += len(words) if current_word_count >= min_size: segments.append(' '.join(current_segment)) current_segment = [] current_word_count = 0 # Ajouter le reste if current_segment: if segments and current_word_count < min_size // 2: segments[-1] += ' ' + ' '.join(current_segment) else: segments.append(' '.join(current_segment)) return segments, [{'vars': []} for _ in segments] def build_dtm(segments): """Construit la matrice termes-documents binaire""" vocabulary = {} vocab_index = 0 # Construire le vocabulaire tokenized_segments = [] for segment in segments: tokens = set(tokenize(segment)) tokenized_segments.append(tokens) for token in tokens: if token not in vocabulary: vocabulary[token] = vocab_index vocab_index += 1 # Construire la matrice binaire n_docs = len(segments) n_terms = len(vocabulary) if n_terms == 0 or n_docs == 0: return None, [], [] # Matrice dense (plus simple pour Pyodide) matrix = np.zeros((n_docs, n_terms), dtype=np.float64) for doc_idx, tokens in enumerate(tokenized_segments): for token in tokens: if token in vocabulary: matrix[doc_idx, vocabulary[token]] = 1.0 vocab_list = [''] * len(vocabulary) for term, idx in vocabulary.items(): vocab_list[idx] = term return matrix, vocab_list, tokenized_segments def compute_chi2_split(matrix, split_point): """Calcule le Chi2 pour un point de division donné""" n_docs, n_terms = matrix.shape group1 = matrix[:split_point] group2 = matrix[split_point:] n1 = len(group1) n2 = len(group2) if n1 == 0 or n2 == 0: return 0.0 # Sommes par groupe sum1 = np.sum(group1, axis=0) sum2 = np.sum(group2, axis=0) total = sum1 + sum2 # Chi2 pour chaque terme expected1 = total * n1 / (n1 + n2) expected2 = total * n2 / (n1 + n2) # Éviter division par zéro with np.errstate(divide='ignore', invalid='ignore'): chi2_1 = np.where(expected1 > 0, (sum1 - expected1)**2 / expected1, 0) chi2_2 = np.where(expected2 > 0, (sum2 - expected2)**2 / expected2, 0) return np.sum(chi2_1 + chi2_2) def reinert_split(matrix, min_members=10, tsj=3, cc_test=0.3): """Division d'une matrice maximisant le Chi2""" n_docs, n_terms = matrix.shape if n_docs < min_members * 2: return None, None, 0.0 # Filtrer les termes trop rares term_freq = np.sum(matrix, axis=0) valid_terms = term_freq >= tsj if np.sum(valid_terms) < 3: return None, None, 0.0 filtered_matrix = matrix[:, valid_terms] # AFC simplifiée via SVD try: # Centrer la matrice row_sums = np.sum(filtered_matrix, axis=1, keepdims=True) col_sums = np.sum(filtered_matrix, axis=0, keepdims=True) total = np.sum(filtered_matrix) if total == 0: return None, None, 0.0 # Matrice des résidus standardisés (approximation AFC) expected = row_sums @ col_sums / total with np.errstate(divide='ignore', invalid='ignore'): residuals = np.where(expected > 0, (filtered_matrix - expected) / np.sqrt(expected), 0) # SVD pour obtenir les coordonnées sur le premier axe if residuals.shape[1] > 1: svd = TruncatedSVD(n_components=1, random_state=42) coords = svd.fit_transform(residuals)[:, 0] else: coords = residuals[:, 0] if residuals.shape[1] > 0 else np.zeros(n_docs) # Ordonner les documents order = np.argsort(coords) ordered_matrix = matrix[order] # Chercher le meilleur point de division best_chi2 = 0.0 best_split = n_docs // 2 for split_point in range(min_members, n_docs - min_members + 1): chi2 = compute_chi2_split(ordered_matrix, split_point) if chi2 > best_chi2: best_chi2 = chi2 best_split = split_point # Récupérer les indices originaux group1_indices = order[:best_split].tolist() group2_indices = order[best_split:].tolist() return group1_indices, group2_indices, best_chi2 except Exception as e: # En cas d'erreur, division simple au milieu mid = n_docs // 2 return list(range(mid)), list(range(mid, n_docs)), 0.0 def hierarchical_classification(matrix, k, min_members): """Classification descendante hiérarchique complète""" n_docs = matrix.shape[0] # Initialisation classes = {0: list(range(n_docs))} tree = [] # [(parent, child1, child2, chi2)] next_class_id = 1 while len(classes) < k: # Trouver la plus grande classe à diviser best_class_id = None best_size = 0 for class_id, members in classes.items(): if len(members) > best_size and len(members) >= min_members * 2: best_size = len(members) best_class_id = class_id if best_class_id is None: break # Extraire la sous-matrice members = classes[best_class_id] sub_matrix = matrix[members] # Diviser group1, group2, chi2 = reinert_split(sub_matrix, min_members) if group1 is None or len(group1) < min_members or len(group2) < min_members: # Marquer comme non divisible classes[best_class_id] = members break # Convertir les indices locaux en indices globaux group1_global = [members[i] for i in group1] group2_global = [members[i] for i in group2] # Mettre à jour les classes del classes[best_class_id] classes[next_class_id] = group1_global classes[next_class_id + 1] = group2_global # Enregistrer dans l'arbre tree.append((best_class_id, next_class_id, next_class_id + 1, chi2)) next_class_id += 2 return tree, classes def get_characteristic_terms(matrix, classes, vocabulary, n_terms=10): """Calcule les termes caractéristiques de chaque classe par Chi2""" results = {} n_docs_total = matrix.shape[0] for class_id, members in classes.items(): if len(members) == 0: continue class_matrix = matrix[members] other_indices = [i for i in range(n_docs_total) if i not in members] if len(other_indices) == 0: continue other_matrix = matrix[other_indices] # Fréquence dans la classe et hors classe freq_class = np.sum(class_matrix, axis=0) freq_other = np.sum(other_matrix, axis=0) n_class = len(members) n_other = len(other_indices) # Chi2 pour chaque terme total_freq = freq_class + freq_other expected_class = total_freq * n_class / (n_class + n_other) with np.errstate(divide='ignore', invalid='ignore'): chi2_values = np.where( expected_class > 0, (freq_class - expected_class)**2 / expected_class, 0 ) # Signe: positif si surreprésenté signs = np.sign(freq_class - expected_class) signed_chi2 = chi2_values * signs # Top termes surreprésentés top_indices = np.argsort(signed_chi2)[-n_terms:][::-1] terms = [] for idx in top_indices: if signed_chi2[idx] > 0: terms.append({ 'term': vocabulary[idx], 'chi2': float(signed_chi2[idx]), 'freq_class': int(freq_class[idx]), 'freq_total': int(total_freq[idx]) }) results[class_id] = terms return results def create_dendrogram_image(tree, classes, n_docs): """Crée une image du dendrogramme""" if len(classes) <= 1: # Créer une image simple si pas assez de classes fig, ax = plt.subplots(figsize=(10, 6)) ax.text(0.5, 0.5, 'Pas assez de données pour créer un dendrogramme', ha='center', va='center', fontsize=14) ax.set_xlim(0, 1) ax.set_ylim(0, 1) ax.axis('off') else: # Construire la matrice de linkage pour scipy # Format: [idx1, idx2, distance, n_members] # Créer un mapping des classes finales final_classes = list(classes.keys()) n_final = len(final_classes) # Créer une matrice de linkage simplifiée basée sur l'arbre if tree: # Utiliser les valeurs Chi2 comme distances (inversées) max_chi2 = max(t[3] for t in tree) if tree else 1 linkage_matrix = [] cluster_sizes = {class_id: len(members) for class_id, members in classes.items()} for parent, child1, child2, chi2 in tree: size1 = cluster_sizes.get(child1, 1) size2 = cluster_sizes.get(child2, 1) # Distance inversée (chi2 élevé = plus similaire) distance = max_chi2 - chi2 + 1 linkage_matrix.append([child1, child2, distance, size1 + size2]) cluster_sizes[parent] = size1 + size2 if linkage_matrix: Z = np.array(linkage_matrix) # Renormaliser les indices pour scipy old_to_new = {} new_idx = 0 for row in Z: for i in [0, 1]: if row[i] not in old_to_new: old_to_new[row[i]] = new_idx new_idx += 1 # Reconstruire avec les bons indices n_leaves = len(classes) Z_fixed = [] next_cluster = n_leaves for i, (parent, child1, child2, chi2) in enumerate(tree): dist = max_chi2 - chi2 + 1 size = cluster_sizes.get(parent, 2) Z_fixed.append([i, i+1 if i+1 < n_leaves else n_leaves + i - 1, dist, size]) fig, ax = plt.subplots(figsize=(12, 8)) # Labels des classes labels = [f"Classe {cid}\n({len(members)} seg.)" for cid, members in sorted(classes.items())] # Dessiner un dendrogramme simple ax.set_title("Dendrogramme - Classification Reinert", fontsize=14, fontweight='bold') # Représentation simplifiée n_classes = len(classes) x_positions = np.linspace(0, 10, n_classes) for i, (class_id, members) in enumerate(sorted(classes.items())): ax.bar(x_positions[i], len(members), width=0.8, color=plt.cm.Set3(i % 12), edgecolor='black') ax.text(x_positions[i], len(members) + 1, f"Cl. {class_id}", ha='center', va='bottom', fontsize=10) ax.set_xlabel("Classes", fontsize=12) ax.set_ylabel("Nombre de segments", fontsize=12) ax.set_xticks([]) else: fig, ax = plt.subplots(figsize=(10, 6)) ax.text(0.5, 0.5, 'Dendrogramme non disponible', ha='center', va='center', fontsize=14) ax.axis('off') plt.tight_layout() # Sauvegarder en PNG buf = io.BytesIO() plt.savefig(buf, format='png', dpi=150, bbox_inches='tight', facecolor='white', edgecolor='none') plt.close() buf.seek(0) return base64.b64encode(buf.getvalue()).decode('utf-8') # ============ MAIN ============ def main(input_type, text_content, iramuteq_file, num_classes, min_segment_size, min_split_members): # Récupérer le texte selon le type d'entrée if input_type == "iramuteq" and iramuteq_file: text = iramuteq_file segments, metadata = parse_iramuteq(text) elif text_content: text = text_content segments, metadata = segment_text(text, min_segment_size or 40) else: return { "error": True, "message": "Veuillez fournir un texte à analyser ou un fichier Iramuteq." } if len(segments) < 5: return { "error": True, "message": f"Pas assez de segments ({len(segments)}). Minimum requis: 5 segments." } # Construire la matrice termes-documents matrix, vocabulary, tokenized = build_dtm(segments) if matrix is None or len(vocabulary) < 10: return { "error": True, "message": f"Vocabulaire insuffisant ({len(vocabulary) if vocabulary else 0} termes). Le texte est peut-être trop court ou contient trop de mots vides." } # Classification k = min(num_classes or 5, len(segments) // (min_split_members or 10)) k = max(2, k) tree, classes = hierarchical_classification( matrix, k, min_split_members or 10 ) if len(classes) < 2: return { "error": True, "message": "Impossible de créer une classification. Le texte est peut-être trop homogène ou trop court." } # Termes caractéristiques char_terms = get_characteristic_terms(matrix, classes, vocabulary, n_terms=15) # Créer le dendrogramme dendrogram_base64 = create_dendrogram_image(tree, classes, len(segments)) # Préparer les résultats classes_info = [] for class_id, members in sorted(classes.items()): class_info = { "id": class_id, "size": len(members), "percentage": round(100 * len(members) / len(segments), 1), "terms": char_terms.get(class_id, [])[:10], "sample_segments": [segments[i][:200] + "..." if len(segments[i]) > 200 else segments[i] for i in members[:3]] } classes_info.append(class_info) return { "error": False, "dendrogramImage": dendrogram_base64, "stats": { "total_segments": len(segments), "total_terms": len(vocabulary), "num_classes": len(classes) }, "classes": classes_info }

Outils