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 .api.comment import Comment

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
    comment = Comment
    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
        Comment.parent = self

        # Some Instance Vars
        self._executable_path = kwargs.get("executable_path", None)
        self.cookie_jar = 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"  # TODO: Randomly generate agents
        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,
        )

        self.cookie_jar = r.cookies

        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 get_data_no_sig(self, path, subdomain="m", **kwargs) -> dict:
        processed = self._process_kwargs(kwargs)
        full_url = f"https://{subdomain}.tiktok.com/" + path
        referrer = self._browser.referrer
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:101.0) Gecko/20100101 Firefox/101.0",
            "authority": "m.tiktok.com",
            "method": "GET",
            "path": full_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",
        }
        self.logger.debug(f"GET: %s\n\theaders: %s", full_url, headers)

        r = requests.get(
            full_url,
            headers=headers,
            cookies=self._get_cookies(**kwargs),
            proxies=self._format_proxy(processed.proxy),
            **self._requests_extra_kwargs,
        )
        return r.json()

    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()
#   class TikTokApi:
View Source
class TikTokApi:
    _is_context_manager = False
    user = User
    search = Search
    sound = Sound
    hashtag = Hashtag
    video = Video
    trending = Trending
    comment = Comment
    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
        Comment.parent = self

        # Some Instance Vars
        self._executable_path = kwargs.get("executable_path", None)
        self.cookie_jar = 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"  # TODO: Randomly generate agents
        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,
        )

        self.cookie_jar = r.cookies

        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 get_data_no_sig(self, path, subdomain="m", **kwargs) -> dict:
        processed = self._process_kwargs(kwargs)
        full_url = f"https://{subdomain}.tiktok.com/" + path
        referrer = self._browser.referrer
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:101.0) Gecko/20100101 Firefox/101.0",
            "authority": "m.tiktok.com",
            "method": "GET",
            "path": full_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",
        }
        self.logger.debug(f"GET: %s\n\theaders: %s", full_url, headers)

        r = requests.get(
            full_url,
            headers=headers,
            cookies=self._get_cookies(**kwargs),
            proxies=self._format_proxy(processed.proxy),
            **self._requests_extra_kwargs,
        )
        return r.json()

    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()
#   TikTokApi( logging_level: int = 30, 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 )
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.

#   logger = <Logger TikTokApi (WARNING)>
#   def get_data(self, path, subdomain='m', **kwargs) -> dict:
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,
        )

        self.cookie_jar = r.cookies

        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.

#   def get_data_no_sig(self, path, subdomain='m', **kwargs) -> dict:
View Source
    def get_data_no_sig(self, path, subdomain="m", **kwargs) -> dict:
        processed = self._process_kwargs(kwargs)
        full_url = f"https://{subdomain}.tiktok.com/" + path
        referrer = self._browser.referrer
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:101.0) Gecko/20100101 Firefox/101.0",
            "authority": "m.tiktok.com",
            "method": "GET",
            "path": full_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",
        }
        self.logger.debug(f"GET: %s\n\theaders: %s", full_url, headers)

        r = requests.get(
            full_url,
            headers=headers,
            cookies=self._get_cookies(**kwargs),
            proxies=self._format_proxy(processed.proxy),
            **self._requests_extra_kwargs,
        )
        return r.json()
#   def external_signer(self, url, custom_device_id=None, verifyFp=None):
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.

#   def get_bytes(self, **kwargs) -> bytes:
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

#  
@staticmethod
def generate_device_id():
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

#   def shutdown(self) -> None:
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())
#   class TikTokApi.user:
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"]

        """
        TODO: There is a route for user info, but uses msToken :\
        processed = self.parent._process_kwargs(kwargs)
        kwargs["custom_device_id"] = processed.device_id

        query = {
            "uniqueId": "therock",
            "secUid": "",
            "msToken": User.parent._get_cookies()["msToken"]
        }

        path = "api/user/detail/?{}&{}".format(
            User.parent._add_url_params(), urlencode(query)
        )

        res = User.parent.get_data(path, subdomain="m", **kwargs)
        print(res)

        return res["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", [])
            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 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", [])
            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)
#   class TikTokApi.search:
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.

#   class TikTokApi.sound:
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')
#   class TikTokApi.hashtag:
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.
        """

        if name is not None:
            self.name = name
        if id is not None:
            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
        ```
        """
        cursor = offset
        page_size = 30
        while cursor - offset < count:
            query = {
                "aid": 1988,
                "count": page_size,
                "challengeID": self.id,
                "cursor": cursor,
            }
            path = "api/challenge/item_list/?{}".format(urlencode(query))
            res = self.parent.get_data_no_sig(path, subdomain="us", **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')
#   class TikTokApi.video:
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, headers={"user-agent": self.parent._user_agent}
            )

        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 comments(self, count=20, offset=0, **kwargs) -> Iterator[Comment]:
        """
        Returns Comments from the video

        - Parameters:
            - count (int): The amount of videos you want returned.
            - offset (int): The offset you want to check comments of
        """

        processed = Video.parent._process_kwargs(kwargs)
        kwargs["custom_device_id"] = processed.device_id
        cursor = offset

        spawn = requests.head(
            "https://www.tiktok.com",
            proxies=Video.parent._format_proxy(processed.proxy),
            **Video.parent._requests_extra_kwargs,
        )
        ttwid = spawn.cookies["ttwid"]

        while cursor - offset <= count:
            query = {
                "aweme_id": self.id,
                "cursor": cursor,
                "app_language": Video.parent._language,
                "count": 30,
            }
            path = "api/comment/list/?{}&{}".format(
                Video.parent._add_url_params(), urlencode(query)
            )

            api_response = Video.parent.get_data(
                path, subdomain="www", ttwid=ttwid, **kwargs
            )

            for comment_data in api_response.get("comments", []):
                yield self.parent.comment(data=comment_data)

            if api_response.get("has_more", 0) == 0:
                Video.parent.logger.info(
                    "TikTok is not sending comments beyond this point."
                )
                return

            cursor = int(api_response.get("cursor", cursor))

    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)

        if name in ["comments"]:
            # Requires a different request to produce the comments
            self.__extract_comments()
            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')
#   class TikTokApi.trending:
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.

#   class TikTokApi.comment:
View Source
class Comment:
    """
    A TikTok Comment.

    Example Usage
    ```py
    for comment in video.comments:
        print(comment.text)
    ```
    """

    parent: ClassVar[TikTokApi]

    id: str
    """The id of the comment"""
    author: ClassVar[User]
    """The author of the comment"""
    text: str
    """The contents of the comment"""
    likes_count: int
    """The amount of likes of the comment"""
    as_dict: dict
    """The raw data associated with this comment"""

    def __init__(self, data: Optional[dict] = None):
        if data is not None:
            self.as_dict = data
            self.__extract_from_data()

    def __extract_from_data(self):
        self.id = self.as_dict["cid"]
        self.text = self.as_dict["text"]

        usr = self.as_dict["user"]
        self.author = self.parent.user(
            user_id=usr["uid"], username=usr["unique_id"], sec_uid=usr["sec_uid"]
        )
        self.likes_count = self.as_dict["digg_count"]

    def __repr__(self):
        return self.__str__()

    def __str__(self):
        return f"TikTokApi.comment(comment_id='{self.id}', text='{self.text}')"

    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.Comment")

A TikTok Comment.

Example Usage

for comment in video.comments:
    print(comment.text)