mirror of
https://github.com/zbjdonald/synology-drive-api.git
synced 2025-07-25 16:40:06 +00:00
193 lines
7.1 KiB
Python
193 lines
7.1 KiB
Python
from typing import Optional, List, Union
|
|
|
|
from optionaldict import OptionalDict
|
|
|
|
from synology_drive_api.utils import form_urlencoded
|
|
|
|
|
|
def color_name_to_id(color_name: str) -> str:
|
|
"""
|
|
map color_name to hex color
|
|
:param color_name:
|
|
:return:
|
|
"""
|
|
color_id = {
|
|
'gray': '#A0A5AA',
|
|
'red': '#FA8282',
|
|
'orange': '#FA9C3E',
|
|
'yellow': '#F2CA00',
|
|
'green': '#94BF13',
|
|
'blue': '#499DF2',
|
|
'purple': '#A18AE6',
|
|
}
|
|
try:
|
|
return color_id[color_name.lower()]
|
|
except KeyError:
|
|
raise KeyError('Color name error! Use gray/red/orange/yellow/green/blue/purple')
|
|
|
|
|
|
class LabelsMixin:
|
|
"""
|
|
Drive labels related function
|
|
"""
|
|
# label dict cache
|
|
_label_dict: dict = dict()
|
|
|
|
def get_labels(self, name: Optional[str] = None) -> dict:
|
|
"""
|
|
Get label name and label id. If name is None, return all labels.
|
|
:param name: label name
|
|
:return: {name: label_id, ...}
|
|
"""
|
|
api_name = 'SYNO.SynologyDrive.Labels'
|
|
endpoint = 'entry.cgi'
|
|
params = {'api': api_name, 'version': 1, 'method': 'list'}
|
|
req = self.session.http_get(endpoint, params=params)
|
|
label_items = req['data']['items']
|
|
|
|
if not label_items:
|
|
return dict()
|
|
|
|
label_dict = {item['name']: item['label_id'] for item in label_items}
|
|
if name is None:
|
|
return label_dict
|
|
else:
|
|
try:
|
|
return {name: label_dict[name]}
|
|
except KeyError:
|
|
raise Exception(f'Label <{name}> does not exist.')
|
|
|
|
def create_label(self, name: str, color: str = 'gray', pos: Union[None, str, int] = None) -> dict:
|
|
"""
|
|
:param name: label name
|
|
:param color: color name gray/red/orange/yellow/green/blue/purple
|
|
:param pos: label position
|
|
:return:
|
|
"""
|
|
color = color_name_to_id(color_name=color)
|
|
if name in self.label_dict:
|
|
raise Exception('Label_name already exists, please use another one!')
|
|
api_name = 'SYNO.SynologyDrive.Labels'
|
|
endpoint = 'entry.cgi'
|
|
params = OptionalDict(api=api_name, version=1, method='create', name=name, color=color, position=pos)
|
|
ret_label = self.session.http_put(endpoint, params=params)
|
|
self.set_label_dict(name, ret_label['data']['label_id'])
|
|
return ret_label
|
|
|
|
def delete_label(self, label_name: Optional[str] = None, label_id: Union[None, str, int] = None) -> dict:
|
|
"""
|
|
delete label by name or label_id
|
|
:param label_name: label name
|
|
:param label_id: label id
|
|
:return:
|
|
"""
|
|
params_count = [label_name, label_id].count(None)
|
|
if params_count != 1:
|
|
raise Exception('Wrong params number. Enter label name or label_id')
|
|
|
|
if label_id is None:
|
|
try:
|
|
label_id = self.label_dict[label_name]
|
|
except KeyError:
|
|
raise Exception(f'Label <{label_name}> does not exists!')
|
|
|
|
api_name = 'SYNO.SynologyDrive.Labels'
|
|
endpoint = 'entry.cgi'
|
|
params = {'api': api_name, 'version': 1, 'method': 'delete', 'label_id': label_id}
|
|
return self.session.http_delete(endpoint, params=params)
|
|
|
|
def manage_path_label(self, action: str, path: Union[str, List[str]],
|
|
label: Union[str, List[str], List[dict]]) -> dict:
|
|
"""
|
|
add label/labels to file/files or folder/folders
|
|
:param action: 'add', 'delete'
|
|
:param path: file/files, folder/folders
|
|
1. '/team-folders/test_drive/SCU285/test.xls', '/mydrive/test_sheet_file.osheet'
|
|
2. '505415003021516807'
|
|
3. ['505415003021516807', '505415003021516817']
|
|
4. ["id:505415003021516807", "id:525657984139799470", "id:525657984810888112"]
|
|
5. ['/team-folders/test_drive/SCU285/test.xls', '/team-folders/test_drive/SCU283/test2.xls']
|
|
:param label: label/lables
|
|
1. label_name
|
|
2. ['label_name_1', 'lable_name_2']
|
|
3. [{"action": "add", "label_id": "15"}, {"action": "add", "label_id": "16"}]
|
|
:return:
|
|
"""
|
|
action = action.lower()
|
|
if action not in ('add', 'delete'):
|
|
raise Exception('Wrong action params. Use add or delete.')
|
|
|
|
if isinstance(label, list):
|
|
# label = ['label_name_1', 'label_name_2', 'label_name_3', ...]
|
|
if isinstance(label[0], str):
|
|
label = [{'action': action, 'label_id': self.label_dict.get(single_label)} for single_label in label]
|
|
elif isinstance(label[0], dict):
|
|
# [{"action": "add", "label_id": "15"}, {"action": "add", "label_id": "16"}]
|
|
label = label
|
|
else:
|
|
raise Exception('Wrong label params.')
|
|
elif isinstance(label, str):
|
|
label = [{'action': action, 'label_id': self.label_dict.get(label)}]
|
|
else:
|
|
raise Exception('Wrong label params.')
|
|
|
|
if isinstance(path, list):
|
|
path = [f"id:{single_path}" if single_path.isdigit() else single_path for single_path in path]
|
|
elif isinstance(path, str):
|
|
if path.isdigit():
|
|
path = [f"id:{path}"]
|
|
else:
|
|
path = [path]
|
|
else:
|
|
raise Exception('Wrong path params')
|
|
|
|
api_name = 'SYNO.SynologyDrive.Files'
|
|
endpoint = 'entry.cgi'
|
|
data = {'files': path, 'labels': label, 'api': api_name, 'method': 'label', 'version': '2'}
|
|
urlencoded_data = form_urlencoded(data)
|
|
return self.session.http_post(endpoint, data=urlencoded_data)
|
|
|
|
def list_labelled_files(self, label_name=None, label_id=None, limit=1500) -> dict:
|
|
"""
|
|
list specific label files
|
|
:param label_name: label name
|
|
:param label_id:
|
|
:param limit: return result count
|
|
:return:
|
|
"""
|
|
params_count = [label_name, label_id].count(None)
|
|
if params_count != 1:
|
|
raise Exception('Wrong params number. Enter label name or label_id')
|
|
|
|
if label_id is None:
|
|
try:
|
|
label_id = self.label_dict[label_name]
|
|
except KeyError:
|
|
raise Exception(f'Label <{label_name}> does not exists!')
|
|
|
|
api_name = 'SYNO.SynologyDrive.Files'
|
|
endpoint = 'entry.cgi'
|
|
data = {'api': api_name, 'version': 2, 'method': 'list_labelled', 'label_id': label_id, 'offset': 0,
|
|
'limit': limit, 'sort_by': 'name', 'sort_direction': 'desc', 'filter': {}}
|
|
urlencoded_data = form_urlencoded(data)
|
|
return self.session.http_post(endpoint, data=urlencoded_data)
|
|
|
|
def set_label_dict(self, label_name, label_id):
|
|
self._label_dict[label_name] = label_id
|
|
|
|
@property
|
|
def label_dict(self):
|
|
"""
|
|
label name id map
|
|
:return:
|
|
"""
|
|
# if cache is disabled, get labels from drive server.
|
|
if not self.enable_label_cache:
|
|
self._label_dict = self.get_labels()
|
|
return self._label_dict
|
|
|
|
# if cache is enabled and empty, fill up cache.
|
|
if not self._label_dict:
|
|
self._label_dict = self.get_labels()
|
|
return self._label_dict
|