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.

887 lines
29 KiB
Python

"""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
import httpx
# 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/api/"
def hmac(hmac_key, hmac_secret, server="https://localbitcoins.com"):
conn = Connection()
conn._set_hmac(server, hmac_key, hmac_secret)
return conn
class Connection:
def __init__(self):
self.server = None
# HMAC stuff
self.hmac_key = None
self.hmac_secret = None
def call(self, method, url, params=None, stream=False, files=None):
method = method.upper()
if method not in ["GET", "POST"]:
raise Exception("Invalid method {}!".format(method))
if method == "GET" and files:
raise Exception("You cannot send files with GET method!")
if files and not isinstance(files, dict):
raise Exception('"files" must be a dict of file objects or file contents!')
# If URL is absolute, then convert it
if url.startswith(self.server):
url = url[len(self.server) :] # noqa
if self.hmac_key:
# If nonce fails, retry several times, then give up
for retry in range(10):
nonce = str(int(time.time() * 1000)).encode("ascii")
# Prepare request based on method.
if method == "POST":
api_request = requests.Request("POST", self.server + url, data=params, files=files).prepare()
params_encoded = api_request.body
# GET method
else:
api_request = requests.Request("GET", self.server + url, params=params).prepare()
params_encoded = urlparse(api_request.url).query
# 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()
# Store signature and other stuff to headers
api_request.headers["Apiauth-Key"] = self.hmac_key
api_request.headers["Apiauth-Nonce"] = nonce
api_request.headers["Apiauth-Signature"] = signature
# Send request
session = requests.Session()
response = session.send(api_request, stream=stream)
# If HMAC Nonce is already used, then wait a little and try again
try:
response_json = response.json()
if int(response_json.get("error", {}).get("error_code")) == 42:
time.sleep(0.1)
continue
except: # noqa
# No JSONic response, or interrupt, better just give up
pass
return response
raise Exception("Nonce is too small!")
raise Exception("No HMAC connection initialized!")
def get_access_token(self):
return self.access_token
def _set_hmac(self, server, hmac_key, hmac_secret):
self.server = server
self.hmac_key = hmac_key.encode("ascii")
self.hmac_secret = hmac_secret.encode("ascii")
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, api_key: Optional[str], debug: Optional[bool] = False) -> None:
self.api_key = ""
if api_key:
self.api_key = api_key
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.api_key)
def _api_call(
self,
api_method: str,
http_method: Optional[str] = "GET",
query_values: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
api_call_url = URI_API + api_method
headers = {
"Content-Type": "application/json",
"User-Agent": f"localbitcoins_py/{__version__} " f"https://git.zm.is/Pathogen/pluto",
"Authorization": self.api_key,
}
logger.debug("API Call URL: %s", api_call_url)
logger.debug("Headers : %s", 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,
}
try:
response = None
if http_method == "POST":
if query_values:
response = httpx.post(
url=api_call_url,
headers=headers,
content=json.dumps(query_values),
)
else:
response = httpx.post(
url=api_call_url,
headers=headers,
)
else:
response = httpx.get(url=api_call_url, headers=headers, params=query_values)
logger.debug(response)
result["response"] = json.loads(response.text)
result["status"] = response.status_code
if response.status_code == 200:
result["success"] = True
result["message"] = "OK"
else:
result["message"] = "API ERROR"
return result
except httpx.ConnectError 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 httpx.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"account_info/{username}")
def dashboard(self) -> Dict[str, Any]:
"""See LocalBitcoins API.
https://localbitcoins.com/api-docs/#dashboard
"""
return self._api_call(api_method="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="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="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="dashboard/released")
def logout(self) -> Dict[str, Any]:
"""See LocalBitcoins API.
https://localbitcoins.com/api-docs/#logout
"""
return self._api_call(api_method="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="myself")
def notifications(self) -> Dict[str, Any]:
"""See LocalBitcoins API.
https://localbitcoins.com/api-docs/#notifications
"""
return self._api_call(api_method="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="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( # TODO: fill fields
self,
country_code: str,
currency: str,
trade_type: str,
asset: str,
price_equation: str,
track_max_amount: bool,
require_trusted_by_advertiser: bool,
verified_email_required: Optional[bool] = None,
online_provider: Optional[str] = None,
msg: Optional[str] = None,
min_amount: Optional[float] = None,
max_amount: Optional[float] = None,
limit_to_fiat_amounts: Optional[str] = None,
payment_method_details: Optional[str] = None,
first_time_limit_asset: Optional[float] = None,
require_feedback_score: Optional[int] = None,
account_info: Optional[str] = None,
payment_window_minutes: Optional[int] = None,
floating: Optional[bool] = None,
lat: Optional[float] = None,
lon: Optional[float] = 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,
"asset": asset,
"price_equation": price_equation,
"track_max_amount": 1 if track_max_amount else 0,
"require_trusted_by_advertiser": 1 if require_trusted_by_advertiser else 0,
}
if verified_email_required:
params["verified_email_required"] = 1 if verified_email_required else 0
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 limit_to_fiat_amounts:
params["limit_to_fiat_amounts"] = limit_to_fiat_amounts
if payment_method_details:
params["payment_method_detail"] = payment_method_details
if first_time_limit_asset:
params["first_time_limit_asset"] = first_time_limit_asset
if require_feedback_score:
params["require_feedback_score"] = require_feedback_score
if account_info:
params["account_info"] = account_info
if payment_window_minutes:
params["payment_window_minutes"] = payment_window_minutes
if floating:
params["floating"] = 1 if floating else 0
if lat:
params["lat"] = lat
if lon:
params["lon"] = lon
return self._api_call(
api_method="ad-create",
http_method="POST",
query_values=params,
)
def ad(
self,
ad_id: str,
country_code: Optional[str] = None,
currency: Optional[str] = None,
trade_type: Optional[str] = None,
asset: Optional[str] = None,
price_equation: Optional[str] = None,
track_max_amount: Optional[bool] = None,
require_trusted_by_advertiser: Optional[bool] = None,
verified_email_required: Optional[bool] = None,
online_provider: Optional[str] = None,
msg: Optional[str] = None,
min_amount: Optional[float] = None,
max_amount: Optional[float] = None,
limit_to_fiat_amounts: Optional[str] = None,
payment_method_details: Optional[str] = None,
first_time_limit_asset: Optional[float] = None,
require_feedback_score: Optional[int] = None,
account_info: Optional[str] = None,
payment_window_minutes: Optional[int] = None,
floating: Optional[bool] = None,
lat: Optional[float] = None,
lon: Optional[float] = None,
visible: 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, Union[str, float, bool]] = {}
if country_code:
params["countrycode"] = country_code
if currency:
params["currency"] = currency
if trade_type:
params["trade_type"] = trade_type
if asset:
params["asset"] = asset
if price_equation:
params["price_equation"] = price_equation
if track_max_amount:
params["track_max_amount"] = 1 if track_max_amount else 0
if require_trusted_by_advertiser:
params["require_trusted_by_advertiser"] = 1 if require_trusted_by_advertiser else 0
if verified_email_required:
params["verified_email_required"] = 1 if verified_email_required else 0
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 limit_to_fiat_amounts:
params["limit_to_fiat_amounts"] = limit_to_fiat_amounts
if payment_method_details:
params["payment_method_detail"] = payment_method_details
if first_time_limit_asset:
params["first_time_limit_asset"] = first_time_limit_asset
if require_feedback_score:
params["require_feedback_score"] = require_feedback_score
if account_info:
params["account_info"] = account_info
if payment_window_minutes:
params["payment_window_minutes"] = payment_window_minutes
if floating:
params["floating"] = 1 if floating else 0
if lat:
params["lat"] = lat
if lon:
params["lon"] = lon
if visible:
params["visible"] = True if visible else False
return self._api_call(
api_method=f"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"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"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="ads")
return self._api_call(api_method="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="countrycodes")
def currencies(self) -> Dict[str, Any]:
"""See LocalBitcoins API.
https://localbitcoins.com/api-docs/#currencies
"""
return self._api_call(api_method="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="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="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="wallet")
def wallet_balance(self) -> Dict[str, Any]:
"""See LocalBitcoins API.
https://localbitcoins.com/api-docs/#wallet-balance
"""
return self._api_call(api_method="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="wallet-addr")
def fees(self) -> Dict[str, Any]:
"""See LocalBitcoins API.
https://localbitcoins.com/api-docs/#fees
"""
return self._api_call(api_method="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="wallet-send-pin", http_method="POST", query_values=params)