 Unofficial Resonite API python library. 
"""
.. include:: ../README.md
"""
import sys

if sys.version_info[:2] >= (3, 8):
    from importlib import metadata
else:
    import importlib_metadata as metadata

__version__ = metadata.version(__package__)

del metadata, sys
resonitepy/__init__.py
brodokk/resonitepy
"""
This module define some of the Resonite API json
responce under usable python classes.
"""

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from pathlib import PureWindowsPath
from typing import List, Optional

from resonitepy.secrets import generate
from resonitepy.exceptions import ResoniteException

from urllib.parse import ParseResult


class RecordType(Enum):
    """Representation of the type of a Resonite record."""
    OBJECT = "object"
    LINK = "link"
    DIRECTORY = "directory"
    WORLD = "world"
    TEXTURE = "texture"
    AUDIO = "audio"


@dataclass
class ResoniteRecordVersion:
    globalVersion: int
    localVersion: int
    lastModifyingUserId: Optional[str]
    lastModifyingMachineId: Optional[str]

@dataclass
class ResoniteRecord:
    """Representation of a Resonite record."""
    id: str
    assetUri: Optional[str]
    version: ResoniteRecordVersion
    name: str
    recordType: RecordType
    ownerName: str
    path: Optional[str]
    thumbnailUri: Optional[str]
    isPublic: bool
    isForPatrons: bool
    isListed: bool
    isDeleted: bool
    tags: Optional[list]
    creationTime: Optional[datetime]
    lastModificationTime: datetime
    randomOrder: int
    visits: int
    rating: int
    ownerId: str


@dataclass
class ResoniteLink(ResoniteRecord):
    """Representation of a Resonite link."""
    assetUri: ParseResult


@dataclass
class ResoniteDirectory(ResoniteRecord):
    """Representation of a Resonite directory."""
    lastModifyingMachineId: Optional[str]
    ownerName: str
    tags: List[str]
    creationTime: Optional[datetime]

    @property
    def content_path(self) -> str:
        return str(PureWindowsPath(self.path, self.name))


@dataclass
class ResoniteObject(ResoniteRecord):
    """Represenation of a Resonite object."""
    assetUri: str
    lastModifyingMachineId: Optional[str]
    ownerName: str
    tags: List[str]
    creationTime: datetime

@dataclass
class ResoniteWorld(ResoniteRecord):
    """Representation of a Resonite world."""
    pass

@dataclass
class ResoniteTexture(ResoniteRecord):
    """Representation of a Resonite texture."""
    pass

@dataclass
class ResoniteAudio(ResoniteRecord):
    """Representation of a Resonite audio."""
    pass


recordTypeMapping = {
    RecordType.DIRECTORY: ResoniteDirectory,
    RecordType.LINK: ResoniteLink,
    RecordType.OBJECT: ResoniteObject,
    RecordType.WORLD: ResoniteWorld,
    RecordType.TEXTURE: ResoniteTexture,
    RecordType.AUDIO: ResoniteAudio,
}


@dataclass
class LoginDetailsAuth:
    password: str

    def build_dict(self):
        return {
            "$type": "password",
            "password": self.password,
            "recoveryCode": None
        }


@dataclass
class LoginDetails:
    authentication: LoginDetailsAuth
    """Representation of a Resonite login detail."""
    ownerId: Optional[str] = None
    """The ownerId of a LoginDetails should start with `U-`"""
    username: Optional[str] = None
    email: Optional[str] = None
    secretMachineId: str = field(default_factory=generate)
    """See the generate class in the resonite.secrets module."""
    rememberMe: Optional[str] = False

    def __post_init__(self):
        if not self.ownerId and not self.username and not self.email:
            raise ResoniteException(
                'Either an ownerId, an username or an email is needed')
        if not self.authentication:
            raise ResoniteException('A password is needed')


@dataclass
class ProfileData:
    """Representation of a Resonite profile data."""
    iconUrl: Optional[str]
    tokenOutOut: Optional[List[str]]


@dataclass
class PatreonData:
    """Representation of a Resonite Patreon data."""
    isPatreonSupporter: bool
    patreonId: Optional[str]
    lastPatreonPledgeCents: int
    lastTotalCents: int
    minimumTotalUnits: int
    externalCents: int
    lastExternalCents: int
    hasSupported: bool
    lastIsAnorak: bool
    priorityIssue: int
    lastPlusActivationTime: datetime
    lastActivationTime: datetime
    lastPlusPledgeAmount: int
    lastPaidPledgeAmount: int
    accountName: str
    currentAccountType: int
    currentAccountCents: int
    pledgedAccountType: int


