pluto/handler/lib/localbitcoins_py.py

"""See https://agoradesk.com/api-docs/v1."""
# pylint: disable=too-many-lines
# Large API. Lots of lines can't be avoided.
import json
import logging
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Union
import arrow

# Project imports
import util

import hashlib
import hmac as hmac_lib
import requests
import sys
import time


from urllib.parse import urlparse

__author__ = "marvin8"
__copyright__ = "(C) 2021 https://codeberg.org/MarvinsCryptoTools/agoradesk_py"
__version__ = "0.1.0"

# set logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logging.getLogger("requests.packages.urllib3").setLevel(logging.INFO)
logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO)
logger = util.get_logger(__name__)

URI_API = "https://localbitcoins.com/"
SERVER = "https://localbitcoins.com"


class LocalBitcoins:
    """LocalBitcoins API object.

    Documentation: https://localbitcoins.com/api-docs/
    """

    # pylint: disable=too-many-public-methods
    # API provides this many methods, I can't change that

    def __init__(
        self,
        hmac_key: Optional[str],
        hmac_secret: Optional[str],
        debug: Optional[bool] = False,
    ) -> None:
        self.hmac_key = ""
        self.hmac_secret = ""
        if hmac_key:
            self.hmac_key = hmac_key.encode("ascii")
        if hmac_secret:
            self.hmac_secret = hmac_secret.encode("ascii")
        self.debug = debug

        if self.debug:
            logging.getLogger("requests.packages.urllib3").setLevel(logging.DEBUG)
            logger.setLevel(logging.DEBUG)
        else:
            logger.setLevel(logging.INFO)

        logger.debug(
            "creating instance of LocalBitcoins API with api_key %s",
            self.hmac_key,
        )

    def sign_payload(self, nonce, url, params_encoded):
        """
        Sign the payload with our HMAC keys.
        """
        # Calculate signature
        message = nonce + self.hmac_key + url.encode("ascii")
        if params_encoded:
            if sys.version_info >= (3, 0) and isinstance(params_encoded, str):
                message += params_encoded.encode("ascii")
            else:
                message += params_encoded
        signature = hmac_lib.new(self.hmac_secret, msg=message, digestmod=hashlib.sha256).hexdigest().upper()
        return signature

    def encode_params(self, http_method, api_call_url, query_values):
        if http_method == "POST":
            api_request = requests.Request("POST", api_call_url, data=query_values).prepare()
            params_encoded = api_request.body

        # GET method
        else:
            api_request = requests.Request("GET", api_call_url, params=query_values).prepare()
            params_encoded = urlparse(api_request.url).query
        return (api_request, params_encoded)

    def _api_call(
        self,
        api_method: str,
        http_method: Optional[str] = "GET",
        query_values: Optional[Dict[str, Any]] = None,
        files: Optional[Any] = None,
    ) -> Dict[str, Any]:
        api_call_url = URI_API + api_method

        url = api_call_url
        if url.startswith(SERVER):
            url = url[len(SERVER) :]  # noqa

        # HMAC crypto stuff
        api_request, params_encoded = self.encode_params(http_method, api_call_url, query_values)
        nonce = str(int(time.time() * 1000)).encode("ascii")
        signature = self.sign_payload(nonce, url, params_encoded)

        api_request.headers["Apiauth-Key"] = self.hmac_key
        api_request.headers["Apiauth-Nonce"] = nonce
        api_request.headers["Apiauth-Signature"] = signature

        logger.debug("API Call URL: %s", api_call_url)
        logger.debug("Headers     : %s", api_request.headers)
        logger.debug("HTTP Method : %s", http_method)
        logger.debug("Query Values: %s", query_values)
        logger.debug("Query Values as Json:\n%s", json.dumps(query_values))

        result: Dict[str, Any] = {
            "success": False,
            "message": "Invalid Method",
            "response": None,
            "status": None,
        }
        session = requests.Session()
        try:
            response = session.send(api_request)
            response_json = response.json()
            result["response"] = response_json
            result["status"] = response.status_code
            if response.status_code == 200:
                result["success"] = True
                result["message"] = "OK"
            else:
                result["message"] = "API ERROR"
            # print("RESP", result)
            return result

        except requests.ConnectionError as error:
            result["message"] = str(error)
            result["status"] = 600
            result["response"] = {"error": {"message": error}}
            return result
        except json.decoder.JSONDecodeError:
            result["message"] = "Not JSON"
            if response:
                result["status"] = response.status_code
                result["response"] = {"error": {"message": response.text}}
            return result
        except requests.ReadTimeout:
            result["message"] = "Read timed out"
            if response:
                result["status"] = response.status_code
                result["response"] = {"error": {"message": response.text}}
            return result

    # Account related API Methods
    # ===========================

    def account_info(self, username: str) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#account_info
        """
        return self._api_call(api_method=f"api/account_info/{username}/")

    def dashboard(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#dashboard
        """
        return self._api_call(api_method="api/dashboard/")

    # def dashboard_buyer(self) -> Dict[str, Any]:
    #    """See LocalBitcoins API.
    #
    #    https://agoradesk.com/api-docs/v1#operation/getUserDashboardBuyer
    #    """
    #    return self._api_call(api_method="dashboard/buyer")

    # def dashboard_seller(self) -> Dict[str, Any]:
    #    """See LocalBitcoins API.
    #
    #    https://agoradesk.com/api-docs/v1#operation/getUserDashboardSeller
    #    """
    #    return self._api_call(api_method="dashboard/seller")

    def dashboard_canceled(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#dashboard-canceled
        """
        return self._api_call(api_method="api/dashboard/canceled/")

    def dashboard_closed(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#dashboard-closed
        """
        return self._api_call(api_method="api/dashboard/closed/")

    def dashboard_released(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#dashboard-released
        """

        return self._api_call(api_method="api/dashboard/released/")

    def logout(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#logout
        """

        return self._api_call(api_method="api/logout/", http_method="POST")

    def myself(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#myself
        """

        return self._api_call(api_method="api/myself/")

    def notifications(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#notifications
        """

        return self._api_call(api_method="api/notifications/")

    def notifications_mark_as_read(self, notification_id: str) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#notifications-read
        """

        return self._api_call(
            api_method=f"notifications/mark_as_read/{notification_id}/",
            http_method="POST",
        )

    def recent_messages(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#recent-messages
        """

        return self._api_call(api_method="api/recent_messages/")

    # Trade related API Methods
    # ===========================

    #     post/feedback/{username} • Give feedback to a user
    def feedback(self, username: str, feedback: str, msg: Optional[str]) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#feedback
        """

        params = {"feedback": feedback}
        if msg:
            params["msg"] = msg
        return self._api_call(
            api_method=f"feedback/{username}/",
            http_method="POST",
            query_values=params,
        )

    #     Todo:
    #     post/trade/contact_release/{trade_id} • Release trade escrow
    #     post/contact_fund/{trade_id} • Fund a trade
    #     post/contact_dispute/{trade_id} • Start a trade dispute

    #     post/contact_mark_as_paid/{trade_id} • Mark a trade as paid
    def contact_mark_as_paid(self, trade_id: str) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#contact-paid
        """
        return self._api_call(api_method=f"contact_mark_as_paid/{trade_id}/", http_method="POST")

    #     post/contact_cancel/{trade_id} • Cancel the trade
    def contact_cancel(
        self,
        trade_id: str,
    ) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#contact-cancel
        """
        return self._api_call(
            api_method=f"contact_cancel/{trade_id}",
            http_method="POST",
        )

    #     Todo:
    #     post/contact_escrow/{trade_id} • Enable escrow

    #     get/contact_messages/{trade_id} • Get trade messages
    def contact_messages(self, trade_id: str, after: Optional[arrow.Arrow] = None) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#contact-message
        """
        if after:
            reply = self._api_call(
                api_method=f"contact_messages/{trade_id}/",
                query_values={"after": after.to("UTC").isoformat()},
            )
        else:
            reply = self._api_call(api_method=f"contact_messages/{trade_id}/")

        return reply

    #     post/contact_create/{ad_id} • Start a trade
    def contact_create(
        self,
        ad_id: str,
        amount: float,
        msg: Optional[str] = None,
    ) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#contact-create
        """
        payload: Dict[str, Any] = {"amount": amount}
        if msg:
            payload["msg"] = msg
        return self._api_call(
            api_method=f"contact_create/{ad_id}/",
            http_method="POST",
            query_values=payload,
        )

    #     get/contact_info/{trade_id} • Get a trade by trade ID
    def contact_info(self, trade_ids: Union[str, List[str]]) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#contact-info-id and
        https://localbitcoins.com/api-docs/#contact-info
        """
        api_method = "contact_info/"
        if isinstance(trade_ids, list):
            params = "?contacts="
            for trade_id in trade_ids:
                params += f"{trade_id},"
            params = params[0:-1]
        else:
            params = f"/{trade_ids}"
        api_method += params
        return self._api_call(api_method=api_method)

    #     Todo: Add image upload functionality
    #     post/contact_message_post/{trade_id} • Send a chat message/attachment
    def contact_message_post(self, trade_id: str, msg: Optional[str] = None) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#contact-post
        """
        payload = {"msg": msg}
        return self._api_call(
            api_method=f"contact_message_post/{trade_id}/",
            http_method="POST",
            query_values=payload,
        )

    #     Todo:
    #     get/contact_message_attachment/{trade_id}/{attachment_id}

    # Advertisement related API Methods
    # ================================

    def ad_create(
        self,
        country_code: str,
        currency: str,
        trade_type: str,
        price_equation: str,
        track_max_amount: bool,
        require_trusted_by_advertiser: bool,
        bank_name: str,
        sms_verification_required: Optional[bool] = None,
        require_identification: Optional[bool] = None,
        online_provider: Optional[str] = None,
        msg: Optional[str] = None,
        min_amount: Optional[float] = None,
        max_amount: Optional[float] = None,
        account_info: Optional[str] = None,
        first_time_limit_btc: Optional[float] = None,
        require_feedback_score: Optional[int] = None,
        city: Optional[str] = None,
        location_string: Optional[str] = None,
        opening_hours: Optional[dict] = None,
        visible: Optional[bool] = True,
        require_trade_volume: Optional[float] = None,
        volume_coefficient_btc: Optional[float] = None,
        reference_type: Optional[str] = None,
        display_reference: Optional[bool] = None,
    ) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#ad-create
        """

        # pylint: disable=too-many-arguments
        # pylint: disable=too-many-locals
        # pylint: disable=too-many-branches
        # API takes this many arguments, I can't change that
        # Too many locals and too many branches goes hand in hand
        # with too many arguments
        params: Dict[str, Any] = {
            "countrycode": country_code,
            "currency": currency,
            "trade_type": trade_type,
            "price_equation": price_equation,
            "track_max_amount": track_max_amount,
            "require_trusted_by_advertiser": require_trusted_by_advertiser,
        }
        if sms_verification_required:
            params["sms_verification_required"] = True
        if require_identification:
            params["require_identification"] = True
        if online_provider:
            params["online_provider"] = online_provider
        if msg:
            params["msg"] = msg
        if min_amount:
            params["min_amount"] = min_amount
        if max_amount:
            params["max_amount"] = max_amount
        if first_time_limit_btc:
            params["first_time_limit_btc"] = first_time_limit_btc
        if require_feedback_score:
            params["require_feedback_score"] = require_feedback_score
        if account_info:
            params["account_info"] = account_info
        if opening_hours:
            params["opening_hours"] = opening_hours
        if city:
            params["city"] = city
        if location_string:
            params["location_string"] = location_string
        if bank_name:
            params["bank_name"] = bank_name
        if visible:
            params["visible"] = visible
        if require_trade_volume:
            params["require_trade_volume"] = require_trade_volume
        if volume_coefficient_btc:
            params["volume_coefficient_btc"] = volume_coefficient_btc
        if reference_type:
            params["reference_type"] = reference_type
        if display_reference is not None:
            params["display_reference"] = display_reference

        return self._api_call(
            api_method="api/ad-create/",
            http_method="POST",
            query_values=params,
        )

    def ad(
        self,
        ad_id: str,
        country_code: str,
        currency: str,
        trade_type: str,
        price_equation: str,
        track_max_amount: bool,
        require_trusted_by_advertiser: bool,
        bank_name: str,
        sms_verification_required: Optional[bool] = None,
        require_identification: Optional[bool] = None,
        online_provider: Optional[str] = None,
        msg: Optional[str] = None,
        min_amount: Optional[float] = None,
        max_amount: Optional[float] = None,
        account_info: Optional[str] = None,
        first_time_limit_btc: Optional[float] = None,
        require_feedback_score: Optional[int] = None,
        city: Optional[str] = None,
        location_string: Optional[str] = None,
        opening_hours: Optional[dict] = None,
        visible: Optional[bool] = True,
        require_trade_volume: Optional[float] = None,
        volume_coefficient_btc: Optional[float] = None,
        reference_type: Optional[str] = None,
        display_reference: Optional[bool] = None,
    ) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#ad-id
        """

        # pylint: disable=invalid-name
        # Don't want to change the name of the method from what the API call is
        # pylint: disable=too-many-arguments
        # pylint: disable=too-many-locals
        # pylint: disable=too-many-branches
        # API takes this many arguments, I can't change that
        # Too many locals and too many branches goes hand in hand
        # with too many arguments
        params: Dict[str, Any] = {
            "countrycode": country_code,
            "currency": currency,
            "trade_type": trade_type,
            "price_equation": price_equation,
            "track_max_amount": track_max_amount,
            "require_trusted_by_advertiser": require_trusted_by_advertiser,
        }
        if sms_verification_required:
            params["sms_verification_required"] = True
        if require_identification:
            params["require_identification"] = True
        if online_provider:
            params["online_provider"] = online_provider
        if msg:
            params["msg"] = msg
        if min_amount:
            params["min_amount"] = min_amount
        if max_amount:
            params["max_amount"] = max_amount
        if first_time_limit_btc:
            params["first_time_limit_btc"] = first_time_limit_btc
        if require_feedback_score:
            params["require_feedback_score"] = require_feedback_score
        if account_info:
            params["account_info"] = account_info
        if opening_hours:
            params["opening_hours"] = opening_hours
        if city:
            params["city"] = city
        if location_string:
            params["location_string"] = location_string
        if bank_name:
            params["bank_name"] = bank_name
        if visible:
            params["visible"] = visible
        if require_trade_volume:
            params["require_trade_volume"] = require_trade_volume
        if volume_coefficient_btc:
            params["volume_coefficient_btc"] = volume_coefficient_btc
        if reference_type:
            params["reference_type"] = reference_type
        if display_reference is not None:
            params["display_reference"] = display_reference

        return self._api_call(
            api_method=f"api/ad/{ad_id}/",
            http_method="POST",
            query_values=params,
        )

    def ad_equation(self, ad_id: str, price_equation: str) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#ad-equation-id
        """
        return self._api_call(
            api_method=f"api/ad-equation/{ad_id}/",
            http_method="POST",
            query_values={"price_equation": price_equation},
        )

    def ad_delete(self, ad_id: str) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#ad-delete
        """
        return self._api_call(api_method=f"api/ad-delete/{ad_id}/", http_method="POST")

    def ads(
        self,
        country_code: Optional[str] = None,
        currency: Optional[str] = None,
        trade_type: Optional[str] = None,
        visible: Optional[bool] = None,
        asset: Optional[str] = None,
        payment_method_code: Optional[str] = None,
    ) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#ads
        """

        # pylint: disable=too-many-arguments
        # API takes this many arguments, I can't change that

        params = {}
        if country_code:
            params["countrycode"] = country_code
        if currency:
            params["currency"] = currency
        if trade_type:
            params["trade_type"] = trade_type
        if visible is not None and visible:
            params["visible"] = "1"
        elif visible is not None and not visible:
            params["visible"] = "0"
        if asset:
            params["asset"] = asset
        if payment_method_code:
            params["payment_method_code"] = payment_method_code

        if len(params) == 0:
            return self._api_call(api_method="api/ads/")

        return self._api_call(api_method="api/ads/", query_values=params)

    def ad_get(self, ad_ids: List[str]) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#ad-get and
        hhttps://localbitcoins.com/api-docs/#ad-get-id
        """
        api_method = "ad-get/"
        params = None
        ids = str(ad_ids)[1:-1].replace(" ", "").replace("'", "")

        if len(ad_ids) == 1:
            api_method += f"/{ids}"
        else:
            params = {"ads": ids}
        return self._api_call(api_method=api_method, query_values=params)

    def payment_methods(self, country_code: Optional[str] = None) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#payment-methods and
        https://localbitcoins.com/api-docs/#payment_methods-cc
        """
        api_method = "payment_methods/"
        if country_code:
            api_method += f"/{country_code}"
        return self._api_call(api_method=api_method)

    def country_codes(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#countrycodes
        """
        return self._api_call(api_method="api/countrycodes/")

    def currencies(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#currencies
        """
        return self._api_call(api_method="api/currencies/")

    def equation(self, price_equation: str, currency: str) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#equation
        """
        return self._api_call(
            api_method="api/equation/",
            http_method="POST",
            query_values={
                "price_equation": price_equation,
                "currency": currency,
            },
        )

    # Public ad search related API Methods
    # ====================================

    def _generic_online(
        self,
        direction: str,
        main_currency: str,
        exchange_currency: str,
        country_code: Optional[str] = None,
        payment_method: Optional[str] = None,
        amount: Optional[float] = None,
        page: Optional[int] = None,
    ) -> Dict[str, Any]:

        # pylint: disable=too-many-arguments

        add_to_api_method = ""
        if country_code:
            add_to_api_method = f"/{country_code}"
        if payment_method:
            add_to_api_method += f"/{payment_method}"

        params = self._generic_search_parameters(amount, page)

        return self._api_call(
            api_method=f"{direction}-{main_currency}-online/" f"{exchange_currency}{add_to_api_method}",
            query_values=params,
        )

    @staticmethod
    def _generic_search_parameters(amount, page):
        params = None
        if amount and not page:
            params = {"amount": f"{amount}"}
        elif amount and page:
            params = {"amount": f"{amount}", "page": f"{page}"}
        elif not amount and page:
            params = {"page": f"{page}"}
        return params

    def buy_bitcoins_online(  # TODO: check fields
        self,
        currency_code: str,
        country_code: Optional[str] = None,
        payment_method: Optional[str] = None,
        amount: Optional[float] = None,
        page: Optional[int] = None,
    ) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#online-buy1 and
        https://localbitcoins.com/api-docs/#online-buy2 and
        https://localbitcoins.com/api-docs/#online-buy3 and
        https://localbitcoins.com/api-docs/#online-buy4 and
        https://localbitcoins.com/api-docs/#online-buy5 and
        https://localbitcoins.com/api-docs/#online-buy6
        """

        # pylint: disable=too-many-arguments

        return self._generic_online(
            direction="buy",
            main_currency="bitcoins",
            exchange_currency=currency_code,
            country_code=country_code,
            payment_method=payment_method,
            amount=amount,
            page=page,
        )

    def sell_bitcoins_online(
        self,
        currency_code: str,
        country_code: Optional[str] = None,
        payment_method: Optional[str] = None,
        amount: Optional[float] = None,
        page: Optional[int] = None,
    ) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#online-sell1 and
        https://localbitcoins.com/api-docs/#online-sell2 and
        https://localbitcoins.com/api-docs/#online-sell3 and
        https://localbitcoins.com/api-docs/#online-sell4 and
        https://localbitcoins.com/api-docs/#online-sell5 and
        https://localbitcoins.com/api-docs/#online-sell6
        """

        # pylint: disable=too-many-arguments

        return self._generic_online(
            direction="sell",
            main_currency="bitcoins",
            exchange_currency=currency_code,
            country_code=country_code,
            payment_method=payment_method,
            amount=amount,
            page=page,
        )

    # Statistics related API Methods
    # ==============================

    def bitcoinaverage(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#ticker-all
        """
        return self._api_call(api_method="api/bitcoinaverage/ticket-all-currencies/")

    # Wallet related API Methods
    # ===========================

    def wallet(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#wallet
        """
        return self._api_call(api_method="api/wallet/")

    def wallet_balance(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#wallet-balance
        """
        return self._api_call(api_method="api/wallet-balance/")

    def wallet_addr(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#wallet-addr
        """
        return self._api_call(api_method="api/wallet-addr/")

    def fees(self) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#fees
        """
        return self._api_call(api_method="api/fees/")

    def wallet_send_pin(
        self,
        address: str,
        amount: float,
        password: str,
        fee_level: str,
        pincode: Optional[int] = None,
    ) -> Dict[str, Any]:
        """See LocalBitcoins API.

        https://localbitcoins.com/api-docs/#wallet-send
        """
        # pylint: disable=too-many-arguments

        params = {
            "address": address,
            "amount": amount,
            "password": password,
            "fee_level": fee_level,
        }
        if pincode:
            params["pincode"] = pincode

        return self._api_call(
            api_method="api/wallet-send-pin/",
            http_method="POST",
            query_values=params,
        )