Fix SDK

Author: Evil0ctal

README.en.md

@@ -12,13 +12,13 @@
 <p>调用这些接口会使用你账户中的剩余请求次数！</p>
 
 ```python
-from tikhub.api import *
+from tikhub.__main__ import *
 
 # 初始化（Initialization)
 api = API(
-email='EMAIL@EXAMPLE.COM',
-password='PASSWORD',
-proxy=None,
+    email='EMAIL@EXAMPLE.COM',
+    password='PASSWORD',
+    proxy=None,
 )
 ```
 
@@ -51,7 +51,7 @@ pip install tikhub
 -   Usage
 
 ```python
-from tikhub.api import *
+from tikhub.__main__ import *
 
 
 async def async_test() -> None:

README.md

@@ -12,13 +12,13 @@
 <p>调用这些接口会使用你账户中的剩余请求次数！</p>
 
 ```python
-from tikhub.api import *
+from tikhub.__main__ import *
 
 # 初始化（Initialization)
 api = API(
-email='EMAIL@EXAMPLE.COM',
-password='PASSWORD',
-proxy=None,
+    email='EMAIL@EXAMPLE.COM',
+    password='PASSWORD',
+    proxy=None,
 )
 ```
 
@@ -51,7 +51,7 @@ pip install tikhub
 - Usage
 
 ```python
-from tikhub.api import *
+from tikhub.__main__ import *
 
 
 async def async_test() -> None:

setup.py

