AtHeartEngineer/headphones
1
1
Fork 0
mirror of https://github.com/rembo10/headphones.git synced 2026-01-09 14:48:07 -05:00
Code Issues Packages Projects Releases Wiki Activity

add slskd_api @ v0.1.5

Browse Source
This commit is contained in:
rembo10
2024-05-26 09:47:43 +05:30
parent c0c636d545
commit b8168ec8eb
17 changed files with 1312 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
lib/slskd_api/__init__.py Normal file
View File

@@ -0,0 +1,18 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .client import SlskdClient, MetricsApi
__all__ = ('SlskdClient', 'MetricsApi')

44
lib/slskd_api/apis/__init__.py Normal file
View File

@@ -0,0 +1,44 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .application import ApplicationApi
from .conversations import ConversationsApi
from .logs import LogsApi
from .options import OptionsApi
from .public_chat import PublicChatApi
from .relay import RelayApi
from .rooms import RoomsApi
from .searches import SearchesApi
from .server import ServerApi
from .session import SessionApi
from .shares import SharesApi
from .transfers import TransfersApi
from .users import UsersApi
__all__ = (
'ApplicationApi',
'ConversationsApi',
'LogsApi',
'OptionsApi',
'PublicChatApi',
'RelayApi',
'RoomsApi',
'SearchesApi',
'ServerApi',
'SessionApi',
'SharesApi',
'TransfersApi',
'UsersApi'
)

91
lib/slskd_api/apis/application.py Normal file
View File

@@ -0,0 +1,91 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class ApplicationApi(BaseApi):
"""
This class contains the methods to interact with the Application API.
"""
def state(self) -> dict:
"""
Gets the current state of the application.
"""
url = self.api_url + '/application'
response = self.session.get(url)
return response.json()
def stop(self) -> bool:
"""
Stops the application. Only works with token (usr/pwd login). 'Unauthorized' with API-Key.
:return: True if successful.
"""
url = self.api_url + '/application'
response = self.session.delete(url)
return response.ok
def restart(self) -> bool:
"""
Restarts the application. Only works with token (usr/pwd login). 'Unauthorized' with API-Key.
:return: True if successful.
"""
url = self.api_url + '/application'
response = self.session.put(url)
return response.ok
def version(self) -> str:
"""
Gets the current application version.
"""
url = self.api_url + '/application/version'
response = self.session.get(url)
return response.json()
def check_updates(self, forceCheck: bool = False) -> dict:
"""
Checks for updates.
"""
url = self.api_url + '/application/version/latest'
params = dict(
forceCheck=forceCheck
)
response = self.session.get(url, params=params)
return response.json()
def gc(self) -> bool:
"""
Forces garbage collection.
:return: True if successful.
"""
url = self.api_url + '/application/gc'
response = self.session.post(url)
return response.ok
# Not supposed to be part of the external API
# More info in the Github discussion: https://github.com/slskd/slskd/discussions/910
# def dump(self):
# url = self.api_url + '/application/dump'
# response = self.session.get(url)
# return response.json()

26
lib/slskd_api/apis/base.py Normal file
View File

@@ -0,0 +1,26 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import requests
from urllib.parse import quote
class BaseApi:
"""
Base class where api-url and headers are set for all requests.
"""
def __init__(self, api_url: str, session: requests.Session):
self.api_url = api_url
self.session = session

103
lib/slskd_api/apis/conversations.py Normal file
View File

@@ -0,0 +1,103 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class ConversationsApi(BaseApi):
"""
This class contains the methods to interact with the Conversations API.
"""
def acknowledge(self, username: str, id: int) -> bool:
"""
Acknowledges the given message id for the given username.
:return: True if successful.
"""
url = self.api_url + f'/conversations/{quote(username)}/{id}'
response = self.session.put(url)
return response.ok
def acknowledge_all(self, username: str) -> bool:
"""
Acknowledges all messages from the given username.
:return: True if successful.
"""
url = self.api_url + f'/conversations/{quote(username)}'
response = self.session.put(url)
return response.ok
def delete(self, username: str) -> bool:
"""
Closes the conversation associated with the given username.
:return: True if successful.
"""
url = self.api_url + f'/conversations/{quote(username)}'
response = self.session.delete(url)
return response.ok
def get(self, username: str, includeMessages: bool = True) -> dict:
"""
Gets the conversation associated with the specified username.
"""
url = self.api_url + f'/conversations/{quote(username)}'
params = dict(
includeMessages=includeMessages
)
response = self.session.get(url, params=params)
return response.json()
def send(self, username: str, message: str) -> bool:
"""
Sends a private message to the specified username.
:return: True if successful.
"""
url = self.api_url + f'/conversations/{quote(username)}'
response = self.session.post(url, json=message)
return response.ok
def get_all(self, includeInactive: bool = False, unAcknowledgedOnly : bool = False) -> list:
"""
Gets all active conversations.
"""
url = self.api_url + '/conversations'
params = dict(
includeInactive=includeInactive,
unAcknowledgedOnly=unAcknowledgedOnly
)
response = self.session.get(url, params=params)
return response.json()
def get_messages(self, username: str, unAcknowledgedOnly : bool = False) -> list:
"""
Gets all messages associated with the specified username.
"""
url = self.api_url + f'/conversations/{quote(username)}/messages'
params = dict(
username=username,
unAcknowledgedOnly=unAcknowledgedOnly
)
response = self.session.get(url, params=params)
return response.json()