@dataclass
class QuotaBytesSources:
    base: int
    patreon: int
    paid: int
    mmc21_honorary: int

@dataclass
class ResoniteUserQuotaBytesSources:
    base: int
    patreon: Optional[int]

@dataclass
class ResoniteUserMigrationData:
    username: str
    userId: str
    quotaBytes: int
    usedBytes: int
    quotaBytesSources: ResoniteUserQuotaBytesSources
    registrationDate: datetime


@dataclass
class ResoniteUserEntitlementShoutOut:
    shoutoutType: str
    friendlyDescription: str

@dataclass
class ResoniteUserEntitlementCredits:
    creditType: str
    friendlyDescription: str


@dataclass
class ResoniteUser:
    """Representation of a Resonite user."""
    id: str
    username: str
    normalizedUsername: str
    registrationDate: datetime
    isVerified: bool
    isLocked: bool
    supressBanEvasion: bool
    #2fa_login: bool # TODO
    profile: Optional[ProfileData]
    supporterMetadata: Optional[List[dict]]
    entitlements: Optional[List[ResoniteUserEntitlementShoutOut | ResoniteUserEntitlementCredits]]
    migratedData: Optional[ResoniteUserMigrationData]
    tags: Optional[List[str]] = field(default_factory=list)

@dataclass
class WorldId:
    """Representation of a world id."""
    ownerId: str
    """The ownerId of a WorldId should start with `U-`"""
    recordId: str

@dataclass
class ResoniteSessionUser:
    """Representation of a session user."""
    isPresent: bool
    userID: Optional[str]
    username: str
    userSessionId: Optional[str]
    outputDevice: Optional[int]

@dataclass
class ResoniteSession:
    """Representation of a session."""
    activeSessions: Optional[str]
    activeUsers: int
    compatibilityHash: Optional[str]
    correspondingWorldId: Optional[WorldId]
    description: Optional[str]
    hasEnded: bool
    headlessHost: bool
    hostMachineId: str
    hostUserSessionId: str
    hostUserId: Optional[str]
    hostUsername: str
    isValid: bool
    joinedUsers: int
    lastUpdate: datetime
    maxUsers: int
    mobileFriendly: bool
    name: str
    appVersion: str
    normalizedSessionId: str
    sessionBeginTime: datetime
    sessionId: str
    sessionURLs: List[str]
    sessionUsers: List[ResoniteSessionUser]
    tags: List[str]
    thumbnailUrl: Optional[str]
    totalActiveUsers: int
    totalJoinedUsers: int
    hideFromListing: bool


@dataclass
class PublicRSAKey:
    """Representation of a public key."""
    Exponent: str
    Modulus: str


class OnlineStatus(Enum):
    """Representation of an online status."""
    ONLINE = "Online"
    AWAY = "Away"
    BUSY = "Busy"
    OFFLINE = "Offline"


onlineStatusMapping = {
    OnlineStatus.ONLINE: "Online",
    OnlineStatus.AWAY: "Away",
    OnlineStatus.BUSY: "Busy",
    OnlineStatus.OFFLINE: "Offline",
}


class CurrentResoniteSessionAccessLevel(Enum):
    """Representation of a current session access level."""
    PRIVATE = 0
    LAN = 1
    FRIENDS = 2
    FRIENDSOFFRIENDS = 3
    REGISTEREDUSERS = 4
    ANYONE = 5

    def __str__(self):
        text = {
            'PRIVATE': 'Private',
            'LAN': 'LAN',
            'FRIENDS': 'Contacts',
            'FRIENDSOFFRIENDS': 'Contacts+',
            'REGISTEREDUSERS': 'Registered Users',
            'ANYONE': 'Anyone'
        }
        return text[self.name]


currentResoniteSessionAccessLevelMapping = {
    CurrentResoniteSessionAccessLevel.PRIVATE: 0,
    CurrentResoniteSessionAccessLevel.LAN: 1,
    CurrentResoniteSessionAccessLevel.FRIENDS: 2,
    CurrentResoniteSessionAccessLevel.FRIENDSOFFRIENDS: 3,
    CurrentResoniteSessionAccessLevel.REGISTEREDUSERS: 4,
    CurrentResoniteSessionAccessLevel.ANYONE: 5,
}


