Spaces:
Sleeping
Sleeping
File size: 9,431 Bytes
ed28876 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 |
import gradio as gr
import logging
import sqlite3
from typing import List, Dict
import os
import zipfile
import tempfile
import shutil
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Database connection (you'll need to set this up)
db = None # Replace with your actual database connection
class DatabaseError(Exception):
pass
# Database functions
def fetch_items_by_keyword(search_query: str) -> List[Dict]:
try:
with db.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT m.id, m.title, m.url
FROM Media m
JOIN MediaKeywords mk ON m.id = mk.media_id
JOIN Keywords k ON mk.keyword_id = k.id
WHERE k.keyword LIKE ?
""", (f'%{search_query}%',))
results = cursor.fetchall()
return [{"id": r[0], "title": r[1], "url": r[2]} for r in results]
except sqlite3.Error as e:
logger.error(f"Error fetching items by keyword: {e}")
raise DatabaseError(f"Error fetching items by keyword: {e}")
def fetch_item_details(media_id: int) -> tuple:
try:
with db.get_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT prompt, summary
FROM MediaModifications
WHERE media_id = ?
ORDER BY modification_date DESC
LIMIT 1
""", (media_id,))
prompt_summary_result = cursor.fetchone()
cursor.execute("SELECT content FROM Media WHERE id = ?", (media_id,))
content_result = cursor.fetchone()
prompt = prompt_summary_result[0] if prompt_summary_result else ""
summary = prompt_summary_result[1] if prompt_summary_result else ""
content = content_result[0] if content_result else ""
return content, prompt, summary
except sqlite3.Error as e:
logger.error(f"Error fetching item details: {e}")
return "", "", ""
def browse_items(search_query: str, search_type: str) -> List[Dict]:
try:
with db.get_connection() as conn:
cursor = conn.cursor()
if search_type == 'Title':
cursor.execute("SELECT id, title, url FROM Media WHERE title LIKE ?", (f'%{search_query}%',))
elif search_type == 'URL':
cursor.execute("SELECT id, title, url FROM Media WHERE url LIKE ?", (f'%{search_query}%',))
elif search_type == 'Keyword':
return fetch_items_by_keyword(search_query)
elif search_type == 'Content':
cursor.execute("SELECT id, title, url FROM Media WHERE content LIKE ?", (f'%{search_query}%',))
else:
raise ValueError(f"Invalid search type: {search_type}")
results = cursor.fetchall()
return [{"id": r[0], "title": r[1], "url": r[2]} for r in results]
except sqlite3.Error as e:
logger.error(f"Error fetching items by {search_type}: {e}")
raise DatabaseError(f"Error fetching items by {search_type}: {e}")
# Export functions
def export_item_as_markdown(media_id: int) -> str:
try:
content, prompt, summary = fetch_item_details(media_id)
title = f"Item {media_id}" # You might want to fetch the actual title
markdown_content = f"# {title}\n\n## Prompt\n{prompt}\n\n## Summary\n{summary}\n\n## Content\n{content}"
filename = f"export_item_{media_id}.md"
with open(filename, "w", encoding='utf-8') as f:
f.write(markdown_content)
logger.info(f"Successfully exported item {media_id} to {filename}")
return filename
except Exception as e:
logger.error(f"Error exporting item {media_id}: {str(e)}")
return None
def export_items_by_keyword(keyword: str) -> str:
try:
items = fetch_items_by_keyword(keyword)
if not items:
logger.warning(f"No items found for keyword: {keyword}")
return None
# Create a temporary directory to store individual markdown files
with tempfile.TemporaryDirectory() as temp_dir:
folder_name = f"export_keyword_{keyword}"
export_folder = os.path.join(temp_dir, folder_name)
os.makedirs(export_folder)
for item in items:
content, prompt, summary = fetch_item_details(item['id'])
markdown_content = f"# {item['title']}\n\n## Prompt\n{prompt}\n\n## Summary\n{summary}\n\n## Content\n{content}"
# Create individual markdown file for each item
file_name = f"{item['id']}_{item['title'][:50]}.md" # Limit filename length
file_path = os.path.join(export_folder, file_name)
with open(file_path, "w", encoding='utf-8') as f:
f.write(markdown_content)
# Create a zip file containing all markdown files
zip_filename = f"{folder_name}.zip"
shutil.make_archive(os.path.join(temp_dir, folder_name), 'zip', export_folder)
# Move the zip file to a location accessible by Gradio
final_zip_path = os.path.join(os.getcwd(), zip_filename)
shutil.move(os.path.join(temp_dir, zip_filename), final_zip_path)
logger.info(f"Successfully exported {len(items)} items for keyword '{keyword}' to {zip_filename}")
return final_zip_path
except Exception as e:
logger.error(f"Error exporting items for keyword '{keyword}': {str(e)}")
return None
def export_selected_items(selected_items: List[Dict]) -> str:
try:
if not selected_items:
logger.warning("No items selected for export")
return None
markdown_content = "# Selected Items\n\n"
for item in selected_items:
content, prompt, summary = fetch_item_details(item['id'])
markdown_content += f"## {item['title']}\n\n### Prompt\n{prompt}\n\n### Summary\n{summary}\n\n### Content\n{content}\n\n---\n\n"
filename = "export_selected_items.md"
with open(filename, "w", encoding='utf-8') as f:
f.write(markdown_content)
logger.info(f"Successfully exported {len(selected_items)} selected items to {filename}")
return filename
except Exception as e:
logger.error(f"Error exporting selected items: {str(e)}")
return None
# Gradio interface functions
def display_search_results(search_query: str, search_type: str) -> List[Dict]:
try:
results = browse_items(search_query, search_type)
return [{"name": f"{item['title']} ({item['url']})", "value": item} for item in results]
except DatabaseError as e:
logger.error(f"Error in display_search_results: {str(e)}")
return []
# Gradio interface
with gr.Blocks() as demo:
gr.Markdown("# Content Export Interface")
with gr.Tab("Search and Export"):
search_query = gr.Textbox(label="Search Query")
search_type = gr.Radio(["Title", "URL", "Keyword", "Content"], label="Search By")
search_button = gr.Button("Search")
search_results = gr.CheckboxGroup(label="Search Results")
export_selected_button = gr.Button("Export Selected Items")
keyword_input = gr.Textbox(label="Enter keyword for export")
export_by_keyword_button = gr.Button("Export items by keyword")
export_output = gr.File(label="Download Exported File")
error_output = gr.Textbox(label="Status/Error Messages", interactive=False)
search_button.click(
fn=display_search_results,
inputs=[search_query, search_type],
outputs=[search_results, error_output]
)
export_selected_button.click(
fn=lambda selected: (export_selected_items(selected), "Exported selected items") if selected else (
None, "No items selected"),
inputs=[search_results],
outputs=[export_output, error_output]
)
export_by_keyword_button.click(
fn=lambda keyword: (
export_items_by_keyword(keyword), f"Exported items for keyword: {keyword}") if keyword else (
None, "No keyword provided"),
inputs=[keyword_input],
outputs=[export_output, error_output]
)
# Add functionality to export individual items
search_results.select(
fn=lambda item: (export_item_as_markdown(item['id']), f"Exported item: {item['title']}") if item else (
None, "No item selected"),
inputs=[gr.State(lambda: search_results.value)],
outputs=[export_output, error_output]
)
demo.launch()
# This modified version of export_items_by_keyword does the following:
#
# Creates a temporary directory to store individual markdown files.
# For each item associated with the keyword, it creates a separate markdown file.
# Places all markdown files in a folder named export_keyword_{keyword}.
# Creates a zip file containing the folder with all markdown files.
# Moves the zip file to a location accessible by Gradio for download. |