29
lib/slskd_api/apis/logs.py Normal file
View File

@@ -0,0 +1,29 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class LogsApi(BaseApi):
"""
This class contains the methods to interact with the Logs API.
"""
def get(self) -> list:
"""
Gets the last few application logs.
"""
url = self.api_url + '/logs'
response = self.session.get(url)
return response.json()

93
lib/slskd_api/apis/options.py Normal file
View File

@@ -0,0 +1,93 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class OptionsApi(BaseApi):
"""
This class contains the methods to interact with the Options API.
"""
def get(self) -> dict:
"""
Gets the current application options.
"""
url = self.api_url + '/options'
response = self.session.get(url)
return response.json()
def get_startup(self) -> dict:
"""
Gets the application options provided at startup.
"""
url = self.api_url + '/options/startup'
response = self.session.get(url)
return response.json()
def debug(self) -> str:
"""
Gets the debug view of the current application options.
debug and remote_configuration must be set to true.
Only works with token (usr/pwd login). 'Unauthorized' with API-Key.
"""
url = self.api_url + '/options/debug'
response = self.session.get(url)
return response.json()
def yaml_location(self) -> str:
"""
Gets the path of the yaml config file. remote_configuration must be set to true.
Only works with token (usr/pwd login). 'Unauthorized' with API-Key.
"""
url = self.api_url + '/options/yaml/location'
response = self.session.get(url)
return response.json()
def download_yaml(self) -> str:
"""
Gets the content of the yaml config file as text. remote_configuration must be set to true.
Only works with token (usr/pwd login). 'Unauthorized' with API-Key.
"""
url = self.api_url + '/options/yaml'
response = self.session.get(url)
return response.json()
def upload_yaml(self, yaml_content: str) -> bool:
"""
Sets the content of the yaml config file. remote_configuration must be set to true.
Only works with token (usr/pwd login). 'Unauthorized' with API-Key.
:return: True if successful.
"""
url = self.api_url + '/options/yaml'
response = self.session.post(url, json=yaml_content)
return response.ok
def validate_yaml(self, yaml_content: str) -> str:
"""
Validates the provided yaml string. remote_configuration must be set to true.
Only works with token (usr/pwd login). 'Unauthorized' with API-Key.
:return: Empty string if validation successful. Error message otherwise.
"""
url = self.api_url + '/options/yaml/validate'
response = self.session.post(url, json=yaml_content)
return response.text

42
lib/slskd_api/apis/public_chat.py Normal file
View File

@@ -0,0 +1,42 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class PublicChatApi(BaseApi):
"""
[UNTESTED] This class contains the methods to interact with the PublicChat API.
"""
def start(self) -> bool:
"""
Starts public chat.
:return: True if successful.
"""
url = self.api_url + '/publicchat'
response = self.session.post(url)
return response.ok
def stop(self) -> bool:
"""
Stops public chat.
:return: True if successful.
"""
url = self.api_url + '/publicchat'
response = self.session.delete(url)
return response.ok

75
lib/slskd_api/apis/relay.py Normal file
View File

