Created
May 10, 2023 09:05
-
-
Save Owen-Liuyuxuan/ea2e98f41520b0aa8337f23f3be60a19 to your computer and use it in GitHub Desktop.
Synology helper script. Check and Download files from a NAS server.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""Synology helper script. Check and Download files from a NAS server. | |
Requires synology-api. Could be Installed with | |
pip3 install synology-api | |
Directly run python3 synology.py to check the usage. | |
""" | |
from synology_api import filestation | |
from synology_api.exceptions import FileStationError | |
import getpass | |
import warnings | |
import fire | |
import os | |
def try_login_once(username, password, IP="nas.ram-lab.com", port="5000"): | |
try: | |
fl = filestation.FileStation(IP, port, username, password, | |
secure=False, cert_verify=False, dsm_version=7, | |
debug=True, otp_code=None) | |
return fl | |
except KeyError: | |
warnings.warn("Log in fail") | |
return None | |
def login(*args, **kwargs): | |
username = input("NAS username: ") | |
password = getpass.getpass("NAS password: ") | |
result = try_login_once(username, password, *args, **kwargs) | |
if result is None: | |
print("Please re-enter user name and password") | |
username = input("NAS username: ") | |
password = getpass.getpass("NAS password: ") | |
result = try_login_once(username, password, *args, **kwargs) | |
if result is None: | |
print("Login failed twice, exit program") | |
exit() | |
return_dict = dict( | |
fl=result, | |
username = username, | |
) | |
return return_dict | |
def looping_through_folder(nas_dir_path:str, fl:filestation.FileStation, current_depth=0, max_depth=1): | |
try: | |
result_dict = fl.get_file_list(nas_dir_path) | |
except FileStationError as e: | |
warnings.warn(e.error_message) | |
if current_depth < max_depth: | |
if result_dict['success']: | |
return_data = result_dict['data'] | |
for i in range(return_data['total']): | |
file_obj = return_data['files'][i] | |
yield file_obj | |
if file_obj['isdir']: | |
yield from looping_through_folder(file_obj['path'], | |
fl, | |
current_depth+1, | |
max_depth=max_depth) | |
def check(fl:filestation.FileStation, nas_path:str, is_folder=True, max_depth=1): | |
"""Check the file/folder info from the NAS. | |
Usage: | |
python3 synology.py check --nas_path=$NAS_PATH --is_folder=True/False --max_depth=$MAX_FOLDER_DEPTH [optional] --nas_ip=$NAS_IP --port=$NAS_PORT | |
""" | |
if is_folder: | |
for file_obj in looping_through_folder(nas_path, fl, 0, max_depth=max_depth): | |
print(file_obj) | |
else: | |
file_info = fl.get_file_info(nas_path) | |
print(file_info['data']) | |
def download(fl:filestation.FileStation, nas_path:str, local_path:str, is_folder=True, max_depth=1): | |
"""Download file/folder from the NAS. We can download a single file or download all files inside a folder while keeping its form. | |
Usage: | |
python3 synology.py download --nas_path=$NAS_PATH --local_path=$LOCAL_BASE_DIR --is_folder=True/False --max_depth=$MAX_FOLDER_DEPTH [optional] --nas_ip=$NAS_IP --port=$NAS_PORT | |
""" | |
if is_folder: | |
for file_obj in looping_through_folder(nas_path, fl, 0, max_depth=max_depth): | |
if not file_obj['isdir']: | |
rel_path = os.path.relpath(file_obj['path'], nas_path) | |
dirname = os.path.dirname(rel_path) | |
fl.get_file(file_obj['path'], 'download', | |
os.path.join(local_path, dirname) | |
) | |
else: | |
return fl.get_file(nas_path, 'download', local_path) | |
def upload(*args, **kwargs): | |
raise NotImplementedError | |
OPERATION_DICT={ | |
'check' : check, | |
'download': download, | |
} | |
def main(operation='help', nas_ip="nas.ram-lab.com", port="5000", *args, **kwargs): | |
if operation == 'help': | |
sepration = "-" * 50 | |
print(f"{sepration}\n{__doc__}{sepration}") | |
print(f"Currently, the script supports operations including {list(OPERATION_DICT.keys())}") | |
print(sepration) | |
for key in OPERATION_DICT: | |
print(f"Helper docs for operation {key}:\n {OPERATION_DICT[key].__doc__}{sepration}") | |
return | |
result = login( | |
IP=nas_ip, port=str(port) | |
) | |
fl = result['fl'] | |
username = result['username'] | |
return OPERATION_DICT[operation](fl, *args, **kwargs) | |
if __name__ == '__main__': | |
fire.Fire(main) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment