Merge pull request #708 from davidteather/nightly

V4.0.3

TikTokApi/browser_utilities/__init__.py

@@ -0,0 +1 @@
+from .browser_interface import BrowserInterface

TikTokApi/browser.py → TikTokApi/browser_utilities/browser.py

@@ -3,16 +3,15 @@
 import string
 import requests
 import logging
-from threading import Thread
 import time
-import datetime
 import random
 import json
+import re
+from .browser_interface import BrowserInterface
 from urllib.parse import splitquery, parse_qs, parse_qsl
 
 
 # Import Detection From Stealth
-from .stealth import stealth
 from .get_acrawler import get_acrawler, get_tt_params_script
 from playwright.sync_api import sync_playwright
 
@@ -30,7 +29,7 @@ def get_playwright():
     return playwright
 
 
-class browser:
+class browser(BrowserInterface):
     def __init__(
         self,
         **kwargs,
@@ -43,7 +42,6 @@ def __init__(
         self.language = kwargs.get("language", "en")
         self.executablePath = kwargs.get("executablePath", None)
         self.device_id = kwargs.get("custom_device_id", None)
-        find_redirect = kwargs.get("find_redirect", False)
 
         args = kwargs.get("browser_args", [])
         options = kwargs.get("browser_options", {})
@@ -82,8 +80,8 @@ def __init__(
                 args=self.args, **self.options
             )
         except Exception as e:
-            raise e
             logging.critical(e)
+            raise e
 
         context = self.create_context(set_useragent=True)
         page = context.new_page()
@@ -163,14 +161,21 @@ def gen_verifyFp(self):
 
         return f'verify_{scenario_title.lower()}_{"".join(uuid)}'
 
-    def sign_url(self, **kwargs):
+    def sign_url(self, calc_tt_params=False, **kwargs):
+        def process(route):
+            route.abort()
+
         url = kwargs.get("url", None)
         if url is None:
             raise Exception("sign_url required a url parameter")
+
+        tt_params = None
         context = self.create_context()
         page = context.new_page()
 
-        page.goto(kwargs.get('default_url', 'https://www.tiktok.com/@redbull'), wait_until='load')
+        if calc_tt_params:
+            page.route(re.compile(r"(\.png)|(\.jpeg)|(\.mp4)|(x-expire)"), process)
+            page.goto(kwargs.get('default_url', 'https://www.tiktok.com/@redbull'), wait_until='load')
 
         verifyFp = "".join(
             random.choice(
@@ -208,15 +213,17 @@ def sign_url(self, **kwargs):
         )
 
         url = '{}&_signature={}'.format(url, evaluatedPage)
-        page.add_script_tag(content=get_tt_params_script())
 
-        tt_params = page.evaluate(
-            '''() => {
-                return window.genXTTParams(''' + json.dumps(dict(parse_qsl(splitquery(url)[1]))) + ''');
-        
-            }'''
-        ) 
-
+        if calc_tt_params:
+            page.add_script_tag(content=get_tt_params_script())
+
+            tt_params = page.evaluate(
+                '''() => {
+                    return window.genXTTParams(''' + json.dumps(dict(parse_qsl(splitquery(url)[1]))) + ''');
+            
+                }'''
+            )
+
         context.close()
         return (
             verifyFp,

TikTokApi/browser_utilities/browser_interface.py

@@ -0,0 +1,18 @@
+import abc
+class BrowserInterface(abc.ABC):
+    @abc.abstractmethod
+    def __init__(self, **kwargs):
+        pass
+
+    @abc.abstractmethod
+    def get_params(self, page) -> None:
+        pass
+
+    # Returns verify_fp, device_id, signature, tt_params
+    @abc.abstractmethod
+    def sign_url(self, calc_tt_params=False, **kwargs) -> tuple[str, str, str, str]:
+        pass
+
+    @abc.abstractmethod
+    def clean_up(self) -> None:
+        pass

TikTokApi/browser_selenium.py → TikTokApi/browser_utilities/browser_selenium.py

@@ -1,22 +1,24 @@
 import random
 import time
-import string
 import requests
 import logging
 from threading import Thread
 import time
-import datetime
+import re
 import random
+import json
+from .browser_interface import BrowserInterface
 from selenium_stealth import stealth
 from selenium import webdriver
-from .get_acrawler import get_acrawler
+from .get_acrawler import get_acrawler, get_tt_params_script
+from urllib.parse import splitquery, parse_qs, parse_qsl
 
-
-class browser:
+class browser(BrowserInterface):
     def __init__(
         self,
         **kwargs,
     ):
+        self.kwargs = kwargs
         self.debug = kwargs.get("debug", False)
         self.proxy = kwargs.get("proxy", None)
         self.api_url = kwargs.get("api_url", None)
@@ -81,26 +83,32 @@ def setup_browser(self):
             renderer="Intel Iris OpenGL Engine",
             fix_hairline=True,
         )
+
         self.get_params(self.browser)
+        # NOTE: Slower than playwright at loading this because playwright can ignore unneeded files.
+        self.browser.get("https://www.tiktok.com/@redbull")
         self.browser.execute_script(get_acrawler())
+        self.browser.execute_script(get_tt_params_script())
+
 
     def get_params(self, page) -> None:
-        # self.browser_language = await self.page.evaluate("""() => { return
-        # navigator.language || navigator.userLanguage; }""")
-        self.browser_language = ""
-        # self.timezone_name = await self.page.evaluate("""() => { return
-        # Intl.DateTimeFormat().resolvedOptions().timeZone; }""")
-        self.timezone_name = ""
-        # self.browser_platform = await self.page.evaluate("""() => { return window.navigator.platform; }""")
-        self.browser_platform = ""
-        # self.browser_name = await self.page.evaluate("""() => { return window.navigator.appCodeName; }""")
-        self.browser_name = ""
-        # self.browser_version = await self.page.evaluate("""() => { return window.navigator.appVersion; }""")
-        self.browser_version = ""
-
-        self.width = page.execute_script("""return screen.width""")
-        self.height = page.execute_script("""return screen.height""")
         self.userAgent = page.execute_script("""return navigator.userAgent""")
+        self.browser_language = self.kwargs.get("browser_language", ("""return navigator.language"""))
+        self.browser_version = ("""return window.navigator.appVersion""")
+
+        if len(self.browser_language.split("-")) == 0:
+            self.region = self.kwargs.get("region", "US")
+            self.language = self.kwargs.get("language", "en")
+        elif len(self.browser_language.split("-")) == 1:
+            self.region = self.kwargs.get("region", "US")
+            self.language = self.browser_language.split("-")[0]
+        else:
+            self.region = self.kwargs.get("region", self.browser_language.split("-")[1])
+            self.language = self.kwargs.get("language", self.browser_language.split("-")[0])
+
+        self.timezone_name = self.kwargs.get("timezone_name", ("""return Intl.DateTimeFormat().resolvedOptions().timeZone"""))
+        self.width = ("""return screen.width""")
+        self.height = ("""return screen.height""")
 
     def base36encode(self, number, alphabet="0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"):
         """Converts an integer to a base36 string."""
@@ -139,11 +147,12 @@ def gen_verifyFp(self):
 
         return f'verify_{scenario_title.lower()}_{"".join(uuid)}'
 
-    def sign_url(self, **kwargs):
+    def sign_url(self, calc_tt_params=False, **kwargs):
         url = kwargs.get("url", None)
         if url is None:
             raise Exception("sign_url required a url parameter")
 
+        tt_params = None
         if kwargs.get("gen_new_verifyFp", False):
             verifyFp = self.gen_verifyFp()
         else:
@@ -159,24 +168,37 @@ def sign_url(self, **kwargs):
         else:
             device_id = self.device_id
 
-        return (
-            verifyFp,
-            device_id,
+        url = '{}&verifyFp={}&device_id={}'.format(url, verifyFp, device_id)
+        # self.browser.execute_script(content=get_acrawler())
+        # Should be covered by an earlier addition of get_acrawler.
+        evaluatedPage =  (
             self.browser.execute_script(
                 '''
-        var url = "'''
-                + url
-                + "&verifyFp="
-                + verifyFp
-                + """&device_id="""
-                + device_id
-                + """"
+        var url = "'''+ url + """"
         var token = window.byted_acrawler.sign({url: url});
         return token;
         """
             ),
         )
 
+        url = '{}&_signature={}'.format(url, evaluatedPage)
+        # self.browser.execute_script(content=get_tt_params_script())
+        # Should be covered by an earlier addition of get_acrawler.
+
+        tt_params = self.browser.execute_script(
+            '''() => {
+                return window.genXTTParams(''' + json.dumps(dict(parse_qsl(splitquery(url)[1]))) + ''');
+        
+            }'''
+        ) 
+
+        return (
+            verifyFp,
+            device_id,
+            evaluatedPage,
+            tt_params
+        )
+
     def clean_up(self):
         try:
             self.browser.close()

TikTokApi/exceptions.py

@@ -23,3 +23,9 @@ class JSONDecodeFailure(Exception):
     def __init__(self, message="TikTok sent invalid JSON back"):
         self.message = message
         super().__init__(self.message)
+
+
+class TikTokNotAvailableError(Exception):
+    def __init__(self, message="The requested object is not available in this region"):
+        self.message = message
+        super().__init__(self.message)

TikTokApi/tiktok.py

@@ -52,9 +52,9 @@ def __init__(self, **kwargs):
             global BASE_URL
             BASE_URL = "https://t.tiktok.com/"
         if kwargs.get("use_selenium", False):
-            from .browser_selenium import browser
+            from .browser_utilities.browser_selenium import browser
         else:
-            from .browser import browser
+            from .browser_utilities.browser import browser
 
         if kwargs.get("generate_static_device_id", False):
             self.custom_device_id = "".join(random.choice(string.digits) for num in range(19))
@@ -100,8 +100,8 @@ class to prevent issues from arising with playwright
         * 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 
+            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
@@ -110,7 +110,7 @@ class to prevent issues from arising with playwright
 
         * custom_verifyFp: 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 
+            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.
@@ -231,9 +231,11 @@ def get_data(self, **kwargs) -> dict:
             verifyFp = kwargs.get("custom_verifyFp")
 
         tt_params = None
+        send_tt_params = kwargs.get("send_tt_params", False)
+
         if self.signer_url is None:
             kwargs["custom_verifyFp"] = verifyFp
-            verify_fp, device_id, signature, tt_params = self.browser.sign_url(**kwargs)
+            verify_fp, device_id, signature, tt_params = self.browser.sign_url(calc_tt_params=send_tt_params, **kwargs)
             userAgent = self.browser.userAgent
             referrer = self.browser.referrer
         else:
@@ -245,7 +247,6 @@ def get_data(self, **kwargs) -> dict:
 
         if not kwargs.get("send_tt_params", False):
             tt_params = None
-
 
         query = {"verifyFp": verify_fp, "device_id": device_id, "_signature": signature}
         url = "{}&{}".format(kwargs["url"], urlencode(query))
@@ -300,6 +301,12 @@ def get_data(self, **kwargs) -> dict:
                 raise TikTokNotFoundError(
                     "TikTok returned a response indicating the entity is invalid"
                 )
+            if json.get("statusCode", 200) == 10219:
+                # not available in this region
+                raise TikTokNotAvailableError(
+                    "Content not available for this region"
+                )
+
             return r.json()
         except ValueError as e:
             text = r.text
@@ -359,7 +366,7 @@ def get_bytes(self, **kwargs) -> bytes:
         ) = self.__process_kwargs__(kwargs)
         kwargs["custom_device_id"] = device_id
         if self.signer_url is None:
-            verify_fp, device_id, signature = self.browser.sign_url(**kwargs)
+            verify_fp, device_id, signature, _ = self.browser.sign_url(calc_tt_params=False, **kwargs)
             userAgent = self.browser.userAgent
             referrer = self.browser.referrer
         else:
@@ -870,6 +877,39 @@ def by_sound(self, id, count=30, offset=0, **kwargs) -> dict:
 
         return response[:count]
 
+
+    def by_sound_page(self, id, page_size=30, cursor=0, **kwargs) -> dict:
+        """Returns a page of tiktoks with a specific sound.
+
+        Parameters
+        ----------
+        id: The sound id to search by
+            Note: Can be found in the URL of the sound specific page or with other methods.
+        cursor: offset for pagination
+        page_size: The number of posts to return
+        """
+        (
+            region,
+            language,
+            proxy,
+            maxCount,
+            device_id,
+        ) = self.__process_kwargs__(kwargs)
+        kwargs["custom_device_id"] = device_id
+
+        query = {
+            "musicID": str(id),
+            "count": str(page_size),
+            "cursor": cursor,
+            "language": language,
+        }
+        api_url = "{}api/music/item_list/?{}&{}".format(
+            BASE_URL, self.__add_url_params__(), urlencode(query)
+        )
+
+        return self.get_data(url=api_url, send_tt_params=True, **kwargs)
+
+
     def get_music_object(self, id, **kwargs) -> dict:
         """Returns a music object for a specific sound id.
 
@@ -934,6 +974,10 @@ def get_music_object_full_by_api(self, id, **kwargs):
             BASE_URL, id, self.__add_url_params__()
         )
         res = self.get_data(url=api_url, **kwargs)
+
+        if res.get("statusCode", 200) == 10203:
+            raise TikTokNotFoundError()
+
         return res["musicInfo"]
 
     def by_hashtag(self, hashtag, count=30, offset=0, **kwargs) -> dict: