rocket.py

#! /usr/bin/env python3
# -*- coding: utf-8; py-indent-offset: 4 -*-
#
# Author:  Linuxfabrik GmbH, Zurich, Switzerland
# Contact: info (at) linuxfabrik (dot) ch
#          https://www.linuxfabrik.ch/
# License: The Unlicense, see LICENSE file.

# https://github.com/Linuxfabrik/lib/blob/main/CONTRIBUTING.md

"""This library collects some Rocket.Chat related functions that are
needed by more than one Rocket.Chat consumer.

Typical use-case:

.. code-block:: python

    credentials = lib.base.coe(
        lib.rocket.get_token(
            args.URL,
            args.USERNAME,
            args.PASSWORD,
            insecure=args.INSECURE,
            no_proxy=args.NO_PROXY,
            timeout=args.TIMEOUT,
        )
    )
    auth_token, user_id = credentials.split(':')
    result = lib.base.coe(
        lib.rocket.get_stats(
            args.URL,
            auth_token,
            user_id,
            insecure=args.INSECURE,
            no_proxy=args.NO_PROXY,
            timeout=args.TIMEOUT,
        )
    )
"""

__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
__version__ = '2026092101'

from . import url


def _flatten_params(params):
    """
    Recursively flatten a dictionary of parameters into a URL-style query string.

    This helper function takes a dictionary where values may be primitives (e.g., strings, numbers)
    or nested dictionaries, and converts it into a single query string. Nested dictionaries are
    rendered inside braces `{}` with their own `key=value` pairs joined by `&`.

    Parameters
    ----------
    params : dict
        Mapping of parameter names to values. Values can be:

        - Primitive types (`str`, `int`, etc.), rendered as `key=value`.
        - Nested `dict`, rendered as `key={inner_key1=inner_val1&inner_key2=inner_val2}`.

    Returns
    -------
    str
        A query string with `&`-separated `key=value` pairs. Nested dicts are enclosed in
        `{}`.

    Examples
    --------
    >>> params = {
    ...     'key1': 'value1',
    ...     'key2': 'value2',
    ...     'key3': {
    ...         'subkey1': 'subvalue1',
    ...         'subkey2': 'subvalue2',
    ...     },
    ... }
    >>> _flatten_params(params)
    'key1=value1&key2=value2&key3={subkey1=subvalue1&subkey2=subvalue2}'
    """
    parts = []
    for key, val in params.items():
        if isinstance(val, dict):
            inner = _flatten_params(val)
            parts.append(f'{key}={{{inner}}}')
        else:
            parts.append(f'{key}={val}')
    return '&'.join(parts)


def get_groups_history(
    rc_url,
    auth_token,
    user_id,
    room_id=None,
    params=None,
    insecure=False,
    no_proxy=False,
    proxy=None,
    timeout=3,
):
    """
    Retrieve message history for a private group via Rocket.Chat's `groups.history` API.

    This function constructs the correct endpoint URL for `groups.history`, injects the required
    `roomId` into the query parameters along with any additional options, attaches the
    authentication headers (`X-Auth-Token` and `X-User-Id`), and performs a GET request to fetch
    the message history.

    Equivalent to:

    .. code-block:: bash

        curl -H "X-Auth-Token: <auth_token>" \\
             -H "X-User-Id: <user_id>" \\
             "https://chat.example.com/api/v1/groups.history?roomId=<roomId>&count=20&offset=0"

    Parameters
    ----------
    rc_url : str
        Rocket.Chat base URL or full endpoint URL. If it does not already end with
        `/groups.history`, any trailing slashes will be stripped and `/groups.history` appended.
    auth_token : str
        Authentication token from login (for the `X-Auth-Token` header).
    user_id : str
        User ID from login (for the `X-User-Id` header).
    room_id : str
        ID of the private group whose history you want to fetch. Required.
    params : dict, optional
        Additional query parameters for pagination and date filtering,
        such as:

        - `count` (`int`): Number of messages to return.
        - `offset` (`int`): Number of messages to skip.
        - `oldest` (`str`): ISO8601 timestamp for the earliest message.
        - `latest` (`str`): ISO8601 timestamp for the latest message.
        Defaults to `{}`.
    insecure : bool, optional
        Allow insecure SSL connections. Defaults to `False`.
    no_proxy : bool, optional
        Bypass proxy settings. Defaults to `False`.
    proxy : str, optional
        Proxy to reach the target through, overriding the one the environment names.
        Defaults to `None`, which leaves the choice to the environment.
    timeout : int, optional
        Request timeout in seconds. Defaults to `3`.

    Returns
    -------
    tuple (bool, dict or str)
        - On success: `(True, response_json)` where `response_json` contains the history payload.
        - On failure: `(False, 'Error getting groups.history: <error message>')`.

    Examples
    --------
    >>> success, history = get_groups_history(
    ...     'https://chat.example.com',
    ...     'authTokenHere',
    ...     'userIdHere',
    ...     room_id='ABC123',
    ...     params={'count': 50, 'offset': 0}
    ... )
    >>> if success:
    ...     for msg in history.get('messages', []):
    ...         print(msg['u']['username'], msg['msg'])
    """
    if not rc_url.endswith('/groups.history'):
        rc_url = rc_url.rstrip('/') + '/groups.history'

    if params is None:
        params = {}
    params['roomId'] = room_id  # add the required room_id to the parameter list
    query = _flatten_params(params)
    rc_url = rc_url + f'{"?" + query if query else ""}'

    headers = {
        'X-Auth-Token': auth_token,
        'X-User-Id': user_id,
    }

    success, result = url.fetch_json(
        rc_url,
        header=headers,
        insecure=insecure,
        no_proxy=no_proxy,
        proxy=proxy,
        timeout=timeout,
    )

    if not success or not result:
        return False, f'Error getting groups.history: {result}'

    return True, result


def get_rooms_get(
    rc_url, auth_token, user_id, insecure=False, no_proxy=False, proxy=None, timeout=3
):
    """
    Retrieve the list of chat rooms accessible to the authenticated user via Rocket.Chat's
    `rooms.get` API.

    This function constructs the correct endpoint URL for `rooms.get`, attaches the required
    authentication headers (`X-Auth-Token` and `X-User-Id`), and performs a GET request to fetch
    metadata about all chat rooms that the user can access.

    Equivalent to:

    .. code-block:: bash

        curl -H "X-Auth-Token: <auth_token>" \\
             -H "X-User-Id: <user_id>" \\
             https://chat.example.com/api/v1/rooms.get

    Parameters
    ----------
    rc_url : str
        Rocket.Chat base URL or full endpoint URL. If it does not already end with
        `/rooms.get`, any trailing slashes will be stripped and `/rooms.get` appended.
    auth_token : str
        Authentication token obtained from `login` (for the `X-Auth-Token`
        header).
    user_id : str
        User ID obtained from `login` (for the `X-User-Id` header).
    insecure : bool, optional
        Allow insecure SSL connections. Defaults to `False`.
    no_proxy : bool, optional
        Bypass proxy settings. Defaults to `False`.
    proxy : str, optional
        Proxy to reach the target through, overriding the one the environment names.
        Defaults to `None`, which leaves the choice to the environment.
    timeout : int, optional
        Request timeout in seconds. Defaults to `3`.

    Returns
    -------
    tuple (bool, dict or str)
        - On success: `(True, response_json)` where `response_json` is the parsed JSON result from
          the API.
        - On failure: `(False, 'Error getting rooms.get: <error message>')`.

    Examples
    --------
    >>> success, rooms = get_rooms_get(
    ...     'https://chat.example.com',
    ...     'authTokenHere',
    ...     'userIdHere'
    ... )
    >>> if success:
    ...     print(rooms)
    """
    if not rc_url.endswith('/rooms.get'):
        rc_url = rc_url.rstrip('/') + '/rooms.get'

    headers = {
        'X-Auth-Token': auth_token,
        'X-User-Id': user_id,
    }

    success, result = url.fetch_json(
        rc_url,
        header=headers,
        insecure=insecure,
        no_proxy=no_proxy,
        proxy=proxy,
        timeout=timeout,
    )

    if not success or not result:
        return False, f'Error getting rooms.get: {result}'

    return True, result


