from datetime import datetime from aiogram.utils.web_app import safe_parse_webapp_init_data from sanic import response from app.core._config import TELEGRAM_API_KEY from app.core.models.user import User async def s_api_v1_auth_twa(request): if not request.json: return response.json({"error": "No data provided"}, status=400) if not request.json.get('twa_data'): return response.json({"error": "No TWA data provided"}, status=400) twa_data = request.json['twa_data'] twa_data = safe_parse_webapp_init_data(token=TELEGRAM_API_KEY, init_data=twa_data) assert twa_data known_user = request.ctx.db_session.query(User).filter(User.telegram_id == twa_data.user.id).first() if not known_user: new_user = User( telegram_id=twa_data.user.id, username=twa_data.user.username, meta={ "first_name": twa_data.user.first_name, "last_name": twa_data.user.last_name, "photo_url": twa_data.user.photo_url }, lang_code=twa_data.user.language_code, last_use=datetime.now(), created=datetime.now() ) request.ctx.db_session.add(new_user) request.ctx.db_session.commit() known_user = request.ctx.db_session.query(User).filter(User.telegram_id == twa_data.user.id).first() assert known_user, "User not created" new_user_key = await known_user.create_api_token_v1(request.ctx.db_session, "USER_API_V1") return response.json({ 'user': known_user.json_format(), 'auth_v1_token': new_user_key['auth_v1_token'] })