personal_development/ServerManagerWebApp/backend/app/samba_manager.py

from smb.SMBConnection import SMBConnection
import logging

logger = logging.getLogger(__name__)

class SambaManager:
    def __init__(self, server_ip, share_name, username, password, client_machine_name="smbclient"):
        self.server_ip = server_ip
        self.share_name = share_name
        self.username = username
        self.password = password
        self.client_machine_name = client_machine_name
        self.conn = SMBConnection(username, password, client_machine_name, "server_name", use_ntlm_v2=True)
        self.conn.connect(server_ip, 139)

    def clone(self):
        """Creates a new independent connection with the same credentials."""
        return SambaManager(self.server_ip, self.share_name, self.username, self.password, self.client_machine_name)

    def list_path(self, current_dir_path="/"):
        try:
            files = self.conn.listPath(self.share_name, current_dir_path)
            items = []
            for f in files:
                if f.filename in ['.', '..']:
                    continue

                normalized_current_path = current_dir_path
                if normalized_current_path != '/' and not normalized_current_path.endswith('/'):
                    normalized_current_path += '/'

                full_item_path = f"{normalized_current_path}{f.filename}".replace("\\\\", "/")

                items.append({
                    "name": f.filename,
                    "is_directory": f.isDirectory,
                    "size": f.file_size,
                    "path": full_item_path,
                    "last_modified": f.last_write_time
                })
            return items
        except Exception as e:
            logger.error(f"Error listing path '{current_dir_path}' on share '{self.share_name}': {e}", exc_info=True)
            return []

    def search_files(self, query, start_path="/"):
        """
        Recursively searches for files matching query (partial, case-insensitive).
        """
        matches = []
        try:
            stack = [start_path]
            while stack:
                current_path = stack.pop()
                try:
                    files = self.conn.listPath(self.share_name, current_path)
                except Exception:
                    continue # Skip unreadable dirs

                for f in files:
                    if f.filename in ['.', '..']: continue
                    
                    normalized_current_path = current_path
                    if normalized_current_path != '/' and not normalized_current_path.endswith('/'):
                        normalized_current_path += '/'
                    
                    full_item_path = f"{normalized_current_path}{f.filename}".replace("\\", "/")
                    
                    if query.lower() in f.filename.lower():
                        matches.append({
                            "name": f.filename,
                            "is_directory": f.isDirectory,
                            "size": f.file_size,
                            "path": full_item_path,
                            "last_modified": f.last_write_time
                        })
                    
                    if f.isDirectory:
                        stack.append(full_item_path)
                        
        except Exception as e:
            logger.error(f"Error searching path '{start_path}': {e}")
            
        return matches


    def delete_file(self, path):
        try:
            self.conn.deleteFiles(self.share_name, path)
            return {"success": True}
        except Exception as e:
            return {"error": str(e)}

    def upload_file(self, path, file_obj):
        try:
            logger.info(f"Uploading file to share '{self.share_name}' at path: '{path}'")
            self.conn.storeFile(self.share_name, path, file_obj)
            return {"success": True}
        except Exception as e:
            return {"error": str(e)}

    def download_file(self, path, file_obj):
        try:
            self.conn.retrieveFile(self.share_name, path, file_obj)
            return {"success": True}
        except Exception as e:
            if "Scan canceled by user" in str(e):
                raise e
            return {"error": str(e)}

    def download_file_range(self, path, file_obj, offset, max_length):
        try:
            self.conn.retrieveFileFromOffset(self.share_name, path, file_obj, offset, max_length)
            return {"success": True}
        except Exception as e:
            return {"error": str(e)}

    def get_attributes(self, path):
        try:
            return self.conn.getAttributes(self.share_name, path)
        except Exception as e:
            return {"error": str(e)}

    def rename_file(self, old_path, new_path):
        try:
            self.conn.rename(self.share_name, old_path, new_path)
            return {"success": True}
        except Exception as e:
            return {"error": str(e)}

    def create_directory(self, path):
        try:
            self.conn.createDirectory(self.share_name, path)
            return {"success": True}
        except Exception as e:
            return {"error": str(e)}

    def delete_directory(self, path):
        try:
            self.conn.deleteDirectory(self.share_name, path)
            return {"success": True}
        except Exception as e:
            return {"error": str(e)}

    def delete_directory_recursive(self, path):
        try:
            items = self.list_path(path)
            for item in items:
                item_path = item["path"]
                if item["is_directory"]:
                    self.delete_directory_recursive(item_path)
                else:
                    self.delete_file(item_path)
            self.delete_directory(path)
            return {"success": True}
        except Exception as e:
            return {"error": str(e)}

    def close(self):
        self.conn.close()