def get_stats(
    rc_url, auth_token, user_id, insecure=False, no_proxy=False, proxy=None, timeout=3
):
    """
    Retrieve Rocket.Chat statistics using an API token.

    This function calls the `api/v1/statistics` endpoint to retrieve server stats after
    authentication.

    Equivalent to:

    .. code-block:: bash

        # https://rocket.chat/docs/developer-guides/rest-api/miscellaneous/statistics/
        curl -H "X-Auth-Token: linuxfabrik"
             -H "X-User-Id: linuxfabrik"
             http://localhost:3000/api/v1/statistics

    Parameters
    ----------
    rc_url : str
        Rocket.Chat base URL.
    auth_token : str
        Authentication token.
    user_id : str
        User ID linked to the token.
    insecure : bool, optional
        Allow insecure SSL. Defaults to `False`.
    no_proxy : bool, optional
        Ignore proxy. Defaults to `False`.
    proxy : str, optional
        Proxy to reach the target through, overriding the one the environment names.
        Defaults to `None`, which leaves the choice to the environment.
    timeout : int, optional
        Timeout in seconds. Defaults to `3`.

    Returns
    -------
    tuple (bool, dict)
        Success flag and stats data or error message.

    Examples
    --------
    >>> get_stats('https://chat.example.com', auth_token, user_id)
    (True, {...})
    """
    if not rc_url.endswith('/statistics'):
        rc_url = rc_url.rstrip('/') + '/statistics'

    headers = {
        'X-Auth-Token': auth_token,
        'X-User-Id': user_id,
    }

    success, result = url.fetch_json(
        rc_url,
        header=headers,
        insecure=insecure,
        no_proxy=no_proxy,
        proxy=proxy,
        timeout=timeout,
    )

    if not success or not result:
        return False, f'Error getting statistics: {result}'

    return True, result


def get_token(
    rc_url, user, password, insecure=False, no_proxy=False, proxy=None, timeout=3
):
    """
    Retrieve an API token from Rocket.Chat using user credentials.

    This function authenticates against a Rocket.Chat instance and retrieves an `authToken` and
    `userId` for future authenticated API calls.

    Equivalent to:

    .. code-block:: bash

        curl -X "POST"
             -d "user=admin&password=linuxfabrik"
             http://localhost:3000/api/v1/login

    Parameters
    ----------
    rc_url : str
        Rocket.Chat base URL.
    user : str
        Username.
    password : str
        Password.
    insecure : bool, optional
        Allow insecure SSL. Defaults to `False`.
    no_proxy : bool, optional
        Ignore proxy. Defaults to `False`.
    proxy : str, optional
        Proxy to reach the target through, overriding the one the environment names.
        Defaults to `None`, which leaves the choice to the environment.
    timeout : int, optional
        Timeout in seconds. Defaults to `3`.

    Returns
    -------
    tuple (bool, str)
        Success flag and result string or error.

    Examples
    --------
    >>> get_token('https://chat.example.com', 'admin', 'linuxfabrik')
    (True, 'authToken:userId')
    """
    if not rc_url.endswith('/login'):
        rc_url = rc_url.rstrip('/') + '/login'

    data = {'user': user, 'password': password}

    success, result = url.fetch_json(
        rc_url,
        data=data,
        insecure=insecure,
        no_proxy=no_proxy,
        proxy=proxy,
        timeout=timeout,
    )

    if not success or not result:
        return False, f'Error getting token: {result}'

    data = result.get('data', {})
    auth_token = data.get('authToken')
    user_id = data.get('userId')

    if not auth_token or not user_id:
        return False, 'Authentication failed or user unauthorized.'

    return True, f'{auth_token}:{user_id}'


