TikTokApi.tiktok
View Source
import json import logging import os import threading import asyncio import random import string import time from typing import ClassVar, Optional from urllib import request from urllib.parse import quote, urlencode import requests from .api.sound import Sound from .api.user import User from .api.search import Search from .api.hashtag import Hashtag from .api.video import Video from .api.trending import Trending from playwright.sync_api import sync_playwright from .exceptions import * from .utilities import LOGGER_NAME, update_messager from .browser_utilities.browser import browser from dataclasses import dataclass os.environ["no_proxy"] = "127.0.0.1,localhost" BASE_URL = "https://m.tiktok.com/" DESKTOP_BASE_URL = "https://www.tiktok.com/" _thread_lock = threading.Lock() class TikTokApi: _is_context_manager = False user = User search = Search sound = Sound hashtag = Hashtag video = Video trending = Trending logger = logging.getLogger(LOGGER_NAME) def __init__( self, logging_level: int = logging.WARNING, request_delay: Optional[int] = None, custom_device_id: Optional[str] = None, generate_static_device_id: Optional[bool] = False, custom_verify_fp: Optional[str] = None, use_test_endpoints: Optional[bool] = False, proxy: Optional[str] = None, executable_path: Optional[str] = None, *args, **kwargs, ): """The TikTokApi class. Used to interact with TikTok. This is a singleton class to prevent issues from arising with playwright ##### Parameters * logging_level: The logging level you want the program to run at, optional These are the standard python logging module's levels. * request_delay: The amount of time in seconds to wait before making a request, optional This is used to throttle your own requests as you may end up making too many requests to TikTok for your IP. * custom_device_id: A TikTok parameter needed to download videos, optional The code generates these and handles these pretty well itself, however for some things such as video download you will need to set a consistent one of these. All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. * generate_static_device_id: A parameter that generates a custom_device_id at the instance level Use this if you want to download videos from a script but don't want to generate your own custom_device_id parameter. * custom_verify_fp: A TikTok parameter needed to work most of the time, optional To get this parameter look at [this video](https://youtu.be/zwLmLfVI-VQ?t=117) I recommend watching the entire thing, as it will help setup this package. All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. You can use the following to generate `"".join(random.choice(string.digits) for num in range(19))` * use_test_endpoints: Send requests to TikTok's test endpoints, optional This parameter when set to true will make requests to TikTok's testing endpoints instead of the live site. I can't guarantee this will work in the future, however currently basically any custom_verify_fp will work here which is helpful. * proxy: A string containing your proxy address, optional If you want to do a lot of scraping of TikTok endpoints you'll likely need a proxy. Ex: "https://0.0.0.0:8080" All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. * executable_path: The location of the driver, optional This shouldn't be needed if you're using playwright * **kwargs Parameters that are passed on to basically every module and methods that interact with this main class. These may or may not be documented in other places. """ self.logger.setLevel(logging_level) with _thread_lock: self._initialize( request_delay=request_delay, custom_device_id=custom_device_id, generate_static_device_id=generate_static_device_id, custom_verify_fp=custom_verify_fp, use_test_endpoints=use_test_endpoints, proxy=proxy, executable_path=executable_path, *args, **kwargs, ) def _initialize(self, **kwargs): # Add classes from the api folder User.parent = self Search.parent = self Sound.parent = self Hashtag.parent = self Video.parent = self Trending.parent = self # Some Instance Vars self._executable_path = kwargs.get("executable_path", None) if kwargs.get("custom_did") != None: raise Exception("Please use 'custom_device_id' instead of 'custom_did'") self._custom_device_id = kwargs.get("custom_device_id", None) self._user_agent = "5.0 (iPhone; CPU iPhone OS 14_8 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.2 Mobile/15E148 Safari/604.1" self._proxy = kwargs.get("proxy", None) self._custom_verify_fp = kwargs.get("custom_verify_fp") self._signer_url = kwargs.get("external_signer", None) self._request_delay = kwargs.get("request_delay", None) self._requests_extra_kwargs = kwargs.get("requests_extra_kwargs", {}) if kwargs.get("use_test_endpoints", False): global BASE_URL BASE_URL = "https://t.tiktok.com/" if kwargs.get("generate_static_device_id", False): self._custom_device_id = "".join( random.choice(string.digits) for num in range(19) ) if self._signer_url is None: self._browser = asyncio.get_event_loop().run_until_complete( asyncio.gather(browser.create(**kwargs)) )[0] self._user_agent = self._browser.user_agent try: self._timezone_name = self._browser.timezone_name self._browser_language = self._browser.browser_language self._width = self._browser.width self._height = self._browser.height self._region = self._browser.region self._language = self._browser.language except Exception as e: self.logger.exception( "An error occurred while opening your browser, but it was ignored\n", "Are you sure you ran python -m playwright install?", ) self._timezone_name = "" self._browser_language = "" self._width = "1920" self._height = "1080" self._region = "US" self._language = "en" raise e from e def get_data(self, path, subdomain="m", **kwargs) -> dict: """Makes requests to TikTok and returns their JSON. This is all handled by the package so it's unlikely you will need to use this. """ processed = self._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if self._request_delay is not None: time.sleep(self._request_delay) if self._proxy is not None: proxy = self._proxy if kwargs.get("custom_verify_fp") == None: if self._custom_verify_fp != None: verifyFp = self._custom_verify_fp else: verifyFp = "verify_khr3jabg_V7ucdslq_Vrw9_4KPb_AJ1b_Ks706M8zIJTq" else: verifyFp = kwargs.get("custom_verify_fp") tt_params = None send_tt_params = kwargs.get("send_tt_params", False) full_url = f"https://{subdomain}.tiktok.com/" + path if self._signer_url is None: kwargs["custom_verify_fp"] = verifyFp ( verify_fp, device_id, signature, tt_params, ) = asyncio.get_event_loop().run_until_complete( asyncio.gather( self._browser.sign_url( full_url, calc_tt_params=send_tt_params, **kwargs ) ) )[ 0 ] user_agent = self._browser.user_agent referrer = self._browser.referrer else: ( verify_fp, device_id, signature, user_agent, referrer, ) = self.external_signer( full_url, custom_device_id=kwargs.get("custom_device_id"), verifyFp=kwargs.get("custom_verify_fp", verifyFp), ) if not kwargs.get("send_tt_params", False): tt_params = None query = {"verifyFp": verify_fp, "device_id": device_id, "_signature": signature} url = "{}&{}".format(full_url, urlencode(query)) h = requests.head( url, headers={"x-secsdk-csrf-version": "1.2.5", "x-secsdk-csrf-request": "1"}, proxies=self._format_proxy(processed.proxy), **self._requests_extra_kwargs, ) csrf_token = None if subdomain == "m": csrf_session_id = h.cookies["csrf_session_id"] csrf_token = h.headers["X-Ware-Csrf-Token"].split(",")[1] kwargs["csrf_session_id"] = csrf_session_id headers = { "authority": "m.tiktok.com", "method": "GET", "path": url.split("tiktok.com")[1], "scheme": "https", "accept": "application/json, text/plain, */*", "accept-encoding": "gzip", "accept-language": "en-US,en;q=0.9", "origin": referrer, "referer": referrer, "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "none", "sec-gpc": "1", "user-agent": user_agent, "x-secsdk-csrf-token": csrf_token, "x-tt-params": tt_params, } self.logger.debug(f"GET: %s\n\theaders: %s", url, headers) r = requests.get( url, headers=headers, cookies=self._get_cookies(**kwargs), proxies=self._format_proxy(processed.proxy), **self._requests_extra_kwargs, ) try: parsed_data = r.json() if ( parsed_data.get("type") == "verify" or parsed_data.get("verifyConfig", {}).get("type", "") == "verify" ): self.logger.error( "Tiktok wants to display a captcha.\nResponse:\n%s\nCookies:\n%s\nURL:\n%s", r.text, self._get_cookies(**kwargs), url, ) raise CaptchaException( "TikTok blocks this request displaying a Captcha \nTip: Consider using a proxy or a custom_verify_fp as method parameters" ) # statusCode from props->pageProps->statusCode thanks @adiantek on #403 error_codes = { "0": "OK", "450": "CLIENT_PAGE_ERROR", "10000": "VERIFY_CODE", "10101": "SERVER_ERROR_NOT_500", "10102": "USER_NOT_LOGIN", "10111": "NET_ERROR", "10113": "SHARK_SLIDE", "10114": "SHARK_BLOCK", "10119": "LIVE_NEED_LOGIN", "10202": "USER_NOT_EXIST", "10203": "MUSIC_NOT_EXIST", "10204": "VIDEO_NOT_EXIST", "10205": "HASHTAG_NOT_EXIST", "10208": "EFFECT_NOT_EXIST", "10209": "HASHTAG_BLACK_LIST", "10210": "LIVE_NOT_EXIST", "10211": "HASHTAG_SENSITIVITY_WORD", "10212": "HASHTAG_UNSHELVE", "10213": "VIDEO_LOW_AGE_M", "10214": "VIDEO_LOW_AGE_T", "10215": "VIDEO_ABNORMAL", "10216": "VIDEO_PRIVATE_BY_USER", "10217": "VIDEO_FIRST_REVIEW_UNSHELVE", "10218": "MUSIC_UNSHELVE", "10219": "MUSIC_NO_COPYRIGHT", "10220": "VIDEO_UNSHELVE_BY_MUSIC", "10221": "USER_BAN", "10222": "USER_PRIVATE", "10223": "USER_FTC", "10224": "GAME_NOT_EXIST", "10225": "USER_UNIQUE_SENSITIVITY", "10227": "VIDEO_NEED_RECHECK", "10228": "VIDEO_RISK", "10229": "VIDEO_R_MASK", "10230": "VIDEO_RISK_MASK", "10231": "VIDEO_GEOFENCE_BLOCK", "10404": "FYP_VIDEO_LIST_LIMIT", "undefined": "MEDIA_ERROR", } statusCode = parsed_data.get("statusCode", 0) self.logger.debug(f"TikTok Returned: %s", json) if statusCode == 10201: # Invalid Entity raise NotFoundException( "TikTok returned a response indicating the entity is invalid" ) elif statusCode == 10219: # Not available in this region raise NotAvailableException("Content not available for this region") elif statusCode != 0 and statusCode != -1: raise TikTokException( error_codes.get( statusCode, f"TikTok sent an unknown StatusCode of {statusCode}" ) ) return r.json() except ValueError as e: text = r.text self.logger.debug("TikTok response: %s", text) if len(text) == 0: raise EmptyResponseException( "Empty response from Tiktok to " + url ) from None else: raise InvalidJSONException("TikTok sent invalid JSON") from e def __del__(self): """A basic cleanup method, called automatically from the code""" if not self._is_context_manager: self.logger.debug( "TikTokAPI was shutdown improperlly. Ensure the instance is terminated with .shutdown()" ) self.shutdown() return def external_signer(self, url, custom_device_id=None, verifyFp=None): """Makes requests to an external signer instead of using a browser. ##### Parameters * url: The server to make requests to This server is designed to sign requests. You can find an example of this signature server in the examples folder. * custom_device_id: A TikTok parameter needed to download videos The code generates these and handles these pretty well itself, however for some things such as video download you will need to set a consistent one of these. * custom_verify_fp: A TikTok parameter needed to work most of the time, To get this parameter look at [this video](https://youtu.be/zwLmLfVI-VQ?t=117) I recommend watching the entire thing, as it will help setup this package. """ if custom_device_id is not None: query = { "url": url, "custom_device_id": custom_device_id, "verifyFp": verifyFp, } else: query = {"url": url, "verifyFp": verifyFp} data = requests.get( self._signer_url + "?{}".format(urlencode(query)), **self._requests_extra_kwargs, ) parsed_data = data.json() return ( parsed_data["verifyFp"], parsed_data["device_id"], parsed_data["_signature"], parsed_data["user_agent"], parsed_data["referrer"], ) def _get_cookies(self, **kwargs): """Extracts cookies from the kwargs passed to the function for get_data""" device_id = kwargs.get( "custom_device_id", "".join(random.choice(string.digits) for num in range(19)), ) if kwargs.get("custom_verify_fp") is None: if self._custom_verify_fp is not None: verifyFp = self._custom_verify_fp else: verifyFp = None else: verifyFp = kwargs.get("custom_verify_fp") if kwargs.get("force_verify_fp_on_cookie_header", False): return { "tt_webid": device_id, "tt_webid_v2": device_id, "csrf_session_id": kwargs.get("csrf_session_id"), "tt_csrf_token": "".join( random.choice(string.ascii_uppercase + string.ascii_lowercase) for i in range(16) ), "s_v_web_id": verifyFp, "ttwid": kwargs.get("ttwid"), } else: return { "tt_webid": device_id, "tt_webid_v2": device_id, "csrf_session_id": kwargs.get("csrf_session_id"), "tt_csrf_token": "".join( random.choice(string.ascii_uppercase + string.ascii_lowercase) for i in range(16) ), "ttwid": kwargs.get("ttwid"), } def get_bytes(self, **kwargs) -> bytes: """Returns TikTok's response as bytes, similar to get_data""" processed = self._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if self._signer_url is None: ( verify_fp, device_id, signature, _, ) = asyncio.get_event_loop().run_until_complete( asyncio.gather(self._browser.sign_url(calc_tt_params=False, **kwargs)) )[ 0 ] user_agent = self._browser.user_agent referrer = self._browser.referrer else: ( verify_fp, device_id, signature, user_agent, referrer, ) = self.external_signer( kwargs["url"], custom_device_id=kwargs.get("custom_device_id", None) ) query = {"verifyFp": verify_fp, "_signature": signature} url = "{}&{}".format(kwargs["url"], urlencode(query)) r = requests.get( url, headers={ "Accept": "*/*", "Accept-Encoding": "identity;q=1, *;q=0", "Accept-Language": "en-US;en;q=0.9", "Cache-Control": "no-cache", "Connection": "keep-alive", "Host": url.split("/")[2], "Pragma": "no-cache", "Range": "bytes=0-", "Referer": "https://www.tiktok.com/", "User-Agent": user_agent, }, proxies=self._format_proxy(processed.proxy), cookies=self._get_cookies(**kwargs), ) return r.content @staticmethod def generate_device_id(): """Generates a valid device_id for other methods. Pass this as the custom_device_id field to download videos""" return "".join([random.choice(string.digits) for num in range(19)]) # # PRIVATE METHODS # def _format_proxy(self, proxy) -> Optional[dict]: """ Formats the proxy object """ if proxy is None and self._proxy is not None: proxy = self._proxy if proxy is not None: return {"http": proxy, "https": proxy} else: return None # Process the kwargs def _process_kwargs(self, kwargs): region = kwargs.get("region", "US") language = kwargs.get("language", "en") proxy = kwargs.get("proxy", None) if kwargs.get("custom_device_id", None) != None: device_id = kwargs.get("custom_device_id") else: if self._custom_device_id != None: device_id = self._custom_device_id else: device_id = "".join(random.choice(string.digits) for num in range(19)) @dataclass class ProcessedKwargs: region: str language: str proxy: str device_id: str return ProcessedKwargs( region=region, language=language, proxy=proxy, device_id=device_id ) def _add_url_params(self) -> str: try: region = self._region browser_language = self._browser_language.lower() timezone = self._timezone_name language = self._language except AttributeError as e: self.logger.debug("Attribute Error on add_url_params", exc_info=e) region = "US" browser_language = "en-us" timezone = "America/Chicago" language = "en" query = { "aid": 1988, "app_name": "tiktok_web", "device_platform": "web_mobile", "region": region, "priority_region": "", "os": "ios", "referer": "", "cookie_enabled": "true", "screen_width": self._width, "screen_height": self._height, "browser_language": browser_language, "browser_platform": "iPhone", "browser_name": "Mozilla", "browser_version": self._user_agent, "browser_online": "true", "timezone_name": timezone, "is_page_visible": "true", "focus_state": "true", "is_fullscreen": "false", "history_len": random.randint(1, 5), "language": language, } return urlencode(query) def shutdown(self) -> None: with _thread_lock: self.logger.debug("Shutting down Playwright") asyncio.get_event_loop().run_until_complete(self._browser._clean_up()) def __enter__(self): with _thread_lock: self._is_context_manager = True return self def __exit__(self, type, value, traceback): self.shutdown()
View Source
class TikTokApi: _is_context_manager = False user = User search = Search sound = Sound hashtag = Hashtag video = Video trending = Trending logger = logging.getLogger(LOGGER_NAME) def __init__( self, logging_level: int = logging.WARNING, request_delay: Optional[int] = None, custom_device_id: Optional[str] = None, generate_static_device_id: Optional[bool] = False, custom_verify_fp: Optional[str] = None, use_test_endpoints: Optional[bool] = False, proxy: Optional[str] = None, executable_path: Optional[str] = None, *args, **kwargs, ): """The TikTokApi class. Used to interact with TikTok. This is a singleton class to prevent issues from arising with playwright ##### Parameters * logging_level: The logging level you want the program to run at, optional These are the standard python logging module's levels. * request_delay: The amount of time in seconds to wait before making a request, optional This is used to throttle your own requests as you may end up making too many requests to TikTok for your IP. * custom_device_id: A TikTok parameter needed to download videos, optional The code generates these and handles these pretty well itself, however for some things such as video download you will need to set a consistent one of these. All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. * generate_static_device_id: A parameter that generates a custom_device_id at the instance level Use this if you want to download videos from a script but don't want to generate your own custom_device_id parameter. * custom_verify_fp: A TikTok parameter needed to work most of the time, optional To get this parameter look at [this video](https://youtu.be/zwLmLfVI-VQ?t=117) I recommend watching the entire thing, as it will help setup this package. All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. You can use the following to generate `"".join(random.choice(string.digits) for num in range(19))` * use_test_endpoints: Send requests to TikTok's test endpoints, optional This parameter when set to true will make requests to TikTok's testing endpoints instead of the live site. I can't guarantee this will work in the future, however currently basically any custom_verify_fp will work here which is helpful. * proxy: A string containing your proxy address, optional If you want to do a lot of scraping of TikTok endpoints you'll likely need a proxy. Ex: "https://0.0.0.0:8080" All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. * executable_path: The location of the driver, optional This shouldn't be needed if you're using playwright * **kwargs Parameters that are passed on to basically every module and methods that interact with this main class. These may or may not be documented in other places. """ self.logger.setLevel(logging_level) with _thread_lock: self._initialize( request_delay=request_delay, custom_device_id=custom_device_id, generate_static_device_id=generate_static_device_id, custom_verify_fp=custom_verify_fp, use_test_endpoints=use_test_endpoints, proxy=proxy, executable_path=executable_path, *args, **kwargs, ) def _initialize(self, **kwargs): # Add classes from the api folder User.parent = self Search.parent = self Sound.parent = self Hashtag.parent = self Video.parent = self Trending.parent = self # Some Instance Vars self._executable_path = kwargs.get("executable_path", None) if kwargs.get("custom_did") != None: raise Exception("Please use 'custom_device_id' instead of 'custom_did'") self._custom_device_id = kwargs.get("custom_device_id", None) self._user_agent = "5.0 (iPhone; CPU iPhone OS 14_8 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.2 Mobile/15E148 Safari/604.1" self._proxy = kwargs.get("proxy", None) self._custom_verify_fp = kwargs.get("custom_verify_fp") self._signer_url = kwargs.get("external_signer", None) self._request_delay = kwargs.get("request_delay", None) self._requests_extra_kwargs = kwargs.get("requests_extra_kwargs", {}) if kwargs.get("use_test_endpoints", False): global BASE_URL BASE_URL = "https://t.tiktok.com/" if kwargs.get("generate_static_device_id", False): self._custom_device_id = "".join( random.choice(string.digits) for num in range(19) ) if self._signer_url is None: self._browser = asyncio.get_event_loop().run_until_complete( asyncio.gather(browser.create(**kwargs)) )[0] self._user_agent = self._browser.user_agent try: self._timezone_name = self._browser.timezone_name self._browser_language = self._browser.browser_language self._width = self._browser.width self._height = self._browser.height self._region = self._browser.region self._language = self._browser.language except Exception as e: self.logger.exception( "An error occurred while opening your browser, but it was ignored\n", "Are you sure you ran python -m playwright install?", ) self._timezone_name = "" self._browser_language = "" self._width = "1920" self._height = "1080" self._region = "US" self._language = "en" raise e from e def get_data(self, path, subdomain="m", **kwargs) -> dict: """Makes requests to TikTok and returns their JSON. This is all handled by the package so it's unlikely you will need to use this. """ processed = self._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if self._request_delay is not None: time.sleep(self._request_delay) if self._proxy is not None: proxy = self._proxy if kwargs.get("custom_verify_fp") == None: if self._custom_verify_fp != None: verifyFp = self._custom_verify_fp else: verifyFp = "verify_khr3jabg_V7ucdslq_Vrw9_4KPb_AJ1b_Ks706M8zIJTq" else: verifyFp = kwargs.get("custom_verify_fp") tt_params = None send_tt_params = kwargs.get("send_tt_params", False) full_url = f"https://{subdomain}.tiktok.com/" + path if self._signer_url is None: kwargs["custom_verify_fp"] = verifyFp ( verify_fp, device_id, signature, tt_params, ) = asyncio.get_event_loop().run_until_complete( asyncio.gather( self._browser.sign_url( full_url, calc_tt_params=send_tt_params, **kwargs ) ) )[ 0 ] user_agent = self._browser.user_agent referrer = self._browser.referrer else: ( verify_fp, device_id, signature, user_agent, referrer, ) = self.external_signer( full_url, custom_device_id=kwargs.get("custom_device_id"), verifyFp=kwargs.get("custom_verify_fp", verifyFp), ) if not kwargs.get("send_tt_params", False): tt_params = None query = {"verifyFp": verify_fp, "device_id": device_id, "_signature": signature} url = "{}&{}".format(full_url, urlencode(query)) h = requests.head( url, headers={"x-secsdk-csrf-version": "1.2.5", "x-secsdk-csrf-request": "1"}, proxies=self._format_proxy(processed.proxy), **self._requests_extra_kwargs, ) csrf_token = None if subdomain == "m": csrf_session_id = h.cookies["csrf_session_id"] csrf_token = h.headers["X-Ware-Csrf-Token"].split(",")[1] kwargs["csrf_session_id"] = csrf_session_id headers = { "authority": "m.tiktok.com", "method": "GET", "path": url.split("tiktok.com")[1], "scheme": "https", "accept": "application/json, text/plain, */*", "accept-encoding": "gzip", "accept-language": "en-US,en;q=0.9", "origin": referrer, "referer": referrer, "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "none", "sec-gpc": "1", "user-agent": user_agent, "x-secsdk-csrf-token": csrf_token, "x-tt-params": tt_params, } self.logger.debug(f"GET: %s\n\theaders: %s", url, headers) r = requests.get( url, headers=headers, cookies=self._get_cookies(**kwargs), proxies=self._format_proxy(processed.proxy), **self._requests_extra_kwargs, ) try: parsed_data = r.json() if ( parsed_data.get("type") == "verify" or parsed_data.get("verifyConfig", {}).get("type", "") == "verify" ): self.logger.error( "Tiktok wants to display a captcha.\nResponse:\n%s\nCookies:\n%s\nURL:\n%s", r.text, self._get_cookies(**kwargs), url, ) raise CaptchaException( "TikTok blocks this request displaying a Captcha \nTip: Consider using a proxy or a custom_verify_fp as method parameters" ) # statusCode from props->pageProps->statusCode thanks @adiantek on #403 error_codes = { "0": "OK", "450": "CLIENT_PAGE_ERROR", "10000": "VERIFY_CODE", "10101": "SERVER_ERROR_NOT_500", "10102": "USER_NOT_LOGIN", "10111": "NET_ERROR", "10113": "SHARK_SLIDE", "10114": "SHARK_BLOCK", "10119": "LIVE_NEED_LOGIN", "10202": "USER_NOT_EXIST", "10203": "MUSIC_NOT_EXIST", "10204": "VIDEO_NOT_EXIST", "10205": "HASHTAG_NOT_EXIST", "10208": "EFFECT_NOT_EXIST", "10209": "HASHTAG_BLACK_LIST", "10210": "LIVE_NOT_EXIST", "10211": "HASHTAG_SENSITIVITY_WORD", "10212": "HASHTAG_UNSHELVE", "10213": "VIDEO_LOW_AGE_M", "10214": "VIDEO_LOW_AGE_T", "10215": "VIDEO_ABNORMAL", "10216": "VIDEO_PRIVATE_BY_USER", "10217": "VIDEO_FIRST_REVIEW_UNSHELVE", "10218": "MUSIC_UNSHELVE", "10219": "MUSIC_NO_COPYRIGHT", "10220": "VIDEO_UNSHELVE_BY_MUSIC", "10221": "USER_BAN", "10222": "USER_PRIVATE", "10223": "USER_FTC", "10224": "GAME_NOT_EXIST", "10225": "USER_UNIQUE_SENSITIVITY", "10227": "VIDEO_NEED_RECHECK", "10228": "VIDEO_RISK", "10229": "VIDEO_R_MASK", "10230": "VIDEO_RISK_MASK", "10231": "VIDEO_GEOFENCE_BLOCK", "10404": "FYP_VIDEO_LIST_LIMIT", "undefined": "MEDIA_ERROR", } statusCode = parsed_data.get("statusCode", 0) self.logger.debug(f"TikTok Returned: %s", json) if statusCode == 10201: # Invalid Entity raise NotFoundException( "TikTok returned a response indicating the entity is invalid" ) elif statusCode == 10219: # Not available in this region raise NotAvailableException("Content not available for this region") elif statusCode != 0 and statusCode != -1: raise TikTokException( error_codes.get( statusCode, f"TikTok sent an unknown StatusCode of {statusCode}" ) ) return r.json() except ValueError as e: text = r.text self.logger.debug("TikTok response: %s", text) if len(text) == 0: raise EmptyResponseException( "Empty response from Tiktok to " + url ) from None else: raise InvalidJSONException("TikTok sent invalid JSON") from e def __del__(self): """A basic cleanup method, called automatically from the code""" if not self._is_context_manager: self.logger.debug( "TikTokAPI was shutdown improperlly. Ensure the instance is terminated with .shutdown()" ) self.shutdown() return def external_signer(self, url, custom_device_id=None, verifyFp=None): """Makes requests to an external signer instead of using a browser. ##### Parameters * url: The server to make requests to This server is designed to sign requests. You can find an example of this signature server in the examples folder. * custom_device_id: A TikTok parameter needed to download videos The code generates these and handles these pretty well itself, however for some things such as video download you will need to set a consistent one of these. * custom_verify_fp: A TikTok parameter needed to work most of the time, To get this parameter look at [this video](https://youtu.be/zwLmLfVI-VQ?t=117) I recommend watching the entire thing, as it will help setup this package. """ if custom_device_id is not None: query = { "url": url, "custom_device_id": custom_device_id, "verifyFp": verifyFp, } else: query = {"url": url, "verifyFp": verifyFp} data = requests.get( self._signer_url + "?{}".format(urlencode(query)), **self._requests_extra_kwargs, ) parsed_data = data.json() return ( parsed_data["verifyFp"], parsed_data["device_id"], parsed_data["_signature"], parsed_data["user_agent"], parsed_data["referrer"], ) def _get_cookies(self, **kwargs): """Extracts cookies from the kwargs passed to the function for get_data""" device_id = kwargs.get( "custom_device_id", "".join(random.choice(string.digits) for num in range(19)), ) if kwargs.get("custom_verify_fp") is None: if self._custom_verify_fp is not None: verifyFp = self._custom_verify_fp else: verifyFp = None else: verifyFp = kwargs.get("custom_verify_fp") if kwargs.get("force_verify_fp_on_cookie_header", False): return { "tt_webid": device_id, "tt_webid_v2": device_id, "csrf_session_id": kwargs.get("csrf_session_id"), "tt_csrf_token": "".join( random.choice(string.ascii_uppercase + string.ascii_lowercase) for i in range(16) ), "s_v_web_id": verifyFp, "ttwid": kwargs.get("ttwid"), } else: return { "tt_webid": device_id, "tt_webid_v2": device_id, "csrf_session_id": kwargs.get("csrf_session_id"), "tt_csrf_token": "".join( random.choice(string.ascii_uppercase + string.ascii_lowercase) for i in range(16) ), "ttwid": kwargs.get("ttwid"), } def get_bytes(self, **kwargs) -> bytes: """Returns TikTok's response as bytes, similar to get_data""" processed = self._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if self._signer_url is None: ( verify_fp, device_id, signature, _, ) = asyncio.get_event_loop().run_until_complete( asyncio.gather(self._browser.sign_url(calc_tt_params=False, **kwargs)) )[ 0 ] user_agent = self._browser.user_agent referrer = self._browser.referrer else: ( verify_fp, device_id, signature, user_agent, referrer, ) = self.external_signer( kwargs["url"], custom_device_id=kwargs.get("custom_device_id", None) ) query = {"verifyFp": verify_fp, "_signature": signature} url = "{}&{}".format(kwargs["url"], urlencode(query)) r = requests.get( url, headers={ "Accept": "*/*", "Accept-Encoding": "identity;q=1, *;q=0", "Accept-Language": "en-US;en;q=0.9", "Cache-Control": "no-cache", "Connection": "keep-alive", "Host": url.split("/")[2], "Pragma": "no-cache", "Range": "bytes=0-", "Referer": "https://www.tiktok.com/", "User-Agent": user_agent, }, proxies=self._format_proxy(processed.proxy), cookies=self._get_cookies(**kwargs), ) return r.content @staticmethod def generate_device_id(): """Generates a valid device_id for other methods. Pass this as the custom_device_id field to download videos""" return "".join([random.choice(string.digits) for num in range(19)]) # # PRIVATE METHODS # def _format_proxy(self, proxy) -> Optional[dict]: """ Formats the proxy object """ if proxy is None and self._proxy is not None: proxy = self._proxy if proxy is not None: return {"http": proxy, "https": proxy} else: return None # Process the kwargs def _process_kwargs(self, kwargs): region = kwargs.get("region", "US") language = kwargs.get("language", "en") proxy = kwargs.get("proxy", None) if kwargs.get("custom_device_id", None) != None: device_id = kwargs.get("custom_device_id") else: if self._custom_device_id != None: device_id = self._custom_device_id else: device_id = "".join(random.choice(string.digits) for num in range(19)) @dataclass class ProcessedKwargs: region: str language: str proxy: str device_id: str return ProcessedKwargs( region=region, language=language, proxy=proxy, device_id=device_id ) def _add_url_params(self) -> str: try: region = self._region browser_language = self._browser_language.lower() timezone = self._timezone_name language = self._language except AttributeError as e: self.logger.debug("Attribute Error on add_url_params", exc_info=e) region = "US" browser_language = "en-us" timezone = "America/Chicago" language = "en" query = { "aid": 1988, "app_name": "tiktok_web", "device_platform": "web_mobile", "region": region, "priority_region": "", "os": "ios", "referer": "", "cookie_enabled": "true", "screen_width": self._width, "screen_height": self._height, "browser_language": browser_language, "browser_platform": "iPhone", "browser_name": "Mozilla", "browser_version": self._user_agent, "browser_online": "true", "timezone_name": timezone, "is_page_visible": "true", "focus_state": "true", "is_fullscreen": "false", "history_len": random.randint(1, 5), "language": language, } return urlencode(query) def shutdown(self) -> None: with _thread_lock: self.logger.debug("Shutting down Playwright") asyncio.get_event_loop().run_until_complete(self._browser._clean_up()) def __enter__(self): with _thread_lock: self._is_context_manager = True return self def __exit__(self, type, value, traceback): self.shutdown()
View Source
def __init__( self, logging_level: int = logging.WARNING, request_delay: Optional[int] = None, custom_device_id: Optional[str] = None, generate_static_device_id: Optional[bool] = False, custom_verify_fp: Optional[str] = None, use_test_endpoints: Optional[bool] = False, proxy: Optional[str] = None, executable_path: Optional[str] = None, *args, **kwargs, ): """The TikTokApi class. Used to interact with TikTok. This is a singleton class to prevent issues from arising with playwright ##### Parameters * logging_level: The logging level you want the program to run at, optional These are the standard python logging module's levels. * request_delay: The amount of time in seconds to wait before making a request, optional This is used to throttle your own requests as you may end up making too many requests to TikTok for your IP. * custom_device_id: A TikTok parameter needed to download videos, optional The code generates these and handles these pretty well itself, however for some things such as video download you will need to set a consistent one of these. All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. * generate_static_device_id: A parameter that generates a custom_device_id at the instance level Use this if you want to download videos from a script but don't want to generate your own custom_device_id parameter. * custom_verify_fp: A TikTok parameter needed to work most of the time, optional To get this parameter look at [this video](https://youtu.be/zwLmLfVI-VQ?t=117) I recommend watching the entire thing, as it will help setup this package. All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. You can use the following to generate `"".join(random.choice(string.digits) for num in range(19))` * use_test_endpoints: Send requests to TikTok's test endpoints, optional This parameter when set to true will make requests to TikTok's testing endpoints instead of the live site. I can't guarantee this will work in the future, however currently basically any custom_verify_fp will work here which is helpful. * proxy: A string containing your proxy address, optional If you want to do a lot of scraping of TikTok endpoints you'll likely need a proxy. Ex: "https://0.0.0.0:8080" All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods. * executable_path: The location of the driver, optional This shouldn't be needed if you're using playwright * **kwargs Parameters that are passed on to basically every module and methods that interact with this main class. These may or may not be documented in other places. """ self.logger.setLevel(logging_level) with _thread_lock: self._initialize( request_delay=request_delay, custom_device_id=custom_device_id, generate_static_device_id=generate_static_device_id, custom_verify_fp=custom_verify_fp, use_test_endpoints=use_test_endpoints, proxy=proxy, executable_path=executable_path, *args, **kwargs, )
The TikTokApi class. Used to interact with TikTok. This is a singleton class to prevent issues from arising with playwright
Parameters
logging_level: The logging level you want the program to run at, optional These are the standard python logging module's levels.
request_delay: The amount of time in seconds to wait before making a request, optional This is used to throttle your own requests as you may end up making too many requests to TikTok for your IP.
custom_device_id: A TikTok parameter needed to download videos, optional The code generates these and handles these pretty well itself, however for some things such as video download you will need to set a consistent one of these. All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods.
generate_static_device_id: A parameter that generates a custom_device_id at the instance level Use this if you want to download videos from a script but don't want to generate your own custom_device_id parameter.
custom_verify_fp: A TikTok parameter needed to work most of the time, optional To get this parameter look at this video I recommend watching the entire thing, as it will help setup this package. All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods.
You can use the following to generate
"".join(random.choice(string.digits) for num in range(19))
use_test_endpoints: Send requests to TikTok's test endpoints, optional This parameter when set to true will make requests to TikTok's testing endpoints instead of the live site. I can't guarantee this will work in the future, however currently basically any custom_verify_fp will work here which is helpful.
proxy: A string containing your proxy address, optional If you want to do a lot of scraping of TikTok endpoints you'll likely need a proxy.
Ex: "https://0.0.0.0:8080"
All the methods take this as a optional parameter, however it's cleaner code to store this at the instance level. You can override this at the specific methods.
executable_path: The location of the driver, optional This shouldn't be needed if you're using playwright
**kwargs Parameters that are passed on to basically every module and methods that interact with this main class. These may or may not be documented in other places.
View Source
def get_data(self, path, subdomain="m", **kwargs) -> dict: """Makes requests to TikTok and returns their JSON. This is all handled by the package so it's unlikely you will need to use this. """ processed = self._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if self._request_delay is not None: time.sleep(self._request_delay) if self._proxy is not None: proxy = self._proxy if kwargs.get("custom_verify_fp") == None: if self._custom_verify_fp != None: verifyFp = self._custom_verify_fp else: verifyFp = "verify_khr3jabg_V7ucdslq_Vrw9_4KPb_AJ1b_Ks706M8zIJTq" else: verifyFp = kwargs.get("custom_verify_fp") tt_params = None send_tt_params = kwargs.get("send_tt_params", False) full_url = f"https://{subdomain}.tiktok.com/" + path if self._signer_url is None: kwargs["custom_verify_fp"] = verifyFp ( verify_fp, device_id, signature, tt_params, ) = asyncio.get_event_loop().run_until_complete( asyncio.gather( self._browser.sign_url( full_url, calc_tt_params=send_tt_params, **kwargs ) ) )[ 0 ] user_agent = self._browser.user_agent referrer = self._browser.referrer else: ( verify_fp, device_id, signature, user_agent, referrer, ) = self.external_signer( full_url, custom_device_id=kwargs.get("custom_device_id"), verifyFp=kwargs.get("custom_verify_fp", verifyFp), ) if not kwargs.get("send_tt_params", False): tt_params = None query = {"verifyFp": verify_fp, "device_id": device_id, "_signature": signature} url = "{}&{}".format(full_url, urlencode(query)) h = requests.head( url, headers={"x-secsdk-csrf-version": "1.2.5", "x-secsdk-csrf-request": "1"}, proxies=self._format_proxy(processed.proxy), **self._requests_extra_kwargs, ) csrf_token = None if subdomain == "m": csrf_session_id = h.cookies["csrf_session_id"] csrf_token = h.headers["X-Ware-Csrf-Token"].split(",")[1] kwargs["csrf_session_id"] = csrf_session_id headers = { "authority": "m.tiktok.com", "method": "GET", "path": url.split("tiktok.com")[1], "scheme": "https", "accept": "application/json, text/plain, */*", "accept-encoding": "gzip", "accept-language": "en-US,en;q=0.9", "origin": referrer, "referer": referrer, "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "none", "sec-gpc": "1", "user-agent": user_agent, "x-secsdk-csrf-token": csrf_token, "x-tt-params": tt_params, } self.logger.debug(f"GET: %s\n\theaders: %s", url, headers) r = requests.get( url, headers=headers, cookies=self._get_cookies(**kwargs), proxies=self._format_proxy(processed.proxy), **self._requests_extra_kwargs, ) try: parsed_data = r.json() if ( parsed_data.get("type") == "verify" or parsed_data.get("verifyConfig", {}).get("type", "") == "verify" ): self.logger.error( "Tiktok wants to display a captcha.\nResponse:\n%s\nCookies:\n%s\nURL:\n%s", r.text, self._get_cookies(**kwargs), url, ) raise CaptchaException( "TikTok blocks this request displaying a Captcha \nTip: Consider using a proxy or a custom_verify_fp as method parameters" ) # statusCode from props->pageProps->statusCode thanks @adiantek on #403 error_codes = { "0": "OK", "450": "CLIENT_PAGE_ERROR", "10000": "VERIFY_CODE", "10101": "SERVER_ERROR_NOT_500", "10102": "USER_NOT_LOGIN", "10111": "NET_ERROR", "10113": "SHARK_SLIDE", "10114": "SHARK_BLOCK", "10119": "LIVE_NEED_LOGIN", "10202": "USER_NOT_EXIST", "10203": "MUSIC_NOT_EXIST", "10204": "VIDEO_NOT_EXIST", "10205": "HASHTAG_NOT_EXIST", "10208": "EFFECT_NOT_EXIST", "10209": "HASHTAG_BLACK_LIST", "10210": "LIVE_NOT_EXIST", "10211": "HASHTAG_SENSITIVITY_WORD", "10212": "HASHTAG_UNSHELVE", "10213": "VIDEO_LOW_AGE_M", "10214": "VIDEO_LOW_AGE_T", "10215": "VIDEO_ABNORMAL", "10216": "VIDEO_PRIVATE_BY_USER", "10217": "VIDEO_FIRST_REVIEW_UNSHELVE", "10218": "MUSIC_UNSHELVE", "10219": "MUSIC_NO_COPYRIGHT", "10220": "VIDEO_UNSHELVE_BY_MUSIC", "10221": "USER_BAN", "10222": "USER_PRIVATE", "10223": "USER_FTC", "10224": "GAME_NOT_EXIST", "10225": "USER_UNIQUE_SENSITIVITY", "10227": "VIDEO_NEED_RECHECK", "10228": "VIDEO_RISK", "10229": "VIDEO_R_MASK", "10230": "VIDEO_RISK_MASK", "10231": "VIDEO_GEOFENCE_BLOCK", "10404": "FYP_VIDEO_LIST_LIMIT", "undefined": "MEDIA_ERROR", } statusCode = parsed_data.get("statusCode", 0) self.logger.debug(f"TikTok Returned: %s", json) if statusCode == 10201: # Invalid Entity raise NotFoundException( "TikTok returned a response indicating the entity is invalid" ) elif statusCode == 10219: # Not available in this region raise NotAvailableException("Content not available for this region") elif statusCode != 0 and statusCode != -1: raise TikTokException( error_codes.get( statusCode, f"TikTok sent an unknown StatusCode of {statusCode}" ) ) return r.json() except ValueError as e: text = r.text self.logger.debug("TikTok response: %s", text) if len(text) == 0: raise EmptyResponseException( "Empty response from Tiktok to " + url ) from None else: raise InvalidJSONException("TikTok sent invalid JSON") from e
Makes requests to TikTok and returns their JSON.
This is all handled by the package so it's unlikely you will need to use this.
View Source
def external_signer(self, url, custom_device_id=None, verifyFp=None): """Makes requests to an external signer instead of using a browser. ##### Parameters * url: The server to make requests to This server is designed to sign requests. You can find an example of this signature server in the examples folder. * custom_device_id: A TikTok parameter needed to download videos The code generates these and handles these pretty well itself, however for some things such as video download you will need to set a consistent one of these. * custom_verify_fp: A TikTok parameter needed to work most of the time, To get this parameter look at [this video](https://youtu.be/zwLmLfVI-VQ?t=117) I recommend watching the entire thing, as it will help setup this package. """ if custom_device_id is not None: query = { "url": url, "custom_device_id": custom_device_id, "verifyFp": verifyFp, } else: query = {"url": url, "verifyFp": verifyFp} data = requests.get( self._signer_url + "?{}".format(urlencode(query)), **self._requests_extra_kwargs, ) parsed_data = data.json() return ( parsed_data["verifyFp"], parsed_data["device_id"], parsed_data["_signature"], parsed_data["user_agent"], parsed_data["referrer"], )
Makes requests to an external signer instead of using a browser.
Parameters
url: The server to make requests to This server is designed to sign requests. You can find an example of this signature server in the examples folder.
custom_device_id: A TikTok parameter needed to download videos The code generates these and handles these pretty well itself, however for some things such as video download you will need to set a consistent one of these.
custom_verify_fp: A TikTok parameter needed to work most of the time, To get this parameter look at this video I recommend watching the entire thing, as it will help setup this package.
View Source
def get_bytes(self, **kwargs) -> bytes: """Returns TikTok's response as bytes, similar to get_data""" processed = self._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if self._signer_url is None: ( verify_fp, device_id, signature, _, ) = asyncio.get_event_loop().run_until_complete( asyncio.gather(self._browser.sign_url(calc_tt_params=False, **kwargs)) )[ 0 ] user_agent = self._browser.user_agent referrer = self._browser.referrer else: ( verify_fp, device_id, signature, user_agent, referrer, ) = self.external_signer( kwargs["url"], custom_device_id=kwargs.get("custom_device_id", None) ) query = {"verifyFp": verify_fp, "_signature": signature} url = "{}&{}".format(kwargs["url"], urlencode(query)) r = requests.get( url, headers={ "Accept": "*/*", "Accept-Encoding": "identity;q=1, *;q=0", "Accept-Language": "en-US;en;q=0.9", "Cache-Control": "no-cache", "Connection": "keep-alive", "Host": url.split("/")[2], "Pragma": "no-cache", "Range": "bytes=0-", "Referer": "https://www.tiktok.com/", "User-Agent": user_agent, }, proxies=self._format_proxy(processed.proxy), cookies=self._get_cookies(**kwargs), ) return r.content
Returns TikTok's response as bytes, similar to get_data
View Source
@staticmethod def generate_device_id(): """Generates a valid device_id for other methods. Pass this as the custom_device_id field to download videos""" return "".join([random.choice(string.digits) for num in range(19)])
Generates a valid device_id for other methods. Pass this as the custom_device_id field to download videos
View Source
def shutdown(self) -> None: with _thread_lock: self.logger.debug("Shutting down Playwright") asyncio.get_event_loop().run_until_complete(self._browser._clean_up())
View Source
class User: """ A TikTok User. Example Usage ```py user = api.user(username='therock') # or user_id = '5831967' sec_uid = 'MS4wLjABAAAA-VASjiXTh7wDDyXvjk10VFhMWUAoxr8bgfO1kAL1-9s' user = api.user(user_id=user_id, sec_uid=sec_uid) ``` """ parent: ClassVar[TikTokApi] user_id: str """The user ID of the user.""" sec_uid: str """The sec UID of the user.""" username: str """The username of the user.""" as_dict: dict """The raw data associated with this user.""" def __init__( self, username: Optional[str] = None, user_id: Optional[str] = None, sec_uid: Optional[str] = None, data: Optional[dict] = None, ): """ You must provide the username or (user_id and sec_uid) otherwise this will not function correctly. """ self.__update_id_sec_uid_username(user_id, sec_uid, username) if data is not None: self.as_dict = data self.__extract_from_data() def info(self, **kwargs): """ Returns a dictionary of TikTok's User object Example Usage ```py user_data = api.user(username='therock').info() ``` """ return self.info_full(**kwargs)["user"] def info_full(self, **kwargs) -> dict: """ Returns a dictionary of information associated with this User. Includes statistics about this user. Example Usage ```py user_data = api.user(username='therock').info_full() ``` """ # TODO: Find the one using only user_id & sec_uid if not self.username: raise TypeError( "You must provide the username when creating this class to use this method." ) quoted_username = quote(self.username) r = requests.get( "https://tiktok.com/@{}?lang=en".format(quoted_username), headers={ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9", "path": "/@{}".format(quoted_username), "Accept-Encoding": "gzip, deflate", "Connection": "keep-alive", "User-Agent": self.parent._user_agent, }, proxies=User.parent._format_proxy(kwargs.get("proxy", None)), cookies=User.parent._get_cookies(**kwargs), **User.parent._requests_extra_kwargs, ) data = extract_tag_contents(r.text) user = json.loads(data) user_props = user["props"]["pageProps"] if user_props["statusCode"] == 404: raise NotFoundException( "TikTok user with username {} does not exist".format(self.username) ) return user_props["userInfo"] def videos(self, count=30, cursor=0, **kwargs) -> Iterator[Video]: """ Returns an iterator yielding Video objects. - Parameters: - count (int): The amount of videos you want returned. - cursor (int): The unix epoch to get uploaded videos since. Example Usage ```py user = api.user(username='therock') for video in user.videos(count=100): # do something ``` """ processed = User.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if not self.user_id and not self.sec_uid: self.__find_attributes() first = True amount_yielded = 0 while amount_yielded < count: query = { "count": 30, "id": self.user_id, "cursor": cursor, "type": 1, "secUid": self.sec_uid, "sourceType": 8, "appId": 1233, "region": processed.region, "priority_region": processed.region, "language": processed.language, } path = "api/post/item_list/?{}&{}".format( User.parent._add_url_params(), urlencode(query) ) res = User.parent.get_data(path, send_tt_params=True, **kwargs) videos = res.get("itemList", []) amount_yielded += len(videos) for video in videos: yield self.parent.video(data=video) if not res.get("hasMore", False) and not first: User.parent.logger.info( "TikTok isn't sending more TikToks beyond this point." ) return cursor = res["cursor"] first = False def liked(self, count: int = 30, cursor: int = 0, **kwargs) -> Iterator[Video]: """ Returns a dictionary listing TikToks that a given a user has liked. **Note**: The user's likes must be **public** (which is not the default option) - Parameters: - count (int): The amount of videos you want returned. - cursor (int): The unix epoch to get uploaded videos since. Example Usage ```py for liked_video in api.user(username='public_likes'): # do something ``` """ processed = User.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id amount_yielded = 0 first = True if self.user_id is None and self.sec_uid is None: self.__find_attributes() while amount_yielded < count: query = { "count": 30, "id": self.user_id, "type": 2, "secUid": self.sec_uid, "cursor": cursor, "sourceType": 9, "appId": 1233, "region": processed.region, "priority_region": processed.region, "language": processed.language, } path = "api/favorite/item_list/?{}&{}".format( User.parent._add_url_params(), urlencode(query) ) res = self.parent.get_data(path, **kwargs) if "itemList" not in res.keys(): if first: User.parent.logger.error("User's likes are most likely private") return videos = res.get("itemList", []) amount_yielded += len(videos) for video in videos: amount_yielded += 1 yield self.parent.video(data=video) if not res.get("hasMore", False) and not first: User.parent.logger.info( "TikTok isn't sending more TikToks beyond this point." ) return cursor = res["cursor"] first = False def __extract_from_data(self): data = self.as_dict keys = data.keys() if "user_info" in keys: self.__update_id_sec_uid_username( data["user_info"]["uid"], data["user_info"]["sec_uid"], data["user_info"]["unique_id"], ) elif "uniqueId" in keys: self.__update_id_sec_uid_username( data["id"], data["secUid"], data["uniqueId"] ) if None in (self.username, self.user_id, self.sec_uid): User.parent.logger.error( f"Failed to create User with data: {data}\nwhich has keys {data.keys()}" ) def __update_id_sec_uid_username(self, id, sec_uid, username): self.user_id = id self.sec_uid = sec_uid self.username = username def __find_attributes(self) -> None: # It is more efficient to check search first, since self.user_object() makes HTML request. found = False for u in self.parent.search.users(self.username): if u.username == self.username: found = True self.__update_id_sec_uid_username(u.user_id, u.sec_uid, u.username) break if not found: user_object = self.info() self.__update_id_sec_uid_username( user_object["id"], user_object["secUid"], user_object["uniqueId"] ) def __repr__(self): return self.__str__() def __str__(self): return f"TikTokApi.user(username='{self.username}', user_id='{self.user_id}', sec_uid='{self.sec_uid}')" def __getattr__(self, name): if name in ["as_dict"]: self.as_dict = self.info() self.__extract_from_data() return self.__getattribute__(name) raise AttributeError(f"{name} doesn't exist on TikTokApi.api.User")
A TikTok User.
Example Usage
user = api.user(username='therock')
# or
user_id = '5831967'
sec_uid = 'MS4wLjABAAAA-VASjiXTh7wDDyXvjk10VFhMWUAoxr8bgfO1kAL1-9s'
user = api.user(user_id=user_id, sec_uid=sec_uid)
View Source
class Search: """Contains static methods about searching.""" parent: TikTokApi @staticmethod def videos(search_term, count=28, offset=0, **kwargs) -> Iterator[Video]: """ Searches for Videos - Parameters: - search_term (str): The phrase you want to search for. - count (int): The amount of videos you want returned. - offset (int): The offset of videos from your data you want returned. Example Usage ```py for video in api.search.videos('therock'): # do something ``` """ return Search.search_type( search_term, "item", count=count, offset=offset, **kwargs ) @staticmethod def users(search_term, count=28, offset=0, **kwargs) -> Iterator[User]: """ Searches for users using an alternate endpoint than Search.users - Parameters: - search_term (str): The phrase you want to search for. - count (int): The amount of videos you want returned. Example Usage ```py for user in api.search.users_alternate('therock'): # do something ``` """ return Search.search_type( search_term, "user", count=count, offset=offset, **kwargs ) @staticmethod def search_type(search_term, obj_type, count=28, offset=0, **kwargs) -> Iterator: """ Searches for users using an alternate endpoint than Search.users - Parameters: - search_term (str): The phrase you want to search for. - count (int): The amount of videos you want returned. - obj_type (str): user | item Just use .video & .users ``` """ processed = Search.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id cursor = offset spawn = requests.head( "https://www.tiktok.com", proxies=Search.parent._format_proxy(processed.proxy), **Search.parent._requests_extra_kwargs ) ttwid = spawn.cookies["ttwid"] # For some reason when <= it can be off by one. while cursor - offset <= count: query = { "keyword": search_term, "cursor": cursor, "app_language": Search.parent._language, } path = "api/search/{}/full/?{}&{}".format( obj_type, Search.parent._add_url_params(), urlencode(query) ) if obj_type == "user": subdomain = "www" elif obj_type == "item": subdomain = "us" else: raise TypeError("invalid obj_type") api_response = Search.parent.get_data( path, subdomain=subdomain, ttwid=ttwid, **kwargs ) # When I move to 3.10+ support make this a match switch. for result in api_response.get("user_list", []): yield User(data=result) for result in api_response.get("item_list", []): yield Video(data=result) if api_response.get("has_more", 0) == 0: Search.parent.logger.info( "TikTok is not sending videos beyond this point." ) return cursor = int(api_response.get("cursor", cursor))
Contains static methods about searching.
Inherited Members
View Source
class Sound: """ A TikTok Sound/Music/Song. Example Usage ```py song = api.song(id='7016547803243022337') ``` """ parent: ClassVar[TikTokApi] id: str """TikTok's ID for the sound""" title: Optional[str] """The title of the song.""" author: Optional[User] """The author of the song (if it exists)""" def __init__(self, id: Optional[str] = None, data: Optional[str] = None): """ You must provide the id of the sound or it will not work. """ if data is not None: self.as_dict = data self.__extract_from_data() elif id is None: raise TypeError("You must provide id parameter.") else: self.id = id def info(self, use_html=False, **kwargs) -> dict: """ Returns a dictionary of TikTok's Sound/Music object. - Parameters: - use_html (bool): If you want to perform an HTML request or not. Defaults to False to use an API call, which shouldn't get detected as often as an HTML request. Example Usage ```py sound_data = api.sound(id='7016547803243022337').info() ``` """ self.__ensure_valid() if use_html: return self.info_full(**kwargs)["musicInfo"] processed = self.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id path = "node/share/music/-{}?{}".format(self.id, self.parent._add_url_params()) res = self.parent.get_data(path, **kwargs) if res.get("statusCode", 200) == 10203: raise NotFoundException() return res["musicInfo"]["music"] def info_full(self, **kwargs) -> dict: """ Returns all the data associated with a TikTok Sound. This makes an API request, there is no HTML request option, as such with Sound.info() Example Usage ```py sound_data = api.sound(id='7016547803243022337').info_full() ``` """ self.__ensure_valid() r = requests.get( "https://www.tiktok.com/music/-{}".format(self.id), headers={ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9", "Accept-Encoding": "gzip, deflate", "Connection": "keep-alive", "User-Agent": self.parent._user_agent, }, proxies=self.parent._format_proxy(kwargs.get("proxy", None)), cookies=self.parent._get_cookies(**kwargs), **self.parent._requests_extra_kwargs, ) data = extract_tag_contents(r.text) return json.loads(data)["props"]["pageProps"]["musicInfo"] def videos(self, count=30, offset=0, **kwargs) -> Iterator[Video]: """ Returns Video objects of videos created with this sound. - Parameters: - count (int): The amount of videos you want returned. - offset (int): The offset of videos you want returned. Example Usage ```py for video in api.sound(id='7016547803243022337').videos(): # do something ``` """ self.__ensure_valid() processed = self.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id cursor = offset page_size = 30 while cursor - offset < count: query = { "secUid": "", "musicID": self.id, "cursor": cursor, "shareUid": "", "count": page_size, } path = "api/music/item_list/?{}&{}".format( self.parent._add_url_params(), urlencode(query) ) res = self.parent.get_data(path, send_tt_params=True, **kwargs) for result in res.get("itemList", []): yield self.parent.video(data=result) if not res.get("hasMore", False): self.parent.logger.info( "TikTok isn't sending more TikToks beyond this point." ) return cursor = int(res["cursor"]) def __extract_from_data(self): data = self.as_dict keys = data.keys() if data.get("id") == "": self.id = "" if "authorName" in keys: self.id = data["id"] self.title = data["title"] if data.get("authorName") is not None: self.author = self.parent.user(username=data["authorName"]) if self.id is None: Sound.parent.logger.error( f"Failed to create Sound with data: {data}\nwhich has keys {data.keys()}" ) def __ensure_valid(self): if self.id == "": raise SoundRemovedException("This sound has been removed!") def __repr__(self): return self.__str__() def __str__(self): return f"TikTokApi.sound(id='{self.id}')" def __getattr__(self, name): if name in ["title", "author", "as_dict"]: self.as_dict = self.info() self.__extract_from_data() return self.__getattribute__(name) raise AttributeError(f"{name} doesn't exist on TikTokApi.api.Sound")
A TikTok Sound/Music/Song.
Example Usage
song = api.song(id='7016547803243022337')
View Source
class Hashtag: """ A TikTok Hashtag/Challenge. Example Usage ```py hashtag = api.hashtag(name='funny') ``` """ parent: ClassVar[TikTokApi] id: Optional[str] """The ID of the hashtag""" name: Optional[str] """The name of the hashtag (omiting the #)""" as_dict: dict """The raw data associated with this hashtag.""" def __init__( self, name: Optional[str] = None, id: Optional[str] = None, data: Optional[dict] = None, ): """ You must provide the name or id of the hashtag. """ self.name = name self.id = id if data is not None: self.as_dict = data self.__extract_from_data() def info(self, **kwargs) -> dict: """ Returns TikTok's dictionary representation of the hashtag object. """ return self.info_full(**kwargs)["challengeInfo"]["challenge"] def info_full(self, **kwargs) -> dict: """ Returns all information sent by TikTok related to this hashtag. Example Usage ```py hashtag_data = api.hashtag(name='funny').info_full() ``` """ processed = self.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if self.name is not None: query = {"challengeName": self.name} elif self.id is not None: query = {"challengeId": self.id} else: self.parent.logger.warning("Malformed Hashtag Object") return {} path = "api/challenge/detail/?{}&{}".format( self.parent._add_url_params(), urlencode(query) ) data = self.parent.get_data(path, **kwargs) if data["challengeInfo"].get("challenge") is None: raise NotFoundException("Challenge {} does not exist".format(self.name)) return data def videos(self, count=30, offset=0, **kwargs) -> Iterator[Video]: """Returns a dictionary listing TikToks with a specific hashtag. - Parameters: - count (int): The amount of videos you want returned. - offset (int): The the offset of videos from 0 you want to get. Example Usage ```py for video in api.hashtag(name='funny').videos(): # do something ``` """ processed = self.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id if self.id is None: self.id = self.info()["id"] cursor = offset page_size = 30 while cursor - offset < count: query = { "count": page_size, "challengeID": self.id, "cursor": cursor, } path = "api/challenge/item_list/?{}&{}".format( self.parent._add_url_params(), urlencode(query) ) res = self.parent.get_data(path, **kwargs) for result in res.get("itemList", []): yield self.parent.video(data=result) if not res.get("hasMore", False): self.parent.logger.info( "TikTok isn't sending more TikToks beyond this point." ) return cursor = int(res["cursor"]) def __extract_from_data(self): data = self.as_dict keys = data.keys() if "title" in keys: self.id = data["id"] self.name = data["title"] if None in (self.name, self.id): Hashtag.parent.logger.error( f"Failed to create Hashtag with data: {data}\nwhich has keys {data.keys()}" ) def __repr__(self): return self.__str__() def __str__(self): return f"TikTokApi.hashtag(id='{self.id}', name='{self.name}')" def __getattr__(self, name): # TODO: Maybe switch to using @property instead if name in ["id", "name", "as_dict"]: self.as_dict = self.info() self.__extract_from_data() return self.__getattribute__(name) raise AttributeError(f"{name} doesn't exist on TikTokApi.api.Hashtag")
A TikTok Hashtag/Challenge.
Example Usage
hashtag = api.hashtag(name='funny')
View Source
class Video: """ A TikTok Video class Example Usage ```py video = api.video(id='7041997751718137094') ``` """ parent: ClassVar[TikTokApi] id: Optional[str] """TikTok's ID of the Video""" create_time: Optional[datetime] """The creation time of the Video""" stats: Optional[dict] """TikTok's stats of the Video""" author: Optional[User] """The User who created the Video""" sound: Optional[Sound] """The Sound that is associated with the Video""" hashtags: Optional[list[Hashtag]] """A List of Hashtags on the Video""" as_dict: dict """The raw data associated with this Video.""" def __init__( self, id: Optional[str] = None, url: Optional[str] = None, data: Optional[dict] = None, ): """ You must provide the id or a valid url, else this will fail. """ self.id = id if data is not None: self.as_dict = data self.__extract_from_data() elif url is not None: self.id = extract_video_id_from_url(url) if self.id is None: raise TypeError("You must provide id or url parameter.") def info(self, **kwargs) -> dict: """ Returns a dictionary of TikTok's Video object. Example Usage ```py video_data = api.video(id='7041997751718137094').info() ``` """ return self.info_full(**kwargs)["itemInfo"]["itemStruct"] def info_full(self, **kwargs) -> dict: """ Returns a dictionary of all data associated with a TikTok Video. Example Usage ```py video_data = api.video(id='7041997751718137094').info_full() ``` """ processed = self.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id device_id = kwargs.get("custom_device_id", None) query = { "itemId": self.id, } path = "api/item/detail/?{}&{}".format( self.parent._add_url_params(), urlencode(query) ) return self.parent.get_data(path, **kwargs) def bytes(self, **kwargs) -> bytes: """ Returns the bytes of a TikTok Video. Example Usage ```py video_bytes = api.video(id='7041997751718137094').bytes() # Saving The Video with open('saved_video.mp4', 'wb') as output: output.write(video_bytes) ``` """ processed = self.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id video_data = self.info(**kwargs) download_url = video_data["video"]["playAddr"] return self.parent.get_bytes(url=download_url, **kwargs) def __extract_from_data(self) -> None: data = self.as_dict keys = data.keys() if "author" in keys: self.id = data["id"] self.create_time = datetime.fromtimestamp(data["createTime"]) self.stats = data["stats"] self.author = self.parent.user(data=data["author"]) self.sound = self.parent.sound(data=data["music"]) self.hashtags = [ self.parent.hashtag(data=hashtag) for hashtag in data.get("challenges", []) ] if self.id is None: Video.parent.logger.error( f"Failed to create Video with data: {data}\nwhich has keys {data.keys()}" ) def __repr__(self): return self.__str__() def __str__(self): return f"TikTokApi.video(id='{self.id}')" def __getattr__(self, name): # Handle author, sound, hashtags, as_dict if name in ["author", "sound", "hashtags", "stats", "create_time", "as_dict"]: self.as_dict = self.info() self.__extract_from_data() return self.__getattribute__(name) raise AttributeError(f"{name} doesn't exist on TikTokApi.api.Video")
A TikTok Video class
Example Usage
video = api.video(id='7041997751718137094')
Inherited Members
View Source
class Trending: """Contains static methods related to trending.""" parent: TikTokApi @staticmethod def videos(count=30, **kwargs) -> Iterator[Video]: """ Returns Videos that are trending on TikTok. - Parameters: - count (int): The amount of videos you want returned. """ processed = Trending.parent._process_kwargs(kwargs) kwargs["custom_device_id"] = processed.device_id spawn = requests.head( "https://www.tiktok.com", proxies=Trending.parent._format_proxy(processed.proxy), **Trending.parent._requests_extra_kwargs, ) ttwid = spawn.cookies["ttwid"] first = True amount_yielded = 0 while amount_yielded < count: query = { "count": 30, "id": 1, "sourceType": 12, "itemID": 1, "insertedItemID": "", "region": processed.region, "priority_region": processed.region, "language": processed.language, } path = "api/recommend/item_list/?{}&{}".format( Trending.parent._add_url_params(), urlencode(query) ) res = Trending.parent.get_data(path, ttwid=ttwid, **kwargs) for result in res.get("itemList", []): yield Video(data=result) amount_yielded += len(res.get("itemList", [])) if not res.get("hasMore", False) and not first: Trending.parent.logger.info( "TikTok isn't sending more TikToks beyond this point." ) return first = False
Contains static methods related to trending.