Add files via upload

DriveX/__init__.py

@@ -0,0 +1,2 @@
+from DriveX.core import *
+from DriveX.error.error import *

DriveX/core/Authenticator/Authenticator.py

@@ -0,0 +1,48 @@
+from google.auth.transport.requests import Request
+from google_auth_oauthlib.flow import InstalledAppFlow
+import os, pickle, json
+
+class Authenticator:
+    """Responsible for handling authentication with Google Drive."""
+
+    def __init__(self, credentials_path, token_path, scopes):
+        """Initialize the Authenticator.
+
+        Args:
+            credentials_path (str): Path to the credentials file.
+            token_path (str): Path to the token file.
+            scopes (str or list of str): The scope or scopes to be requested during the authorization flow.
+        """
+        self.credentials_path = credentials_path
+        self.token_path = token_path
+        self.scopes = scopes
+
+    def get_credentials(self):
+        """Retrieve or generate OAuth2 credentials for accessing Google Drive.
+
+        Returns:
+            google.oauth2.credentials.Credentials: OAuth2 credentials.
+        """
+        creds = None
+        if os.path.exists(self.token_path):
+            with open(self.token_path, 'rb') as token:
+                creds = pickle.load(token)
+        if not creds or not creds.valid:
+            if creds and creds.expired and creds.refresh_token:
+                creds.refresh(Request())
+            else:
+                flow = InstalledAppFlow.from_client_config(
+                    self._read_credentials_file(), self.scopes)
+                creds = flow.run_local_server(port=0)
+            with open(self.token_path, 'wb') as token:
+                pickle.dump(creds, token)
+        return creds
+
+    def _read_credentials_file(self):
+        """Read the credentials file and return its content.
+
+        Returns:
+            dict: Content of the credentials file.
+        """
+        with open(self.credentials_path) as f:
+            return json.load(f)

DriveX/core/Authenticator/__init__.py

@@ -0,0 +1 @@
+from .Authenticator import Authenticator

DriveX/core/FileManager/FileManager.py

@@ -0,0 +1,108 @@
+import os, io
+from ...error import *
+from googleapiclient.http import MediaFileUpload, MediaIoBaseDownload
+
+
+class FileManager:
+    """Handles file-related operations in Google Drive."""
+
+    def __init__(self, service):
+        self.service = service
+
+    def upload_file(self, file_path, folder_id=None):
+        """
+        Upload a file to Google Drive.
+
+        Args:
+            file_path (str): Path to the file to upload.
+            folder_id (str, optional): ID of the folder to upload the file to. Defaults to None.
+
+        Returns:
+            str: ID of the uploaded file.
+        """
+        try:
+            file_metadata = {'name': os.path.basename(file_path)}
+            if folder_id:
+                file_metadata['parents'] = [folder_id]
+            media = MediaFileUpload(file_path, resumable=True)
+            file = self.service.files().create(body=file_metadata, media_body=media, fields='id').execute()
+            return file.get('id')
+        except Exception as e:
+            raise FileError(f"Failed to upload file: {str(e)}")
+
+    def download_file(self, file_id, destination_path):
+        """
+        Download a file from Google Drive.
+
+        Args:
+            file_id (str): ID of the file to download.
+            destination_path (str): Path to save the downloaded file.
+
+        Returns:
+            None
+        """
+        try:
+            request = self.service.files().get_media(fileId=file_id)
+            with open(destination_path, 'wb') as fh:
+                downloader = MediaIoBaseDownload(fh, request)
+                done = False
+                while not done:
+                    status, done = downloader.next_chunk()
+        except Exception as e:
+            raise FileError(f"Failed to download file: {str(e)}")
+
+    def read_file(self, file_id):
+        """
+        Read the contents of a file from Google Drive.
+
+        Args:
+            file_id (str): ID of the file to read.
+
+        Returns:
+            str: Contents of the file.
+        """
+        try:
+            request = self.service.files().get_media(fileId=file_id)
+            downloaded_file = io.BytesIO()
+            downloader = MediaIoBaseDownload(downloaded_file, request)
+            done = False
+            while not done:
+                status, done = downloader.next_chunk()
+            return downloaded_file.getvalue().decode('utf-8')
+        except Exception as e:
+            raise FileError(f"Failed to read file: {str(e)}")
+
+    def delete_file(self, file_id):
+        """
+        Delete a file from Google Drive.
+
+        Args:
+            file_id (str): ID of the file to delete.
+
+        Returns:
+            None
+        """
+        try:
+            self.service.files().delete(fileId=file_id).execute()
+        except Exception as e:
+            raise FileError(f"Failed to delete file: {str(e)}")
+
+    def copy_file(self, file_id, new_parent_id=None):
+        """
+        Copy a file in Google Drive.
+
+        Args:
+            file_id (str): ID of the file to copy.
+            new_parent_id (str, optional): ID of the new parent folder. Defaults to None.
+
+        Returns:
+            str: ID of the newly copied file.
+        """
+        try:
+            file = {'parents': []}
+            if new_parent_id:
+                file['parents'].append(new_parent_id)
+            copied_file = self.service.files().copy(fileId=file_id, body=file).execute()
+            return copied_file.get('id')
+        except Exception as e:
+            raise FolderError(f"Failed to copy file: {str(e)}")

DriveX/core/FileManager/__init__.py

@@ -0,0 +1 @@
+from .FileManager import FileManager

DriveX/core/FolderManager/FolderManager.py

@@ -0,0 +1,89 @@
+from ...error.error import *
+
+class FolderManager:
+    """Handles folder-related operations in Google Drive."""
+
+    def __init__(self, service):
+        self.service = service
+
+    def create_folder(self, folder_name, parent_folder_id=None):
+        """
+        Create a new folder in Google Drive.
+
+        Args:
+            folder_name (str): Name of the new folder.
+            parent_folder_id (str, optional): ID of the parent folder. Defaults to None.
+
+        Returns:
+            str: ID of the newly created folder.
+        """
+        try:
+            folder_metadata = {
+                'name': folder_name,
+                'mimeType': 'application/vnd.google-apps.folder'
+            }
+            if parent_folder_id:
+                folder_metadata['parents'] = [parent_folder_id]
+            folder = self.service.files().create(body=folder_metadata, fields='id').execute()
+            return folder.get('id')
+        except Exception as e:
+            raise FolderError(f"Failed to create folder: {str(e)}")
+
+    def move_file(self, file_id, new_parent_id):
+        """
+        Move a file to a different folder.
+
+        Args:
+            file_id (str): ID of the file to move.
+            new_parent_id (str): ID of the new parent folder.
+
+        Returns:
+            dict: Updated file metadata.
+        """
+        try:
+            file = self.service.files().get(fileId=file_id, fields='parents').execute()
+            previous_parents = ",".join(file.get('parents'))
+            file = self.service.files().update(
+                fileId=file_id,
+                addParents=new_parent_id,
+                removeParents=previous_parents,
+                fields='id, parents'
+            ).execute()
+            return file
+        except Exception as e:
+            raise FolderError(f"Failed to move file: {str(e)}")
+
+    def copy_folder(self, folder_id, new_parent_id=None):
+        """
+        Copy a folder in Google Drive.
+
+        Args:
+            folder_id (str): ID of the folder to copy.
+            new_parent_id (str, optional): ID of the new parent folder. Defaults to None.
+
+        Returns:
+            str: ID of the newly copied folder.
+        """
+        try:
+            folder = {'parents': []}
+            if new_parent_id:
+                folder['parents'].append(new_parent_id)
+            copied_folder = self.service.files().copy(fileId=folder_id, body=folder).execute()
+            return copied_folder.get('id')
+        except Exception as e:
+            raise FolderError(f"Failed to copy folder: {str(e)}")
+
+    def delete_folder(self, folder_id):
+        """
+        Delete a folder from Google Drive.
+
+        Args:
+            folder_id (str): ID of the folder to delete.
+
+        Returns:
+            None
+        """
+        try:
+            self.service.files().delete(fileId=folder_id).execute()
+        except Exception as e:
+            raise FolderError(f"Failed to delete folder: {str(e)}")

DriveX/core/FolderManager/__init__.py

@@ -0,0 +1 @@
+from .FolderManager import FolderManager

DriveX/core/GoogleDriveAPI/GoogleDriveAPI.py