@dataclass
class UserStatusData:
    """Representation of a Resonite user status data."""
    activeSessions: Optional[List[ResoniteSession]]
    currentSession: Optional[ResoniteSession]
    compatibilityHash: Optional[str]
    currentHosting: bool
    currentSessionAccessLevel: CurrentResoniteSessionAccessLevel
    currentSessionHidden: bool
    currentSessionId: Optional[str]
    isMobile: bool
    lastStatusChange: datetime
    neosVersion: Optional[str]
    onlineStatus: OnlineStatus
    OutputDevice: Optional[str]
    publicRSAKey: Optional[PublicRSAKey]

@dataclass
class ResoniteUserStatus:
    """Represenation of a Resonite user status."""
    onlineStatus: OnlineStatus
    lastStatusChange: datetime
    currentSessionAccessLevel: int
    currentSessionHidden: bool
    currentHosting: bool
    compatibilityHash: Optional[str]
    neosVersion: Optional[str]
    publicRSAKey: Optional[PublicRSAKey]
    OutputDevice: Optional[str]
    isMobile: bool

class ContactStatus(Enum):
    """Representation of a ContactStatus."""
    ACCEPTED = "Accepted"
    IGNORED = "Ignored"
    REQUESTED = "Requested"
    NONE = "None"


contactStatusMapping = {
    ContactStatus.ACCEPTED: "Accepted",
    ContactStatus.IGNORED: "Ignored",
    ContactStatus.REQUESTED: "Requested",
    ContactStatus.NONE: "None",
}


@dataclass
class ResoniteContact:
    """Representation of a Resonite contact."""
    id: str
    contactUsername: str
    contactStatus: ContactStatus
    isAccepted: bool
    profile: Optional[ProfileData]
    latestMessageTime: datetime
    isMigrated: bool
    isCounterpartMigrated: bool
    ownerId: str

class ResoniteMessageType(Enum):
    TEXT = "Text"
    OBJECT = "Object"
    SOUND = "Sound"
    SESSIONINVITE = "SessionInvite"
    CREDITTRANSFER = "CreditTransfer"
    SUGARCUBES = "SugarCubes"

ResoniteMessageTypeMapping = {
    ResoniteMessageType.TEXT: "Text",
    ResoniteMessageType.OBJECT: "Object",
    ResoniteMessageType.SOUND: "Sound",
    ResoniteMessageType.SESSIONINVITE: "SessionInvite",
    ResoniteMessageType.CREDITTRANSFER: "CreditTransfer",
    ResoniteMessageType.SUGARCUBES: "SugarCubes",
}


class ResoniteMessageContentText(str):
    def __new__(cls, *args, **kw):
        return str.__new__(cls, *args, **kw)


@dataclass
class ResoniteMessageContentObject:
    id: str
    ownerId: str
    assetUri: str
    version: Optional[ResoniteRecordVersion]
    name: str
    recordType: RecordType
    ownerName: Optional[str]
    tags: List[str]
    path: Optional[str]
    thumbnailUri: str
    isPublic: bool
    isForPatrons: bool
    isListed: bool
    lastModificationTime: datetime
    creationTime: datetime
    firstPublishTime: Optional[datetime]
    isDeleted: Optional[bool]
    visits: int
    rating: float
    randomOrder: int
    sumbissions: Optional[str]

@dataclass
class ResoniteMessageContentSessionInvite:
    name: str
    description: Optional[str]
    correspondingWorldId: Optional[WorldId]
    tags: List[str]
    sessionId: str
    normalizedSessionId: str
    hostMachineId: str
    hostUsername: str
    compatibilityHash: Optional[str]
    universeId: Optional[str]
    appVersion: Optional[str]
    headlessHost: Optional[bool]
    sessionURLs: List[str]
    parentSessionIds: Optional[List[str]]
    nestedSessionIds: Optional[List[str]]
    sessionUsers: List[ResoniteSessionUser]
    thumbnail: Optional[str]
    joinedUsers: int
    activeUsers: int
    totalActiveUsers: int
    totalJoinedUsers: int
    maxUsers: int
    mobileFriendly: bool
    sessionBeginTime: datetime
    lastUpdate: datetime
    accessLevel: str
    broadcastKey: Optional[str]


