406 lines
19 KiB
Python
406 lines
19 KiB
Python
from base64 import b64encode, b32decode
|
||
from datetime import datetime
|
||
import traceback
|
||
|
||
from sanic import response
|
||
from sqlalchemy import and_, select, func
|
||
from tonsdk.boc import begin_cell, begin_dict
|
||
from tonsdk.utils import Address
|
||
|
||
from base58 import b58encode, b58decode
|
||
from app.core._blockchain.ton.connect import TonConnect, wallet_obj_by_name
|
||
from app.core._blockchain.ton.platform import platform
|
||
from app.core._config import PROJECT_HOST
|
||
from app.core.logger import make_log
|
||
from app.core._utils.resolve_content import resolve_content
|
||
from app.core.content.utils import create_metadata_for_item
|
||
from app.core._crypto.content import create_encrypted_content
|
||
from app.core.models.content.user_content import UserContent
|
||
from app.core.models.node_storage import StoredContent
|
||
from app.core.models._telegram import Wrapped_CBotChat
|
||
from app.core._keyboards import get_inline_keyboard
|
||
from app.core.models.promo import PromoAction
|
||
from app.core.models.tasks import BlockchainTask
|
||
|
||
|
||
def valid_royalty_params(royalty_params):
|
||
assert sum([x['value'] for x in royalty_params]) == 10000, "Values of royalties should sum to 10000"
|
||
for royalty_param in royalty_params:
|
||
for field_key, field_value in {
|
||
'address': lambda x: isinstance(x, str),
|
||
'value': lambda x: (isinstance(x, int) and 0 <= x <= 10000)
|
||
}.items():
|
||
assert field_key in royalty_param, f"No {field_key} provided"
|
||
assert field_value(royalty_param[field_key]), f"Invalid {field_key} provided"
|
||
return True
|
||
|
||
|
||
async def s_api_v1_blockchain_send_new_content_message(request):
|
||
try:
|
||
assert request.json, "No data provided"
|
||
assert request.ctx.user, "No authorized user provided"
|
||
|
||
if not request.json['hashtags']:
|
||
request.json['hashtags'] = []
|
||
|
||
for field_key, field_value in {
|
||
'title': lambda x: isinstance(x, str),
|
||
'authors': lambda x: isinstance(x, list),
|
||
'content': lambda x: isinstance(x, str), # may be plaintext CID (legacy) or encrypted IPFS CID (bafy...)
|
||
'image': lambda x: isinstance(x, str),
|
||
'description': lambda x: isinstance(x, str),
|
||
'price': lambda x: (isinstance(x, str) and x.isdigit()),
|
||
'allowResale': lambda x: isinstance(x, bool),
|
||
'royaltyParams': lambda x: (isinstance(x, list) and valid_royalty_params(x)),
|
||
'hashtags': lambda x: isinstance(x, list) and all([isinstance(y, str) for y in x])
|
||
}.items():
|
||
assert field_key in request.json, f"No {field_key} provided"
|
||
assert field_value(request.json[field_key]), f"Invalid {field_key} provided"
|
||
|
||
# Support legacy: 'content' as decrypted ContentId; and new: 'content' as encrypted IPFS CID
|
||
encrypted_content_cid = None
|
||
try:
|
||
# Legacy path
|
||
decrypted_content_cid, err = resolve_content(request.json['content'])
|
||
assert not err
|
||
decrypted_content = (await request.ctx.db_session.execute(
|
||
select(StoredContent).where(StoredContent.hash == decrypted_content_cid.content_hash_b58)
|
||
)).scalars().first()
|
||
assert decrypted_content and decrypted_content.type == "local/content_bin"
|
||
encrypted_content = await create_encrypted_content(request.ctx.db_session, decrypted_content)
|
||
encrypted_content_cid = encrypted_content.cid
|
||
except BaseException:
|
||
# New path: treat provided string as encrypted IPFS CID (ENCF v1)
|
||
encrypted_ipfs_cid = request.json['content']
|
||
|
||
class _EC:
|
||
"""Adapter to provide ContentId-like interface for IPFS CID strings."""
|
||
|
||
def __init__(self, cid_str: str):
|
||
self._cid = cid_str
|
||
self.content_hash = self._extract_content_hash(cid_str)
|
||
self._content_hash_b58 = None
|
||
|
||
@staticmethod
|
||
def _decode_multibase(cid_str: str) -> bytes:
|
||
if not cid_str:
|
||
raise ValueError("empty CID")
|
||
prefix = cid_str[0]
|
||
if prefix in ('b', 'B'):
|
||
payload = cid_str[1:]
|
||
padding = (-len(payload)) % 8
|
||
return b32decode(payload.upper() + ('=' * padding), casefold=True)
|
||
if prefix in ('z', 'Z'):
|
||
return b58decode(cid_str[1:])
|
||
# CIDv0 (base58btc without explicit multibase prefix)
|
||
return b58decode(cid_str)
|
||
|
||
@staticmethod
|
||
def _read_varint(data: bytes, offset: int):
|
||
result = 0
|
||
shift = 0
|
||
while True:
|
||
if offset >= len(data):
|
||
raise ValueError("truncated varint")
|
||
byte = data[offset]
|
||
offset += 1
|
||
result |= (byte & 0x7F) << shift
|
||
if not (byte & 0x80):
|
||
break
|
||
shift += 7
|
||
if shift > 63:
|
||
raise ValueError("varint overflow")
|
||
return result, offset
|
||
|
||
@classmethod
|
||
def _extract_content_hash(cls, cid_str: str) -> bytes:
|
||
data = cls._decode_multibase(cid_str)
|
||
offset = 0
|
||
if data and data[0] == 0x01:
|
||
version, offset = cls._read_varint(data, offset)
|
||
if version != 1:
|
||
raise ValueError("unsupported CID version")
|
||
_, offset = cls._read_varint(data, offset) # skip codec
|
||
code, offset = cls._read_varint(data, offset)
|
||
length, offset = cls._read_varint(data, offset)
|
||
digest = data[offset:offset + length]
|
||
if len(digest) != length:
|
||
raise ValueError("truncated multihash digest")
|
||
if code != 0x12 or length != 32:
|
||
raise ValueError("unsupported multihash (expect sha2-256)")
|
||
return digest
|
||
|
||
def serialize_v2(self, include_accept_type: bool = False):
|
||
return self._cid
|
||
|
||
@property
|
||
def content_hash_b58(self) -> str:
|
||
if self._content_hash_b58 is None:
|
||
self._content_hash_b58 = b58encode(self.content_hash).decode()
|
||
return self._content_hash_b58
|
||
|
||
try:
|
||
encrypted_content_cid = _EC(encrypted_ipfs_cid)
|
||
except Exception as exc:
|
||
make_log("Blockchain", f"Provided encrypted IPFS CID is invalid: {exc}", level='error')
|
||
raise AssertionError("Invalid encrypted content CID provided") from exc
|
||
|
||
if request.json['image']:
|
||
image_content_cid, err = resolve_content(request.json['image'])
|
||
assert not err, f"Invalid image CID"
|
||
image_content = (await request.ctx.db_session.execute(
|
||
select(StoredContent).where(StoredContent.hash == image_content_cid.content_hash_b58)
|
||
)).scalars().first()
|
||
assert image_content, "No image locally found"
|
||
else:
|
||
image_content_cid = None
|
||
image_content = None
|
||
|
||
|
||
content_title = f"{', '.join(request.json['authors'])} – {request.json['title']}" if request.json['authors'] else request.json['title']
|
||
|
||
metadata_content = await create_metadata_for_item(
|
||
request.ctx.db_session,
|
||
title=content_title,
|
||
cover_url=f"{PROJECT_HOST}/api/v1.5/storage/{image_content_cid.serialize_v2()}" if image_content_cid else None,
|
||
authors=request.json['authors'],
|
||
hashtags=request.json['hashtags'],
|
||
downloadable=request.json['downloadable'] if 'downloadable' in request.json else False,
|
||
)
|
||
|
||
# Try to update ContentIndexItem with cover_url for this encrypted content
|
||
try:
|
||
from app.core.models.content_v3 import ContentIndexItem
|
||
ecid_str = encrypted_content_cid.serialize_v2()
|
||
row = (await request.ctx.db_session.execute(select(ContentIndexItem).where(ContentIndexItem.encrypted_cid == ecid_str))).scalars().first()
|
||
if row:
|
||
payload = row.payload or {}
|
||
payload['cover_url'] = f"{PROJECT_HOST}/api/v1.5/storage/{image_content_cid.serialize_v2()}" if image_content_cid else None
|
||
row.payload = payload
|
||
await request.ctx.db_session.commit()
|
||
except Exception as _e:
|
||
make_log("Blockchain", f"index cover update failed: {_e}", level='warning')
|
||
|
||
royalties_dict = begin_dict(8)
|
||
i = 0
|
||
for royalty_param in request.json['royaltyParams']:
|
||
royalties_dict.store_ref(
|
||
i, begin_cell()
|
||
.store_address(Address(royalty_param['address']))
|
||
.store_uint(royalty_param['value'], 16)
|
||
.end_cell()
|
||
)
|
||
i += 1
|
||
|
||
_cnt = (await request.ctx.db_session.execute(
|
||
select(func.count()).select_from(PromoAction).where(
|
||
and_(
|
||
PromoAction.user_internal_id == request.ctx.user.id,
|
||
PromoAction.action_type == 'freeUpload'
|
||
)
|
||
)
|
||
)).scalar()
|
||
promo_free_upload_available = 3 - int(_cnt or 0)
|
||
|
||
has_pending_task = (await request.ctx.db_session.execute(
|
||
select(BlockchainTask).where(
|
||
and_(BlockchainTask.user_id == request.ctx.user.id, BlockchainTask.status != 'done')
|
||
)
|
||
)).scalars().first()
|
||
if has_pending_task:
|
||
make_log("Blockchain", f"User {request.ctx.user.id} already has a pending task", level='warning')
|
||
promo_free_upload_available = 0
|
||
|
||
make_log("Blockchain", f"User {request.ctx.user.id} has {promo_free_upload_available} free uploads available", level='info')
|
||
|
||
if promo_free_upload_available > 0:
|
||
promo_action = PromoAction(
|
||
user_id = str(request.ctx.user.id),
|
||
user_internal_id=request.ctx.user.id,
|
||
action_type='freeUpload',
|
||
action_ref=str(encrypted_content_cid.content_hash),
|
||
created=datetime.now()
|
||
)
|
||
request.ctx.db_session.add(promo_action)
|
||
|
||
blockchain_task = BlockchainTask(
|
||
destination=platform.address.to_string(1, 1, 1),
|
||
amount=str(int(0.03 * 10 ** 9)),
|
||
payload=b64encode(
|
||
begin_cell()
|
||
.store_uint(0x5491d08c, 32)
|
||
.store_uint(int.from_bytes(encrypted_content_cid.content_hash, "big", signed=False), 256)
|
||
.store_address(Address(await request.ctx.user.wallet_address_async(request.ctx.db_session)))
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_coins(int(0))
|
||
.store_coins(int(0))
|
||
.store_coins(int(request.json['price']))
|
||
.end_cell()
|
||
)
|
||
.store_maybe_ref(royalties_dict.end_dict())
|
||
.store_uint(0, 1)
|
||
.end_cell()
|
||
)
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_bytes(f"{PROJECT_HOST}/api/v1.5/storage/{metadata_content.cid.serialize_v2(include_accept_type=True)}".encode())
|
||
.end_cell()
|
||
)
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_ref(begin_cell().store_bytes(f"{encrypted_content_cid.serialize_v2()}".encode()).end_cell())
|
||
.store_ref(begin_cell().store_bytes(f"{image_content_cid.serialize_v2() if image_content_cid else ''}".encode()).end_cell())
|
||
.store_ref(begin_cell().store_bytes(f"{metadata_content.cid.serialize_v2()}".encode()).end_cell())
|
||
.end_cell()
|
||
)
|
||
.end_cell()
|
||
)
|
||
.end_cell().to_boc(False)
|
||
).decode(),
|
||
epoch=None, seqno=None,
|
||
created = datetime.now(),
|
||
status='wait',
|
||
user_id = request.ctx.user.id
|
||
)
|
||
request.ctx.db_session.add(blockchain_task)
|
||
await request.ctx.db_session.commit()
|
||
|
||
await request.ctx.user_uploader_wrapper.send_message(
|
||
request.ctx.user.translated('p_uploadContentTxPromo').format(
|
||
title=content_title,
|
||
free_count=(promo_free_upload_available - 1)
|
||
), message_type='hint', message_meta={
|
||
'encrypted_content_hash': b58encode(encrypted_content_cid.content_hash).decode(),
|
||
'hint_type': 'uploadContentTxRequested'
|
||
}
|
||
)
|
||
return response.json({
|
||
'address': "free",
|
||
'amount': str(int(0.03 * 10 ** 9)),
|
||
'payload': ""
|
||
})
|
||
|
||
await request.ctx.user_uploader_wrapper.send_message(
|
||
request.ctx.user.translated('p_uploadContentTxRequested').format(
|
||
title=content_title,
|
||
), message_type='hint', message_meta={
|
||
'encrypted_content_hash': b58encode(encrypted_content_cid.content_hash).decode(),
|
||
'hint_type': 'uploadContentTxRequested'
|
||
}
|
||
)
|
||
|
||
return response.json({
|
||
'address': platform.address.to_string(1, 1, 1),
|
||
'amount': str(int(0.03 * 10 ** 9)),
|
||
'payload': b64encode(
|
||
begin_cell()
|
||
.store_uint(0x5491d08c, 32)
|
||
.store_uint(int.from_bytes(encrypted_content_cid.content_hash, "big", signed=False), 256)
|
||
.store_uint(0, 2)
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_coins(int(0))
|
||
.store_coins(int(0))
|
||
.store_coins(int(request.json['price']))
|
||
.end_cell()
|
||
)
|
||
.store_maybe_ref(royalties_dict.end_dict())
|
||
.store_uint(0, 1)
|
||
.end_cell()
|
||
)
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_bytes(f"{PROJECT_HOST}/api/v1.5/storage/{metadata_content.cid.serialize_v2(include_accept_type=True)}".encode())
|
||
.end_cell()
|
||
)
|
||
.store_ref(
|
||
begin_cell()
|
||
.store_ref(begin_cell().store_bytes(f"{encrypted_content_cid.serialize_v2()}".encode()).end_cell())
|
||
.store_ref(begin_cell().store_bytes(f"{image_content_cid.serialize_v2() if image_content_cid else ''}".encode()).end_cell())
|
||
.store_ref(begin_cell().store_bytes(f"{metadata_content.cid.serialize_v2()}".encode()).end_cell())
|
||
.end_cell()
|
||
)
|
||
.end_cell()
|
||
)
|
||
.end_cell().to_boc(False)
|
||
).decode()
|
||
})
|
||
except BaseException as e:
|
||
make_log("Blockchain", f"Error while sending new content message: {e}" + '\n' + traceback.format_exc(), level='error')
|
||
return response.json({"error": str(e)}, status=400)
|
||
|
||
|
||
async def s_api_v1_blockchain_send_purchase_content_message(request):
|
||
assert request.json, "No data provided"
|
||
|
||
for field_key, field_value in {
|
||
'content_address': lambda x: isinstance(x, str),
|
||
'license_type': lambda x: x in ['resale']
|
||
}.items():
|
||
assert field_key in request.json, f"No {field_key} provided"
|
||
assert field_value(request.json[field_key]), f"Invalid {field_key} provided"
|
||
|
||
if not (await request.ctx.user.wallet_address_async(request.ctx.db_session)):
|
||
return response.json({"error": "No wallet address provided"}, status=400)
|
||
|
||
from sqlalchemy import select
|
||
license_exist = (await request.ctx.db_session.execute(select(UserContent).where(
|
||
UserContent.onchain_address == request.json['content_address']
|
||
))).scalars().first()
|
||
if license_exist:
|
||
from app.core.content.content_id import ContentId
|
||
_cid = ContentId.deserialize(license_exist.content.cid.serialize_v2())
|
||
r_content = (await request.ctx.db_session.execute(select(StoredContent).where(StoredContent.hash == _cid.content_hash_b58))).scalars().first()
|
||
else:
|
||
from app.core.content.content_id import ContentId
|
||
_cid = ContentId.deserialize(request.json['content_address'])
|
||
r_content = (await request.ctx.db_session.execute(select(StoredContent).where(StoredContent.hash == _cid.content_hash_b58))).scalars().first()
|
||
|
||
async def open_content_async(session, sc: StoredContent):
|
||
if not sc.encrypted:
|
||
decrypted = sc
|
||
encrypted = (await session.execute(select(StoredContent).where(StoredContent.decrypted_content_id == sc.id))).scalars().first()
|
||
else:
|
||
encrypted = sc
|
||
decrypted = (await session.execute(select(StoredContent).where(StoredContent.id == sc.decrypted_content_id))).scalars().first()
|
||
assert decrypted and encrypted, "Can't open content"
|
||
ctype = decrypted.json_format().get('content_type', 'application/x-binary')
|
||
try:
|
||
content_type = ctype.split('/')[0]
|
||
except Exception:
|
||
content_type = 'application'
|
||
return {'encrypted_content': encrypted, 'decrypted_content': decrypted, 'content_type': content_type}
|
||
content = await open_content_async(request.ctx.db_session, r_content)
|
||
|
||
licenses_cost = content['encrypted_content'].json_format()['license']
|
||
assert request.json['license_type'] in licenses_cost
|
||
|
||
return response.json({
|
||
'address': (
|
||
license_exist.onchain_address if license_exist else content['encrypted_content'].json_format()['item_address']
|
||
),
|
||
'amount': str(int(licenses_cost['resale']['price'])),
|
||
'payload': b64encode((
|
||
begin_cell()
|
||
.store_uint(0x2a319593, 32)
|
||
.store_uint(0, 64)
|
||
.store_uint(3, 8)
|
||
# .store_uint({
|
||
# 'listen': 1,
|
||
# 'resale': 3
|
||
# }[request.json['license_type']], 8)
|
||
.store_uint(0, 256)
|
||
.store_uint(0, 2)
|
||
.end_cell()
|
||
).to_boc(False)).decode()
|
||
})
|