def send2webhook(
    rc_url, webhook, data, insecure=False, no_proxy=False, proxy=None, timeout=3
):
    """
    Send a JSON payload to a Rocket.Chat incoming webhook.

    This function posts `data` to a Rocket.Chat incoming webhook endpoint, constructing the full URL
    from the provided `rc_url` and `webhook` ID or token.

    Equivalent to:

    .. code-block:: bash

        curl -X POST \\
             -H 'Content-type: application/json' \\
             -d '{"text":"Hello"}' \\
             https://chat.example.com/hooks/<webhook>

    Parameters
    ----------
    rc_url : str
        Rocket.Chat base URL. May include `/api/v1`; if so, it will be stripped.
    webhook : str
        Incoming webhook identifier or token (e.g. `CWaA.../Zbpj...`).
    data : dict
        JSON-serializable payload to send (e.g., `{'text': 'message'}`).
    insecure : bool, optional
        Allow insecure SSL connections. Defaults to `False`.
    no_proxy : bool, optional
        Bypass any proxy settings. Defaults to `False`.
    proxy : str, optional
        Proxy to reach the target through, overriding the one the environment names.
        Defaults to `None`, which leaves the choice to the environment.
    timeout : int, optional
        Request timeout in seconds. Defaults to `3`.

    Returns
    -------
    tuple (bool, bool or str)
        - On success: `(True, True)`.
        - On failure: `(False, 'Error: <error message>')`.

    Examples
    --------
    >>> data = { 'text': '\\n'.join(['bitte beantworten:'] + output) }
    >>> send2webhook('https://chat.example.com/api/v1', 'CWaA.../Zbpj...', data)
    (True, True)
    """
    rc_url = rc_url.replace('/api/v1', '').rstrip('/') + '/hooks/' + webhook
    success, result = url.fetch_json(
        rc_url,
        data=data,
        insecure=insecure,
        no_proxy=no_proxy,
        proxy=proxy,
        timeout=timeout,
    )

    if not success or not result:
        return False, f'Error: {result}'

    return True, True


def send_message(
    webhook_url,
    data,
    header=None,
    insecure=False,
    no_proxy=False,
    proxy=None,
    timeout=3,
):
    """
    Post a JSON message payload to a Rocket.Chat incoming webhook (complete URL).

    Unlike `send2webhook()`, which composes the endpoint from a base URL plus a webhook id and
    posts form-encoded data, this function posts to an already complete incoming-webhook URL and
    serialises `data` as JSON. This lets callers send rich payloads (custom emoji, attachments)
    together with their own headers, and receive the parsed response.

    Parameters
    ----------
    webhook_url : str
        Complete Rocket.Chat incoming-webhook URL, for example
        `https://chat.example.com/hooks/<id>/<token>`.
    data : dict
        JSON-serializable payload to send (e.g. `{'text': 'message'}`).
    header : dict, optional
        Request headers. Defaults to
        `{'Content-Type': 'application/json'}` when omitted.
    insecure : bool, optional
        Allow insecure SSL connections. Defaults to `False`.
    no_proxy : bool, optional
        Bypass any proxy settings. Defaults to `False`.
    proxy : str, optional
        Proxy to reach the target through, overriding the one the environment names.
        Defaults to `None`, which leaves the choice to the environment.
    timeout : int, optional
        Request timeout in seconds. Defaults to `3`.

    Returns
    -------
    tuple (bool, dict or str)
        - On success: `(True, <response dict>)`.
        - On failure: `(False, '<error message>')`.

    Examples
    --------
    >>> send_message('https://chat.example.com/hooks/abc/def', {'text': 'hello'})
    (True, {...})
    """
    if header is None:
        header = {'Content-Type': 'application/json'}
    return url.fetch_json(
        webhook_url,
        header=header,
        data=data,
        encoding='serialized-json',
        insecure=insecure,
        no_proxy=no_proxy,
        proxy=proxy,
        timeout=timeout,
        extended=True,
    )