@@ -0,0 +1,75 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class RelayApi(BaseApi):
"""
[UNTESTED] This class contains the methods to interact with the Relay API.
"""
def connect(self) -> bool:
"""
Connects to the configured controller.
:return: True if successful.
"""
url = self.api_url + '/relay/agent'
response = self.session.put(url)
return response.ok
def disconnect(self) -> bool:
"""
Disconnects from the connected controller.
:return: True if successful.
"""
url = self.api_url + '/relay/agent'
response = self.session.delete(url)
return response.ok
def download_file(self, token: str) -> bool:
"""
Downloads a file from the connected controller.
:return: True if successful.
"""
url = self.api_url + f'/relay/controller/downloads/{token}'
response = self.session.get(url)
return response.ok
def upload_file(self, token: str) -> bool:
"""
Uploads a file from the connected controller.
:return: True if successful.
"""
url = self.api_url + f'/relay/controller/files/{token}'
response = self.session.post(url)
return response.ok
def upload_share_info(self, token: str) -> bool:
"""
Uploads share information to the connected controller.
:return: True if successful.
"""
url = self.api_url + f'/relay/controller/shares/{token}'
response = self.session.post(url)
return response.ok

124
lib/slskd_api/apis/rooms.py Normal file
View File

@@ -0,0 +1,124 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class RoomsApi(BaseApi):
"""
This class contains the methods to interact with the Rooms API.
"""
def get_all_joined(self) -> list:
"""
Gets all joined rooms.
:return: Names of the joined rooms.
"""
url = self.api_url + '/rooms/joined'
response = self.session.get(url)
return response.json()
def join(self, roomName: str) -> dict:
"""
Joins a room.
:return: room info: name, isPrivate, users, messages
"""
url = self.api_url + '/rooms/joined'
response = self.session.post(url, json=roomName)
return response.json()
def get_joined(self, roomName: str) -> dict:
"""
Gets the specified room.
:return: room info: name, isPrivate, users, messages
"""
url = self.api_url + f'/rooms/joined/{quote(roomName)}'
response = self.session.get(url)
return response.json()
def leave(self, roomName: str) -> bool:
"""
Leaves a room.
:return: True if successful.
"""
url = self.api_url + f'/rooms/joined/{quote(roomName)}'
response = self.session.delete(url)
return response.ok
def send(self, roomName: str, message: str) -> bool:
"""
Sends a message to the specified room.
:return: True if successful.
"""
url = self.api_url + f'/rooms/joined/{quote(roomName)}/messages'
response = self.session.post(url, json=message)
return response.ok
def get_messages(self, roomName: str) -> list:
"""
Gets the current list of messages for the specified room.
"""
url = self.api_url + f'/rooms/joined/{quote(roomName)}/messages'
response = self.session.get(url)
return response.json()
def set_ticker(self, roomName: str, ticker: str) -> bool:
"""
Sets a ticker for the specified room.
:return: True if successful.
"""
url = self.api_url + f'/rooms/joined/{quote(roomName)}/ticker'
response = self.session.post(url, json=ticker)
return response.ok
def add_member(self, roomName: str, username: str) -> bool:
"""
Adds a member to a private room.
:return: True if successful.
"""
url = self.api_url + f'/rooms/joined/{quote(roomName)}/members'
response = self.session.post(url, json=username)
return response.ok
def get_users(self, roomName: str) -> list:
"""
Gets the current list of users for the specified joined room.
"""
url = self.api_url + f'/rooms/joined/{quote(roomName)}/users'
response = self.session.get(url)
return response.json()
def get_all(self) -> list:
"""
Gets a list of rooms from the server.
"""
url = self.api_url + '/rooms/available'
response = self.session.get(url)
return response.json()

126
lib/slskd_api/apis/searches.py Normal file
View File

