123 lines
4.9 KiB
Python
123 lines
4.9 KiB
Python
from datetime import datetime, timedelta
|
|
from sanic import response
|
|
from aiogram import Bot, types
|
|
from sqlalchemy import and_
|
|
from app.core.logger import make_log
|
|
from app.core.models.node_storage import StoredContent
|
|
from app.core.models.keys import KnownKey
|
|
from app.core.models import StarsInvoice
|
|
from app.core.models.content.user_content import UserContent
|
|
from app.core._config import CLIENT_TELEGRAM_API_KEY
|
|
import json
|
|
import uuid
|
|
|
|
|
|
async def s_api_v1_content_list(request):
|
|
offset = int(request.args.get('offset', 0))
|
|
limit = int(request.args.get('limit', 100))
|
|
assert 0 <= offset, "Invalid offset"
|
|
assert 0 < limit <= 1000, "Invalid limit"
|
|
|
|
store = request.args.get('store', 'local')
|
|
assert store in ('local', 'onchain'), "Invalid store"
|
|
|
|
content_list = request.ctx.db_session.query(StoredContent).filter(
|
|
StoredContent.type.like(store + '%'),
|
|
StoredContent.disabled == False
|
|
).order_by(StoredContent.created.desc()).offset(offset).limit(limit)
|
|
make_log("Content", f"Listed {content_list.count()} contents", level='info')
|
|
result = {}
|
|
for content in content_list.all():
|
|
content_json = content.json_format()
|
|
result[content_json["cid"]] = content_json
|
|
|
|
return response.json(result)
|
|
|
|
|
|
async def s_api_v1_content_view(request, content_address: str):
|
|
# content_address can be CID or TON address
|
|
r_content = StoredContent.from_cid(request.ctx.db_session, content_address)
|
|
content = r_content.open_content(request.ctx.db_session)
|
|
|
|
opts = {'content_type': content['content_type'], 'content_address': content['encrypted_content'].meta.get('item_address', '')}
|
|
if content['encrypted_content'].key_id:
|
|
known_key = request.ctx.db_session.query(KnownKey).filter(
|
|
KnownKey.id == content['encrypted_content'].key_id
|
|
).first()
|
|
if known_key:
|
|
opts['key_hash'] = known_key.seed_hash
|
|
|
|
opts['have_licenses'] = []
|
|
opts['invoice'] = None
|
|
|
|
have_access = False
|
|
if request.ctx.user:
|
|
user_wallet_address = request.ctx.user.wallet_address(request.ctx.db_session)
|
|
have_access = (
|
|
(content['encrypted_content'].owner_address == user_wallet_address)
|
|
or bool(request.ctx.db_session.query(UserContent).filter_by(owner_address=user_wallet_address, status='active',
|
|
content_id=content['encrypted_content'].id).first())
|
|
)
|
|
if not have_access:
|
|
stars_cost = 1
|
|
|
|
exist_invoice = request.ctx.db_session.query(StarsInvoice).filter(
|
|
and_(
|
|
StarsInvoice.user_id == request.ctx.user.id,
|
|
StarsInvoice.created > datetime.now() - timedelta(minutes=25),
|
|
)
|
|
).first()
|
|
if exist_invoice:
|
|
invoice_url = exist_invoice.invoice_url
|
|
else:
|
|
invoice_id = f"access_{uuid.uuid4().hex}"
|
|
try:
|
|
invoice_url = await Bot(token=CLIENT_TELEGRAM_API_KEY).create_invoice_link(
|
|
'Lifetime access to content',
|
|
'You will receive NFT with lifetime access to content',
|
|
f"access_{invoice_id}", "", "XTR",
|
|
[
|
|
types.LabeledPrice(label='Lifetime access', amount=stars_cost),
|
|
],
|
|
)
|
|
request.ctx.db_session.add(
|
|
StarsInvoice(
|
|
external_id=invoice_id,
|
|
type='access',
|
|
amount=stars_cost,
|
|
user_id=request.ctx.user.id,
|
|
content_hash=content['encrypted_content'].hash,
|
|
invoice_url=invoice_url
|
|
)
|
|
)
|
|
request.ctx.db_session.commit()
|
|
except BaseException as e:
|
|
make_log("Content", f"Can't create invoice link: {e}", level='warning')
|
|
|
|
opts['invoice'] = {
|
|
'url': invoice_url,
|
|
'amount': stars_cost,
|
|
}
|
|
|
|
display_options = {
|
|
'content_url': content['decrypted_content'].web_url + (
|
|
'?seconds_limit=30' if not have_access else ''
|
|
)
|
|
}
|
|
if have_access:
|
|
opts['have_licenses'].append('listen')
|
|
|
|
content_meta = content['encrypted_content'].json_format()
|
|
content_metadata = StoredContent.from_cid(request.ctx.db_session, content_meta.get('metadata_cid') or None)
|
|
with open(content_metadata.filepath, 'r') as f:
|
|
content_metadata_json = json.loads(f.read())
|
|
|
|
display_options['metadata'] = content_metadata_json
|
|
|
|
return response.json({
|
|
**opts,
|
|
'encrypted': content['encrypted_content'].json_format(),
|
|
'display_options': display_options
|
|
})
|
|
|