# 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}

{formatted_content}
""" 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', '
') return f"""

{title}

{formatted_content}
""" 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_html += "" table_html += "" for item_id, title, url in paginated_items: table_html += f""" """ table_html += "
TitleURL
{html.escape(title)} {html.escape(url)}
" # 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"

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 = "" table_html += """ """ 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""" """ table_html += "
Title Keywords Notes Rating
{html.escape(str(title))} {html.escape(', '.join(keywords))} {len(notes)} note(s) {html.escape(str(rating))}
" 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"

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', '
') return f"""

{title}

{formatted_content}
""" def format_chat_history(messages): html_content = "
" for role, content in messages: role_class = "assistant" if role.lower() == "assistant" else "user" html_content += f"""
{html.escape(role)}:
{html.escape(content)}
""" html_content += "
" 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 = "

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 = "
" for conv_id, title in conversations_data: html_content += f"""

{html.escape(title)}

Conversation ID: {html.escape(conv_id)}

Keywords: {', '.join(html.escape(k) for k in get_keywords_for_conversation(conv_id))}

""" html_content += "
" return html_content def format_notes_html(notes_data): """Format notes data as HTML.""" if not notes_data: return "

No notes found for selected keywords.

" html_content = "
" for note_id, title, content, timestamp in notes_data: keywords = get_keywords_for_note(note_id) html_content += f"""

{html.escape(title)}

Created: {timestamp}

Keywords: {', '.join(html.escape(k) for k in keywords)}

{html.escape(content)}
""" html_content += "
" return html_content def view_items(keywords, page, entries_per_page): if not keywords or (isinstance(keywords, list) and len(keywords) == 0): return ( "

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 ################################################################ # #######################################################################################################################