@@ -0,0 +1,126 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
import uuid
from typing import Optional
class SearchesApi(BaseApi):
"""
Class that handles operations on searches.
"""
def search_text(self,
searchText: str,
id: Optional[str] = None,
fileLimit: int = 10000,
filterResponses: bool = True,
maximumPeerQueueLength: int = 1000000,
minimumPeerUploadSpeed: int = 0,
minimumResponseFileCount: int = 1,
responseLimit: int = 100,
searchTimeout: int = 15000
) -> dict:
"""
Performs a search for the specified request.
:param searchText: Search query
:param id: uuid of the search. One will be generated if None.
:param fileLimit: Max number of file results
:param filterResponses: Filter unreachable users from the results
:param maximumPeerQueueLength: Max queue length
:param minimumPeerUploadSpeed: Min upload speed in bit/s
:param minimumResponseFileCount: Min number of matching files per user
:param responseLimit: Max number of users results
:param searchTimeout: Search timeout in ms
:return: Info about the search (no results!)
"""
url = self.api_url + '/searches'
try:
id = str(uuid.UUID(id)) # check if given id is a valid uuid
except:
id = str(uuid.uuid1()) # otherwise generate a new one
data = {
"id": id,
"fileLimit": fileLimit,
"filterResponses": filterResponses,
"maximumPeerQueueLength": maximumPeerQueueLength,
"minimumPeerUploadSpeed": minimumPeerUploadSpeed,
"minimumResponseFileCount": minimumResponseFileCount,
"responseLimit": responseLimit,
"searchText": searchText,
"searchTimeout": searchTimeout,
}
response = self.session.post(url, json=data)
return response.json()
def get_all(self) -> list:
"""
Gets the list of active and completed searches.
"""
url = self.api_url + '/searches'
response = self.session.get(url)
return response.json()
def state(self, id: str, includeResponses: bool = False) -> dict:
"""
Gets the state of the search corresponding to the specified id.
:param id: uuid of the search.
:param includeResponses: Include responses (search result list) in the returned dict
:return: Info about the search
"""
url = self.api_url + f'/searches/{id}'
params = dict(
includeResponses=includeResponses
)
response = self.session.get(url, params=params)
return response.json()
def stop(self, id: str) -> bool:
"""
Stops the search corresponding to the specified id.
:return: True if successful.
"""
url = self.api_url + f'/searches/{id}'
response = self.session.put(url)
return response.ok
def delete(self, id: str):
"""
Deletes the search corresponding to the specified id.
:return: True if successful.
"""
url = self.api_url + f'/searches/{id}'
response = self.session.delete(url)
return response.ok
def search_responses(self, id: str) -> list:
"""
Gets search responses corresponding to the specified id.
"""
url = self.api_url + f'/searches/{id}/responses'
response = self.session.get(url)
return response.json()

51
lib/slskd_api/apis/server.py Normal file
View File

@@ -0,0 +1,51 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class ServerApi(BaseApi):
"""
This class contains the methods to interact with the Server API.
"""
def connect(self) -> bool:
"""
Connects the client.
:return: True if successful.
"""
url = self.api_url + '/server'
response = self.session.put(url)
return response.ok
def disconnect(self) -> bool:
"""
Disconnects the client.
:return: True if successful.
"""
url = self.api_url + '/server'
response = self.session.delete(url, json='')
return response.ok
def state(self) -> dict:
"""
Retrieves the current state of the server.
"""
url = self.api_url + '/server'
response = self.session.get(url)
return response.json()

53
lib/slskd_api/apis/session.py Normal file
View File

@@ -0,0 +1,53 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class SessionApi(BaseApi):
"""
This class contains the methods to interact with the Session API.
"""
def auth_valid(self) -> bool:
"""
Checks whether the provided authentication is valid.
"""
url = self.api_url + '/session'
response = self.session.get(url)
return response.ok
def login(self, username: str, password: str) -> dict:
"""
Logs in.
:return: Session info for the given user incl. token.
"""
url = self.api_url + '/session'
data = {
'username': username,
'password': password
}
response = self.session.post(url, json=data)
return response.json()
def security_enabled(self) -> bool:
"""
Checks whether security is enabled.
"""
url = self.api_url + '/session/enabled'
response = self.session.get(url)
return response.json()

78
lib/slskd_api/apis/shares.py Normal file
View File

@@ -0,0 +1,78 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class SharesApi(BaseApi):
"""
This class contains the methods to interact with the Shares API.
"""
def get_all(self) -> dict:
"""
Gets the current list of shares.
"""
url = self.api_url + '/shares'
response = self.session.get(url)
return response.json()
def start_scan(self) -> bool:
"""
Initiates a scan of the configured shares.
:return: True if successful.
"""
url = self.api_url + '/shares'
response = self.session.put(url)
return response.ok
def cancel_scan(self) -> bool:
"""
Cancels a share scan, if one is running.
:return: True if successful.
"""
url = self.api_url + '/shares'
response = self.session.delete(url)
return response.ok
def get(self, id: str) -> dict:
"""
Gets the share associated with the specified id.
"""
url = self.api_url + f'/shares/{id}'
response = self.session.get(url)
return response.json()
def all_contents(self) -> list:
"""
Returns a list of all shared directories and files.
"""
url = self.api_url + '/shares/contents'
response = self.session.get(url)
return response.json()
def contents(self, id: str) -> list:
"""
Gets the contents of the share associated with the specified id.
"""
url = self.api_url + f'/shares/{id}/contents'
response = self.session.get(url)
return response.json()

