118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
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
|
|
})
|
|
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 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:
|
|
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()
|