YukkiMusic/platforms/YoutubeAPI/InnertubeAPI.py

from urllib.request import urlopen
from urllib.parse import urlencode
import requests
import re
import json
import asyncio
import aiohttp
from .JSON_DATA import *

async def trending_with_post_api(query):
    #res = requests.get('https://www.youtube.com/')
    searchresult_list=[]

    key_value ={
        "now":"",
        "music":"4gINGgt5dG1hX2NoYXJ0cw%3D%3D",
        "gaming":"4gIcGhpnYW1pbmdfY29ycHVzX21vc3RfcG9wdWxhcg%3D%3D",
        "movies":"4gIKGgh0cmFpbGVycw%3D%3D",
    }
    
    final_query={
        "browseId": "FEtrending",
        "params": key_value.get(query),
        "context": {
            "client": {
                "hl":"en",
                "gl":"IN",
            "clientName": "WEB",
            "clientVersion": "2.20230728.00.00"
            
            }
        }
        }

    async with aiohttp.ClientSession() as session:
        async with session.post(
                'https://youtubei.googleapis.com/youtubei/v1/browse',
                json=final_query) as response:
            
            json_data= await response.text()
            json_data=json.loads(json_data)

    
    key_value_list=list(key_value.keys())
    print(key_value_list,key_value_list.index(query))
    all_search_video = json_data.get('contents', {}).get('twoColumnBrowseResultsRenderer',{}).get('tabs',[])[ key_value_list.index(query) ].get('tabRenderer',{}).get('content', {}).get('sectionListRenderer',{}).get('contents', [])[0].get('itemSectionRenderer',{}).get('contents', [])[0].get('shelfRenderer',{}).get('content',{}).get('expandedShelfContentsRenderer',{}).get('items',[])
    #.get('twoColumnSearchResultsRenderer',{}).get('tabs',[])
    #.get('sectionListRenderer',{}).get('contents', [])[0].get('itemSectionRenderer',{}).get('contents', [])[0].get('shelfRenderer',{}).get('content',{}).get('expandedShelfContentsRenderer',{}).get('items',[])
    #return all_search_video
    for video in all_search_video:
        if video.get('videoRenderer', {}):
            searchresult_list.append(JSON_videoRenderer_Formatter(video))

    if query == 'now':
        all_video_Now = json_data.get('contents', {}).get('twoColumnBrowseResultsRenderer',{}).get('tabs',[])[ key_value_list.index(query) ].get('tabRenderer',{}).get('content', {}).get('sectionListRenderer',{}).get('contents', [])[2].get('itemSectionRenderer',{}).get('contents', [])[0].get('shelfRenderer',{}).get('content',{}).get('expandedShelfContentsRenderer',{}).get('items',[])
        for video in all_video_Now:
            if video.get('videoRenderer', {}):
                searchresult_list.append(JSON_videoRenderer_Formatter(video))

    return searchresult_list
                



async def search_videos_with_post_api(query):
    searchresult_list=[]
    final_query={
                "query": query, 
                "key":"AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8",
                "context": {
                    "client": {
                        "clientName": "WEB",
                        "clientVersion": "2.20230728.00.00"
                    }
                }
                }
    async with aiohttp.ClientSession() as session:
        async with session.post(
                'https://youtubei.googleapis.com/youtubei/v1/search',
                json=final_query) as response:
            
            json_data= await response.text()
            json_data=json.loads(json_data)
    
    all_search_video = json_data.get('contents', {}).get(
    'twoColumnSearchResultsRenderer',
    {}).get('primaryContents',
            {}).get('sectionListRenderer',
                    {}).get('contents', [])[0].get('itemSectionRenderer',
                                                   {}).get('contents', [])
    
    nextPageToken = json_data['contents']['twoColumnSearchResultsRenderer']['primaryContents']['sectionListRenderer']['contents'][1]['continuationItemRenderer']['continuationEndpoint']['continuationCommand']['token'] if json_data.get('contents', {}).get(
    'twoColumnSearchResultsRenderer',
    {}).get('primaryContents',
            {}).get('sectionListRenderer',
                    {}).get('contents', [])[1].get('continuationItemRenderer',{}).get('continuationEndpoint',{}).get('continuationCommand',{}).get('token','') else None
    


    for video in all_search_video:
        if video.get('videoRenderer', {}):
            searchresult_list.append(JSON_videoRenderer_Formatter(video))

        elif video.get('childVideoRenderer', {}): 
            searchresult_list.append(JSON_childVideoRenderer_Formatter(video))

        elif video.get('gridVideoRenderer', {}):
            searchresult_list.append(JSON_childVideoRenderer_Formatter(video))

    #print(url,searchresult_list)
    return searchresult_list, nextPageToken