@@ -0,0 +1,95 @@
+from ...error.error import *
+from datetime import datetime
+
+class GoogleDriveAPI:
+    """Provides an interface for interacting with Google Drive."""
+
+    def __init__(self, credentials, service):
+        self.credentials = credentials
+        self.service = service
+
+    def list_files(self, page_size=10, query=None):
+        """
+        List files in Google Drive.
+
+        Args:
+            page_size (int, optional): Number of files to retrieve per page. Defaults to 10.
+            query (str, optional): Query string to filter the files. Defaults to None.
+
+        Returns:
+            list: List of files matching the query.
+        """
+        try:
+            query_params = {'pageSize': page_size, 'fields': 'nextPageToken, files(id, name, createdTime, permissions)'}
+            if query:
+                query_params['q'] = query
+            results = self.service.files().list(**query_params).execute()
+            items = results.get('files', [])
+
+            for item in items:
+                permissions = item.get('permissions', [])
+                if isinstance(permissions, str):
+                    permissions = eval(permissions)
+                item['permissions'] = self.format_permissions(permissions)
+
+            return items
+        except Exception as e:
+            raise FolderError(f"Failed to list files: {str(e)}")
+
+    def _convert_size(self, size_bytes):
+        """
+        Convert file size in bytes to human-readable format.
+
+        Args:
+            size_bytes (int or str): File size in bytes.
+
+        Returns:
+            str: Human-readable file size.
+        """
+        try:
+            size_bytes = int(size_bytes)
+        except ValueError:
+            return size_bytes
+
+        sizes = ['Bytes', 'KB', 'MB', 'GB', 'TB']
+        i = 0
+        while size_bytes >= 1024 and i < len(sizes) - 1:
+            size_bytes /= 1024.0
+            i += 1
+        return "{:.2f} {}".format(size_bytes, sizes[i])
+
+    def get_file_info(self, file_id):
+        """
+        Get detailed information about a file.
+
+        Args:
+            file_id (str): ID of the file.
+
+        Returns:
+            dict: File metadata.
+        """
+        try:
+            file = self.service.files().get(fileId=file_id).execute()
+            return file
+        except Exception as e:
+            raise FileError(f"Failed to get file info: {str(e)}")
+
+    @staticmethod
+    def format_permissions(permissions):
+        """
+        Format permissions for display.
+
+        Args:
+            permissions (list): List of permission dictionaries.
+
+        Returns:
+            str: Formatted permissions string.
+        """
+        if not permissions:
+            return "No permissions"
+        permissions_str = ""
+        for perm in permissions:
+            role = perm.get("role", "Unknown role")
+            email = perm.get("emailAddress", "Unknown email")
+            permissions_str += f"Role: {role}, Email: {email}\n"
+        return permissions_str

DriveX/core/ServiceBuilder/ServiceBuilder.py

@@ -0,0 +1,33 @@
+from ...error.error import *
+from ..Authenticator.Authenticator import Authenticator 
+from googleapiclient.discovery import build
+
+class ServiceBuilder:
+    """Builds the Google Drive service."""
+
+    def __init__(self, credentials_path, token_path, token_name ,scopes="https://www.googleapis.com/auth/drive"):
+        """Initialize the ServiceBuilder.
+
+        Args:
+            credentials_path (str): Path to the credentials file.
+            token_path (str): Path to the token file.
+            scopes (str or list of str): The scope or scopes to be requested during the authorization flow.
+        """
+        self.credentials_path = credentials_path
+        self.token_path = token_path
+        self.token_name = token_name
+        self.scopes = scopes
+
+    def build_service(self):
+        """Builds and returns the Google Drive service.
+
+        Returns:
+            googleapiclient.discovery.Resource: Google Drive service object.
+        """
+        try:
+            authenticator = Authenticator(self.credentials_path, f"{self.token_path}/{self.token_name}.pickle" , self.scopes)
+            credentials = authenticator.get_credentials()
+            service = build('drive', 'v3', credentials=credentials)
+            return service
+        except GoogleDriveError as e:
+            raise GoogleDriveError(f"Failed to build service: {str(e)}")

DriveX/core/ServiceBuilder/__init__.py

@@ -0,0 +1 @@
+from .ServiceBuilder import ServiceBuilder