Spaces:
Paused
Paused
| from typing import List | |
| from fastapi import APIRouter, Query, Body, Request, HTTPException # 导入FastAPI组件 | |
| from app.api.models.APIResponseModel import ResponseModel, ErrorResponseModel # 导入响应模型 | |
| from crawlers.tiktok.web.web_crawler import TikTokWebCrawler # 导入TikTokWebCrawler类 | |
| router = APIRouter() | |
| TikTokWebCrawler = TikTokWebCrawler() | |
| # 获取单个作品数据 | |
| async def fetch_one_video(request: Request, | |
| itemId: str = Query(example="7339393672959757570", description="作品id/Video id")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取单个作品数据 | |
| ### 参数: | |
| - itemId: 作品id | |
| ### 返回: | |
| - 作品数据 | |
| # [English] | |
| ### Purpose: | |
| - Get single video data | |
| ### Parameters: | |
| - itemId: Video id | |
| ### Return: | |
| - Video data | |
| # [示例/Example] | |
| itemId = "7339393672959757570" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_one_video(itemId) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户的个人信息 | |
| async def fetch_user_profile(request: Request, | |
| uniqueId: str = Query(default="tiktok", description="用户uniqueId/User uniqueId"), | |
| secUid: str = Query(default="", description="用户secUid/User secUid"),): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户的个人信息 | |
| ### 参数: | |
| - secUid: 用户secUid | |
| - uniqueId: 用户uniqueId | |
| - secUid和uniqueId至少提供一个, 优先使用uniqueId, 也就是用户主页的链接中的用户名。 | |
| ### 返回: | |
| - 用户的个人信息 | |
| # [English] | |
| ### Purpose: | |
| - Get user profile | |
| ### Parameters: | |
| - secUid: User secUid | |
| - uniqueId: User uniqueId | |
| - At least one of secUid and uniqueId is provided, and uniqueId is preferred, that is, the username in the user's homepage link. | |
| ### Return: | |
| - User profile | |
| # [示例/Example] | |
| secUid = "MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM" | |
| uniqueId = "tiktok" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_user_profile(secUid, uniqueId) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户的作品列表 | |
| async def fetch_user_post(request: Request, | |
| secUid: str = Query(example="MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM", | |
| description="用户secUid/User secUid"), | |
| cursor: int = Query(default=0, description="翻页游标/Page cursor"), | |
| count: int = Query(default=35, description="每页数量/Number per page"), | |
| coverFormat: int = Query(default=2, description="封面格式/Cover format")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户的作品列表 | |
| ### 参数: | |
| - secUid: 用户secUid | |
| - cursor: 翻页游标 | |
| - count: 每页数量 | |
| - coverFormat: 封面格式 | |
| ### 返回: | |
| - 用户的作品列表 | |
| # [English] | |
| ### Purpose: | |
| - Get user posts | |
| ### Parameters: | |
| - secUid: User secUid | |
| - cursor: Page cursor | |
| - count: Number per page | |
| - coverFormat: Cover format | |
| ### Return: | |
| - User posts | |
| # [示例/Example] | |
| secUid = "MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM" | |
| cursor = 0 | |
| count = 35 | |
| coverFormat = 2 | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_user_post(secUid, cursor, count, coverFormat) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户的点赞列表 | |
| async def fetch_user_like(request: Request, | |
| secUid: str = Query( | |
| example="MS4wLjABAAAAq1iRXNduFZpY301UkVpJ1eQT60_NiWS9QQSeNqmNQEDJp0pOF8cpleNEdiJx5_IU", | |
| description="用户secUid/User secUid"), | |
| cursor: int = Query(default=0, description="翻页游标/Page cursor"), | |
| count: int = Query(default=35, description="每页数量/Number per page"), | |
| coverFormat: int = Query(default=2, description="封面格式/Cover format")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户的点赞列表 | |
| - 注意: 该接口需要用户点赞列表为公开状态 | |
| ### 参数: | |
| - secUid: 用户secUid | |
| - cursor: 翻页游标 | |
| - count: 每页数量 | |
| - coverFormat: 封面格式 | |
| ### 返回: | |
| - 用户的点赞列表 | |
| # [English] | |
| ### Purpose: | |
| - Get user likes | |
| - Note: This interface requires that the user's like list be public | |
| ### Parameters: | |
| - secUid: User secUid | |
| - cursor: Page cursor | |
| - count: Number per page | |
| - coverFormat: Cover format | |
| ### Return: | |
| - User likes | |
| # [示例/Example] | |
| secUid = "MS4wLjABAAAAq1iRXNduFZpY301UkVpJ1eQT60_NiWS9QQSeNqmNQEDJp0pOF8cpleNEdiJx5_IU" | |
| cursor = 0 | |
| count = 35 | |
| coverFormat = 2 | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_user_like(secUid, cursor, count, coverFormat) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户的收藏列表 | |
| async def fetch_user_collect(request: Request, | |
| cookie: str = Query(example="Your_Cookie", description="用户cookie/User cookie"), | |
| secUid: str = Query(example="Your_SecUid", description="用户secUid/User secUid"), | |
| cursor: int = Query(default=0, description="翻页游标/Page cursor"), | |
| count: int = Query(default=30, description="每页数量/Number per page"), | |
| coverFormat: int = Query(default=2, description="封面格式/Cover format")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户的收藏列表 | |
| - 注意: 该接口目前只能获取自己的收藏列表,需要提供自己账号的cookie。 | |
| ### 参数: | |
| - cookie: 用户cookie | |
| - secUid: 用户secUid | |
| - cursor: 翻页游标 | |
| - count: 每页数量 | |
| - coverFormat: 封面格式 | |
| ### 返回: | |
| - 用户的收藏列表 | |
| # [English] | |
| ### Purpose: | |
| - Get user favorites | |
| - Note: This interface can currently only get your own favorites list, you need to provide your account cookie. | |
| ### Parameters: | |
| - cookie: User cookie | |
| - secUid: User secUid | |
| - cursor: Page cursor | |
| - count: Number per page | |
| - coverFormat: Cover format | |
| ### Return: | |
| - User favorites | |
| # [示例/Example] | |
| cookie = "Your_Cookie" | |
| secUid = "Your_SecUid" | |
| cursor = 0 | |
| count = 30 | |
| coverFormat = 2 | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_user_collect(cookie, secUid, cursor, count, coverFormat) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户的播放列表 | |
| async def fetch_user_play_list(request: Request, | |
| secUid: str = Query(example="MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM", | |
| description="用户secUid/User secUid"), | |
| cursor: int = Query(default=0, description="翻页游标/Page cursor"), | |
| count: int = Query(default=30, description="每页数量/Number per page")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户的播放列表 | |
| ### 参数: | |
| - secUid: 用户secUid | |
| - cursor: 翻页游标 | |
| - count: 每页数量 | |
| ### 返回: | |
| - 用户的播放列表 | |
| # [English] | |
| ### Purpose: | |
| - Get user play list | |
| ### Parameters: | |
| - secUid: User secUid | |
| - cursor: Page cursor | |
| - count: Number per page | |
| ### Return: | |
| - User play list | |
| # [示例/Eample] | |
| secUid = "MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM" | |
| cursor = 0 | |
| count = 30 | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_user_play_list(secUid, cursor, count) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户的合辑列表 | |
| async def fetch_user_mix(request: Request, | |
| mixId: str = Query(example="7101538765474106158", | |
| description="合辑id/Mix id"), | |
| cursor: int = Query(default=0, description="翻页游标/Page cursor"), | |
| count: int = Query(default=30, description="每页数量/Number per page")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户的合辑列表 | |
| ### 参数: | |
| - mixId: 合辑id | |
| - cursor: 翻页游标 | |
| - count: 每页数量 | |
| ### 返回: | |
| - 用户的合辑列表 | |
| # [English] | |
| ### Purpose: | |
| - Get user mix list | |
| ### Parameters: | |
| - mixId: Mix id | |
| - cursor: Page cursor | |
| - count: Number per page | |
| ### Return: | |
| - User mix list | |
| # [示例/Eample] | |
| mixId = "7101538765474106158" | |
| cursor = 0 | |
| count = 30 | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_user_mix(mixId, cursor, count) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取作品的评论列表 | |
| async def fetch_post_comment(request: Request, | |
| aweme_id: str = Query(example="7304809083817774382", description="作品id/Video id"), | |
| cursor: int = Query(default=0, description="翻页游标/Page cursor"), | |
| count: int = Query(default=20, description="每页数量/Number per page"), | |
| current_region: str = Query(default="", description="当前地区/Current region")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取作品的评论列表 | |
| ### 参数: | |
| - aweme_id: 作品id | |
| - cursor: 翻页游标 | |
| - count: 每页数量 | |
| - current_region: 当前地区,默认为空。 | |
| ### 返回: | |
| - 作品的评论列表 | |
| # [English] | |
| ### Purpose: | |
| - Get video comments | |
| ### Parameters: | |
| - aweme_id: Video id | |
| - cursor: Page cursor | |
| - count: Number per page | |
| - current_region: Current region, default is empty. | |
| ### Return: | |
| - Video comments | |
| # [示例/Eample] | |
| aweme_id = "7304809083817774382" | |
| cursor = 0 | |
| count = 20 | |
| current_region = "" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_post_comment(aweme_id, cursor, count, current_region) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取作品的评论回复列表 | |
| async def fetch_post_comment_reply(request: Request, | |
| item_id: str = Query(example="7304809083817774382", description="作品id/Video id"), | |
| comment_id: str = Query(example="7304877760886588191", | |
| description="评论id/Comment id"), | |
| cursor: int = Query(default=0, description="翻页游标/Page cursor"), | |
| count: int = Query(default=20, description="每页数量/Number per page"), | |
| current_region: str = Query(default="", description="当前地区/Current region")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取作品的评论回复列表 | |
| ### 参数: | |
| - item_id: 作品id | |
| - comment_id: 评论id | |
| - cursor: 翻页游标 | |
| - count: 每页数量 | |
| - current_region: 当前地区,默认为空。 | |
| ### 返回: | |
| - 作品的评论回复列表 | |
| # [English] | |
| ### Purpose: | |
| - Get video comment replies | |
| ### Parameters: | |
| - item_id: Video id | |
| - comment_id: Comment id | |
| - cursor: Page cursor | |
| - count: Number per page | |
| - current_region: Current region, default is empty. | |
| ### Return: | |
| - Video comment replies | |
| # [示例/Eample] | |
| item_id = "7304809083817774382" | |
| comment_id = "7304877760886588191" | |
| cursor = 0 | |
| count = 20 | |
| current_region = "" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_post_comment_reply(item_id, comment_id, cursor, count, current_region) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户的粉丝列表 | |
| async def fetch_user_fans(request: Request, | |
| secUid: str = Query(example="MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM", | |
| description="用户secUid/User secUid"), | |
| count: int = Query(default=30, description="每页数量/Number per page"), | |
| maxCursor: int = Query(default=0, description="最大游标/Max cursor"), | |
| minCursor: int = Query(default=0, description="最小游标/Min cursor")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户的粉丝列表 | |
| ### 参数: | |
| - secUid: 用户secUid | |
| - count: 每页数量 | |
| - maxCursor: 最大游标 | |
| - minCursor: 最小游标 | |
| ### 返回: | |
| - 用户的粉丝列表 | |
| # [English] | |
| ### Purpose: | |
| - Get user followers | |
| ### Parameters: | |
| - secUid: User secUid | |
| - count: Number per page | |
| - maxCursor: Max cursor | |
| - minCursor: Min cursor | |
| ### Return: | |
| - User followers | |
| # [示例/Example] | |
| secUid = "MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM" | |
| count = 30 | |
| maxCursor = 0 | |
| minCursor = 0 | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_user_fans(secUid, count, maxCursor, minCursor) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户的关注列表 | |
| async def fetch_user_follow(request: Request, | |
| secUid: str = Query(example="MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM", | |
| description="用户secUid/User secUid"), | |
| count: int = Query(default=30, description="每页数量/Number per page"), | |
| maxCursor: int = Query(default=0, description="最大游标/Max cursor"), | |
| minCursor: int = Query(default=0, description="最小游标/Min cursor")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户的关注列表 | |
| ### 参数: | |
| - secUid: 用户secUid | |
| - count: 每页数量 | |
| - maxCursor: 最大游标 | |
| - minCursor: 最小游标 | |
| ### 返回: | |
| - 用户的关注列表 | |
| # [English] | |
| ### Purpose: | |
| - Get user followings | |
| ### Parameters: | |
| - secUid: User secUid | |
| - count: Number per page | |
| - maxCursor: Max cursor | |
| - minCursor: Min cursor | |
| ### Return: | |
| - User followings | |
| # [示例/Example] | |
| secUid = "MS4wLjABAAAAv7iSuuXDJGDvJkmH_vz1qkDZYo1apxgzaxdBSeIuPiM" | |
| count = 30 | |
| maxCursor = 0 | |
| minCursor = 0 | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_user_follow(secUid, count, maxCursor, minCursor) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| """-------------------------------------------------------utils接口列表-------------------------------------------------------""" | |
| # 生成真实msToken | |
| async def generate_real_msToken(request: Request): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 生成真实msToken | |
| ### 返回: | |
| - 真实msToken | |
| # [English] | |
| ### Purpose: | |
| - Generate real msToken | |
| ### Return: | |
| - Real msToken | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_real_msToken() | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 生成ttwid | |
| async def generate_ttwid(request: Request, | |
| cookie: str = Query(example="Your_Cookie", description="用户cookie/User cookie")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 生成ttwid | |
| ### 参数: | |
| - cookie: 用户cookie | |
| ### 返回: | |
| - ttwid | |
| # [English] | |
| ### Purpose: | |
| - Generate ttwid | |
| ### Parameters: | |
| - cookie: User cookie | |
| ### Return: | |
| - ttwid | |
| # [示例/Example] | |
| cookie = "Your_Cookie" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.fetch_ttwid(cookie) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 生成xbogus | |
| async def generate_xbogus(request: Request, | |
| url: str = Query( | |
| example="https://www.tiktok.com/api/item/detail/?WebIdLastTime=1712665533&aid=1988&app_language=en&app_name=tiktok_web&browser_language=en-US&browser_name=Mozilla&browser_online=true&browser_platform=Win32&browser_version=5.0%20%28Windows%29&channel=tiktok_web&cookie_enabled=true&device_id=7349090360347690538&device_platform=web_pc&focus_state=true&from_page=user&history_len=4&is_fullscreen=false&is_page_visible=true&language=en&os=windows&priority_region=US&referer=®ion=US&root_referer=https%3A%2F%2Fwww.tiktok.com%2F&screen_height=1080&screen_width=1920&webcast_language=en&tz_name=America%2FTijuana&msToken=AYFCEapCLbMrS8uTLBoYdUMeeVLbCdFQ_QF_-OcjzJw1CPr4JQhWUtagy0k4a9IITAqi5Qxr2Vdh9mgCbyGxTnvWLa4ZVY6IiSf6lcST-tr0IXfl-r_ZTpzvWDoQfqOVsWCTlSNkhAwB-tap5g==&itemId=7339393672959757570", | |
| description="未签名的API URL/Unsigned API URL"), | |
| user_agent: str = Query( | |
| example="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3", | |
| description="用户浏览器User-Agent/User browser User-Agent")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 生成xbogus | |
| ### 参数: | |
| - url: 未签名的API URL | |
| - user_agent: 用户浏览器User-Agent | |
| ### 返回: | |
| - xbogus | |
| # [English] | |
| ### Purpose: | |
| - Generate xbogus | |
| ### Parameters: | |
| - url: Unsigned API URL | |
| - user_agent: User browser User-Agent | |
| ### Return: | |
| - xbogus | |
| # [示例/Example] | |
| url = "https://www.tiktok.com/api/item/detail/?WebIdLastTime=1712665533&aid=1988&app_language=en&app_name=tiktok_web&browser_language=en-US&browser_name=Mozilla&browser_online=true&browser_platform=Win32&browser_version=5.0%20%28Windows%29&channel=tiktok_web&cookie_enabled=true&device_id=7349090360347690538&device_platform=web_pc&focus_state=true&from_page=user&history_len=4&is_fullscreen=false&is_page_visible=true&language=en&os=windows&priority_region=US&referer=®ion=US&root_referer=https%3A%2F%2Fwww.tiktok.com%2F&screen_height=1080&screen_width=1920&webcast_language=en&tz_name=America%2FTijuana&msToken=AYFCEapCLbMrS8uTLBoYdUMeeVLbCdFQ_QF_-OcjzJw1CPr4JQhWUtagy0k4a9IITAqi5Qxr2Vdh9mgCbyGxTnvWLa4ZVY6IiSf6lcST-tr0IXfl-r_ZTpzvWDoQfqOVsWCTlSNkhAwB-tap5g==&itemId=7339393672959757570" | |
| user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.gen_xbogus(url, user_agent) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 提取列表用户id | |
| async def get_sec_user_id(request: Request, | |
| url: str = Query( | |
| example="https://www.tiktok.com/@tiktok", | |
| description="用户主页链接/User homepage link")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 提取列表用户id | |
| ### 参数: | |
| - url: 用户主页链接 | |
| ### 返回: | |
| - 用户id | |
| # [English] | |
| ### Purpose: | |
| - Extract list user id | |
| ### Parameters: | |
| - url: User homepage link | |
| ### Return: | |
| - User id | |
| # [示例/Example] | |
| url = "https://www.tiktok.com/@tiktok" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.get_sec_user_id(url) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 提取列表用户id | |
| async def get_all_sec_user_id(request: Request, | |
| url: List[str] = Body( | |
| example=["https://www.tiktok.com/@tiktok"], | |
| description="用户主页链接/User homepage link")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 提取列表用户id | |
| ### 参数: | |
| - url: 用户主页链接 | |
| ### 返回: | |
| - 用户id | |
| # [English] | |
| ### Purpose: | |
| - Extract list user id | |
| ### Parameters: | |
| - url: User homepage link | |
| ### Return: | |
| - User id | |
| # [示例/Example] | |
| url = ["https://www.tiktok.com/@tiktok"] | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.get_all_sec_user_id(url) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 提取单个作品id | |
| async def get_aweme_id(request: Request, | |
| url: str = Query( | |
| example="https://www.tiktok.com/@owlcitymusic/video/7218694761253735723", | |
| description="作品链接/Video link")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 提取单个作品id | |
| ### 参数: | |
| - url: 作品链接 | |
| ### 返回: | |
| - 作品id | |
| # [English] | |
| ### Purpose: | |
| - Extract single video id | |
| ### Parameters: | |
| - url: Video link | |
| ### Return: | |
| - Video id | |
| # [示例/Example] | |
| url = "https://www.tiktok.com/@owlcitymusic/video/7218694761253735723" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.get_aweme_id(url) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 提取列表作品id | |
| async def get_all_aweme_id(request: Request, | |
| url: List[str] = Body( | |
| example=["https://www.tiktok.com/@owlcitymusic/video/7218694761253735723"], | |
| description="作品链接/Video link")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 提取列表作品id | |
| ### 参数: | |
| - url: 作品链接 | |
| ### 返回: | |
| - 作品id | |
| # [English] | |
| ### Purpose: | |
| - Extract list video id | |
| ### Parameters: | |
| - url: Video link | |
| ### Return: | |
| - Video id | |
| # [示例/Example] | |
| url = ["https://www.tiktok.com/@owlcitymusic/video/7218694761253735723"] | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.get_all_aweme_id(url) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取用户unique_id | |
| async def get_unique_id(request: Request, | |
| url: str = Query( | |
| example="https://www.tiktok.com/@tiktok", | |
| description="用户主页链接/User homepage link")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取用户unique_id | |
| ### 参数: | |
| - url: 用户主页链接 | |
| ### 返回: | |
| - unique_id | |
| # [English] | |
| ### Purpose: | |
| - Get user unique_id | |
| ### Parameters: | |
| - url: User homepage link | |
| ### Return: | |
| - unique_id | |
| # [示例/Example] | |
| url = "https://www.tiktok.com/@tiktok" | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.get_unique_id(url) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |
| # 获取列表unique_id列表 | |
| async def get_all_unique_id(request: Request, | |
| url: List[str] = Body( | |
| example=["https://www.tiktok.com/@tiktok"], | |
| description="用户主页链接/User homepage link")): | |
| """ | |
| # [中文] | |
| ### 用途: | |
| - 获取列表unique_id | |
| ### 参数: | |
| - url: 用户主页链接 | |
| ### 返回: | |
| - unique_id | |
| # [English] | |
| ### Purpose: | |
| - Get list unique_id | |
| ### Parameters: | |
| - url: User homepage link | |
| ### Return: | |
| - unique_id | |
| # [示例/Example] | |
| url = ["https://www.tiktok.com/@tiktok"] | |
| """ | |
| try: | |
| data = await TikTokWebCrawler.get_all_unique_id(url) | |
| return ResponseModel(code=200, | |
| router=request.url.path, | |
| data=data) | |
| except Exception as e: | |
| status_code = 400 | |
| detail = ErrorResponseModel(code=status_code, | |
| router=request.url.path, | |
| params=dict(request.query_params), | |
| ) | |
| raise HTTPException(status_code=status_code, detail=detail.dict()) | |