import spaces import gradio as gr import pandas as pd import numpy as np import os import re import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import seaborn as sns import networkx as nx from datetime import datetime from huggingface_hub import HfApi, hf_hub_download from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity from sklearn.manifold import TSNE # ========================================== # 1. ИНИЦИАЛИЗАЦИЯ ИИ (ЭТОГО НЕ ХВАТАЛО) # ========================================== model = SentenceTransformer('all-MiniLM-L6-v2') def split_into_sentences(text): # Разбиваем по точкам, восклицательным и вопросительным знакам sentences = re.split(r'(?<=[.!?])\s+', text.strip()) return [s for s in sentences if len(s) > 10] # ========================================== # 2. КОНФИГУРАЦИЯ БАЗЫ ДАННЫХ # ========================================== DATASET_REPO_ID = "Masterogon/dream-database" DB_FILE = "dream_database.csv" HF_TOKEN = os.environ.get("HF_TOKEN") api = HfApi() def init_db(): if HF_TOKEN: try: print("Downloading database from Hugging Face Hub...") file_path = hf_hub_download(repo_id=DATASET_REPO_ID, filename=DB_FILE, repo_type="dataset", token=HF_TOKEN) import shutil shutil.copy(file_path, DB_FILE) print("Database successfully loaded.") return except Exception as e: print("Could not download DB. It might be empty or missing. Error:", e) if not os.path.exists(DB_FILE): df = pd.DataFrame(columns=["Timestamp", "Alias", "ASC_Type", "Emotion", "Intensity", "Narrative"]) df.to_csv(DB_FILE, index=False) print("Created a fresh local database.") init_db() # ========================================== # 3. ФУНКЦИЯ ЗАПИСИ # ========================================== def process_entry(alias, asc_type, emotion, intensity, narrative): if not narrative.strip(): return "Error: Please describe your experience.", pd.read_csv(DB_FILE).tail(5) timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") new_data = pd.DataFrame([[timestamp, alias, asc_type, emotion, intensity, narrative]], columns=["Timestamp", "Alias", "ASC_Type", "Emotion", "Intensity", "Narrative"]) new_data.to_csv(DB_FILE, mode='a', header=False, index=False) if HF_TOKEN: try: api.upload_file( path_or_fileobj=DB_FILE, path_in_repo=DB_FILE, repo_id=DATASET_REPO_ID, repo_type="dataset", token=HF_TOKEN, commit_message=f"Added new report by {alias}" ) backup_status = "Successfully synced to cloud." except Exception as e: backup_status = f"Warning: Cloud sync failed ({e})" else: backup_status = "Warning: No HF_TOKEN. Data is local." return f"Success! {alias} added. {backup_status}", pd.read_csv(DB_FILE).tail(10) # ДАЛЕЕ ИДУТ ВАШИ ФУНКЦИИ macro_analysis, micro_analysis и hybrid_pattern_discovery... # 3. ФУНКЦИИ МАКРО-АНАЛИЗА (Tab 2: Heatmap & Graph) @spaces.GPU def macro_analysis(): df = pd.read_csv(DB_FILE).dropna(subset=['Narrative']) if len(df) < 2: return None, None texts = df['Narrative'].tolist() aliases = df['Alias'].tolist() # Векторизация целых документов для общей картины embeddings = model.encode(texts) sim_matrix = cosine_similarity(embeddings) # 3.1 Построение Heatmap fig_heat, ax_heat = plt.subplots(figsize=(8, 6)) sns.heatmap(sim_matrix, xticklabels=aliases, yticklabels=aliases, annot=True, cmap="YlOrRd", fmt=".2f", ax=ax_heat) ax_heat.set_title("Document Cosine Similarity Matrix") plt.tight_layout() # 3.2 Построение Network Graph fig_graph, ax_graph = plt.subplots(figsize=(8, 6)) G = nx.Graph() for i, alias in enumerate(aliases): G.add_node(alias) threshold = 0.40 # Порог сходства для создания связи for i in range(len(aliases)): for j in range(i + 1, len(aliases)): if sim_matrix[i, j] > threshold: G.add_edge(aliases[i], aliases[j], weight=sim_matrix[i, j]) pos = nx.spring_layout(G, seed=42) # Рисуем только вершины nx.draw_networkx_nodes( G, pos, node_color="skyblue", node_size=2000, ax=ax_graph ) # Подписи nx.draw_networkx_labels( G, pos, font_size=10, font_weight="bold", ax=ax_graph ) #nx.draw(G, pos, with_labels=True, node_color='skyblue', #node_size=2000, font_size=10, font_weight='bold', ax=ax_graph) # Рисуем толщину линий в зависимости от силы сходства edges = G.edges() weights = [G[u][v]['weight'] * 5 for u,v in edges] nx.draw_networkx_edges(G, pos, edgelist=edges, width=weights, edge_color='red', alpha=0.5, ax=ax_graph) ax_graph.set_title(f"Semantic Network Graph (Threshold > {threshold})") return fig_heat, fig_graph # 4. ФУНКЦИИ МИКРО-АНАЛИЗА ПО ПРЕДЛОЖЕНИЯМ (Tab 3: Sentence t-SNE) @spaces.GPU def micro_analysis(): df = pd.read_csv(DB_FILE).dropna(subset=['Narrative']) if len(df) < 2: return None, "Not enough data." all_sentences = [] parent_aliases = [] for _, row in df.iterrows(): sents = split_into_sentences(row['Narrative']) all_sentences.extend(sents) parent_aliases.extend([row['Alias']] * len(sents)) if len(all_sentences) < 5: return None, "Not enough sentences extracted." # Векторизация предложений sent_embeddings = model.encode(all_sentences) # 4.1 t-SNE по предложениям perplexity = min(30, len(all_sentences) - 1) tsne = TSNE(n_components=2, random_state=42, perplexity=perplexity) vecs_2d = tsne.fit_transform(sent_embeddings) fig_tsne, ax_tsne = plt.subplots(figsize=(10, 8)) sns.scatterplot(x=vecs_2d[:,0], y=vecs_2d[:,1], hue=parent_aliases, palette="tab10", s=100, ax=ax_tsne) ax_tsne.set_title("Sentence-Level Semantic Projections (t-SNE)") plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left') plt.tight_layout() # 4.2 Поиск самых центральных (архетипичных) предложений # Считаем среднее косинусное сходство каждого предложения со всеми остальными sim_matrix = cosine_similarity(sent_embeddings) mean_sims = sim_matrix.mean(axis=1) # Берем топ-5 top_indices = mean_sims.argsort()[-5:][::-1] central_text = "### Top 5 Most Central Semantic Fragments\n\n" for idx in top_indices: central_text += f"> **{parent_aliases[idx]}**: \"{all_sentences[idx]}\" *(Centrality Score: {mean_sims[idx]:.2f})*\n\n" return fig_tsne, central_text # 5. ГИБРИДНЫЙ ДВИЖОК ПОИСКА ПАТТЕРНОВ (Tab 4) @spaces.GPU def hybrid_pattern_discovery(mode, custom_motifs_text): df = pd.read_csv(DB_FILE).dropna(subset=['Narrative']) if len(df) < 2: return "Not enough data. Need at least 2 reports." all_sentences = [] parent_aliases = [] # Сбор всех предложений for _, row in df.iterrows(): sents = split_into_sentences(row['Narrative']) all_sentences.extend(sents) parent_aliases.extend([row['Alias']] * len(sents)) if len(all_sentences) < 5: return "Not enough detailed sentences." # ========================================== # РЕЖИМ 1: СЛЕПОЙ ПОИСК (Unsupervised) # ========================================== if mode == "Blind Extraction (Unsupervised)": sent_embeddings = model.encode(all_sentences) num_clusters = max(2, min(8, len(all_sentences) // 10)) from sklearn.cluster import KMeans kmeans = KMeans(n_clusters=num_clusters, random_state=42) labels = kmeans.fit_predict(sent_embeddings) results = "### 👁️ Blind AI Extraction: Emergent Signals from Noise\n" results += f"*Analyzed {len(all_sentences)} sentences. Extracted {num_clusters} hidden thematic clusters without human prompts.*\n\n" for i in range(num_clusters): cluster_indices = np.where(labels == i)[0] if len(cluster_indices) < 2: continue cluster_embeddings = sent_embeddings[cluster_indices] centroid = kmeans.cluster_centers_[i] from sklearn.metrics.pairwise import cosine_distances distances = cosine_distances([centroid], cluster_embeddings)[0] global_central_idx = cluster_indices[np.argmin(distances)] central_sentence = all_sentences[global_central_idx] authors_in_cluster = set([parent_aliases[idx] for idx in cluster_indices]) if len(authors_in_cluster) > 1: results += f"#### 🟢 Discovered Archetype: *«{central_sentence[:100]}...»*\n" results += f"**Cross-validation:** Found in {len(authors_in_cluster)} independent subjects.\n" count = 0 for idx in cluster_indices: if idx != global_central_idx and count < 3: results += f"- *{parent_aliases[idx]}*: \"{all_sentences[idx]}\"\n" count += 1 results += "---\n" return results # ========================================== # РЕЖИМ 2: ЦЕЛЕВОЙ ПОИСК (Zero-Shot) # ========================================== else: seed_motifs = [m.strip() for m in re.split(r'[,|\n]', custom_motifs_text) if m.strip()] if not seed_motifs: return "Please enter at least one motif to search for." motif_embeddings = model.encode(seed_motifs) results = "### 🎯 Targeted AI Search: Hypothesis Testing\n\n" for idx, motif in enumerate(seed_motifs): motif_emb = motif_embeddings[idx] reports_with_motif = 0 best_matches = [] for _, row in df.iterrows(): sents = split_into_sentences(row['Narrative']) if not sents: continue sent_embs = model.encode(sents) sims = cosine_similarity([motif_emb], sent_embs)[0] max_sim = np.max(sims) if max_sim > 0.40: # Порог совпадения reports_with_motif += 1 best_idx = np.argmax(sims) best_matches.append(f"*{row['Alias']}*: \"{sents[best_idx]}\"") results += f"#### ✔ Target: '{motif}'\n" results += f"**Found in {reports_with_motif} out of {len(df)} reports.**\n" if best_matches: for match in best_matches[:4]: results += f"- {match}\n" results += "---\n" return results # Проверяем каждый отчет for idx, motif in enumerate(seed_motifs): motif_emb = motif_embeddings[idx] reports_with_motif = 0 best_matches = [] for _, row in df.iterrows(): sents = split_into_sentences(row['Narrative']) if not sents: continue sent_embs = model.encode(sents) sims = cosine_similarity([motif_emb], sent_embs)[0] max_sim = np.max(sims) if max_sim > 0.40: # Порог обнаружения мотива reports_with_motif += 1 best_idx = np.argmax(sims) best_matches.append(f"*{row['Alias']}*: \"{sents[best_idx]}\"") results += f"#### ✔ Motif: '{motif}'\n" results += f"**Found in {reports_with_motif} out of {len(df)} reports.**\n" if best_matches: results += "Top examples:\n" for match in best_matches[:3]: # Показываем топ-3 примера results += f"- {match}\n" results += "---\n" return results # ----------------- ИНТЕРФЕЙС GRADIO ----------------- with gr.Blocks(theme=gr.themes.Monochrome()) as app: gr.Markdown("# 🌌 DreamCode") with gr.Tabs(): # TAB 1: Data Ingestion with gr.TabItem("1. Data Ingestion"): with gr.Row(): with gr.Column(): alias = gr.Textbox(label="Participant ID") asc_type = gr.Dropdown(choices=["Ordinary Dream", "Lucid Dream (LD)", "OBE", "NDE", "Other"], label="State") emotion = gr.Radio(choices=["Positive", "Neutral", "Negative"], label="Emotional Tone") intensity = gr.Slider(minimum=1, maximum=3, step=1, label="Intensity") narrative = gr.Textbox(label="Narrative", lines=7) submit_btn = gr.Button("Submit Experience", variant="primary") with gr.Column(): status_output = gr.Textbox(label="Status") data_preview = gr.Dataframe(headers=["Timestamp", "Alias", "ASC_Type", "Emotion", "Intensity", "Narrative"]) submit_btn.click(fn=process_entry, inputs=[alias, asc_type, emotion, intensity, narrative], outputs=[status_output, data_preview]) # TAB 2: Document Level Analysis with gr.TabItem("2. Document Similarity"): gr.Markdown("Analyze macro-connections between entire reports using Cosine Similarity.") analyze_macro_btn = gr.Button("Generate Matrix & Graph", variant="primary") with gr.Row(): heat_plot = gr.Plot(label="Cosine Similarity Heatmap") network_plot = gr.Plot(label="Semantic Network Graph") analyze_macro_btn.click(fn=macro_analysis, inputs=[], outputs=[heat_plot, network_plot]) # TAB 3: Sentence Level Analysis with gr.TabItem("3. Scene & Sentence t-SNE"): gr.Markdown("AI splits narratives into individual sentences to cluster specific scenes (e.g., separating the 'red planet' scene from the 'tunnel' scene).") analyze_micro_btn = gr.Button("Process Sentences", variant="primary") with gr.Row(): tsne_plot = gr.Plot(label="Sentence t-SNE") central_text = gr.Markdown(label="Central Fragments") analyze_micro_btn.click(fn=micro_analysis, inputs=[], outputs=[tsne_plot, central_text]) # TAB 4: AI Pattern Discovery (HYBRID) with gr.TabItem("4. AI Pattern Discovery"): gr.Markdown("### Semantic Search Engine\nChoose between extracting unknown signals automatically or testing specific hypotheses against the database.") # Переключатель режимов search_mode = gr.Radio( choices=["Blind Extraction (Unsupervised)", "Targeted Search (Zero-Shot)"], value="Blind Extraction (Unsupervised)", label="Select Analysis Mode" ) # Поле для ручного ввода (по умолчанию скрыто) motif_input = gr.Textbox( label="Enter custom phrases, motifs, or fragments from your own dream (separated by commas or new lines)", lines=3, value="A red celestial body, Huge planet in the sky, Feeling of global catastrophe", visible=False ) # Логика показа/скрытия текстового поля в зависимости от выбора def toggle_input(mode): if mode == "Targeted Search (Zero-Shot)": return gr.update(visible=True) else: return gr.update(visible=False) search_mode.change(fn=toggle_input, inputs=[search_mode], outputs=[motif_input]) analyze_btn = gr.Button("Run Analysis Engine", variant="primary") output_md = gr.Markdown() analyze_btn.click(fn=hybrid_pattern_discovery, inputs=[search_mode, motif_input], outputs=[output_md]) app.launch()