wldmr/transcriptifier-st-hf7

1import json2import requests3from tqdm import tqdm4import isodate5 6class YTstats:7 8    def __init__(self, api_key):9        self.api_key = api_key10        self.channel_statistics = None11        self.video_data = None12 13    def extract_all(self, channel_id):14        self.get_channel_statistics(channel_id)15        self.get_channel_video_data(channel_id)16 17    def get_channel_statistics(self, channel_id):18        """Extract the channel statistics"""19        print('get channel statistics...')20        url = f'https://www.googleapis.com/youtube/v3/channels?part=statistics&id={channel_id}&key={self.api_key}'21        #pbar = tqdm(total=1)22        23        json_url = requests.get(url)24        data = json.loads(json_url.text)25        try:26            data = data['items'][0]['statistics']27        except KeyError:28            print('Could not get channel statistics')29            data = {}30 31        self.channel_statistics = data32        #pbar.update()33        #pbar.close()34        return data35 36    def get_channel_video_data(self, channel_id, df_sheet, loading_bar, progress_text, item_limit=3):37        "Extract all video information of the channel"38        print('get video data...')39        channel_videos, channel_playlists = self._get_channel_content(channel_id, limit=50)40 41        channel_videos_out = dict()42 43        total_items = len(channel_videos)44        item = 045        step_size=046        step=047        if total_items!=0:48            step_size=round(1/total_items,4)49        #step = step_size50        parts=["snippet", "statistics","contentDetails", "topicDetails"]51        for video_id in tqdm(channel_videos):52            if item == item_limit:53                break54            55            loading_bar.progress(step, text=progress_text)56 57            for part in parts:58                data = self._get_single_video_data(video_id, part)59                channel_videos[video_id].update(data)60            61            duration = isodate.parse_duration(channel_videos[video_id]['duration'])62            short_duration = isodate.parse_duration('PT4M')63 64            if duration > short_duration and video_id not in list(df_sheet.ID):65                item = item+166                step = step +step_size67                channel_videos_out[video_id] = channel_videos[video_id]68                69 70        step=1.071        loading_bar.progress(step, text=progress_text)72        self.video_data = channel_videos_out73        74 75    def _get_single_video_data(self, video_id, part):76        """77        Extract further information for a single video78        parts can be: 'snippet', 'statistics', 'contentDetails', 'topicDetails'79        """80 81        url = f"https://www.googleapis.com/youtube/v3/videos?part={part}&id={video_id}&key={self.api_key}"82        json_url = requests.get(url)83        data = json.loads(json_url.text)84        try:85            data = data['items'][0][part]86        except KeyError as e:87            print(f'Error! Could not get {part} part of data: \n{data}')88            data = dict()89        return data90 91    def _get_channel_content(self, channel_id, limit=None, check_all_pages=True):92        """93        Extract all videos and playlists, can check all available search pages94        channel_videos = videoId: title, publishedAt95        channel_playlists = playlistId: title, publishedAt96        return channel_videos, channel_playlists97        """98        url = f"https://www.googleapis.com/youtube/v3/search?key={self.api_key}&channelId={channel_id}&part=snippet,id&order=date"99        if limit is not None and isinstance(limit, int):100            url += "&maxResults=" + str(limit)101 102        vid, pl, npt = self._get_channel_content_per_page(url)103        idx = 0104        while(check_all_pages and npt is not None and idx < 10):105            nexturl = url + "&pageToken=" + npt106            next_vid, next_pl, npt = self._get_channel_content_per_page(nexturl)107            vid.update(next_vid)108            pl.update(next_pl)109            idx += 1110 111        return vid, pl112 113    def _get_channel_content_per_page(self, url):114        """115        Extract all videos and playlists per page116        return channel_videos, channel_playlists, nextPageToken117        """118        json_url = requests.get(url)119        data = json.loads(json_url.text)120        channel_videos = dict()121        channel_playlists = dict()122        if 'items' not in data:123            print('Error! Could not get correct channel data!\n', data)124            return channel_videos, channel_videos, None125 126        nextPageToken = data.get("nextPageToken", None)127 128        item_data = data['items']129        for item in item_data:130            try:131                kind = item['id']['kind']132                published_at = item['snippet']['publishedAt']133                title = item['snippet']['title']134                if kind == 'youtube#video':135                    video_id = item['id']['videoId']136                    channel_videos[video_id] = {'publishedAt': published_at, 'title': title}137                elif kind == 'youtube#playlist':138                    playlist_id = item['id']['playlistId']139                    channel_playlists[playlist_id] = {'publishedAt': published_at, 'title': title}140            except KeyError as e:141                print('Error! Could not extract data from item:\n', item)142 143        return channel_videos, channel_playlists, nextPageToken144 145    def dump(self, channel_id):146        """Dumps channel statistics and video data in a single json file"""147        if self.channel_statistics is None or self.video_data is None:148            print('data is missing!\nCall get_channel_statistics() and get_channel_video_data() first!')149            return150 151        fused_data = {channel_id: {"channel_statistics": self.channel_statistics,152                              "video_data": self.video_data}}153 154        channel_title = self.video_data.popitem()[1].get('channelTitle', channel_id)155        channel_title = channel_title.replace(" ", "_").lower()156        filename = channel_title + '.json'157        with open(filename, 'w') as f:158            json.dump(fused_data, f, indent=4)159        160        print('file dumped to', filename)