XF
/
pluto
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
master
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '70d0aad046'
${ noResults }
pluto/handler/lib/localbitcoins_py.py

885 lines
28 KiB
Python
Raw Blame History

"""See https://agoradesk.com/api-docs/v1."""
# pylint: disable=too-many-lines
# Large API. Lots of lines can't be avoided.
import hashlib
import hmac as hmac_lib
import json
import logging
import sys
import time
from typing import Any, Dict, List, Optional, Union
from urllib.parse import urlparse
import arrow
import requests
import treq
# Project imports
import util
from twisted.internet.defer import inlineCallbacks
__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)
@inlineCallbacks
def callback_api_call(self, response, result):
logger.debug(response)
text = yield response.content()
try:
result["response"] = json.loads(text)
except json.decoder.JSONDecodeError:
result["success"] = "ERROR"
result["message"] = "Error parsing JSON."
return result
result["status"] = response.code
if response.code == 200:
result["success"] = True
result["message"] = "OK"
else:
result["message"] = "API ERROR"
return result
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)
headers = {
"Apiauth-Key": self.hmac_key,
"Apiauth-Nonce": nonce,
"Apiauth-Signature": signature,
}
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,
}
response = None
if http_method == "POST":
if query_values:
# response = httpx.post(
# url=api_call_url,
# headers=headers,
# content=json.dumps(query_values),
# )
response = treq.post(
api_call_url,
headers=headers,
data=query_values,
)
else:
# response = httpx.post(
# url=api_call_url,
# headers=headers,
# )
response = treq.post(
api_call_url,
headers=headers,
)
else:
# response = httpx.get(url=api_call_url, headers=headers, params=query_values)
response = treq.get(api_call_url, headers=headers, params=query_values)
if response:
response.addCallback(self.callback_api_call, result)
return response
# 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"api/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,
page: Optional[int] = 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 page:
params["page"] = page
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}/.json",
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,
)
Reference in New Issue View Git Blame Copy Permalink