@@ -14,15 +14,15 @@
 setuptools.setup(
     name='tikhub',
     author='TikHub.io',
-    version='1.0.2',
+    version='1.0.3',
     license='Apache V2.0 License',
-    description='Douyin/TikTok async data scraper.',
+    description='TikHub Unofficial 抖音/TikTok/小红书 RESTful API',
     long_description=long_description,
     long_description_content_type="text/markdown",
     author_email='tikhub.io@proton.me',
     url='https://github.com/orgs/TikHubIO',
     packages=setuptools.find_packages(),
-    keywords='TikTok, Douyin, 抖音, Scraper, Crawler, API, Download, Video, No Watermark, Async',
+    keywords='TikTok, Douyin, 抖音, 小红书, Scraper, Crawler, API, Download, Video, No Watermark, Async',
     # 依赖包
     install_requires=[
         'aiohttp',

test/test.py

@@ -1,87 +1,49 @@
-from tikhub.api import *
+import asyncio
 
+from tikhub import TikTokAPI, DouyinAPI
 
-async def async_test() -> None:
-    # 异步测试/Async test
 
-    tiktok_url = 'https://www.tiktok.com/@evil0ctal/video/7156033831819037994'
-
-    tiktok_music_url = 'https://www.tiktok.com/music/original-sound-7128362040359488261'
-
-    douyin_url = 'https://www.douyin.com/video/7153585499477757192'
+if __name__ == '__main__':
+    token = input('Please enter your TikTok token: ')
 
-    douyin_user_url = 'https://www.douyin.com/user/MS4wLjABAAAAaNJuvXC83kL5nhaZHubKdjsRJQovgz58wXzlLnJUsslG-Kb24TM1QJlf_2HMaUJk'
+    tiktok_api = TikTokAPI(token)
+    douyin_api = DouyinAPI(token)
 
-    print("Test start...\n")
-    start_time = time.time()
+    tiktok_video_url = "https://www.tiktok.com/@evil0ctal/video/7201344014984006954"
 
-    # 获取TikHub请求头/Get TikHub request header
-    r = await api.user_login()
-    print("Running test : API.user_login()")
-    print(r)
+    r = None
 
-    # 获取TikHub用户信息/Get TikHub user information
-    print("Running test : API.get_user_info()")
-    r = await api.get_user_info()
+    # 读取用户信息/Read user information
+    r = asyncio.run(tiktok_api.get_user_info())
     print(r)
 
-    print("\nRunning ALL TikTok methods test...\n")
+    tiktok_video_url = 'https://www.tiktok.com/@evil0ctal/video/7156033831819037994'
+    tiktok_music_url = 'https://www.tiktok.com/music/original-sound-7128362040359488261'
 
-    # 获取单个视频数据/Get single video data
-    print("Running test : API.get_tiktok_video_data()")
-    r = await api.get_tiktok_video_data(tiktok_url)
+    # 解析单一tiktok视频/Parse a single tiktok video
+    # r = asyncio.run(tiktok_api.get_tiktok_video_data(tiktok_video_url))
     # print(r)
 
-    # 获取获取用户主页的所有视频数据/Get all video data on the user's homepage
-    print("Running test : API.get_tiktok_profile_videos()")
-    r = await api.get_tiktok_profile_videos(tiktok_url, cursor=None, count=None, get_all=False)
-    print(f'Get {len(r)} videos from profile')
-
-    # 获取用户主页的所有点赞视频数据/Get all liked video data on the user's homepage
-    print("Running test : API.get_tiktok_profile_liked_videos()")
-    r = await api.get_tiktok_profile_liked_videos(tiktok_url, cursor=None, count=None, get_all=False)
-    print(f'Get {len(r)} liked videos from profile')
-
-    # 获取TikTok视频的所有评论数据/Get all comment data of TikTok video
-    print("Running test : API.get_tiktok_video_comments()")
-    r = await api.get_tiktok_video_comments(tiktok_url, cursor=None, count=None, get_all=False)
-    print(f'Get {len(r)} comments from video')
-
-    # 获取音乐页面上的所有(理论上能抓取到的)视频数据/Get all (theoretically) video data on the music page
-    print("Running test : API.get_tiktok_music_videos()")
-    r = await api.get_tiktok_music_videos(tiktok_music_url, cursor=None, count=None, get_all=False)
-    print(f'Get {len(r)} videos from music')
-
-    print("\nRunning ALL Douyin methods test...\n")
-
-    # 获取单个视频数据/Get single video data
-    print("Running test : API.get_douyin_video_data()")
-    r = await api.get_douyin_video_data(douyin_url)
-
-    # 获取获取用户主页的所有视频数据/Get all video data on the user's homepage
-    print("Running test : API.get_douyin_profile_videos()")
-    r = await api.get_douyin_profile_videos(douyin_user_url, cursor=None, count=None, get_all=False)
-    print(f'Get {len(r)} videos from profile')
+    # 解析tiktok用户信息/Parse tiktok user information
+    # r = asyncio.run(tiktok_api.get_tiktok_user_data(tiktok_video_url))
+    # print(r)
 
-    # 获取用户主页的所有点赞视频数据/Get all liked video data on the user's homepage
-    print("Running test : API.get_douyin_profile_liked_videos()")
-    r = await api.get_douyin_profile_liked_videos(douyin_user_url, cursor=None, count=None, get_all=False)
-    print(f'Get {len(r)} liked videos from profile')
+    # 解析tiktok用户的所有视频数据/Parse all video data of tiktok user
+    # r = asyncio.run(tiktok_api.get_tiktok_profile_videos(tiktok_video_url, count=1))
+    # print(r)
 
-    # 获取抖音视频的所有评论数据/Get all comment data of Douyin video
-    print("Running test : API.get_douyin_video_comments()")
-    r = await api.get_douyin_video_comments(douyin_url, cursor=None, count=None, get_all=False)
-    print(f'Get {len(r)} comments from video')
+    # 解析tiktok用户的所有喜欢视频数据/Parse all liked video data of tiktok user
+    # r = asyncio.run(tiktok_api.get_tiktok_profile_liked_videos(tiktok_video_url, count=1))
+    # print(r)
 
-    # 总耗时/Total time
-    total_time = round(time.time() - start_time, 2)
-    print("\nTest completed, total time: {}s".format(total_time))
+    # 解析tiktok视频的所有评论数据/Parse all comment data of tiktok video
+    # r = asyncio.run(tiktok_api.get_tiktok_video_comments(tiktok_video_url, count=1))
+    # print(r)
 
+    # 解析配乐页视频数据/Parse video data of music page
+    # r = asyncio.run(tiktok_api.get_tiktok_music_videos(tiktok_music_url, count=1))
+    # print(r)
 
-if __name__ == '__main__':
-    api = API(
-        email='EMAIL@EXAMPLE.COM',
-        password='PASSWORD',
-        proxy=None,
-    )
-    asyncio.run(async_test())
+    # 解析tiktok热搜/Parse tiktok hot search
+    r = asyncio.run(tiktok_api.get_tiktok_search_data_hot('evil0ctal'))
+    print(r)

tikhub/AiohttpClient.py

@@ -0,0 +1,99 @@
+import aiohttp
+import asyncio
+
+
+class AiohttpClient:
+    _instance = None
+
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
+    async def request(self, method, url, headers=None, params=None, data=None, json=None):
+        """
+        发起单独的HTTP请求
+        :param method: 请求方法
+        :param url: 请求URL
+        :param headers: 请求头
+        :param params: URL参数
+        :param data: 表单数据
+        :param json: JSON数据
+        :return: 包含响应状态码、文本和JSON的字典。
+        """
+        # 发起一个单独的HTTP请求
+        async with aiohttp.ClientSession() as session:
+            async with session.request(method, url, headers=headers, params=params, data=data, json=json) as response:
+                # 获取响应内容
+                response_text = await response.text()
+                response_json = None
+                if response_text:
+                    try:
+                        response_json = await response.json()
+                    except Exception as e:
+                        print("Error in AiohttpClient.request when response.json():", e)
+                # 返回响应状态码、文本和JSON
+                # return response.status, response_text, response_json
+                result = {
+                    'status': response.status,
+                    'text': response_text,
+                    'json': response_json
+                }
+                return result
+
+    async def make_requests(self, requests: list):
+        """
+        批量发起HTTP请求
+        :param requests: 请求列表, 列表内是元组格式，[(method, url, headers, params, data, json)]。
+        :return: 包含所有请求结果的列表。
+        """
+        # 批量发起HTTP请求
+        async with aiohttp.ClientSession() as session:
+            # 创建异步任务列表
+            tasks = []
+            for request in requests:
+                method, url, headers, params, data, json = request
+                task = session.request(method, url, headers=headers, params=params, data=data, json=json)
+                tasks.append(task)
+            # 并发处理所有请求
+            results = []
+            for task in asyncio.as_completed(tasks):
+                # 等待异步任务完成
+                response = await task
+                # 获取响应内容
+                response_text = await response.text()
+                if response_text:
+                    try:
+                        response_json = await response.json()
+                    except Exception as e:
+                        response_json = None
+                # 将结果添加到列表中
+                results.append({
+                    str(response.url): {
+                        'method': response.method,
+                        'status': response.status,
+                        'text': response_text,
+                        'json': response_json
+                    }
+                })
+            # 返回所有请求的结果列表
+            return results
+
+
+if __name__ == '__main__':
+    # 批量发起HTTP请求
+    requests = [
+        # GET请求
+        ('GET',  # method
+         'https://api.tikhub.io/',  # url
+         {'Authorization': 'Bearer TOKEN'},  # headers
+         None, None, None),  # params, data, json
+        # POST请求
+        ('POST',
+         'https://api.tikhub.io/users/login',
+         {'Authorization': 'Bearer TOKEN'},
+         None, {'title': 'New post', 'content': 'Hello world!'}, None)
+    ]
+    client = AiohttpClient()
+    results = asyncio.run(client.make_requests(requests))
+    print(results)

tikhub/Auth.py

@@ -0,0 +1,58 @@
+import asyncio
+import platform
+
+from tikhub.AiohttpClient import AiohttpClient
+
+
+class Auth:
+    # 初始化/initialization
+    def __init__(self, token: str, domain: str = 'https://api.tikhub.io', proxy: str = None):
+        super().__init__()
+        self.token = token  # Token/token
+        self.domain = domain  # 域名/domain
+        self.proxies = proxy  # HTTP代理/HTTP proxy
+        self.headers = {'Authorization': f'Bearer {token}'} if 'bearer ' not in token.lower() else {'Authorization': token}
+        self.client = AiohttpClient()
+        if platform.system() == 'Windows':
+            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+    """__________________________________________⬇️️认证(authenticate)⬇️️__________________________________________"""
+
+    # 检查Token是否过期或无效/Check if the token is expired or invalid
+    @staticmethod
+    async def check_token(response: dict):
+        if response['status'] != 200:
+            error = Exception(f'Error: {response}')
+            raise error
+
+    # 获取TikHub用户信息/Get TikHub user information
+    async def get_user_info(self) -> dict:
+        url = f'{self.domain}/users/me/'
+        result = await self.client.request('GET', url, self.headers, None, None, None)
+        await self.check_token(result)
+        return result.get('json')
+
+    # 每日签到/Daily check-in
+    async def daily_check_in(self) -> dict:
+        url = f'{self.domain}/promotion/daily_check_in'
+        result = await self.client.request('GET', url, self.headers, None, None, None)
+        await self.check_token(result)
+        return result.get('json')
+
+
+if __name__ == '__main__':
+
+    token = input('请输入token: ')
+
+    auth = Auth(token)
+
+    r = None
+
+    # 读取用户信息/Read user information
+    r = asyncio.run(auth.get_user_info())
+    print(r)
+
+    # 每日签到/Daily check-in
+    r = asyncio.run(auth.daily_check_in())
+    print(r)
+