@dataclass
class ResoniteMessageContentSound:
    id: str
    OwnerId: Optional[str]
    assetUri: str
    globalVersion: Optional[int]
    localVersion: Optional[int]
    lastModifyingUserId: Optional[str]
    lastModifyingMachineId: Optional[str]
    name: str
    recordType: RecordType
    ownerName: Optional[str]
    tags: List[str]
    path: Optional[str]
    isPublic: bool
    isForPatron: Optional[bool]
    isListed: bool
    lastModificationTime: datetime
    creationTime: datetime
    firstPublishTime: Optional[datetime]
    visits: int
    rating: float
    randomOrder: int
    submissions: Optional[str]
    neosDBmanifest: Optional[list]



@dataclass
class ResoniteMessage:
    """Representation of a Resonite message."""
    id: str
    senderId: str
    ownerId: str
    """The ownerId of a ResoniteMessage should start with `U-`"""
    sendTime: str
    recipientId: str
    messageType: ResoniteMessageType
    content: ResoniteMessageContentText | ResoniteMessageContentSessionInvite | ResoniteMessageContentObject | ResoniteMessageContentSound

@dataclass
class ResoniteCloudVar:
    """Representation of Resonite clound variable."""
    ownerId: str
    """The ownerId of a ResoniteCloudVar should start with `U-`"""
    path: str
    """The path of a ResoniteCloudVar should start with a `U-` for a user owned path and a `G-` for a group owned path."""
    value: Optional[str]
    partitionKey: str
    rowKey: str
    timestamp: str
    eTag: Optional[str]


class OwnerType(Enum):
    MACHING = "Machine"
    USER = "User"
    GROUP = "Group"
    INVALID = "Invalid"


@dataclass
class ResoniteCloudVarDefs:
    definitionOwnerId: str
    subpath: str
    variableType: str
    defaultValue: Optional[str]
    deleteScheduled: bool
    readPermissions: List[str]
    writePermissions: List[str]
    listPermissions: List[str]
    partitionKey: str
    rowKey: str
    timestamp: str
    eTag: str
resonitepy/classes.py
brodokk/resonitepy
"""
This module define the client who will do the request to the
Resonite API.
"""

import os
import dataclasses
import json
import logging
from datetime import datetime
from hashlib import sha256
from os import path as OSpath
from typing import Dict, List
from urllib.parse import ParseResult, urlparse

import dacite
from requests import Session
from requests import exceptions as requests_exceptions
from dateutil.parser import isoparse

from . import __version__
from .classes import (
    LoginDetails,
    ResoniteDirectory,
    ResoniteContact,
    ResoniteLink,
    ResoniteRecord,
    ResoniteUser,
    ResoniteUserEntitlementShoutOut,
    ResoniteUserEntitlementCredits,
    ResoniteMessage,
    ResoniteMessageType,
    ResoniteMessageContentSessionInvite,
    ResoniteMessageContentObject,
    ResoniteMessageContentSound,
    ResoniteMessageContentText,
    ResoniteCloudVar,
    RecordType,
    recordTypeMapping,
    OnlineStatus,
    ResoniteSession,
    CurrentResoniteSessionAccessLevel,
    ContactStatus,
    OwnerType,
    ResoniteCloudVarDefs,
)
from .utils import (
    nested_asdict_factory,
    getOwnerType,
)

from .endpoints import API_URL, ASSETS_URL
from resonitepy import exceptions as resonite_exceptions

DACITE_CONFIG = dacite.Config(
    cast=[
        ResoniteMessageType,
        RecordType,
        OnlineStatus,
        CurrentResoniteSessionAccessLevel,
        ContactStatus,
    ],
    type_hooks={
        datetime: isoparse,
        ParseResult: urlparse,
    },
)

AUTHFILE_NAME = "auth.token"


