# 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', '
')
return f"""
Title | " table_html += "URL |
---|---|
{html.escape(title)} | {html.escape(url)} |
Error: {str(e)}
", "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 = "Title | Keywords | Notes | Rating |
---|---|---|---|
{html.escape(str(title))} | {html.escape(', '.join(keywords))} | {len(notes)} note(s) | {html.escape(str(rating))} |
Error: {str(e)}
", "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', 'No notes associated with this conversation.
" 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 "No conversations found for selected keywords.
" html_content = "Conversation ID: {html.escape(conv_id)}
Keywords: {', '.join(html.escape(k) for k in get_keywords_for_conversation(conv_id))}
No notes found for selected keywords.
" html_content = "Created: {timestamp}
Keywords: {', '.join(html.escape(k) for k in keywords)}
Please select at least one keyword.
", "Please select at least one keyword.
", "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"Error: {str(e)}
", f"Error: {str(e)}
", "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 ################################################################ # #######################################################################################################################