Initial commit

app/bitcoin_rpc.py

@@ -0,0 +1,126 @@
+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