127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
import json
|
|
import re
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
import requests
|
|
|
|
from app.config import get_config
|
|
|
|
|
|
class BitcoinRPCError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class BitcoinRPCClient:
|
|
def __init__(
|
|
self,
|
|
rpc_url: str,
|
|
username: str,
|
|
password: str,
|
|
wallet: str = "",
|
|
timeout: int | None = None,
|
|
) -> None:
|
|
cfg = get_config()
|
|
self.rpc_url = rpc_url.rstrip("/")
|
|
self.username = username
|
|
self.password = password
|
|
self.wallet = wallet.strip()
|
|
self.timeout = timeout or cfg.request_timeout_seconds
|
|
|
|
def _endpoint(self, wallet_override: str | None = None) -> str:
|
|
wallet = self.wallet if wallet_override is None else wallet_override.strip()
|
|
if wallet:
|
|
return f"{self.rpc_url}/wallet/{quote(wallet, safe='')}"
|
|
return self.rpc_url
|
|
|
|
def call(
|
|
self,
|
|
method: str,
|
|
params: list[Any] | None = None,
|
|
wallet_override: str | None = None,
|
|
) -> Any:
|
|
payload = {
|
|
"jsonrpc": "1.0",
|
|
"id": "tellscoin-dashboard",
|
|
"method": method,
|
|
"params": params or [],
|
|
}
|
|
try:
|
|
response = requests.post(
|
|
self._endpoint(wallet_override),
|
|
json=payload,
|
|
auth=(self.username, self.password),
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
body = response.json()
|
|
except requests.RequestException as exc:
|
|
raise BitcoinRPCError(f"RPC request failed: {exc}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise BitcoinRPCError("RPC response was not valid JSON") from exc
|
|
|
|
if body.get("error"):
|
|
raise BitcoinRPCError(str(body["error"]))
|
|
return body.get("result")
|
|
|
|
def batch_call(self, calls: list[tuple[str, list[Any]]]) -> dict[str, Any]:
|
|
payload = []
|
|
for index, (method, params) in enumerate(calls):
|
|
payload.append(
|
|
{
|
|
"jsonrpc": "1.0",
|
|
"id": str(index),
|
|
"method": method,
|
|
"params": params,
|
|
}
|
|
)
|
|
|
|
try:
|
|
response = requests.post(
|
|
self._endpoint(),
|
|
json=payload,
|
|
auth=(self.username, self.password),
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
body = response.json()
|
|
except requests.RequestException as exc:
|
|
raise BitcoinRPCError(f"Batch RPC request failed: {exc}") from exc
|
|
|
|
by_id = {entry.get("id"): entry for entry in body}
|
|
results = {}
|
|
for index, (method, _) in enumerate(calls):
|
|
key = str(index)
|
|
entry = by_id.get(key)
|
|
if not entry:
|
|
raise BitcoinRPCError(f"Missing batch result for method {method}")
|
|
if entry.get("error"):
|
|
raise BitcoinRPCError(f"{method}: {entry['error']}")
|
|
results[method] = entry.get("result")
|
|
return results
|
|
|
|
|
|
COMMAND_LINE_RE = re.compile(r"^([a-z][a-z0-9_]*)\b")
|
|
|
|
|
|
|
|
def parse_help_output(help_text: str) -> list[dict[str, str]]:
|
|
commands: list[dict[str, str]] = []
|
|
seen = set()
|
|
|
|
for raw in help_text.splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("=="):
|
|
continue
|
|
match = COMMAND_LINE_RE.match(line)
|
|
if not match:
|
|
continue
|
|
method = match.group(1)
|
|
if method in seen:
|
|
continue
|
|
seen.add(method)
|
|
commands.append({"method": method, "synopsis": line})
|
|
|
|
commands.sort(key=lambda item: item["method"])
|
|
return commands
|