Spaces:
Running
Running
# View_DB_Items_tab.py | |
# Description: This file contains the code for the search tab in the Gradio UI | |
# | |
# Imports | |
import html | |
import logging | |
# | |
# External Imports | |
import gradio as gr | |
# | |
# Local Imports | |
from App_Function_Libraries.DB.DB_Manager import view_database, get_all_document_versions, \ | |
fetch_paginated_data, fetch_item_details, get_latest_transcription, list_prompts, fetch_prompt_details | |
from App_Function_Libraries.DB.RAG_QA_Chat_DB import get_keywords_for_note, search_conversations_by_keywords, \ | |
get_notes_by_keywords, get_keywords_for_conversation, get_db_connection, get_all_conversations, load_chat_history, \ | |
get_notes | |
from App_Function_Libraries.DB.SQLite_DB import get_document_version, fetch_items_by_keyword, fetch_all_keywords | |
# | |
#################################################################################################### | |
# | |
# Functions | |
def format_as_html(content, title): | |
escaped_content = html.escape(content) | |
formatted_content = escaped_content.replace('\n', '<br>') | |
return f""" | |
<div style="border: 1px solid #ddd; padding: 10px; margin-bottom: 10px;"> | |
<h3>{title}</h3> | |
<div style="max-height: 700px; overflow-y: auto;"> | |
{formatted_content} | |
</div> | |
</div> | |
""" | |
def extract_prompt_and_summary(content: str): | |
# Implement this function based on how prompt and summary are stored in your DocumentVersions content | |
# This is a placeholder implementation | |
parts = content.split('\n\n', 2) | |
prompt = parts[0] if len(parts) > 0 else "No prompt available." | |
summary = parts[1] if len(parts) > 1 else "No summary available." | |
return prompt, summary | |
def create_view_all_mediadb_with_versions_tab(): | |
with gr.TabItem("View All MediaDB Items", visible=True): | |
gr.Markdown("# View All Media Database Entries with Version Selection") | |
with gr.Row(): | |
with gr.Column(scale=1): | |
entries_per_page = gr.Dropdown(choices=[10, 20, 50, 100], label="Entries per Page", value=10) | |
page_number = gr.Number(value=1, label="Page Number", precision=0) | |
view_button = gr.Button("View Page") | |
next_page_button = gr.Button("Next Page") | |
previous_page_button = gr.Button("Previous Page") | |
with gr.Column(scale=2): | |
items_output = gr.Dropdown(label="Select Item to View Details", choices=[]) | |
version_dropdown = gr.Dropdown(label="Select Version", choices=[], visible=False) | |
with gr.Row(): | |
with gr.Column(scale=1): | |
pagination_info = gr.Textbox(label="Pagination Info", interactive=False) | |
with gr.Column(scale=2): | |
prompt_output = gr.Textbox(label="Prompt Used", visible=True) | |
summary_output = gr.HTML(label="Summary", visible=True) | |
transcription_output = gr.HTML(label="Transcription", visible=True) | |
item_mapping = gr.State({}) | |
def update_page(page, entries_per_page): | |
results, total_entries = fetch_paginated_data(page, entries_per_page) | |
total_pages = (total_entries + entries_per_page - 1) // entries_per_page | |
pagination = f"Page {page} of {total_pages} (Total items: {total_entries})" | |
choices = [f"{item[1]} (ID: {item[0]})" for item in results] | |
new_item_mapping = {f"{item[1]} (ID: {item[0]})": item[0] for item in results} | |
next_disabled = page >= total_pages | |
prev_disabled = page <= 1 | |
return (gr.update(choices=choices, value=None), | |
pagination, | |
page, | |
gr.update(interactive=not next_disabled), | |
gr.update(interactive=not prev_disabled), | |
gr.update(visible=False, choices=[]), | |
"", "", "", | |
new_item_mapping) | |
def format_as_html(content, title): | |
if content is None: | |
content = "No content available." | |
escaped_content = html.escape(str(content)) | |
formatted_content = escaped_content.replace('\n', '<br>') | |
return f""" | |
<div style="border: 1px solid #ddd; padding: 10px; margin-bottom: 10px;"> | |
<h3>{title}</h3> | |
<div style="max-height: 700px; overflow-y: auto;"> | |
{formatted_content} | |
</div> | |
</div> | |
""" | |
def display_item_details(selected_item, item_mapping): | |
if selected_item and item_mapping and selected_item in item_mapping: | |
media_id = item_mapping[selected_item] | |
prompt, summary, transcription = fetch_item_details(media_id) | |
versions = get_all_document_versions(media_id) | |
# Filter out duplicate versions and sort them | |
unique_versions = list(set((v['version_number'], v['created_at']) for v in versions)) | |
unique_versions.sort(key=lambda x: x[0], reverse=True) | |
version_choices = [f"Version {v[0]} ({v[1]})" for v in unique_versions] | |
summary_html = format_as_html(summary, "Summary") | |
transcription_html = format_as_html(transcription, "Transcription") | |
return ( | |
gr.update(visible=True, choices=version_choices, | |
value=version_choices[0] if version_choices else None), | |
prompt if prompt is not None else "", | |
summary_html, | |
transcription_html | |
) | |
return gr.update(visible=False, choices=[]), "", "", "" | |
def update_version_content(selected_item, item_mapping, selected_version): | |
if selected_item and item_mapping and selected_item in item_mapping and selected_version: | |
media_id = item_mapping[selected_item] | |
version_number = int(selected_version.split()[1].split('(')[0]) | |
version_data = get_document_version(media_id, version_number) | |
if 'error' not in version_data: | |
content = version_data['content'] | |
prompt, summary = extract_prompt_and_summary(content) | |
transcription = get_latest_transcription(media_id) | |
summary_html = format_as_html(summary, "Summary") | |
transcription_html = format_as_html(transcription, "Transcription") | |
return prompt if prompt is not None else "", summary_html, transcription_html | |
return gr.update(value=selected_item), gr.update(), gr.update() | |
view_button.click( | |
fn=update_page, | |
inputs=[page_number, entries_per_page], | |
outputs=[items_output, pagination_info, page_number, next_page_button, previous_page_button, | |
version_dropdown, prompt_output, summary_output, transcription_output, item_mapping] | |
) | |
next_page_button.click( | |
fn=lambda page, entries: update_page(page + 1, entries), | |
inputs=[page_number, entries_per_page], | |
outputs=[items_output, pagination_info, page_number, next_page_button, previous_page_button, | |
version_dropdown, prompt_output, summary_output, transcription_output, item_mapping] | |
) | |
previous_page_button.click( | |
fn=lambda page, entries: update_page(max(1, page - 1), entries), | |
inputs=[page_number, entries_per_page], | |
outputs=[items_output, pagination_info, page_number, next_page_button, previous_page_button, | |
version_dropdown, prompt_output, summary_output, transcription_output, item_mapping] | |
) | |
items_output.change( | |
fn=display_item_details, | |
inputs=[items_output, item_mapping], | |
outputs=[version_dropdown, prompt_output, summary_output, transcription_output] | |
) | |
version_dropdown.change( | |
fn=update_version_content, | |
inputs=[items_output, item_mapping, version_dropdown], | |
outputs=[prompt_output, summary_output, transcription_output] | |
) | |
def create_mediadb_keyword_search_tab(): | |
with gr.TabItem("Search MediaDB by Keyword", visible=True): | |
gr.Markdown("# List Media Database Items by Keyword") | |
with gr.Row(): | |
with gr.Column(scale=1): | |
# Keyword selection dropdown - initialize with empty list, will be populated on load | |
keyword_dropdown = gr.Dropdown( | |
label="Select Keyword", | |
choices=fetch_all_keywords(), # Initialize with keywords on creation | |
value=None | |
) | |
entries_per_page = gr.Dropdown( | |
choices=[10, 20, 50, 100], | |
label="Entries per Page", | |
value=10 | |
) | |
page_number = gr.Number( | |
value=1, | |
label="Page Number", | |
precision=0 | |
) | |
# Navigation buttons | |
refresh_keywords_button = gr.Button("Refresh Keywords") | |
view_button = gr.Button("View Results") | |
next_page_button = gr.Button("Next Page") | |
previous_page_button = gr.Button("Previous Page") | |
# Pagination information | |
pagination_info = gr.Textbox( | |
label="Pagination Info", | |
interactive=False | |
) | |
with gr.Column(scale=2): | |
# Results area | |
results_table = gr.HTML( | |
label="Search Results" | |
) | |
item_details = gr.HTML( | |
label="Item Details", | |
visible=True | |
) | |
def update_keyword_choices(): | |
try: | |
keywords = fetch_all_keywords() | |
return gr.update(choices=keywords) | |
except Exception as e: | |
return gr.update(choices=[], value=None) | |
def search_items(keyword, page, entries_per_page): | |
try: | |
# Calculate offset for pagination | |
offset = (page - 1) * entries_per_page | |
# Fetch items for the selected keyword | |
items = fetch_items_by_keyword(keyword) | |
total_items = len(items) | |
total_pages = (total_items + entries_per_page - 1) // entries_per_page | |
# Paginate results | |
paginated_items = items[offset:offset + entries_per_page] | |
# Generate HTML table for results | |
table_html = "<table style='width:100%; border-collapse: collapse;'>" | |
table_html += "<tr><th style='border: 1px solid black; padding: 8px;'>Title</th>" | |
table_html += "<th style='border: 1px solid black; padding: 8px;'>URL</th></tr>" | |
for item_id, title, url in paginated_items: | |
table_html += f""" | |
<tr> | |
<td style='border: 1px solid black; padding: 8px;'>{html.escape(title)}</td> | |
<td style='border: 1px solid black; padding: 8px;'>{html.escape(url)}</td> | |
</tr> | |
""" | |
table_html += "</table>" | |
# Update pagination info | |
pagination = f"Page {page} of {total_pages} (Total items: {total_items})" | |
# Determine button states | |
next_disabled = page >= total_pages | |
prev_disabled = page <= 1 | |
return ( | |
table_html, | |
pagination, | |
gr.update(interactive=not next_disabled), | |
gr.update(interactive=not prev_disabled) | |
) | |
except Exception as e: | |
return ( | |
f"<p>Error: {str(e)}</p>", | |
"Error in pagination", | |
gr.update(interactive=False), | |
gr.update(interactive=False) | |
) | |
def go_to_next_page(keyword, current_page, entries_per_page): | |
next_page = current_page + 1 | |
return search_items(keyword, next_page, entries_per_page) + (next_page,) | |
def go_to_previous_page(keyword, current_page, entries_per_page): | |
previous_page = max(1, current_page - 1) | |
return search_items(keyword, previous_page, entries_per_page) + (previous_page,) | |
# Event handlers | |
refresh_keywords_button.click( | |
fn=update_keyword_choices, | |
inputs=[], | |
outputs=[keyword_dropdown] | |
) | |
view_button.click( | |
fn=search_items, | |
inputs=[keyword_dropdown, page_number, entries_per_page], | |
outputs=[results_table, pagination_info, next_page_button, previous_page_button] | |
) | |
next_page_button.click( | |
fn=go_to_next_page, | |
inputs=[keyword_dropdown, page_number, entries_per_page], | |
outputs=[results_table, pagination_info, next_page_button, previous_page_button, page_number] | |
) | |
previous_page_button.click( | |
fn=go_to_previous_page, | |
inputs=[keyword_dropdown, page_number, entries_per_page], | |
outputs=[results_table, pagination_info, next_page_button, previous_page_button, page_number] | |
) | |
def create_viewing_mediadb_tab(): | |
with gr.TabItem("View Media Database Entries", visible=True): | |
gr.Markdown("# View Media Database Entries") | |
with gr.Row(): | |
with gr.Column(): | |
entries_per_page = gr.Dropdown(choices=[10, 20, 50, 100], label="Entries per Page", value=10) | |
page_number = gr.Number(value=1, label="Page Number", precision=0) | |
view_button = gr.Button("View Page") | |
next_page_button = gr.Button("Next Page") | |
previous_page_button = gr.Button("Previous Page") | |
pagination_info = gr.Textbox(label="Pagination Info", interactive=False) | |
with gr.Column(): | |
results_display = gr.HTML() | |
def update_page(page, entries_per_page): | |
results, pagination, total_pages = view_database(page, entries_per_page) | |
next_disabled = page >= total_pages | |
prev_disabled = page <= 1 | |
return results, pagination, page, gr.update(interactive=not next_disabled), gr.update(interactive=not prev_disabled) | |
def go_to_next_page(current_page, entries_per_page): | |
next_page = current_page + 1 | |
return update_page(next_page, entries_per_page) | |
def go_to_previous_page(current_page, entries_per_page): | |
previous_page = max(1, current_page - 1) | |
return update_page(previous_page, entries_per_page) | |
view_button.click( | |
fn=update_page, | |
inputs=[page_number, entries_per_page], | |
outputs=[results_display, pagination_info, page_number, next_page_button, previous_page_button] | |
) | |
next_page_button.click( | |
fn=go_to_next_page, | |
inputs=[page_number, entries_per_page], | |
outputs=[results_display, pagination_info, page_number, next_page_button, previous_page_button] | |
) | |
previous_page_button.click( | |
fn=go_to_previous_page, | |
inputs=[page_number, entries_per_page], | |
outputs=[results_display, pagination_info, page_number, next_page_button, previous_page_button] | |
) | |
##################################################################### | |
# | |
# RAG DB Viewing Functions: | |
def create_viewing_ragdb_tab(): | |
with gr.TabItem("View RAG Database Entries", visible=True): | |
gr.Markdown("# View RAG Database Entries") | |
with gr.Row(): | |
with gr.Column(): | |
entries_per_page = gr.Dropdown(choices=[10, 20, 50, 100], label="Entries per Page", value=10) | |
page_number = gr.Number(value=1, label="Page Number", precision=0) | |
view_button = gr.Button("View Page") | |
next_page_button = gr.Button("Next Page") | |
previous_page_button = gr.Button("Previous Page") | |
pagination_info = gr.Textbox(label="Pagination Info", interactive=False) | |
with gr.Column(): | |
results_display = gr.HTML() | |
def format_conversations_table(conversations): | |
table_html = "<table style='width:100%; border-collapse: collapse;'>" | |
table_html += """ | |
<tr> | |
<th style='border: 1px solid black; padding: 8px;'>Title</th> | |
<th style='border: 1px solid black; padding: 8px;'>Keywords</th> | |
<th style='border: 1px solid black; padding: 8px;'>Notes</th> | |
<th style='border: 1px solid black; padding: 8px;'>Rating</th> | |
</tr> | |
""" | |
for conversation in conversations: | |
conv_id = conversation['conversation_id'] | |
title = conversation['title'] | |
rating = conversation.get('rating', '') # Use get() to handle cases where rating might not exist | |
keywords = get_keywords_for_conversation(conv_id) | |
notes = get_notes(conv_id) | |
table_html += f""" | |
<tr> | |
<td style='border: 1px solid black; padding: 8px;'>{html.escape(str(title))}</td> | |
<td style='border: 1px solid black; padding: 8px;'>{html.escape(', '.join(keywords))}</td> | |
<td style='border: 1px solid black; padding: 8px;'>{len(notes)} note(s)</td> | |
<td style='border: 1px solid black; padding: 8px;'>{html.escape(str(rating))}</td> | |
</tr> | |
""" | |
table_html += "</table>" | |
return table_html | |
def update_page(page, entries_per_page): | |
try: | |
conversations, total_pages, total_count = get_all_conversations(page, entries_per_page) | |
results_html = format_conversations_table(conversations) | |
pagination = f"Page {page} of {total_pages} (Total conversations: {total_count})" | |
next_disabled = page >= total_pages | |
prev_disabled = page <= 1 | |
return ( | |
results_html, | |
pagination, | |
page, | |
gr.update(interactive=not next_disabled), | |
gr.update(interactive=not prev_disabled) | |
) | |
except Exception as e: | |
return ( | |
f"<p>Error: {str(e)}</p>", | |
"Error in pagination", | |
page, | |
gr.update(interactive=False), | |
gr.update(interactive=False) | |
) | |
def go_to_next_page(current_page, entries_per_page): | |
return update_page(current_page + 1, entries_per_page) | |
def go_to_previous_page(current_page, entries_per_page): | |
return update_page(max(1, current_page - 1), entries_per_page) | |
view_button.click( | |
fn=update_page, | |
inputs=[page_number, entries_per_page], | |
outputs=[results_display, pagination_info, page_number, next_page_button, previous_page_button] | |
) | |
next_page_button.click( | |
fn=go_to_next_page, | |
inputs=[page_number, entries_per_page], | |
outputs=[results_display, pagination_info, page_number, next_page_button, previous_page_button] | |
) | |
previous_page_button.click( | |
fn=go_to_previous_page, | |
inputs=[page_number, entries_per_page], | |
outputs=[results_display, pagination_info, page_number, next_page_button, previous_page_button] | |
) | |
def create_view_all_rag_notes_tab(): | |
with gr.TabItem("View All RAG notes/Conversation Items", visible=True): | |
gr.Markdown("# View All RAG Notes/Conversation Entries") | |
with gr.Row(): | |
with gr.Column(scale=1): | |
entries_per_page = gr.Dropdown(choices=[10, 20, 50, 100], label="Entries per Page", value=10) | |
page_number = gr.Number(value=1, label="Page Number", precision=0) | |
view_button = gr.Button("View Page") | |
next_page_button = gr.Button("Next Page") | |
previous_page_button = gr.Button("Previous Page") | |
with gr.Column(scale=2): | |
items_output = gr.Dropdown(label="Select Conversation to View Details", choices=[]) | |
conversation_title = gr.Textbox(label="Conversation Title", visible=True) | |
with gr.Row(): | |
with gr.Column(scale=1): | |
pagination_info = gr.Textbox(label="Pagination Info", interactive=False) | |
with gr.Column(scale=2): | |
keywords_output = gr.Textbox(label="Keywords", visible=True) | |
chat_history_output = gr.HTML(label="Chat History", visible=True) | |
notes_output = gr.HTML(label="Associated Notes", visible=True) | |
item_mapping = gr.State({}) | |
def update_page(page, entries_per_page): | |
try: | |
conversations, total_pages, total_count = get_all_conversations(page, entries_per_page) | |
pagination = f"Page {page} of {total_pages} (Total conversations: {total_count})" | |
# Handle the dictionary structure correctly | |
choices = [f"{conv['title']} (ID: {conv['conversation_id']})" for conv in conversations] | |
new_item_mapping = { | |
f"{conv['title']} (ID: {conv['conversation_id']})": conv['conversation_id'] | |
for conv in conversations | |
} | |
next_disabled = page >= total_pages | |
prev_disabled = page <= 1 | |
return ( | |
gr.update(choices=choices, value=None), | |
pagination, | |
page, | |
gr.update(interactive=not next_disabled), | |
gr.update(interactive=not prev_disabled), | |
"", # conversation_title | |
"", # keywords_output | |
"", # chat_history_output | |
"", # notes_output | |
new_item_mapping | |
) | |
except Exception as e: | |
logging.error(f"Error in update_page: {str(e)}", exc_info=True) | |
return ( | |
gr.update(choices=[], value=None), | |
f"Error: {str(e)}", | |
page, | |
gr.update(interactive=False), | |
gr.update(interactive=False), | |
"", "", "", "", | |
{} | |
) | |
def format_as_html(content, title): | |
if content is None: | |
content = "No content available." | |
escaped_content = html.escape(str(content)) | |
formatted_content = escaped_content.replace('\n', '<br>') | |
return f""" | |
<div style="border: 1px solid #ddd; padding: 10px; margin-bottom: 10px;"> | |
<h3>{title}</h3> | |
<div style="max-height: 700px; overflow-y: auto;"> | |
{formatted_content} | |
</div> | |
</div> | |
""" | |
def format_chat_history(messages): | |
html_content = "<div style='max-height: 500px; overflow-y: auto;'>" | |
for role, content in messages: | |
role_class = "assistant" if role.lower() == "assistant" else "user" | |
html_content += f""" | |
<div class='{role_class}-message' style='margin: 10px 0; padding: 10px; border-radius: 5px; | |
background-color: {"#f0f0f0" if role_class == "user" else "#e3f2fd"}'> | |
<strong>{html.escape(role)}:</strong><br> | |
{html.escape(content)} | |
</div> | |
""" | |
html_content += "</div>" | |
return html_content | |
def display_conversation_details(selected_item, item_mapping): | |
if selected_item and item_mapping and selected_item in item_mapping: | |
conv_id = item_mapping[selected_item] | |
# Get keywords | |
keywords = get_keywords_for_conversation(conv_id) | |
keywords_text = ", ".join(keywords) if keywords else "No keywords" | |
# Get chat history | |
chat_messages, _, _ = load_chat_history(conv_id) | |
chat_html = format_chat_history(chat_messages) | |
# Get associated notes | |
notes = get_notes(conv_id) | |
notes_html = "" | |
for note in notes: | |
notes_html += format_as_html(note, "Note") | |
if not notes: | |
notes_html = "<p>No notes associated with this conversation.</p>" | |
return ( | |
selected_item.split(" (ID:")[0], # Conversation title | |
keywords_text, | |
chat_html, | |
notes_html | |
) | |
return "", "", "", "" | |
view_button.click( | |
fn=update_page, | |
inputs=[page_number, entries_per_page], | |
outputs=[ | |
items_output, | |
pagination_info, | |
page_number, | |
next_page_button, | |
previous_page_button, | |
conversation_title, | |
keywords_output, | |
chat_history_output, | |
notes_output, | |
item_mapping | |
] | |
) | |
next_page_button.click( | |
fn=lambda page, entries: update_page(page + 1, entries), | |
inputs=[page_number, entries_per_page], | |
outputs=[items_output, pagination_info, page_number, next_page_button, previous_page_button, | |
conversation_title, keywords_output, chat_history_output, notes_output, item_mapping] | |
) | |
previous_page_button.click( | |
fn=lambda page, entries: update_page(max(1, page - 1), entries), | |
inputs=[page_number, entries_per_page], | |
outputs=[items_output, pagination_info, page_number, next_page_button, previous_page_button, | |
conversation_title, keywords_output, chat_history_output, notes_output, item_mapping] | |
) | |
items_output.change( | |
fn=display_conversation_details, | |
inputs=[items_output, item_mapping], | |
outputs=[conversation_title, keywords_output, chat_history_output, notes_output] | |
) | |
def create_ragdb_keyword_items_tab(): | |
with gr.TabItem("View RAG Notes/Conversations by Keyword", visible=True): | |
gr.Markdown("# View RAG Notes and Conversations by Keyword") | |
with gr.Row(): | |
with gr.Column(scale=1): | |
# Keyword selection | |
keyword_dropdown = gr.Dropdown( | |
label="Select Keyword", | |
choices=[], | |
value=None, | |
multiselect=True | |
) | |
entries_per_page = gr.Dropdown( | |
choices=[10, 20, 50, 100], | |
label="Entries per Page", | |
value=10 | |
) | |
page_number = gr.Number( | |
value=1, | |
label="Page Number", | |
precision=0 | |
) | |
# Navigation buttons | |
refresh_keywords_button = gr.Button("Refresh Keywords") | |
view_button = gr.Button("View Items") | |
next_page_button = gr.Button("Next Page") | |
previous_page_button = gr.Button("Previous Page") | |
pagination_info = gr.Textbox( | |
label="Pagination Info", | |
interactive=False | |
) | |
with gr.Column(scale=2): | |
# Results tabs for conversations and notes | |
with gr.Tabs(): | |
with gr.Tab("Notes"): | |
notes_results = gr.HTML() | |
with gr.Tab("Conversations"): | |
conversation_results = gr.HTML() | |
def update_keyword_choices(): | |
"""Fetch all available keywords for the dropdown.""" | |
try: | |
query = "SELECT keyword FROM rag_qa_keywords ORDER BY keyword" | |
with get_db_connection() as conn: | |
cursor = conn.cursor() | |
cursor.execute(query) | |
keywords = [row[0] for row in cursor.fetchall()] | |
return gr.update(choices=keywords) | |
except Exception as e: | |
return gr.update(choices=[], value=None) | |
def format_conversations_html(conversations_data): | |
"""Format conversations data as HTML.""" | |
if not conversations_data: | |
return "<p>No conversations found for selected keywords.</p>" | |
html_content = "<div class='results-container'>" | |
for conv_id, title in conversations_data: | |
html_content += f""" | |
<div style='border: 1px solid #ddd; padding: 10px; margin-bottom: 10px;'> | |
<h3>{html.escape(title)}</h3> | |
<p>Conversation ID: {html.escape(conv_id)}</p> | |
<p><strong>Keywords:</strong> {', '.join(html.escape(k) for k in get_keywords_for_conversation(conv_id))}</p> | |
</div> | |
""" | |
html_content += "</div>" | |
return html_content | |
def format_notes_html(notes_data): | |
"""Format notes data as HTML.""" | |
if not notes_data: | |
return "<p>No notes found for selected keywords.</p>" | |
html_content = "<div class='results-container'>" | |
for note_id, title, content, timestamp in notes_data: | |
keywords = get_keywords_for_note(note_id) | |
html_content += f""" | |
<div style='border: 1px solid #ddd; padding: 10px; margin-bottom: 10px;'> | |
<h3>{html.escape(title)}</h3> | |
<p><strong>Created:</strong> {timestamp}</p> | |
<p><strong>Keywords:</strong> {', '.join(html.escape(k) for k in keywords)}</p> | |
<div style='background: #f5f5f5; padding: 10px; margin-top: 10px;'> | |
{html.escape(content)} | |
</div> | |
</div> | |
""" | |
html_content += "</div>" | |
return html_content | |
def view_items(keywords, page, entries_per_page): | |
if not keywords or (isinstance(keywords, list) and len(keywords) == 0): | |
return ( | |
"<p>Please select at least one keyword.</p>", | |
"<p>Please select at least one keyword.</p>", | |
"No results", | |
gr.update(interactive=False), | |
gr.update(interactive=False) | |
) | |
try: | |
# Ensure keywords is a list | |
keywords_list = keywords if isinstance(keywords, list) else [keywords] | |
# Get conversations for selected keywords | |
conversations, conv_total_pages, conv_count = search_conversations_by_keywords( | |
keywords_list, page, entries_per_page | |
) | |
# Get notes for selected keywords | |
notes, notes_total_pages, notes_count = get_notes_by_keywords( | |
keywords_list, page, entries_per_page | |
) | |
# Format results as HTML | |
conv_html = format_conversations_html(conversations) | |
notes_html = format_notes_html(notes) | |
# Create pagination info | |
pagination = f"Page {page} of {max(conv_total_pages, notes_total_pages)} " | |
pagination += f"(Conversations: {conv_count}, Notes: {notes_count})" | |
# Determine button states | |
max_pages = max(conv_total_pages, notes_total_pages) | |
next_disabled = page >= max_pages | |
prev_disabled = page <= 1 | |
return ( | |
conv_html, | |
notes_html, | |
pagination, | |
gr.update(interactive=not next_disabled), | |
gr.update(interactive=not prev_disabled) | |
) | |
except Exception as e: | |
logging.error(f"Error in view_items: {str(e)}") | |
return ( | |
f"<p>Error: {str(e)}</p>", | |
f"<p>Error: {str(e)}</p>", | |
"Error in retrieval", | |
gr.update(interactive=False), | |
gr.update(interactive=False) | |
) | |
def go_to_next_page(keywords, current_page, entries_per_page): | |
return view_items(keywords, current_page + 1, entries_per_page) | |
def go_to_previous_page(keywords, current_page, entries_per_page): | |
return view_items(keywords, max(1, current_page - 1), entries_per_page) | |
# Event handlers | |
refresh_keywords_button.click( | |
fn=update_keyword_choices, | |
inputs=[], | |
outputs=[keyword_dropdown] | |
) | |
view_button.click( | |
fn=view_items, | |
inputs=[keyword_dropdown, page_number, entries_per_page], | |
outputs=[ | |
conversation_results, | |
notes_results, | |
pagination_info, | |
next_page_button, | |
previous_page_button | |
] | |
) | |
next_page_button.click( | |
fn=go_to_next_page, | |
inputs=[keyword_dropdown, page_number, entries_per_page], | |
outputs=[ | |
conversation_results, | |
notes_results, | |
pagination_info, | |
next_page_button, | |
previous_page_button | |
] | |
) | |
previous_page_button.click( | |
fn=go_to_previous_page, | |
inputs=[keyword_dropdown, page_number, entries_per_page], | |
outputs=[ | |
conversation_results, | |
notes_results, | |
pagination_info, | |
next_page_button, | |
previous_page_button | |
] | |
) | |
# Initialize keyword dropdown on page load | |
keyword_dropdown.value = update_keyword_choices() | |
# | |
# End of RAG DB Viewing tabs | |
################################################################ | |
# | |
####################################################################################################################### | |