75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
from base58 import b58encode, b58decode
|
|
|
|
from app.core._config import ALLOWED_CONTENT_TYPES
|
|
from app.core._utils.string_binary import string_to_bytes_fixed_size, bytes_to_string
|
|
|
|
|
|
# cid_v1#_ cid_version:int8 accept_type:uint120 content_sha256:uint256 onchain_index:uint128 = CIDv1;
|
|
|
|
|
|
class ContentId:
|
|
def __init__(
|
|
self,
|
|
content_hash: bytes = None, # only SHA256
|
|
onchain_index: int = None,
|
|
accept_type: str = 'image/jpeg'
|
|
):
|
|
self.content_hash = content_hash
|
|
self.onchain_index = onchain_index or -1
|
|
|
|
self.accept_type = accept_type
|
|
|
|
@property
|
|
def content_hash_b58(self) -> str:
|
|
return b58encode(self.content_hash).decode()
|
|
|
|
def serialize_v1(self) -> str:
|
|
at_bin = string_to_bytes_fixed_size(self.accept_type, 15)
|
|
assert len(self.content_hash) == 32, "Invalid hash length"
|
|
if self.onchain_index < 0:
|
|
oi_bin = b''
|
|
else:
|
|
oi_bin = self.onchain_index.to_bytes(16, 'big', signed=False)
|
|
assert len(oi_bin) == 16, "Invalid onchain_index"
|
|
|
|
return b58encode(
|
|
(1).to_bytes(1, 'big') # cid version
|
|
+ at_bin
|
|
+ self.content_hash
|
|
+ oi_bin
|
|
).decode()
|
|
|
|
@classmethod
|
|
def from_v1(cls, cid: str):
|
|
cid_bin = b58decode(cid)
|
|
(
|
|
cid_version,
|
|
accept_type,
|
|
content_sha256,
|
|
onchain_index
|
|
) = (
|
|
int.from_bytes(cid_bin[0:1], 'big'),
|
|
bytes_to_string(cid_bin[1:16]),
|
|
cid_bin[16:48],
|
|
int.from_bytes(cid_bin[48:], 'big') if len(cid_bin) > 48 else -1
|
|
)
|
|
assert cid_version == 1, "Invalid version"
|
|
content_type = accept_type.split('/')
|
|
assert '/'.join(content_type[0:2]) in ALLOWED_CONTENT_TYPES, "Invalid accept type"
|
|
assert len(content_sha256) == 32, "Invalid hash length"
|
|
return cls(
|
|
content_hash=content_sha256,
|
|
onchain_index=onchain_index,
|
|
accept_type=accept_type
|
|
)
|
|
|
|
@classmethod
|
|
def deserialize(cls, cid: str):
|
|
cid_version = int.from_bytes(b58decode(cid)[0:1], 'big')
|
|
if cid_version == 1:
|
|
return cls.from_v1(cid)
|
|
else:
|
|
raise ValueError("Invalid cid version")
|
|
|
|
|