diff --git a/fetch_from_api.py b/fetch_from_api.py new file mode 100644 index 0000000..c258251 --- /dev/null +++ b/fetch_from_api.py @@ -0,0 +1,120 @@ +import requests, os, re, json, sys, argparse + +argparser = argparse.ArgumentParser(description='Download streamer emotes from Twitch.') + +class TwApi: + session = requests.Session() + def __init__(self): + resp = self.session.get("https://www.twitch.tv/") + client_id = re.search('clientId="(.*?)",', resp.content.decode()).group(1) + self.client_id = client_id + # return client_id + + def gqlPlaybackAccessToken(self, channel: str) -> str: + 'input channel name, return channel_id' + resp = self.session.post( + url="https://gql.twitch.tv/gql", + json={ + "operationName": "PlaybackAccessToken_Template", + "query": 'query PlaybackAccessToken_Template($login: String!, $isLive: Boolean!, $vodID: ID!, $isVod: Boolean!, $playerType: String!, $platform: String!) { streamPlaybackAccessToken(channelName: $login, params: {platform: $platform, playerBackend: "mediaplayer", playerType: $playerType}) @include(if: $isLive) { value signature authorization { isForbidden forbiddenReasonCode } __typename } videoPlaybackAccessToken(id: $vodID, params: {platform: $platform, playerBackend: "mediaplayer", playerType: $playerType}) @include(if: $isVod) { value signature __typename }}', + "variables": { + "isLive": True, + "login": channel, + "isVod": False, + "vodID": "", + "playerType": "site", + "platform": "web", + }, + }, + headers={"Client-Id": self.client_id}, + ) + r_dict: dict = resp.json() + v = json.loads(r_dict['data']['streamPlaybackAccessToken']['value']) + r = v['channel_id'] + # print(r) + return r.__str__() + + + def gqlEmotePicker(self, channelOwnerID: str) -> dict: + 'return a dict like `{\'pewdiepieLegendBroFist\': \'emotesv2_b6e72807df1b4c78a0b70c8bb534b2fc\'}`' + resp = self.session.post( + url="https://gql.twitch.tv/gql", + json=[ + { + "operationName": "EmotePicker_EmotePicker_UserSubscriptionProducts", + "variables": {"channelOwnerID": channelOwnerID}, + "extensions": { + "persistedQuery": { + "version": 1, + "sha256Hash": "71b5f829a4576d53b714c01d3176f192cbd0b14973eb1c3d0ee23d5d1b78fd7e", + } + }, + } + ], + headers={"Client-Id": self.client_id}, + ) + resp_dict = resp.json() + r_dict = dict() + if resp_dict[0]['data']['channel']['localEmoteSets'] != None: + for emo_set in resp_dict[0]['data']['channel']['localEmoteSets']: + for emo in emo_set['emotes']: + r_dict[emo['token']] = emo['id'] + + for emo_set in resp_dict[0]['data']['user']['subscriptionProducts']: + for emo in emo_set['emotes']: + # if emo['assetType'] == 'ANIMATED': + # ext = '.gif' + # elif emo['assetType'] == 'STATIC': + # ext = '.png' + r_dict[emo['token']] = emo['id'] + + return r_dict + + def downloadEmote(self, filename: str, url: str): + resp = self.session.get(url) + ext = resp.headers['Content-Type'].split('/')[-1] + # if ('.png' not in filename) and ('.gif' not in filename): + # from PIL import Image + # from io import BytesIO + # img = Image.open(BytesIO(resp.content)) + # ext = img.format.lower() + filename += f'.{ext}' + with open(filename, mode='wb') as f: + f.write(resp.content) + print(f'download as {filename}') + + + def downloadEmotes(self, emote_dict: dict, max_workers: int = 20, dir: str = ''): + if not os.path.exists(dir): + os.mkdir(dir) + import concurrent.futures + executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) + thr_set = set() + for filename, url in emote_dict.items(): + thr = executor.submit( + self.downloadEmote, + filename=os.path.join(dir, filename), + url=f'https://static-cdn.jtvnw.net/emoticons/v2/{url}/default/light/3.0') + thr_set.add(thr) + executor.shutdown() + + for thr in thr_set: + thr: concurrent.futures.Future + e = thr.exception() + if e != None: + print(thr.result()) + +if __name__ == '__main__': + argparser.add_argument('-c', '--channel', dest='channel', help='Channel name.', required=True) + argparser.add_argument('-d', '--dir', dest='dir', help='Download files into this directory. (default: \'./emotes\', will create if not exist.)', default='./emotes') + argparser.add_argument('-p', '--proxy', dest='proxy', help='Use specified HTTP proxy server. (e.g. \'http://localhost:10809\', or you can set https_proxy or all_proxy in the terminal)', default='') + args = argparser.parse_args() + + if args.proxy != '': + os.environ["all_proxy"] = args.proxy + + api = TwApi() + channel_id = api.gqlPlaybackAccessToken(args.channel) + emote_dict = api.gqlEmotePicker(channel_id) + api.downloadEmotes(emote_dict=emote_dict, dir=args.dir) + # print(emote_dict) \ No newline at end of file