157
lib/slskd_api/apis/transfers.py Normal file
View File

@@ -0,0 +1,157 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
from typing import Union
class TransfersApi(BaseApi):
"""
This class contains the methods to interact with the Transfers API.
"""
def cancel_download(self, username: str, id:str, remove: bool = False) -> bool:
"""
Cancels the specified download.
:return: True if successful.
"""
url = self.api_url + f'/transfers/downloads/{quote(username)}/{id}'
params = dict(
remove=remove
)
response = self.session.delete(url, params=params)
return response.ok
def get_download(self, username: str, id: str) -> dict:
"""
Gets the specified download.
"""
url = self.api_url + f'/transfers/downloads/{quote(username)}/{id}'
response = self.session.get(url)
return response.json()
def remove_completed_downloads(self) -> bool:
"""
Removes all completed downloads, regardless of whether they failed or succeeded.
:return: True if successful.
"""
url = self.api_url + '/transfers/downloads/all/completed'
response = self.session.delete(url)
return response.ok
def cancel_upload(self, username: str, id: str, remove: bool = False) -> bool:
"""
Cancels the specified upload.
:return: True if successful.
"""
url = self.api_url + f'/transfers/uploads/{quote(username)}/{id}'
params = dict(
remove=remove
)
response = self.session.delete(url, params=params)
return response.ok
def get_upload(self, username: str, id: str) -> dict:
"""
Gets the specified upload.
"""
url = self.api_url + f'/transfers/uploads/{quote(username)}/{id}'
response = self.session.get(url)
return response.json()
def remove_completed_uploads(self) -> bool:
"""
Removes all completed uploads, regardless of whether they failed or succeeded.
:return: True if successful.
"""
url = self.api_url + '/transfers/uploads/all/completed'
response = self.session.delete(url)
return response.ok
def enqueue(self, username: str, files: list) -> bool:
"""
Enqueues the specified download.
:param username: User to download from.
:param files: A list of dictionaries in the same form as what's returned
by :py:func:`~slskd_api.apis.SearchesApi.search_responses`:
[{'filename': <filename>, 'size': <filesize>}...]
:return: True if successful.
"""
url = self.api_url + f'/transfers/downloads/{quote(username)}'
response = self.session.post(url, json=files)
return response.ok
def get_downloads(self, username: str) -> dict:
"""
Gets all downloads for the specified username.
"""
url = self.api_url + f'/transfers/downloads/{quote(username)}'
response = self.session.get(url)
return response.json()
def get_all_downloads(self, includeRemoved: bool = False) -> list:
"""
Gets all downloads.
"""
url = self.api_url + '/transfers/downloads/'
params = dict(
includeRemoved=includeRemoved
)
response = self.session.get(url, params=params)
return response.json()
def get_queue_position(self, username: str, id: str) -> Union[int,str]:
"""
Gets the download for the specified username matching the specified filename, and requests the current place in the remote queue of the specified download.
:return: Queue position or error message
"""
url = self.api_url + f'/transfers/downloads/{quote(username)}/{id}/position'
response = self.session.get(url)
return response.json()
def get_all_uploads(self, includeRemoved: bool = False) -> list:
"""
Gets all uploads.
"""
url = self.api_url + '/transfers/uploads/'
params = dict(
includeRemoved=includeRemoved
)
response = self.session.get(url, params=params)
return response.json()
def get_uploads(self, username: str) -> dict:
"""
Gets all uploads for the specified username.
"""
url = self.api_url + f'/transfers/uploads/{quote(username)}'
response = self.session.get(url)
return response.json()

79
lib/slskd_api/apis/users.py Normal file
View File

