newTao/modules/synology_folder_manager.py

112 lines
4.0 KiB
Python

from datetime import datetime
import requests
class SynologyFolderManager:
def __init__(self, account, password, base_url):
self.account = account
self.password = password
self.base_url = base_url
self.session_id = None
def login(self):
params = {
'api': 'SYNO.API.Auth',
'version': '7',
'method': 'login',
'account': self.account,
'passwd': self.password,
'session': 'FileStation',
'format': 'sid'
}
resp = requests.get(f'{self.base_url}/webapi/auth.cgi', params=params)
resp_json = resp.json()
self.session_id = resp_json['data']['sid']
if resp.status_code == 200:
result = resp.json()
if result.get("success", False):
self.session_id = result['data']['sid']
print(f"Login successful. Session ID: {self.session_id}")
else:
print(f"Login failed. Response: {result}")
else:
print(f"HTTP error. Status code: {resp.status_code}. Response: {resp.text}")
def list_folder(self, folder_path):
params = {
'api': 'SYNO.FileStation.List',
'version': '2',
'method': 'list',
'folder_path': folder_path,
'_sid': self.session_id
}
response = requests.get(f'{self.base_url}/webapi/entry.cgi', params=params)
if response.status_code == 200:
result = response.json()
if result.get("success", False):
print(f"Contents of {folder_path}:")
for item in result['data']['files']:
print(f"- {item['name']} ({'Folder' if item['isdir'] else 'File'})")
else:
error_code = result.get("error", {}).get("code", "")
print(f"Failed to list folder contents. Error code: {error_code}.")
else:
print(f"Failed to communicate with Synology NAS. HTTP status code: {response.status_code}.")
def create_folder(self, path, name):
params = {
'api': 'SYNO.FileStation.CreateFolder',
'version': '2',
'method': 'create',
'folder_path': path,
'name' : name,
'_sid': self.session_id
}
response = requests.get(f'{self.base_url}/webapi/entry.cgi', params=params)
if response.status_code == 200:
result = response.json()
if result.get("success", False):
print(f"Folder created successfully at {path}.")
else:
error_code = result.get("error", {}).get("code", "")
print(f"Failed to create folder. Error code: {error_code}.")
else:
print(f"Failed to communicate with Synology NAS. HTTP status code: {response.status_code}.")
def logout(self):
params = {
'api': 'SYNO.API.Auth',
'version': '1',
'method': 'logout',
'session': 'FileStation',
'_sid': self.session_id
}
requests.get(f'{self.base_url}/webapi/auth.cgi', params=params)
def create_user_directory(self, email):
# 이메일에서 사용자 ID 추출
user_id = email.split('@')[0]
# 현재 날짜 기준 연도 및 월 폴더 이름 설정
current_year = datetime.now().year
current_month = datetime.now().month
# 사용자 기본 폴더 경로 구성
base_path = "/volume1/Listup"
user_path = f"/Listup_id_{user_id}"
year_path = f"{user_path}/{current_year}"
month_path = f"{year_path}/{current_month}"
# 사용자 폴더 생성
self.drive.create_folder(user_path)
# 연도별 폴더 생성
self.drive.create_folder(year_path)
# 월별 폴더 생성
self.drive.create_folder(month_path)
print(f"{email}에 대한 폴더 생성 완료: {month_path}")