communex.util

  1import ipaddress
  2import json
  3import os.path
  4import re
  5from typing import Any, Callable, Generic, Optional, Protocol, TypeVar
  6
  7import requests
  8
  9
 10def check_str(x: Any) -> str:
 11    assert isinstance(x, str)
 12    return x
 13
 14
 15def ensure_dir_exists(path: str) -> None:
 16    if not os.path.exists(path):
 17        os.makedirs(path, exist_ok=True)
 18
 19
 20def ensure_parent_dir_exists(path: str) -> None:
 21    ensure_dir_exists(os.path.dirname(path))
 22
 23
 24def bytes_to_hex(value: str | bytes) -> str:
 25    """
 26    Converts a string or bytes object to its hexadecimal representation.
 27
 28    If the input is already a string, it assumes that the string is already in
 29    hexadecimal format and returns it as is. If the input is bytes, it converts
 30    the bytes to their hexadecimal string representation.
 31
 32    Args:
 33        x: The input string or bytes object to be converted to hexadecimal.
 34
 35    Returns:
 36        The hexadecimal representation of the input.
 37
 38    Examples:
 39        _to_hex(b'hello') returns '68656c6c6f'
 40        _to_hex('68656c6c6f') returns '68656c6c6f'
 41    """
 42    if isinstance(value, bytes):
 43        return value.hex()
 44    assert isinstance(value, str)
 45    # TODO: Check that `value` is a valid hexadecimal string
 46    return value
 47
 48
 49def is_ip_valid(ip: str) -> bool:
 50    """
 51    Checks if an ip address is valid
 52    """
 53    try:
 54        ipaddress.ip_address(ip)
 55        return True
 56    except ValueError:
 57        return False
 58
 59
 60T = TypeVar("T")
 61
 62
 63class SetterGetterFn(Generic[T], Protocol):
 64    def __call__(self, x: T = ..., /) -> T:
 65        ...
 66
 67
 68def create_state_fn(default: Callable[..., T]) -> SetterGetterFn[T]:
 69    """
 70    Creates a state function that can be used to get or set a value.
 71    """
 72    value = default()
 73
 74    def state_function(input: Optional[T] = None):
 75        nonlocal value
 76        if input is not None:
 77            value = input
 78        return value
 79
 80    return state_function
 81
 82
 83def get_json_from_cid(cid: str) -> dict[Any, Any] | None:
 84    gateway = "https://ipfs.io/ipfs/"
 85    try:
 86        result = requests.get(gateway + cid)
 87        if result.ok:
 88            return result.json()
 89        return None
 90    except Exception:
 91        return None
 92
 93
 94def convert_cid_on_proposal(proposals: dict[int, dict[str, Any]]):
 95    unwrapped: dict[int, dict[str, Any]] = {}
 96    for prop_id, proposal in proposals.items():
 97        data = proposal.get("data")
 98        if data and "Custom" in data:
 99            metadata = proposal["metadata"]
100            cid = metadata.split("ipfs://")[-1]
101            queried_cid = get_json_from_cid(cid)
102            if queried_cid:
103                body = queried_cid.get("body")
104                if body:
105                    try:
106                        queried_cid["body"] = json.loads(body)
107                    except Exception:
108                        pass
109            proposal["Custom"] = queried_cid
110        unwrapped[prop_id] = proposal
111    return unwrapped
112
113
114HEX_PATTERN = re.compile(r"^[0-9a-fA-F]+$")
115
116
117# TODO: merge `is_hex_string` into `parse_hex`
118def is_hex_string(string: str):
119    return bool(HEX_PATTERN.match(string))
120
121
122def parse_hex(hex_str: str) -> bytes:
123    if hex_str[0:2] == "0x":
124        return bytes.fromhex(hex_str[2:])
125    else:
126        return bytes.fromhex(hex_str)