async def search_scroll_videos_with_post_api(query):
    searchresult_list=[]
    
    final_query={
        "continuation": query, 
        "key":"AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8", 
        "context": {
            "client": {
                "clientName": "WEB", 
                "clientVersion": "2.20230728.00.00"
                }
                }
            }

    async with aiohttp.ClientSession() as session:
        async with session.post(
                'https://youtubei.googleapis.com/youtubei/v1/search',
                json=final_query) as response:
            
            json_data= await response.text()
            json_data=json.loads(json_data)
    #return json_data
    all_search_video = json_data.get('onResponseReceivedCommands', [])[0].get(
    'appendContinuationItemsAction',
    {}).get('continuationItems',
            [])[0].get('itemSectionRenderer',
                    {}).get('contents', [])
    

    nextPageToken =json_data['onResponseReceivedCommands'][0]['appendContinuationItemsAction']['continuationItems'][1]['continuationItemRenderer']['continuationEndpoint']['continuationCommand']['token'] if len(json_data.get('onResponseReceivedCommands', [])) > 0 and json_data.get('onResponseReceivedCommands', [])[0].get(
    'appendContinuationItemsAction',
    {}).get('continuationItems',
            [])[1].get('continuationItemRenderer',{}).get('continuationEndpoint',{}).get('continuationCommand',{}).get('token','') else None
    

    for video in all_search_video:
        if video.get('videoRenderer', {}):
            searchresult_list.append(JSON_videoRenderer_Formatter(video))

        elif video.get('childVideoRenderer', {}): 
            searchresult_list.append(JSON_childVideoRenderer_Formatter(video))

        elif video.get('gridVideoRenderer', {}):
            searchresult_list.append(JSON_childVideoRenderer_Formatter(video))

    #print(url,searchresult_list)
    return searchresult_list, nextPageToken



#####-------------RETURN VIDEO LIST WITH STREAMING URL BY VIDEO ID------######
async def search_player_data_with_post_api(query):
    searchresult_list=[]
    
    
    final_query={
                    "context":{
                        "client":{"hl":"en",
                                  "gl":"IN",
                                  "deviceMake":"",
                                  "deviceModel":"",
                                  "userAgent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:132.0) Gecko/20100101 Firefox/132.0,gzip(gfe)",
                                  "clientName":"IOS",
                                  "clientVersion":"19.16.3",
                                  "screenPixelDensity":1,
                                  "timeZone":"Asia/Kolkata",
                                  "browserName":"Firefox",
                                  "browserVersion":"132.0",
                                  "acceptHeader":"text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                                  "deviceExperimentId":"ChxOelF6TkRNNU1UTTJPRE01T0RrNE9EZzVNQT09EI-HsbkGGI-HsbkG",
                                  "screenWidthPoints":1534,
                                  "screenHeightPoints":334,
                                  "utcOffsetMinutes":330,
                                  "clientScreen":"WATCH",
                                  "mainAppWebInfo":{
                                      "pwaInstallabilityStatus":"PWA_INSTALLABILITY_STATUS_UNKNOWN",
                                      "webDisplayMode":"WEB_DISPLAY_MODE_BROWSER","isWebNativeShareAvailable":"false"
                                      }
                                },
                                "user":{"lockedSafetyMode":"false"},
                                "request":{"useSsl":"true","internalExperimentFlags":[],"consistencyTokenJars":[]},
                                "clickTracking":{"clickTrackingParams":"CNQCENwwIhMI1emaz7LJiQMVN-lMAh3IOCkMMgpnLWhpZ2gtcmVjWg9GRXdoYXRfdG9fd2F0Y2iaAQYQjh4YngE="}
                            },
                            "videoId":query,
                            "params":"YAHIAQE%3D",
                            "playbackContext":{
                                "contentPlaybackContext":{
                                    "vis":5,
                                    "splay":"false",
                                    "autoCaptionsDefaultOn":"false",
                                    "autonavState":"STATE_NONE",
                                    "html5Preference":"HTML5_PREF_WANTS",
                                    "signatureTimestamp":20032,
                                    "autoplay":"true",
                                    "autonav":"true",
                                    "referer":"https://www.youtube.com/",
                                    "lactMilliseconds":"-1",
                                    "watchAmbientModeContext":{
                                        "hasShownAmbientMode":"true","watchAmbientModeEnabled":"true"}
                                        }
                                    },
                                    "racyCheckOk":"false",
                                    "contentCheckOk":"false"
                                    }

    async with aiohttp.ClientSession() as session:
        async with session.post(
                'https://youtubei.googleapis.com/youtubei/v1/player',
                json=final_query) as response:
            
            json_data= await response.text()
            json_data=json.loads(json_data)
    #return json_data
    all_formats_video = json_data.get('streamingData', {}).get(
    'formats',
    [])
    all_adaptive_formats_video = json_data.get('streamingData', {}).get(
    'adaptiveFormats',
    [])
    videodetails = json_data.get('videoDetails', {})


    for video in all_formats_video:
        
        searchresult_list.append(video)
    for video in all_adaptive_formats_video:
        
        searchresult_list.append(video)
        
    #print(searchresult_list, videodetails)
    return searchresult_list, videodetails