from base64 import b64encode from datetime import datetime import traceback from sanic import response from tonsdk.boc import begin_cell, begin_dict from tonsdk.utils import Address from app.core._blockchain.ton.connect import TonConnect, TON_CONNECT_WALLETS_LIST 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.node_storage import StoredContent from app.core.models._telegram import Wrapped_CBotChat from app.core._keyboards import get_inline_keyboard 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" for field_key, field_value in { 'title': lambda x: isinstance(x, str), 'authors': lambda x: isinstance(x, list), 'content': lambda x: isinstance(x, str), '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)), }.items(): assert field_key in request.json, f"No {field_key} provided" assert field_value(request.json[field_key]), f"Invalid {field_key} provided" ton_connect, ton_connection = TonConnect.by_user(request.ctx.db_session, request.ctx.user, callback_fn=()) await ton_connect.restore_connection() assert ton_connect.connected, "No connected wallet" decrypted_content_cid, err = resolve_content(request.json['content']) assert not err, f"Invalid content CID" decrypted_content = request.ctx.db_session.query(StoredContent).filter( StoredContent.hash == decrypted_content_cid.content_hash_b58 ).first() assert decrypted_content, "No content locally found" assert decrypted_content.type == "local/content_bin", "Invalid content type" encrypted_content = await create_encrypted_content(request.ctx.db_session, decrypted_content) encrypted_content_cid = encrypted_content.cid if request.json['image']: image_content_cid, err = resolve_content(request.json['image']) assert not err, f"Invalid image CID" image_content = request.ctx.db_session.query(StoredContent).filter( StoredContent.hash == image_content_cid.content_hash_b58 ).first() assert image_content, "No image locally found" else: image_content_cid = None image_content = None metadata_content = await create_metadata_for_item( request.ctx.db_session, title=request.json['title'], cover_url=request.json['image'], authors=request.json['authors'] ) 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 request.app.add_task(ton_connect._sdk_client.send_transaction({ 'valid_until': int(datetime.now().timestamp() + 120), 'messages': [ { 'address': platform.address.to_string(1, 1, 1), 'amount': str(int(0.15 * 10 ** 9)), 'payload': b64encode( begin_cell() .store_uint(0x5491d08c, 32) .store_uint(int(encrypted_content_cid.content_hash.hex()[2:], 16), 256) .store_ref( begin_cell() .store_ref( begin_cell() .store_coins(int(request.json['price'])) .store_coins(int(1000000 * 10 ** 9)) .store_coins(int(5000000 * 10 ** 9)) .end_cell() ) .store_maybe_ref(royalties_dict.end_dict()) .store_uint(0, 1) .end_cell() ) .store_ref( begin_cell() .store_ref( begin_cell() .store_uint(1, 8) .store_bytes(f"{PROJECT_HOST}/api/v1/storage/{metadata_content.cid.serialize_v2()}".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() } ] })) await (Wrapped_CBotChat(request.app.ctx.memory._telegram_bot, chat_id=request.ctx.user.telegram_id)).send_message( request.ctx.user.translated('p_tonconnectTransactionRequested'), reply_markup=get_inline_keyboard([ [{ 'text': request.ctx.user.translated('gotoWallet_button'), 'url': next(x['universal_url'] for x in TON_CONNECT_WALLETS_LIST if x['app_name'] == ton_connection.wallet_key) }], [{ 'text': request.ctx.user.translated('home_button'), 'callback_data': 'home' }] ]) ) return response.json({"message": "Transaction requested"}) 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), 'price': lambda x: (isinstance(x, str) and x.isdigit()), }.items(): assert field_key in request.json, f"No {field_key} provided" assert field_value(request.json[field_key]), f"Invalid {field_key} provided" return response.json({"message": "Transaction requested"})