from django.contrib.auth import get_user_model from django.db.models import Q from drf_yasg import openapi from drf_yasg.utils import swagger_auto_schema from django.forms.models import model_to_dict from oauth2_provider.oauth2_validators import Application from rest_framework import ( generics, pagination, permissions, status, views, viewsets, ) from rest_framework.parsers import FormParser, MultiPartParser from rest_framework.response import Response from apps.artiste.models import Artiste, AudioMedia from apps.lib.constants import USER_TYPE_ARTISTE, USER_TYPE_FAN from apps.lib.models import TermsAndConditions from profiles import serializers from profiles.models import Media, Profile,TriviaGame,TriviaScore,SupportArtistPacks,UserPacksTranscation from users.constants import ACCOUNT_TYPE_ARTISTE, ACCOUNT_TYPE_FAN from users.models import Fan from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from openai import OpenAI from django.conf import settings import stripe import os from rest_framework.pagination import PageNumberPagination User = get_user_model() def generateTranscationsId(): lastObj= UserPacksTranscation.objects.all().last() if lastObj: if not lastObj.transcation_id: return 'TRANS000001' theId=lastObj.transcation_id theId=theId[5:] theId=int(theId)+1 theId=str(theId) theId=theId.zfill(5) return "TRANS"+str(theId) else: return 'TRANS000001' class UserProfileViewe(generics.RetrieveUpdateAPIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated] queryset = Profile.objects.all() def get_object(self): return Profile.objects.get_or_create(user=self.request.user)[0] class UserProfileView(generics.RetrieveUpdateAPIView): permission_classes = [permissions.IsAuthenticated] def get_serializer_class(self): if self.request.user.account_type == ACCOUNT_TYPE_FAN: return serializers.FanProfileSerializer elif self.request.user.account_type == ACCOUNT_TYPE_ARTISTE: return serializers.ArtisteProfileSerializer def get_queryset(self): if self.request.user.account_type == ACCOUNT_TYPE_FAN: return Fan.objects.all() elif self.request.user.account_type == ACCOUNT_TYPE_ARTISTE: return Artiste.objects.all() def get_object(self): if self.request.user.account_type == ACCOUNT_TYPE_FAN: # type: ignore return Fan.objects.get_or_create(user=self.request.user)[0] elif self.request.user.account_type == ACCOUNT_TYPE_ARTISTE: # type: ignore return Artiste.objects.get_or_create(user=self.request.user)[0] def get(self, request, *args, **kwargs): if self.request.user.account_type == ACCOUNT_TYPE_FAN: # type: ignore obj, _ = Fan.objects.get_or_create(user=self.request.user) elif self.request.user.account_type == ACCOUNT_TYPE_ARTISTE: # type: ignore obj, _ = Fan.objects.get_or_create(user=self.request.user) obj, _ = Artiste.objects.get_or_create(user=self.request.user) response=self.retrieve(request, *args, **kwargs) UserData=User.objects.get(email=response.data.get("email_id")) user_dict = model_to_dict(UserData) response.data["profile_flag"]=user_dict.get("profile_flag") return response class UserProfilePictureUpdate(views.APIView): parser_classes = ( MultiPartParser, FormParser, ) def get_profile(self, user): if user.account_type == ACCOUNT_TYPE_FAN: return Fan.objects.get_or_create(user=user)[0] elif user.account_type == ACCOUNT_TYPE_ARTISTE: return Artiste.objects.get_or_create(user=user)[0] @swagger_auto_schema( operation_id='Upload Profile Picture', operation_description='Upload Profile Picture', required=['profile_picture'], manual_parameters=[ openapi.Parameter( 'profile_picture', openapi.IN_FORM, type=openapi.TYPE_FILE, description='Profile Picture' ), ], ) def patch(self, request, *args, **kwargs): user = request.user _user=User.objects.get(email=user.email) profile = self.get_profile(user) if profile is not None: _user.profile_flag=True _user.save() profile.profile_picture = request.data.get("profile_picture") profile.save() user_dict = model_to_dict(_user) return Response({"detail": "Profile picture updated", "profile_flag": user_dict.get("profile_flag")}) # return Response({"detail": "Profile picture updated","data":(json.dumps(_user))}) class LikedSongsListView(generics.ListAPIView): serializer_class = serializers.MediaSerializer queryset = Media.objects.all() permission_classes = [ permissions.IsAuthenticated, ] pagination_class = pagination.LimitOffsetPagination def get_queryset(self): return self.request.user.liked_media.all() class FollowedArtistsView(generics.ListAPIView): serializer_class = serializers.ArtisteProfileSerializer queryset = Profile.objects.all() permission_classes = [ permissions.IsAuthenticated, ] pagination_class = pagination.LimitOffsetPagination def get_queryset(self): followed_artists = self.request.user.followed.all() profiles = Profile.objects.filter(user__in=followed_artists) return profiles # return self.request.user.liked_profiles.all() class MoreOfWhatYouLikeViewset(viewsets.ReadOnlyModelViewSet): serializer_class = serializers.MoreOfWhatYouLikeSerializer permission_classes = [ permissions.IsAuthenticated, ] queryset = Media.objects.all() pagination_class = pagination.PageNumberPagination class PopularArtistViewSet(viewsets.ModelViewSet): serializer_class = serializers.ArtisteProfileSerializer permission_classes = [permissions.IsAuthenticated] queryset = Profile.objects.filter(user__user_type="artist") class TrendingSongView(generics.ListAPIView): queryset = Media.objects.all() serializer_class = serializers.MediaSerializer permission_classes = [ permissions.IsAuthenticated, ] class SearchAPIView(generics.GenericAPIView): permission_classes = [ permissions.IsAuthenticated, ] serializer_class = serializers.MediaSearchSerializer @swagger_auto_schema( manual_parameters=[ openapi.Parameter(in_=openapi.IN_QUERY, type=openapi.TYPE_STRING, name='search', description='SearchAPI'), ] ) def get(self, request, format=None): search_query = request.query_params.get('search', None) if search_query is None: return Response({'error': 'Please provide a search term'}, status=status.HTTP_400_BAD_REQUEST) songs = AudioMedia.objects.filter( Q(title__icontains=search_query) | Q(album__album_name__icontains=search_query) ) songs_serializer = serializers.SuggestionsSerializer(songs, context={"request": request}, many=True) artiste = Artiste.objects.filter( Q(stage_name__icontains=search_query) | Q(user__first_name__icontains=search_query) | Q(user__last_name__icontains=search_query) ) artiste_serializer = serializers.ArtisteSerializer(artiste, context={"request": request}, many=True) data = { 'songs': songs_serializer.data, 'artistes': artiste_serializer.data, } return Response(data, status=status.HTTP_200_OK) class SuggestionView(views.APIView): permission_classes = [ permissions.IsAuthenticated, ] def get(self, request, format=None): most_liked_songs = AudioMedia.objects.prefetch_related('likes').all().order_by('-likes')[:20] serializer = serializers.SuggestionsSerializer(most_liked_songs, context={"request": request}, many=True) most_liked_artists = Artiste.objects.prefetch_related('followers').all().order_by('-followers')[:20] most_liked_artists_serializer = serializers.ArtisteSerializer( most_liked_artists, context={"request": request}, many=True ) top_trending_songs = AudioMedia.objects.prefetch_related('comments').all().order_by('-comments')[:20] trending_serializer = serializers.SuggestionsSerializer( top_trending_songs, context={"request": request}, many=True ) data = { 'most_liked_songs': serializer.data, 'most_liked_artists': most_liked_artists_serializer.data, 'top_trending_songs': trending_serializer.data, } return Response(data, status=status.HTTP_200_OK) class TermsAndConditionsAPIView(generics.GenericAPIView): serializer_class = serializers.TermsAndConditionsSerializer queryset = TermsAndConditions.objects.filter(is_active=True).all() permission_classes = [ permissions.IsAuthenticated, ] def get_queryset(self): if self.request.user.user_type == "artist": return super().get_queryset().filter(user_type=USER_TYPE_ARTISTE).order_by('-updated_datetime').first() return super().get_queryset().filter(user_type=USER_TYPE_FAN).order_by('-updated_datetime').first() @swagger_auto_schema(responses={200: serializers.TermsAndConditionsSerializer()}) def get(self, request, format=None): serializer = self.serializer_class(self.get_queryset()) return Response(serializer.data, status=status.HTTP_200_OK) import random class CreateAppAPIView(generics.GenericAPIView): authentication_classes = [] permission_classes = [] def get(self, request, format=None): try: client_secret = request.GET.get("client_secret") if not client_secret: return Response( {"error": "client_secret is required as a query parameter."}, status=status.HTTP_400_BAD_REQUEST ) Application.objects.create( name="Saigon", client_type=Application.CLIENT_CONFIDENTIAL, authorization_grant_type="password", client_secret=client_secret, algorithm="RS256" ) return Response({"data": "App added successfully."}, status=status.HTTP_201_CREATED) except Exception as e: return Response( {"error": "Failed to create application.", "details": str(e)}, status=status.HTTP_400_BAD_REQUEST ) import random def calculate_points(correct: bool, time_taken: float) -> int: if not correct: return 0 base_points = 100 bonus = max(0, int(50 - time_taken)) # Faster answer = higher bonus return base_points + bonus # Temporary storage for previously generated questions (you can replace this with a database in production) generated_questions_set = set() def generate_unique_question(genre, difficulty): # Define the prompt template prompt_template = f""" You are a music trivia quiz master. Your job is to generate one unique and non-repeating multiple-choice question about {genre} music at the {difficulty} level. Make sure the question is fresh and not overused — do not include widely repeated examples like “Who is known as the King of Pop?” or “What is the best-selling album of all time?” Difficulty levels are defined as: Easy: Well-known mainstream music knowledge (popular artists, hit songs, chart-toppers). Hard: Deeper knowledge (lesser-known songs, producers, album tracks, genre history, etc.). Expert: Extremely challenging, niche, or technical questions (e.g., rare live recordings, deep discography cuts, studio credits, music theory, etc.). Ensure topic variety — rotate across artists, albums, songs, history, and notable events within the genre. Output only ONE question at a time, using the following exact format: Question: A. Option 1 B. Option 2 C. Option 3 D. Option 4 Answer: Important: Only one correct answer per question. Randomize the placement of the correct option. Ensure no topic or question is repeated frequently. Each question must feel original and fresh. """ client = OpenAI(api_key=settings.OPEN_AI_KEY) # Generate a single question using the OpenAI client response = client.chat.completions.create( model="gpt-3.5-turbo", # You can use gpt-4 if available messages=[{"role": "user", "content": prompt_template}], temperature=0.7 ) # Extract the generated question question_text = response.choices[0].message.content.strip() # Check if the question is unique (not already generated) if question_text in generated_questions_set: # If the question was already generated, generate another one return generate_unique_question(genre, difficulty) # Store the generated question in the set generated_questions_set.add(question_text) return question_text class GetTrivia(generics.GenericAPIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated,] def post(self, request): user = request.user _user=User.objects.get(email=user.email) genre = request.data.get("genre") difficulty = request.data.get("difficulty") if not genre or not difficulty: return Response({"error": "Genre and difficulty are required"}, status=400) # Generate one unique question at a time question_data = generate_unique_question(genre, difficulty) # Return the question in JSON format return Response({"question": question_data}, status=200) class StartGame(generics.GenericAPIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated,] def post(self, request): user = request.user _user=User.objects.get(email=user.email) game = TriviaGame.objects.create(user=_user, duration=30) # Default 30 seconds return Response({"message": "Game started", "game_id": game.id}, status=200) class SubmitAnswer(views.APIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated,] def post(self, request): user = request.user _user=User.objects.get(email=user.email) game_id = request.data.get("game_id") correct = request.data.get("correct") time_taken = float(request.data.get("time_taken", 0)) if not game_id: return Response({"error": "Game ID required"}, status=400) try: game = TriviaGame.objects.get(id=game_id, user=_user, status="ongoing") except TriviaGame.DoesNotExist: return Response({"error": "Game not found or already completed"}, status=404) points = calculate_points(correct, time_taken) # Update game stats if correct: game.correct_answers += 1 else: game.incorrect_answers += 1 game.score += points game.save() return Response({ "message": "Answer recorded", "current_score": game.score, "correct_answers": game.correct_answers, "incorrect_answers": game.incorrect_answers }, status=200) # Finish the Game class FinishGame(views.APIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated,] def post(self, request): user = request.user _user=User.objects.get(email=user.email) game_id = request.data.get("game_id") if not game_id: return Response({"error": "Game ID required"}, status=400) try: game = TriviaGame.objects.get(id=game_id, user=_user, status="ongoing") except TriviaGame.DoesNotExist: return Response({"error": "Game not found or already completed"}, status=404) # Complete the game game.status = 'completed' game.save() # Save to TriviaScore TriviaScore.objects.create( user=user, game=game, score=game.score ) return Response({ "message": "Game completed", "final_score": game.score, "correct_answers": game.correct_answers, "incorrect_answers": game.incorrect_answers }, status=200) class Leaderboard(views.APIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated,] def get(self, request): top_scores = TriviaScore.objects.order_by('-score')[:10] leaderboard_data = [] for score in top_scores: leaderboard_data.append({ "user": score.user.first_name, "score": score.score, "date_played": score.date_played }) return Response({"leaderboard": leaderboard_data}, status=200) class GetArtistPack(views.APIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated,] def get(self, request): sub_obj = SupportArtistPacks.objects.order_by('-id') subscription_data = [] for sub in sub_obj: subscription_data.append({ "id": sub.id, "title": sub.title, "amount": sub.amount, "description": sub.description, "created_at": sub.created_at, }) return Response({"subscription_data": subscription_data}, status=200) # class CreateStripePaymentIntent(APIView): # permission_classes = [permissions.IsAuthenticated] # def post(self, request): # stripe.api_key = settings.STRIPE_SECRET_KEY # user = request.user # _user=User.objects.get(email=user.email) # amount = request.data.get('amount') # You must pass amount from frontend # if not amount: # return Response({'message': 'Amount is required'}, status=status.HTTP_400_BAD_REQUEST) # pack_id = request.data.get('pack_id') # if not pack_id: # return Response({'message': 'pack_id is required'}, status=status.HTTP_400_BAD_REQUEST) # pack_obj = SupportArtistPacks.objects.filter(id=pack_id).first() # if not pack_obj: # return Response({'message': 'No artist pack found'}, status=status.HTTP_400_BAD_REQUEST) # artist_id = request.data.get('artist_id') # if not artist_id: # return Response({'message': 'artiste_id is required'}, status=status.HTTP_400_BAD_REQUEST) # artist = Artiste.objects.filter(id=artist_id).first() # if not artist: # return Response({'message': 'No artist found'}, status=status.HTTP_400_BAD_REQUEST) # try: # payment_method = 'pm_1J3R6dF4kPmt5xJhZnkJ2HG5' # Replace this with the actual payment method ID # intent = stripe.PaymentIntent.create( # amount=int(float(amount) * 100), # convert dollars to cents # currency='usd', # payment_method_types=['card'], # ) # # if intent['status'] == 'succeeded': # # Save transaction details # CustTransaction = UserPacksTranscation.objects.create( # transcation_id=generateTranscationsId(), # user=_user, # pack=pack_obj, # artiste =artist, # amount=str(pack_obj.amount), # txn_id=intent['id'], # # recipet_url=intent['receipt_url'], # payment_mode='card', # transcation_type="pay" # ) # return Response({'client_secret': intent.client_secret}, status=status.HTTP_200_OK) # except stripe.error.StripeError as e: # return Response({'message': str(e)}, status=status.HTTP_400_BAD_REQUEST) class CreateStripePaymentIntent(APIView): permission_classes = [permissions.IsAuthenticated] def post(self, request): stripe.api_key = settings.STRIPE_SECRET_KEY user = request.user amount = request.data.get('amount') artist_id = request.data.get('artist_id') # Validation if not amount: return Response({'message': 'Amount is required'}, status=400) if not artist_id: return Response({'message': 'artist_id is required'}, status=400) try: # Get related objects artist = Artiste.objects.get(id=artist_id) _user = User.objects.get(email=user.email) if not artist.stripe_account_id: return Response({'message': 'Artist does not have a Stripe account'}, status=400) # Convert amount to cents amount_cents = int(float(amount) * 100) # Create payment intent with transfer to artist intent = stripe.PaymentIntent.create( amount=amount_cents, currency='usd', payment_method_types=['card'], application_fee_amount=int(amount_cents * 0.05), # Example: 5% fee, adjust as necessary transfer_data={ 'destination': artist.stripe_account_id, } ) # Save transaction details CustTransaction = UserPacksTranscation.objects.create( transcation_id=generateTranscationsId(), user=_user, artiste=artist, amount=amount, txn_id=intent['id'], payment_mode='card', transcation_type="pay" ) return Response({'client_secret': intent.client_secret}, status=200) except SupportArtistPacks.DoesNotExist: return Response({'message': 'No artist pack found'}, status=400) except Artiste.DoesNotExist: return Response({'message': 'No artist found'}, status=400) except User.DoesNotExist: return Response({'message': 'User not found'}, status=400) except stripe.error.StripeError as e: return Response({'message': str(e)}, status=400) def get_file_path(file_name): # Construct the full file path file_path = f"/opt/webapp/mediafiles/{file_name}" # Check if the file exists if os.path.exists(file_path): return file_path else: return "Image not found" class CustomPagination(PageNumberPagination): page_size = 10 # Adjust the number of items per page page_size_query_param = 'page_size' max_page_size = 100 class myPaymentHistory(APIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated] def get(self, request): user = request.user _user=User.objects.get(email=user.email) pay_obj = UserPacksTranscation.objects.filter(user = _user).order_by('-id') if not pay_obj: return Response({'message': 'No transactions found'}, status=status.HTTP_400_BAD_REQUEST) # Apply pagination paginator = CustomPagination() # Use custom or default pagination class paginated_data = paginator.paginate_queryset(pay_obj, request) all_data = [] for pay in paginated_data: artist_image = get_file_path(pay.artiste.profile_picture) all_data.append({ 'id': pay.id, 'transcation_id': pay.transcation_id, 'artist': pay.artiste.stage_name, 'artist_image': artist_image, 'date': pay.created_at, 'amount': pay.amount, }) # Return paginated response return paginator.get_paginated_response({'message': 'Success', 'all_data': all_data}) class PurchaseArtistPack(views.APIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated,] def post(self,request): user = request.user _user=User.objects.get(email=user.email) pack_id = request.data.get('pack_id') if not pack_id: return Response({'message': 'pack_id is required'}, status=status.HTTP_400_BAD_REQUEST) pack_obj = SupportArtistPacks.objects.filter(id=pack_id).first() if not pack_obj: return Response({'message': 'No artist pack found'}, status=status.HTTP_400_BAD_REQUEST) # Assuming Apple Pay token is sent instead of card_id apple_pay_token = request.data.get('apple_pay_token') if not apple_pay_token: return Response({'message': 'apple_pay_token is required'}, status=status.HTTP_400_BAD_REQUEST) # Function to process Apple Pay payment def process_apple_pay_payment(token, amount): stripe.api_key = settings.STRIPE_SECRET_KEY charge = stripe.Charge.create( amount=int(amount) * 100, # Convert amount to cents currency="usd", source=token, # 'source' instead of 'payment_method' description="Paying for artist pack" ) return charge # Process Apple Pay payment payment_response = process_apple_pay_payment(apple_pay_token, pack_obj.amount) if payment_response['status'] == 'succeeded': # Save transaction details CustTransaction = UserPacksTranscation.objects.create( transcation_id=generateTranscationsId(), user=_user, pack=pack_obj, amount=str(pack_obj.amount), txn_id=payment_response['id'], recipet_url=payment_response['receipt_url'], payment_mode=payment_response['payment_method_details']['type'], transcation_type="pay" ) return Response({'message': 'Success'},status=200) else: return Response({'message': payment_response.get('error', 'Payment failed')}, status=status.HTTP_400_BAD_REQUEST) class PurchaseArtistPackGpay(views.APIView): serializer_class = serializers.UserProfileSerializer permission_classes = [permissions.IsAuthenticated,] def post(self, request): user = request.user _user = User.objects.get(email=user.email) pack_id = request.data.get('pack_id') if not pack_id: return Response({'message': 'pack_id is required'}, status=status.HTTP_400_BAD_REQUEST) pack_obj = SupportArtistPacks.objects.filter(id=pack_id).first() if not pack_obj: return Response({'message': 'No artist pack found'}, status=status.HTTP_400_BAD_REQUEST) # Assuming Google Pay token is sent instead of card_id google_pay_token = request.data.get('google_pay_token') if not google_pay_token: return Response({'message': 'google_pay_token is required'}, status=status.HTTP_400_BAD_REQUEST) # Function to process Google Pay payment def process_google_pay_payment(token, amount): stripe.api_key = settings.STRIPE_SECRET_KEY charge = stripe.Charge.create( amount=int(amount) * 100, # Convert amount to cents currency="usd", source=token, # 'source' instead of 'payment_method' description="Paying for artist pack" ) return charge # Process Google Pay payment payment_response = process_google_pay_payment(google_pay_token, pack_obj.amount) if payment_response['status'] == 'succeeded': # Save transaction details CustTransaction = UserPacksTranscation.objects.create( transcation_id=generateTranscationsId(), user=_user, pack=pack_obj, amount=str(pack_obj.amount), txn_id=payment_response['id'], recipet_url=payment_response['receipt_url'], payment_mode=payment_response['payment_method_details']['type'], transcation_type="pay" ) return Response({'message': 'Success'}, status=200) else: return Response({'message': payment_response.get('error', 'Payment failed')}, status=status.HTTP_400_BAD_REQUEST)