LifeAdmin-AI / tools /file_manager_server.py
Maheen001's picture
Update tools/file_manager_server.py
e9105b9 verified
raw
history blame
5.43 kB
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}