@dataclasses.dataclass
class Client:
    """Representation of a resonitepy Resonite client."""
    userId: str = None
    token: str = None
    expire: datetime = None  # This don't seems to be use by the API.
    rememberMe: bool = False
    lastUpdate: datetime = None
    secretMachineIdHash: str = None
    secretMachineIdSalt: str = None
    session: Session = Session()
    session.headers['UID'] = sha256(os.urandom(16)).hexdigest().upper()

    @property
    def headers(self) -> dict:
        default = {"User-Agent": f"resonitepy/{__version__}"}
        if not self.userId or not self.token:
            logging.warning("WARNING: headers sections not set. this might throw an error soon...")
            return default
        default["Authorization"] = f"res {self.userId}:{self.token}"
        return default

    @staticmethod
    def processRecordList(data: List[dict]):
        ret = []
        for raw_item in data:
            item = dacite.from_dict(ResoniteRecord, raw_item, DACITE_CONFIG)
            x = dacite.from_dict(recordTypeMapping[item.recordType], raw_item, DACITE_CONFIG)
            ret.append(x)
        return ret

    def _request(
            self, verb: str, path: str, data: dict = None, json: dict = None,
            params: dict = None, ignoreUpdate: bool = False
        ) -> Dict:
        if self.lastUpdate and not ignoreUpdate:
            lastUpdate = self.lastUpdate
            # From PolyLogix/CloudX.js, the token seems to expire after 3600000 seconds
            if (datetime.now() - lastUpdate).total_seconds() <= 3600000:
                self._request('patch', '/userSessions', ignoreUpdate=True)
                self.lastUpdate = datetime.now()
            # While the API dont seems to implement more security, official client behavior must be respected.
            # Only disconnect after 1 day of inactivity for now.
            # TODO: Implement disconnection after 1 week of inactivity when implementing the rememberMe feature.
            #if 64800 >= (datetime.now() - lastUpdate).total_seconds() >= 85536:
            #    self._request('patch', '/userSessions', ignoreUpdate=True)
            else:
                raise resonite_exceptions.InvalidToken("Token expired")
        args = {'url': API_URL + path}
        if data: args['data'] = data
        if json: args['json'] = json
        if params: args['params'] = params
        func = getattr(self.session, verb, None)
        with func(**args) as req:
            logging.debug("ResoniteAPI: [{}] {}".format(req.status_code, args))
            if req.status_code not in [200, 204]:
                if "Invalid credentials" in req.text:
                    raise resonite_exceptions.InvalidCredentials(req.text)
                elif req.status_code == 403:
                    raise resonite_exceptions.InvalidToken(req.headers)
                else:
                    raise resonite_exceptions.ResoniteAPIException(req)
            if req.status_code == 200:
                try:
                    response = req.json()
                    if "message" in response:
                        raise resonite_exceptions.ResoniteAPIException(req, message=response["message"])
                    return response
                except requests_exceptions.JSONDecodeError:
                    return req.text
            # In case of a 204 response
            return

    def login(self, data: LoginDetails) -> None:
        json=dataclasses.asdict(data)
        json['authentication'] = data.authentication.build_dict()
        response = self._request('post', "/userSessions", json=json)
        self.userId = response["entity"]["userId"]
        self.token = response["entity"]["token"]
        self.secretMachineIdHash = response["entity"]["secretMachineIdHash"]
        self.secretMachineIdSalt = response["entity"]["secretMachineIdSalt"]
        self.expire = isoparse(response["entity"]["expire"])
        self.lastUpdate = datetime.now()
        self.session.headers.update(self.headers)

    def logout(self) -> None:
        self._request('delete',
            "/userSessions/{}/{}".format(self.userId, self.token),
            ignoreUpdate=True,
        )
        self.clean_session()

    def clean_session(self) -> None:
        self.userId = None
        self.token = None
        self.expire = None
        self.secretMachineIdHash = None
        self.secretMachineIdSalt = None
        self.lastUpdate = None
        del self.session.headers["Authorization"]
        self.session.headers.update(self.headers)

    def load_token(self):
        if OSpath.exists(AUTHFILE_NAME):
            with open(AUTHFILE_NAME, "r") as f:
                session = json.load(f)
                expire = datetime.fromisoformat(session["expire"])
                if datetime.now().timestamp() < expire.timestamp():
                    self.token = session["token"]
                    self.userId = session["userId"]
                    self.expire = expire
                    self.secretMachineIdHash = session["secretMachineIdHash"]
                    self.secretMachineIdSalt = session["secretMachineIdSalt"]
                    self.session.headers.update(self.headers)
                else:
                    raise resonite_exceptions.NoTokenError
        else:
            raise resonite_exceptions.NoTokenError

    def save_token(self):
        with open(AUTHFILE_NAME, "w+") as f:
            json.dump(
                {
                    "userId": self.userId,
                    "expire": self.expire.isoformat(),
                    "token": self.token,
                    "secretMachineIdHash": self.secretMachineIdHash,
                    "secretMachineIdSalt": self.secretMachineIdSalt,
                },
                f,
            )

    def resDBSignature(self, resUrl: str) -> str:
        return resUrl.split("//")[1].split(".")[0]

    def resDbToHttp(self, resUrl: str) -> str:
        url = ASSETS_URL + self.resDBSignature(resUrl)
        return url

    def getUserData(self, user: str = None) -> ResoniteUser:
        if user is None:
            user = self.userId
        response = self._request('get', "/users/" + user)
        return dacite.from_dict(ResoniteUser, response, DACITE_CONFIG)

    def getSession(self, session_id: str) -> ResoniteSession:
        """ Return session information.
        """
        response = self._request('get', f'/sessions/{session_id}')
        return dacite.from_dict(ResoniteSession, response, DACITE_CONFIG)

    def getContacts(self):
        """
        returns the contacts you have.

        Note: does not create contact out of thin air. you need to do that yourself.
        """
        response = self._request('get', f"/users/{self.userId}/contacts")
        return [dacite.from_dict(ResoniteContact, user, DACITE_CONFIG) for user in response]

    def getInventory(self) -> List[ResoniteRecord]:
        """
        The typical entrypoint to the inventory system.
        """
        response = self._request(
            'get',
            f"/users/{self.userId}/records",
            params={"path": "Inventory"},
        )
        return self.processRecordList(response)

    def getDirectory(self, directory: ResoniteDirectory) -> List[ResoniteRecord]:
        """
        given a directory, return it's contents.
        """
        response = self._request(
            'get',
            f"/users/{directory.ownerId}/records",
            params={"path": directory.content_path},
        )
        return self.processRecordList(response)

    def resolveLink(self, link: ResoniteLink) -> ResoniteDirectory:
        """
        given a link type record, will return it's directory. directoy can be passed to getDirectory
        """
        if link.assetUri.scheme != 'resrec':
            raise resonite_exceptions.ResoniteException(f'Not supported link type {link}')
        import re
        m = re.search('\/(U-.*)\/(R-.*)', link.assetUri.path)
        if not m:
            raise resonite_exceptions.ResoniteException(f'Not supported link type {link}')
        user = m.group(1)
        record = m.group(2)
        response = self._request(
            'get',
            f"/users/{user}/records/{record}",
        )
        return dacite.from_dict(ResoniteDirectory, response, DACITE_CONFIG)


    def getMessageLegacy(
        self, fromTime: str = None, maxItems: int = 100,
        user: str = None, unreadOnly: bool = False
    ) -> List[ResoniteMessage]:
        params = {}
        if fromTime:
            raise ValueError('fromTime is not yet implemented')
        params['maxItems'] = maxItems
        params['unreadOnly'] = unreadOnly
        if user:
            params['user'] = user
        response = self._request(
            'get',
            f'/users/{self.userId}/messages',
            params=params
        )
        messages = []
        for message in response:
            if message['messageType'] == 'SessionInvite':
                message['content'] = json.loads(message['content'])
                ResoniteMessageContentType = ResoniteMessageContentSessionInvite
            elif message['messageType'] == 'Object':
                message['content'] = json.loads(message['content'])
                ResoniteMessageContentType = ResoniteMessageContentObject
            elif message['messageType'] == 'Sound':
                message['content'] = json.loads(message['content'])
                ResoniteMessageContentType = ResoniteMessageContentSound
            elif  message['messageType'] == 'Text':
                ResoniteMessageContentType = ResoniteMessageContentText
            else:
                raise ValueError(f'Non supported type {message["messageType"]}')
            message['content'] = dacite.from_dict(
                ResoniteMessageContentType, message['content'], DACITE_CONFIG
            )
            messages.append(
                dacite.from_dict(ResoniteMessage, message, DACITE_CONFIG)
            )
        return messages

    def getOwnerPath(self, ownerId: str) -> str:
        ownerType = getOwnerType(ownerId)
        if ownerType == OwnerType.USER:
            return "users"
        elif ownerType == OwnerType.GROUP:
            return "groups"
        else:
            raise ValueError(f"invalid ownerType for {ownerId}")

    def listCloudVar(self, ownerId: str) -> List[ResoniteCloudVar]:
        response = self._request(
            'get',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars'
        )
        return [dacite.from_dict(ResoniteCloudVar, cloud_var, DACITE_CONFIG) for cloud_var in response]

    def getCloudVar(self, ownerId: str, path: str) -> ResoniteCloudVar:
        response = self._request(
            'get',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars/{path}'
        )
        return dacite.from_dict(ResoniteCloudVar, response, DACITE_CONFIG)

    def getCloudVarDefs(self, ownerId: str, path: str) -> ResoniteCloudVarDefs:
        json=[{
            "ownerId": ownerId,
            "path": path,
        }]
        response = self._request(
            'post',
            f'/readvars',
            json=json,
        )
        if not response:
            raise resonite_exceptions.ResoniteException(f"{ownerId} {path} doesn't exist")
        return dacite.from_dict(ResoniteCloudVarDefs, response[0]['definition'], DACITE_CONFIG)

    def setCloudVar(self, ownerId: str, path: str, value: str) -> None:
        return self._request(
            'put',
            f'/{self.getOwnerPath(ownerId)}/{ownerId}/vars/{path}',
            json = {
                "ownerId": ownerId,
                "path": path,
                "value": value,
            }
        )

    def searchUser(self, username: str) -> List[ResoniteUser]:
        """
        return a list of user based on username.

        This is not the U- Resonite user id, the API will search over usernames not ids.
        """
        response = self._request(
            'get',
            '/users',
            params = {'name': username}
        )
        users = []
        for user in response:
            if 'entitlements' in user:
                entitlements = []
                for entitlement in user['entitlements']:
                    if entitlement['$type'] == 'shoutOut':
                        del entitlement['$type']
                        entitlements.append(
                            ResoniteUserEntitlementShoutOut(**entitlement)
                        )
                    elif entitlement['$type'] == 'credits':
                        del entitlement['$type']
                        entitlements.append(
                            ResoniteUserEntitlementCredits(**entitlement)
                        )
                    else:
                        print('Warning: {entitlement["$type"]} unknown')
                user['entitlements'] = entitlements
            users.append(dacite.from_dict(ResoniteUser, user, DACITE_CONFIG))
        return users

