513 lines
19 KiB
Python
513 lines
19 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
Модуль для взаимодействия с файловой системой Synology NAS через API FileStation
|
||
"""
|
||
|
||
import os
|
||
import logging
|
||
import requests
|
||
from typing import Dict, Any, Optional, List, Union
|
||
|
||
from src.api.synology import SynologyAPI
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
def add_file_manager_methods_to_synology_api(api_class):
|
||
"""Добавляет методы для работы с файловой системой к классу SynologyAPI"""
|
||
|
||
def list_files(self, folder_path: str = "/") -> List[Dict[str, Any]]:
|
||
"""Получение списка файлов и папок в указанной директории
|
||
|
||
Args:
|
||
folder_path: Путь к директории для просмотра
|
||
|
||
Returns:
|
||
Список файлов и папок в указанной директории
|
||
"""
|
||
logger.info(f"Listing files in directory: {folder_path}")
|
||
|
||
# Аутентифицируемся если нужно
|
||
if not self.sid and not self.login():
|
||
logger.error("Failed to authenticate for file listing")
|
||
return []
|
||
|
||
try:
|
||
# Если это корневая папка, получаем список общих папок
|
||
if folder_path == "/":
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.List",
|
||
"list_share",
|
||
version=2
|
||
)
|
||
|
||
if not result:
|
||
# Пробуем версию 1
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.List",
|
||
"list_share",
|
||
version=1
|
||
)
|
||
|
||
if not result:
|
||
logger.error("Failed to list shared folders")
|
||
return []
|
||
|
||
return result.get("shares", [])
|
||
else:
|
||
# Получаем список файлов в указанной директории
|
||
params = {
|
||
"folder_path": folder_path,
|
||
"sort_by": "name",
|
||
"sort_direction": "ASC"
|
||
}
|
||
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.List",
|
||
"list",
|
||
version=2,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
# Пробуем версию 1
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.List",
|
||
"list",
|
||
version=1,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
logger.error(f"Failed to list files in {folder_path}")
|
||
return []
|
||
|
||
return result.get("files", [])
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error listing files in {folder_path}: {str(e)}")
|
||
return []
|
||
|
||
def get_file_info(self, file_path: str) -> Dict[str, Any]:
|
||
"""Получение подробной информации о файле
|
||
|
||
Args:
|
||
file_path: Полный путь к файлу
|
||
|
||
Returns:
|
||
Информация о файле
|
||
"""
|
||
logger.info(f"Getting file info: {file_path}")
|
||
|
||
# Аутентифицируемся если нужно
|
||
if not self.sid and not self.login():
|
||
logger.error("Failed to authenticate for file info request")
|
||
return {}
|
||
|
||
try:
|
||
params = {
|
||
"path": file_path,
|
||
"additional": "real_path,size,owner,time,perm"
|
||
}
|
||
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.List",
|
||
"getinfo",
|
||
version=2,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
# Пробуем версию 1
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.List",
|
||
"getinfo",
|
||
version=1,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
logger.error(f"Failed to get file info for {file_path}")
|
||
return {}
|
||
|
||
# Возвращаем информацию о первом файле в результате
|
||
files = result.get("files", [])
|
||
if files and len(files) > 0:
|
||
return files[0]
|
||
|
||
return {}
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error getting file info for {file_path}: {str(e)}")
|
||
return {}
|
||
|
||
def download_file(self, file_path: str, local_path: str) -> bool:
|
||
"""Скачивание файла с NAS
|
||
|
||
Args:
|
||
file_path: Путь к файлу на NAS
|
||
local_path: Локальный путь для сохранения файла
|
||
|
||
Returns:
|
||
True если успешно, False в противном случае
|
||
"""
|
||
logger.info(f"Downloading file from {file_path} to {local_path}")
|
||
|
||
# Аутентифицируемся если нужно
|
||
if not self.sid and not self.login():
|
||
logger.error("Failed to authenticate for file download")
|
||
return False
|
||
|
||
try:
|
||
# Получаем URL для скачивания файла
|
||
params = {
|
||
"path": file_path,
|
||
"mode": "download"
|
||
}
|
||
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.Download",
|
||
"download",
|
||
version=2,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
# Пробуем версию 1
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.Download",
|
||
"download",
|
||
version=1,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
logger.error(f"Failed to get download URL for {file_path}")
|
||
return False
|
||
|
||
# URL для скачивания
|
||
download_url = result.get("url")
|
||
if not download_url:
|
||
logger.error("No download URL received")
|
||
return False
|
||
|
||
# Добавляем базовый URL, если URL относительный
|
||
if not download_url.startswith("http"):
|
||
protocol = "https" if self.protocol == "https" else "http"
|
||
download_url = f"{protocol}://{self.base_url}/{download_url}"
|
||
|
||
# Скачиваем файл
|
||
response = self.session.get(download_url, stream=True, verify=False)
|
||
if response.status_code != 200:
|
||
logger.error(f"Failed to download file: HTTP {response.status_code}")
|
||
return False
|
||
|
||
# Сохраняем файл
|
||
with open(local_path, 'wb') as f:
|
||
for chunk in response.iter_content(chunk_size=8192):
|
||
if chunk:
|
||
f.write(chunk)
|
||
|
||
logger.info(f"File successfully downloaded to {local_path}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error downloading file {file_path}: {str(e)}")
|
||
return False
|
||
|
||
def upload_file(self, local_path: str, folder_path: str) -> bool:
|
||
"""Загрузка файла на NAS
|
||
|
||
Args:
|
||
local_path: Локальный путь к файлу
|
||
folder_path: Путь на NAS для загрузки файла
|
||
|
||
Returns:
|
||
True если успешно, False в противном случае
|
||
"""
|
||
logger.info(f"Uploading file from {local_path} to {folder_path}")
|
||
|
||
# Аутентифицируемся если нужно
|
||
if not self.sid and not self.login():
|
||
logger.error("Failed to authenticate for file upload")
|
||
return False
|
||
|
||
try:
|
||
# Проверяем существование файла
|
||
if not os.path.exists(local_path):
|
||
logger.error(f"Local file {local_path} not found")
|
||
return False
|
||
|
||
# Формируем URL для загрузки
|
||
url = f"{self.base_url}/entry.cgi"
|
||
|
||
# Извлекаем имя файла из локального пути
|
||
file_name = os.path.basename(local_path)
|
||
|
||
# Подготавливаем параметры для загрузки
|
||
params = {
|
||
"api": "SYNO.FileStation.Upload",
|
||
"version": "2",
|
||
"method": "upload",
|
||
"path": folder_path,
|
||
"_sid": self.sid
|
||
}
|
||
|
||
# Подготавливаем файл для загрузки
|
||
files = {
|
||
'file': (file_name, open(local_path, 'rb'))
|
||
}
|
||
|
||
# Выполняем запрос
|
||
response = self.session.post(url, params=params, files=files, verify=False)
|
||
|
||
# Закрываем файл
|
||
files['file'][1].close()
|
||
|
||
if response.status_code != 200:
|
||
logger.error(f"Failed to upload file: HTTP {response.status_code}")
|
||
return False
|
||
|
||
# Проверяем ответ
|
||
try:
|
||
data = response.json()
|
||
success = data.get("success", False)
|
||
|
||
if not success:
|
||
error_code = data.get("error", {}).get("code", -1)
|
||
logger.error(f"Failed to upload file: Error code {error_code}")
|
||
return False
|
||
|
||
logger.info(f"File successfully uploaded to {folder_path}/{file_name}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error parsing upload response: {str(e)}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error uploading file {local_path}: {str(e)}")
|
||
return False
|
||
|
||
def delete_file(self, file_path: str) -> bool:
|
||
"""Удаление файла на NAS
|
||
|
||
Args:
|
||
file_path: Путь к файлу для удаления
|
||
|
||
Returns:
|
||
True если успешно, False в противном случае
|
||
"""
|
||
logger.info(f"Deleting file: {file_path}")
|
||
|
||
# Аутентифицируемся если нужно
|
||
if not self.sid and not self.login():
|
||
logger.error("Failed to authenticate for file deletion")
|
||
return False
|
||
|
||
try:
|
||
# Подготавливаем параметры для удаления
|
||
params = {
|
||
"path": [file_path],
|
||
"recursive": True # Удаляем папки рекурсивно
|
||
}
|
||
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.Delete",
|
||
"delete",
|
||
version=2,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
# Пробуем версию 1
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.Delete",
|
||
"delete",
|
||
version=1,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
logger.error(f"Failed to delete file {file_path}")
|
||
return False
|
||
|
||
# Проверяем результат
|
||
task_id = result.get("taskid")
|
||
if not task_id:
|
||
logger.error("No task ID received for deletion")
|
||
return False
|
||
|
||
# Проверяем статус задачи
|
||
task_params = {
|
||
"taskid": task_id
|
||
}
|
||
|
||
# Ждем завершения задачи
|
||
for _ in range(10):
|
||
task_result = self._make_api_request(
|
||
"SYNO.FileStation.Delete",
|
||
"status",
|
||
version=2,
|
||
params=task_params
|
||
)
|
||
|
||
if not task_result:
|
||
task_result = self._make_api_request(
|
||
"SYNO.FileStation.Delete",
|
||
"status",
|
||
version=1,
|
||
params=task_params
|
||
)
|
||
|
||
if not task_result:
|
||
logger.error(f"Failed to check delete task status for {file_path}")
|
||
return False
|
||
|
||
# Проверяем статус задачи
|
||
if task_result.get("finished", False):
|
||
return True
|
||
|
||
# Ждем немного
|
||
import time
|
||
time.sleep(0.5)
|
||
|
||
logger.warning(f"Delete task did not complete in time for {file_path}")
|
||
return True # Возвращаем True, т.к. задача запущена успешно
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error deleting file {file_path}: {str(e)}")
|
||
return False
|
||
|
||
def create_folder(self, parent_path: str, folder_name: str) -> bool:
|
||
"""Создание новой папки на NAS
|
||
|
||
Args:
|
||
parent_path: Родительский путь для новой папки
|
||
folder_name: Имя новой папки
|
||
|
||
Returns:
|
||
True если успешно, False в противном случае
|
||
"""
|
||
logger.info(f"Creating folder {folder_name} in {parent_path}")
|
||
|
||
# Аутентифицируемся если нужно
|
||
if not self.sid and not self.login():
|
||
logger.error("Failed to authenticate for folder creation")
|
||
return False
|
||
|
||
try:
|
||
# Подготавливаем параметры для создания папки
|
||
params = {
|
||
"folder_path": parent_path,
|
||
"name": folder_name
|
||
}
|
||
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.CreateFolder",
|
||
"create",
|
||
version=2,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
# Пробуем версию 1
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.CreateFolder",
|
||
"create",
|
||
version=1,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
logger.error(f"Failed to create folder {folder_name} in {parent_path}")
|
||
return False
|
||
|
||
# Проверяем результат
|
||
folders = result.get("folders", [])
|
||
if not folders:
|
||
logger.error("No folder information received after creation")
|
||
return False
|
||
|
||
logger.info(f"Folder {folder_name} created successfully in {parent_path}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error creating folder {folder_name}: {str(e)}")
|
||
return False
|
||
|
||
def rename_file(self, file_path: str, new_name: str) -> bool:
|
||
"""Переименование файла или папки на NAS
|
||
|
||
Args:
|
||
file_path: Путь к файлу для переименования
|
||
new_name: Новое имя файла (без пути)
|
||
|
||
Returns:
|
||
True если успешно, False в противном случае
|
||
"""
|
||
logger.info(f"Renaming {file_path} to {new_name}")
|
||
|
||
# Аутентифицируемся если нужно
|
||
if not self.sid and not self.login():
|
||
logger.error("Failed to authenticate for file renaming")
|
||
return False
|
||
|
||
try:
|
||
# Получаем путь к родительской директории
|
||
parent_path = os.path.dirname(file_path)
|
||
|
||
# Подготавливаем параметры для переименования
|
||
params = {
|
||
"path": file_path,
|
||
"name": new_name
|
||
}
|
||
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.Rename",
|
||
"rename",
|
||
version=2,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
# Пробуем версию 1
|
||
result = self._make_api_request(
|
||
"SYNO.FileStation.Rename",
|
||
"rename",
|
||
version=1,
|
||
params=params
|
||
)
|
||
|
||
if not result:
|
||
logger.error(f"Failed to rename {file_path} to {new_name}")
|
||
return False
|
||
|
||
# Проверяем результат
|
||
files = result.get("files", [])
|
||
if not files:
|
||
logger.error("No file information received after renaming")
|
||
return False
|
||
|
||
logger.info(f"File {file_path} renamed to {new_name} successfully")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Error renaming file {file_path}: {str(e)}")
|
||
return False
|
||
|
||
# Добавляем все методы в класс API
|
||
api_class.list_files = list_files
|
||
api_class.get_file_info = get_file_info
|
||
api_class.download_file = download_file
|
||
api_class.upload_file = upload_file
|
||
api_class.delete_file = delete_file
|
||
api_class.create_folder = create_folder
|
||
api_class.rename_file = rename_file
|
||
|
||
return api_class
|
||
|
||
# Добавляем методы для работы с файлами к классу SynologyAPI
|
||
add_file_manager_methods_to_synology_api(SynologyAPI)
|