| from json import loads |
| from time import time_ns |
|
|
| from aiohttp import ClientTimeout, ClientSession |
| from substrateinterface import Keypair |
|
|
| from scorevision.utils.settings import get_settings |
|
|
|
|
| async def _sign_batch(payloads: list[str]) -> tuple[str, list[str]]: |
| """ """ |
| settings = get_settings() |
| if settings.SIGNER_URL: |
| try: |
| timeout = ClientTimeout(connect=2, total=30) |
| async with ClientSession(timeout=timeout) as sess: |
| r = await sess.post( |
| f"{settings.SIGNER_URL}/sign", json={"payloads": payloads} |
| ) |
| txt = await r.text() |
| if r.status == 200: |
| data = loads(txt) |
| sigs = data.get("signatures") or [] |
| hk = data.get("hotkey") or "" |
| if len(sigs) == len(payloads) and hk: |
| return hk, sigs |
| except Exception as e: |
| raise Exception(f"signer unavailable, fallback to local: {e}") |
| raise ValueError("No Signer URL set") |
|
|
|
|
| def sign_message(keypair: Keypair, message: str | None) -> str | None: |
| if message is None: |
| return None |
| return f"0x{keypair.sign(message).hex()}" |
|
|
|
|
| def build_validator_query_params(keypair: Keypair) -> dict[str, str]: |
| """ |
| Create validator authentication query parameters required by the ScoreVision API. |
| """ |
| settings = get_settings() |
|
|
| nonce = str(time_ns()) |
| signature = sign_message(keypair, nonce) |
|
|
| return { |
| "validator_hotkey": keypair.ss58_address, |
| "signature": signature, |
| "nonce": nonce, |
| "netuid": str(settings.SCOREVISION_NETUID), |
| } |
|
|