resonitepy/client.py
brodokk/resonitepy
"""
This module define some basic domain use by Resonite API.
"""

API_URL = "https://api.resonite.com"
ASSETS_URL = "https://assets.resonite.com"
resonitepy/endpoints.py
brodokk/resonitepy
"""
This module define the Exceptions classes use by this package.
"""

import requests
import json

class ResoniteException(Exception):
    pass

class ResoniteAPIException(ResoniteException):

    def __init__(self, req, message = None):
        if not message:
            message = "[" + str(req.status_code) + "] API response: " + req.text
            message += "\n HEADERS:\n"
            message += str(req.headers)
        super().__init__(message)
        self.status_code = req.status_code
        try:
            self.json = req.json()
        except json.decoder.JSONDecodeError:
            self.json = {}

class InvalidCredentials(ResoniteException):
    pass


class NoTokenError(ResoniteException):
    pass


class FolderNotFound(ResoniteException):
    pass


class InvalidToken(ResoniteException):
    pass

resonitepy/exceptions.py
brodokk/resonitepy
import random
import string


def generate() -> str:
    """
    direct reimpl of secretID.js from account.resonite.com
    """
    length = 12
    valid_chars = string.ascii_letters + string.digits
    return "".join(random.choices(valid_chars, k=length))

resonitepy/secrets.py
brodokk/resonitepy
"""
This module define some generic function use at multiple
places in the package.
"""

from enum import Enum
from .classes import OwnerType

def nested_asdict_factory(data):
    """Factory for convert dataclasses as dict.

    Can be use when a dataclass have an Enum.
    Example:

    ```
    dataclasses.asdict(
        <dataclass>,
        dict_factory=nested_asdict_factory,
    )
    ```
    """

    def convert_value(obj):
        if isinstance(obj, Enum):
            return obj.value
        return obj

    return dict((k, convert_value(v)) for k, v in data)

def getOwnerType(ownerId: str) -> str:
    if ownerId.startswith('U-'):
        return OwnerType.USER
    elif ownerId.startswith('G-'):
        return OwnerType.GROUP
    elif ownerId.startswith('M-'):
        return OwnerType.MACHINE
    else:
        return OwnerType.INVALID
resonitepy/utils.py
brodokk/resonitepy