Created
November 26, 2024 04:07
-
-
Save shagunmistry/1d0d49197a96df0b9c89362f944a5a73 to your computer and use it in GitHub Desktop.
AI Files Organizer
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import os | |
import shutil | |
from pathlib import Path | |
from typing import List, Dict | |
import mimetypes | |
from datetime import datetime | |
import logging | |
import anthropic | |
import asyncio | |
from getpass import getpass | |
import time | |
class ClaudeFileOrganizer: | |
def __init__(self): | |
"""Initialize the Claude File Organizer.""" | |
# Get API key and directory through user input | |
self.api_key = self._get_api_key() | |
self.source_dir = self._get_directory() | |
self.organized_dir = self.source_dir / "organized" | |
# Initialize Anthropic client | |
self.client = anthropic.Anthropic(api_key=self.api_key) | |
# Setup logging | |
logging.basicConfig( | |
level=logging.INFO, | |
format='%(asctime)s - %(levelname)s - %(message)s', | |
handlers=[ | |
logging.FileHandler('file_organizer.log'), | |
logging.StreamHandler() | |
] | |
) | |
self.logger = logging | |
def _get_api_key(self) -> str: | |
# """Get Anthropic API key from user input.""" | |
print("\n=== Claude File Organizer ===") | |
print("Please enter your Anthropic API key (it will be hidden)") | |
return getpass("API Key: ") | |
def _get_directory(self) -> Path: | |
"""Get directory path from user input.""" | |
while True: | |
dir_path = input("\nEnter the full path of the directory to organize: ") | |
path = Path(dir_path).expanduser().resolve() | |
if not path.exists(): | |
print("β Directory doesn't exist. Please enter a valid path.") | |
continue | |
if not path.is_dir(): | |
print("β Path is not a directory. Please enter a directory path.") | |
continue | |
print(f"β Using directory: {path}") | |
return path | |
def get_file_info(self, file_path: Path) -> Dict: | |
"""Get file information including type, size, and creation date.""" | |
stats = file_path.stat() | |
mime_type, _ = mimetypes.guess_type(file_path) | |
return { | |
"name": file_path.name, | |
"extension": file_path.suffix, | |
"size": stats.st_size, | |
"created": datetime.fromtimestamp(stats.st_ctime).strftime("%Y-%m-%d"), | |
"mime_type": mime_type or "unknown" | |
} | |
def classify_file(self, file_info: Dict) -> str: | |
try: | |
prompt = f""" | |
You are a productivity guru! π§ ππ | |
Based on the following file information, suggest a single appropriate category folder name: | |
Filename: {file_info['name']} | |
Type: {file_info['mime_type']} | |
Created: {file_info['created']} | |
Respond with just the category name (one word, lowercase) from these options: | |
documents, images, audio, video, archives, code, data, downloads, other | |
""" | |
message = self.client.messages.create( | |
model="claude-3-sonnet-20240229", | |
max_tokens=1024, | |
messages=[ | |
{"role": "user", "content": prompt} | |
] | |
) | |
# Fix the response handling | |
category = message.content[0].text.strip().lower() | |
return category if category in ["documents", "images", "audio", "video", "archives", "code", "data", "downloads", "other"] else "other" | |
except Exception as e: | |
self.logger.error(f"Error classifying file {file_info['name']}: {str(e)}") | |
return "other" | |
def organize_files(self): | |
"""Main method to organize files using AI classification.""" | |
try: | |
# Create organized directory if it doesn't exist | |
self.organized_dir.mkdir(exist_ok=True) | |
# Get list of files (excluding directories and already organized files) | |
files = [f for f in self.source_dir.iterdir() | |
if f.is_file() and f != "file_organizer.log" | |
and not str(f).startswith(str(self.organized_dir))] | |
if not files: | |
print("No files found to organize!") | |
return | |
print(f"\nFound {len(files)} files to organize") | |
print("Starting organization process...") | |
# Create progress counter | |
total_files = len(files) | |
processed_files = 0 | |
for file_path in files: | |
try: | |
# sleep for 5 seconds to avoid rate limiting | |
time.sleep(5) | |
# Update progress | |
processed_files += 1 | |
print(f"\rProgress: {processed_files}/{total_files} files processed", end="") | |
# Get file information | |
file_info = self.get_file_info(file_path) | |
# Get AI classification | |
category = self.classify_file(file_info) | |
# Create category directory | |
category_dir = self.organized_dir / category | |
category_dir.mkdir(exist_ok=True) | |
# Move file to its category directory | |
new_path = category_dir / file_path.name | |
if new_path.exists(): | |
# If file exists, append timestamp to filename | |
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
new_path = category_dir / f"{file_path.stem}_{timestamp}{file_path.suffix}" | |
shutil.move(str(file_path), str(new_path)) | |
self.logger.info(f"Moved {file_path.name} to {category}") | |
except Exception as e: | |
self.logger.error(f"Error processing file {file_path}: {str(e)}") | |
continue | |
print("\n\n⨠File organization completed successfully!") | |
print(f"Files have been organized in: {self.organized_dir}") | |
except Exception as e: | |
self.logger.error(f"Error during organization process: {str(e)}") | |
print(f"\nβ An error occurred: {str(e)}") | |
raise | |
def main(): | |
"""Main function to run the file organizer.""" | |
try: | |
organizer = ClaudeFileOrganizer() | |
organizer.organize_files() | |
except KeyboardInterrupt: | |
print("\n\nOperation cancelled by user.") | |
except Exception as e: | |
print(f"\nAn unexpected error occurred: {str(e)}") | |
input("\nPress Enter to exit...") | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment