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
 67def create_state_fn(default: Callable[..., T]) -> SetterGetterFn[T]:
 68    """
 69    Creates a state function that can be used to get or set a value.
 70    """
 71    value = default()
 72
 73    def state_function(input: Optional[T] = None):
 74        nonlocal value
 75        if input is not None:
 76            value = input
 77        return value
 78
 79    return state_function
 80
 81
 82def get_json_from_cid(cid: str) -> dict[Any, Any] | None:
 83    gateway = "https://ipfs.io/ipfs/"
 84    try:
 85        result = requests.get(gateway + cid)
 86        if result.ok:
 87            return result.json()
 88        return None
 89    except Exception:
 90        return None
 91
 92
 93def convert_cid_on_proposal(proposals: dict[int, dict[str, Any]]):
 94    unwrapped: dict[int, dict[str, Any]] = {}
 95    for prop_id, proposal in proposals.items():
 96        data = proposal.get("data")
 97        if data and "Custom" in data:
 98            metadata = proposal["metadata"]
 99            cid = metadata.split("ipfs://")[-1]
100            queried_cid = get_json_from_cid(cid)
101            if queried_cid:
102                body = queried_cid.get("body")
103                if body:
104                    try:
105                        queried_cid["body"] = json.loads(body)
106                    except Exception:
107                        pass
108            proposal["Custom"] = queried_cid
109        unwrapped[prop_id] = proposal
110    return unwrapped
111
112
113HEX_PATTERN = re.compile(r"^[0-9a-fA-F]+$")
114
115
116# TODO: merge `is_hex_string` into `parse_hex`
117def is_hex_string(string: str):
118    return bool(HEX_PATTERN.match(string))
119
120
121def parse_hex(hex_str: str) -> bytes:
122    if hex_str[0:2] == "0x":
123        return bytes.fromhex(hex_str[2:])
124    else:
125        return bytes.fromhex(hex_str)