Spaces:
Running
Running
| import shutil | |
| import os | |
| from pathlib import Path | |
| from typing import List | |
| from datetime import datetime | |
| import mimetypes | |
| async def organize_files(directory: str, strategy: str = 'by_type') -> dict: | |
| """ | |
| Organize files in a directory | |
| Args: | |
| directory: Directory to organize | |
| strategy: Organization strategy (by_type, by_date, by_size) | |
| Returns: | |
| Dict with organization results | |
| """ | |
| try: | |
| dir_path = Path(directory) | |
| if not dir_path.exists(): | |
| return {'error': 'Directory not found', 'success': False} | |
| files = [f for f in dir_path.iterdir() if f.is_file()] | |
| if strategy == 'by_type': | |
| return await organize_by_type(files, dir_path) | |
| elif strategy == 'by_date': | |
| return await organize_by_date(files, dir_path) | |
| elif strategy == 'by_size': | |
| return await organize_by_size(files, dir_path) | |
| else: | |
| return {'error': 'Unknown strategy', 'success': False} | |
| except Exception as e: | |
| return {'error': str(e), 'success': False} | |
| async def organize_by_type(files: List[Path], base_dir: Path) -> dict: | |
| """Organize files by file type""" | |
| type_map = { | |
| 'documents': ['.pdf', '.doc', '.docx', '.txt', '.odt'], | |
| 'images': ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.svg'], | |
| 'spreadsheets': ['.xlsx', '.xls', '.csv'], | |
| 'archives': ['.zip', '.rar', '.7z', '.tar', '.gz'], | |
| 'other': [] | |
| } | |
| moved_files = {} | |
| for file in files: | |
| ext = file.suffix.lower() | |
| # Find category | |
| category = 'other' | |
| for cat, extensions in type_map.items(): | |
| if ext in extensions: | |
| category = cat | |
| break | |
| # Create category folder | |
| cat_dir = base_dir / category | |
| cat_dir.mkdir(exist_ok=True) | |
| # Move file | |
| dest = cat_dir / file.name | |
| try: | |
| shutil.move(str(file), str(dest)) | |
| if category not in moved_files: | |
| moved_files[category] = [] | |
| moved_files[category].append(file.name) | |
| except Exception as e: | |
| print(f"Error moving {file.name}: {e}") | |
| return { | |
| 'success': True, | |
| 'strategy': 'by_type', | |
| 'files_moved': sum(len(v) for v in moved_files.values()), | |
| 'categories': moved_files | |
| } | |
| async def organize_by_date(files: List[Path], base_dir: Path) -> dict: | |
| """Organize files by modification date""" | |
| moved_files = {} | |
| for file in files: | |
| # Get modification time | |
| mtime = datetime.fromtimestamp(file.stat().st_mtime) | |
| date_folder = mtime.strftime('%Y-%m') | |
| # Create date folder | |
| date_dir = base_dir / date_folder | |
| date_dir.mkdir(exist_ok=True) | |
| # Move file | |
| dest = date_dir / file.name | |
| try: | |
| shutil.move(str(file), str(dest)) | |
| if date_folder not in moved_files: | |
| moved_files[date_folder] = [] | |
| moved_files[date_folder].append(file.name) | |
| except Exception as e: | |
| print(f"Error moving {file.name}: {e}") | |
| return { | |
| 'success': True, | |
| 'strategy': 'by_date', | |
| 'files_moved': sum(len(v) for v in moved_files.values()), | |
| 'folders': moved_files | |
| } | |
| async def organize_by_size(files: List[Path], base_dir: Path) -> dict: | |
| """Organize files by size""" | |
| size_categories = { | |
| 'small': (0, 1024 * 1024), # < 1MB | |
| 'medium': (1024 * 1024, 10 * 1024 * 1024), # 1-10MB | |
| 'large': (10 * 1024 * 1024, float('inf')) # > 10MB | |
| } | |
| moved_files = {} | |
| for file in files: | |
| size = file.stat().st_size | |
| # Determine category | |
| category = 'small' | |
| for cat, (min_size, max_size) in size_categories.items(): | |
| if min_size <= size < max_size: | |
| category = cat | |
| break | |
| # Create category folder | |
| cat_dir = base_dir / category | |
| cat_dir.mkdir(exist_ok=True) | |
| # Move file | |
| dest = cat_dir / file.name | |
| try: | |
| shutil.move(str(file), str(dest)) | |
| if category not in moved_files: | |
| moved_files[category] = [] | |
| moved_files[category].append(file.name) | |
| except Exception as e: | |
| print(f"Error moving {file.name}: {e}") | |
| return { | |
| 'success': True, | |
| 'strategy': 'by_size', | |
| 'files_moved': sum(len(v) for v in moved_files.values()), | |
| 'categories': moved_files | |
| } | |
| async def rename_file(file_path: str, new_name: str) -> dict: | |
| """Rename file with semantic name""" | |
| try: | |
| old_path = Path(file_path) | |
| if not old_path.exists(): | |
| return {'error': 'File not found', 'success': False} | |
| # Keep extension | |
| ext = old_path.suffix | |
| if not new_name.endswith(ext): | |
| new_name += ext | |
| new_path = old_path.parent / new_name | |
| old_path.rename(new_path) | |
| return { | |
| 'success': True, | |
| 'old_name': old_path.name, | |
| 'new_name': new_name, | |
| 'new_path': str(new_path) | |
| } | |
| except Exception as e: | |
| return {'error': str(e), 'success': False} |