simplified file_drop
This commit is contained in:
@@ -3,8 +3,8 @@ from typing import Dict, Any, List
|
||||
import base64
|
||||
from PIL import Image
|
||||
import io
|
||||
from niceguiex.components import FileDrop, ImageDrop
|
||||
from niceguiex.components.file_drop import FileData, ImageData
|
||||
from niceguiex.components import FileDrop
|
||||
from tempfile import SpooledTemporaryFile
|
||||
|
||||
|
||||
@ui.page('/')
|
||||
@@ -14,82 +14,21 @@ async def main_page():
|
||||
# Example 1: Multiple files with content
|
||||
uploaded_files = ui.column().classes('w-full mt-4')
|
||||
|
||||
async def handle_multiple_files(file: FileData):
|
||||
async def handle_multiple_files(file_name: str, file_type: str, file_content: SpooledTemporaryFile):
|
||||
with uploaded_files:
|
||||
with ui.card().classes('p-2 mb-2'):
|
||||
ui.label(f"📄 {file['name']}").classes('font-medium')
|
||||
ui.label(f"Size: {file['size']:,} bytes").classes('text-sm text-gray-600')
|
||||
ui.label(f"Type: {file['file_type']}").classes('text-sm text-gray-600')
|
||||
ui.label(f"📄 {file_name}").classes('font-medium')
|
||||
ui.label(f"Type: {file_type}").classes('text-sm text-gray-600')
|
||||
|
||||
ui.label('Multiple Files (returns list)').classes('text-h6 mb-2')
|
||||
FileDrop(
|
||||
on_upload=handle_multiple_files,
|
||||
multiple=True,
|
||||
accept='.pdf,.docx,.txt,.jpg,.png',
|
||||
max_size=15 # 5MB limit
|
||||
).classes('w-full max-w-xl mx-auto')
|
||||
|
||||
ui.separator().classes('my-8')
|
||||
|
||||
# Example 2: Single file (returns dict directly)
|
||||
single_file_info = ui.column().classes('w-full mt-4')
|
||||
|
||||
def handle_single_file(file: Dict[str, Any]):
|
||||
single_file_info.clear()
|
||||
with single_file_info:
|
||||
ui.label(f"Single file uploaded: {file['name']}").classes('font-bold')
|
||||
ui.label(f"Size: {file['size']:,} bytes")
|
||||
|
||||
ui.label('Single File (returns dict)').classes('text-h6 mb-2')
|
||||
FileDrop(
|
||||
on_upload=handle_single_file,
|
||||
multiple=False,
|
||||
accept='.pdf,.docx,.txt'
|
||||
).classes('w-full max-w-xl mx-auto')
|
||||
|
||||
ui.separator().classes('my-8')
|
||||
|
||||
# Example 3: Large files with temp path
|
||||
large_file_info = ui.column().classes('w-full mt-4')
|
||||
|
||||
def handle_large_file(file: Dict[str, Any]):
|
||||
large_file_info.clear()
|
||||
with large_file_info:
|
||||
ui.label(f"Large file saved to: {file['path']}").classes('font-bold')
|
||||
ui.label(f"Name: {file['name']}")
|
||||
ui.label(f"Size: {file['size']:,} bytes")
|
||||
|
||||
ui.label('Large Files (returns temp path)').classes('text-h6 mb-2')
|
||||
FileDrop(
|
||||
on_upload=handle_large_file,
|
||||
multiple=False,
|
||||
return_content=False, # Returns path instead of content
|
||||
accept='video/*'
|
||||
).classes('w-full max-w-xl mx-auto')
|
||||
|
||||
ui.separator().classes('my-8')
|
||||
|
||||
# Example 4: Image drop with preview
|
||||
image_preview = ui.column().classes('w-full mt-4')
|
||||
|
||||
async def handle_image(img: ImageData):
|
||||
print(f"Image uploaded: {img['image'].format}, Size: {img['size']}")
|
||||
image_preview.clear()
|
||||
with image_preview:
|
||||
ui.label(f"Image: {img['image'].format} - {img['image'].size[0]}x{img['image'].size[1]}px").classes('mb-2')
|
||||
# Convert PIL image back to base64 for display
|
||||
buffered = io.BytesIO()
|
||||
img['image'].save(buffered, format=img['image'].format or 'PNG')
|
||||
img_data = base64.b64encode(buffered.getvalue()).decode()
|
||||
ui.html(f'<img src="data:image/{img['image'].format.lower() if img['image'].format else "png"};base64,{img_data}" class="max-w-full rounded">')
|
||||
|
||||
ui.label('Image Drop (returns PIL Image)').classes('text-h6 mb-2')
|
||||
ImageDrop(
|
||||
on_upload=handle_image,
|
||||
multiple=False,
|
||||
max_size=15 # 15MB limit for images
|
||||
).classes('w-full max-w-xl mx-auto').style('min-height: 150px')
|
||||
|
||||
if __name__ in {"__main__", "__mp_main__"}:
|
||||
ui.run(
|
||||
title='File Drop Demo',
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from .auto_scroll_area import AutoScrollArea
|
||||
from .chat_input import ChatInput
|
||||
from .file_drop import FileDrop, ImageDrop, FileData, ImageData
|
||||
from .file_drop import FileDrop
|
||||
|
||||
__all__ = ['AutoScrollArea', 'ChatInput', 'FileDrop', 'ImageDrop', 'FileData', 'ImageData']
|
||||
__all__ = ['AutoScrollArea', 'ChatInput', 'FileDrop']
|
||||
|
||||
@@ -6,28 +6,11 @@ from PIL import Image
|
||||
import io
|
||||
|
||||
|
||||
class FileData(TypedDict):
|
||||
name: str
|
||||
size: int
|
||||
file_type: str
|
||||
content: Optional[Any]
|
||||
path: Optional[Path]
|
||||
|
||||
|
||||
class ImageData(TypedDict):
|
||||
name: str
|
||||
size: int
|
||||
file_type: str
|
||||
image: Image.Image
|
||||
|
||||
|
||||
class FileDrop(ui.element):
|
||||
def __init__(self,
|
||||
on_upload: Optional[Callable] = None,
|
||||
multiple: bool = False,
|
||||
accept: Optional[str] = None,
|
||||
max_size: Optional[int] = None, # Max size in MB
|
||||
return_content: bool = True, # If False, returns temp file path
|
||||
*args, **kwargs):
|
||||
|
||||
super().__init__(tag='div', *args, **kwargs)
|
||||
@@ -35,8 +18,6 @@ class FileDrop(ui.element):
|
||||
self.on_upload_callback = on_upload
|
||||
self.multiple = multiple
|
||||
self.accept = accept
|
||||
self.max_size = max_size
|
||||
self.return_content = return_content
|
||||
|
||||
# Style the container
|
||||
self.classes('relative w-full')
|
||||
@@ -63,10 +44,6 @@ class FileDrop(ui.element):
|
||||
ui.label('Drop files here or click to browse').classes('text-lg text-gray-600 mb-2')
|
||||
ui.label('Drag and drop your files').classes('text-sm text-gray-400')
|
||||
|
||||
"""with self.upload.add_slot('list'):
|
||||
with ui.column().classes('w-full'):
|
||||
ui.label('Testing')"""
|
||||
|
||||
# Add drag over styling with JavaScript
|
||||
ui.timer(0.1, lambda: self._add_drag_styling(), once=True)
|
||||
|
||||
@@ -115,84 +92,4 @@ class FileDrop(ui.element):
|
||||
|
||||
# Process single file upload
|
||||
if hasattr(e, 'content'):
|
||||
print('content')
|
||||
file_data = await self._process_file(e)
|
||||
if file_data:
|
||||
# For single file mode, return the dict directly
|
||||
await self.on_upload_callback(file_data)
|
||||
|
||||
async def _process_file(self, file_obj) -> FileData | None:
|
||||
"""Process a single file object"""
|
||||
if not hasattr(file_obj, 'content'):
|
||||
return None
|
||||
|
||||
content = file_obj.content.read()
|
||||
size = len(content)
|
||||
|
||||
# Check file size if max_size is set
|
||||
if self.max_size and size > self.max_size * 1024 * 1024:
|
||||
ui.notify(f"File too large. Max size is {self.max_size}MB", type='negative')
|
||||
return None
|
||||
|
||||
file_data: FileData = {'name': file_obj.name if hasattr(file_obj, 'name') else 'unknown',
|
||||
'size': size,
|
||||
'file_type': file_obj.type if hasattr(file_obj, 'type') else 'application/octet-stream',
|
||||
'content': None,
|
||||
'path': None}
|
||||
|
||||
# Return content or temp file path based on settings
|
||||
if self.return_content:
|
||||
file_data['content'] = content
|
||||
else:
|
||||
# Save to temp file and return path
|
||||
suffix = Path(file_data['name']).suffix
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
|
||||
tmp.write(content)
|
||||
file_data['path'] = Path(tmp.name)
|
||||
|
||||
return file_data
|
||||
|
||||
|
||||
class ImageDrop(FileDrop):
|
||||
"""Specialized file drop for images that returns PIL Image objects"""
|
||||
|
||||
def __init__(self,
|
||||
on_upload: Optional[Callable] = None,
|
||||
multiple: bool = False,
|
||||
max_size: int = 50, # Default 50MB for images
|
||||
*args, **kwargs):
|
||||
|
||||
# Store the user's callback
|
||||
self._user_callback = on_upload
|
||||
|
||||
# Use our own handler that converts to PIL images
|
||||
super().__init__(
|
||||
on_upload=self._handle_image_upload if on_upload else None,
|
||||
multiple=multiple,
|
||||
accept='image/*',
|
||||
max_size=max_size,
|
||||
return_content=True, # We need content to create PIL images
|
||||
*args, **kwargs
|
||||
)
|
||||
|
||||
async def _handle_image_upload(self, data: FileData):
|
||||
"""Convert file data to PIL Images before calling user callback"""
|
||||
if self._user_callback:
|
||||
if isinstance(data, list) and len(data) == 1:
|
||||
img = Image.open(io.BytesIO(data[0]['content']))
|
||||
await self._user_callback(img)
|
||||
elif isinstance(data, list):
|
||||
# Multiple images - convert each to PIL Image
|
||||
images = []
|
||||
for file_data in data:
|
||||
img = Image.open(io.BytesIO(file_data['content']))
|
||||
images.append(img)
|
||||
await self._user_callback(images)
|
||||
else:
|
||||
# Single image - convert to PIL Image
|
||||
image_data: ImageData = {'name': data['name'],
|
||||
'size': data['size'],
|
||||
'file_type': data['file_type'],
|
||||
'image': Image.open(io.BytesIO(data['content']))}
|
||||
# img = Image.open(io.BytesIO(data['content']))
|
||||
await self._user_callback(image_data)
|
||||
await self.on_upload_callback(e.name, e.type, e.content)
|
||||
|
||||
Reference in New Issue
Block a user