@@ -0,0 +1,79 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from .base import *
class UsersApi(BaseApi):
"""
This class contains the methods to interact with the Users API.
"""
def address(self, username: str) -> dict:
"""
Retrieves the address of the specified username.
"""
url = self.api_url + f'/users/{quote(username)}/endpoint'
response = self.session.get(url)
return response.json()
def browse(self, username: str) -> dict:
"""
Retrieves the files shared by the specified username.
"""
url = self.api_url + f'/users/{quote(username)}/browse'
response = self.session.get(url)
return response.json()
def browsing_status(self, username: str) -> dict:
"""
Retrieves the status of the current browse operation for the specified username, if any.
Will return error 404 if called after the browsing operation has ended.
Best called asynchronously while :py:func:`browse` is still running.
"""
url = self.api_url + f'/users/{quote(username)}/browse/status'
response = self.session.get(url)
return response.json()
def directory(self, username: str, directory: str) -> dict:
"""
Retrieves the files from the specified directory from the specified username.
"""
url = self.api_url + f'/users/{quote(username)}/directory'
data = {
"directory": directory
}
response = self.session.post(url, json=data)
return response.json()
def info(self, username: str) -> dict:
"""
Retrieves information about the specified username.
"""
url = self.api_url + f'/users/{quote(username)}/info'
response = self.session.get(url)
return response.json()
def status(self, username: str) -> dict:
"""
Retrieves status for the specified username.
"""
url = self.api_url + f'/users/{quote(username)}/status'
response = self.session.get(url)
return response.json()

123
lib/slskd_api/client.py Normal file
View File

@@ -0,0 +1,123 @@
# Copyright (C) 2023 bigoulours
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
API_VERSION = 'v0'
import requests
from urllib.parse import urljoin
from functools import reduce
from base64 import b64encode
from slskd_api.apis import *
class HTTPAdapterTimeout(requests.adapters.HTTPAdapter):
def __init__(self, timeout=None, **kwargs):
super().__init__(**kwargs)
self.timeout = timeout
def send(self, *args, **kwargs):
kwargs['timeout'] = self.timeout
return super().send(*args, **kwargs)
class SlskdClient:
"""
The main class that allows access to the different APIs of a slskd instance.
An API-Key with appropriate permissions (`readwrite` for most use cases) must be set in slskd config file.
Alternatively, provide your username and password. Requests error status raise corresponding error.
Usage::
slskd = slskd_api.SlskdClient(host, api_key, url_base)
app_status = slskd.application.state()
"""
def __init__(self,
host: str,
api_key: str = None,
url_base: str = '/',
username: str = None,
password: str = None,
token: str = None,
verify_ssl: bool = True,
timeout: float = None # requests timeout in seconds
):
api_url = reduce(urljoin, [f'{host}/', f'{url_base}/', f'api/{API_VERSION}'])
session = requests.Session()
session.adapters['http://'] = HTTPAdapterTimeout(timeout=timeout)
session.adapters['https://'] = HTTPAdapterTimeout(timeout=timeout)
session.hooks = {'response': lambda r, *args, **kwargs: r.raise_for_status()}
session.headers.update({'accept': '*/*'})
session.verify = verify_ssl
header = {}
if api_key:
header['X-API-Key'] = api_key
elif username and password:
header['Authorization'] = 'Bearer ' + \
SessionApi(api_url, session).login(username, password).get('token', '')
elif token:
header['Authorization'] = 'Bearer ' + token
else:
raise ValueError('Please provide an API-Key, a valid token or username/password.')
session.headers.update(header)
base_args = (api_url, session)
self.application = ApplicationApi(*base_args)
self.conversations = ConversationsApi(*base_args)
self.logs = LogsApi(*base_args)
self.options = OptionsApi(*base_args)
self.public_chat = PublicChatApi(*base_args)
self.relay = RelayApi(*base_args)
self.rooms = RoomsApi(*base_args)
self.searches = SearchesApi(*base_args)
self.server = ServerApi(*base_args)
self.session = SessionApi(*base_args)
self.shares = SharesApi(*base_args)
self.transfers = TransfersApi(*base_args)
self.users = UsersApi(*base_args)
class MetricsApi:
"""
Getting the metrics works with a different endpoint. Default: <slskd_url>:5030/metrics.
Metrics should be first activated in slskd config file.
User/pass is independent from the main application and default value (slskd:slskd) should be changed.
Usage::
metrics_api = slskd_api.MetricsApi(host, metrics_usr='slskd', metrics_pwd='slskd')
metrics = metrics_api.get()
"""
def __init__(self,
host: str,
metrics_usr: str = 'slskd',
metrics_pwd: str = 'slskd',
metrics_url_base: str = '/metrics'
):
self.metrics_url = urljoin(host, metrics_url_base)
basic_auth = b64encode(bytes(f'{metrics_usr}:{metrics_pwd}', 'utf-8'))
self.header = {
'accept': '*/*',
'Authorization': f'Basic {basic_auth.decode()}'
}
def get(self) -> str:
"""
Gets the Prometheus metrics as text.
"""
response = requests.get(self.metrics_url, headers=self.header)
return response.text