mirror of
https://github.com/NanmiCoder/MediaCrawler.git
synced 2026-09-26 19:38:13 +08:00
程序员阿江(Relakkes)
380b426000
抖音在边缘网关新挂了 ArgusSecurityPlugin，对 aweme/detail、aweme/post 这批接口做 业务前置校验：缺少 x-tt-argus 时直接 403，响应体为 "Blocked by ArgusSecurityPlugin Uifid Not Found"；只补 uifid 参数但仍没有这个头 则是 "... Signature Not Found"——后者很容易被误判成 a_bogus / verifyFp 的问题。 网关当前不校验该头取值，传固定字符串即可（实测 "1" 就够）。 参考 https://github.com/Johnserf-Seed/f2/issues/443 注意这是权宜之计：网关哪天升级到真校验该值，会重新出现 Signature Not Found， 届时需要改为页面内注入 JS 让抖音自带 SDK 补齐 Argus 头。 验证：真实 cookie 下 get_video_by_id 恢复可用。 - client.py: 默认请求头补 x-tt-argus，cookie 有 UIFID 时带 uifid 头 （两种都没有则不发，避免被当成「有但为空」） - 新增 tests/test_douyin_argus_header.py（不发网络请求）
396 lines
16 KiB
Python
396 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
||
# Copyright (c) 2025 relakkes@gmail.com
|
||
#
|
||
# This file is part of MediaCrawler project.
|
||
# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/media_platform/douyin/client.py
|
||
# GitHub: https://github.com/NanmiCoder
|
||
# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
|
||
#
|
||
|
||
# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
|
||
# 1. 不得用于任何商业用途。
|
||
# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
|
||
# 3. 不得进行大规模爬取或对平台造成运营干扰。
|
||
# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
|
||
# 5. 不得用于任何非法或不当的用途。
|
||
#
|
||
# 详细许可条款请参阅项目根目录下的LICENSE文件。
|
||
# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。
|
||
|
||
import asyncio
|
||
import copy
|
||
import json
|
||
import urllib.parse
|
||
from typing import TYPE_CHECKING, Any, Callable, Dict, Union, Optional
|
||
|
||
import httpx
|
||
from playwright.async_api import BrowserContext
|
||
|
||
from base.base_crawler import AbstractApiClient
|
||
from proxy.proxy_mixin import ProxyRefreshMixin
|
||
from tools import utils
|
||
from tools.httpx_util import make_async_client
|
||
from var import request_keyword_var
|
||
|
||
if TYPE_CHECKING:
|
||
from proxy.proxy_ip_pool import ProxyIpPool
|
||
|
||
from .exception import *
|
||
from .field import *
|
||
from .help import *
|
||
|
||
# 抖音边缘网关 ArgusSecurityPlugin 要求的请求头。网关目前不校验取值，
|
||
# 传固定字符串即可；将来若开始真校验，会重新出现 "Signature Not Found"。
|
||
DOUYIN_ARGUS_HEADER_VALUE = "1"
|
||
|
||
|
||
class DouYinClient(AbstractApiClient, ProxyRefreshMixin):
|
||
|
||
def __init__(
|
||
self,
|
||
timeout=60, # If the crawl media option is turned on, Douyin’s short videos will require a longer timeout.
|
||
proxy=None,
|
||
*,
|
||
headers: Dict,
|
||
playwright_page: Optional[Page],
|
||
cookie_dict: Dict,
|
||
proxy_ip_pool: Optional["ProxyIpPool"] = None,
|
||
):
|
||
self.proxy = proxy
|
||
self.timeout = timeout
|
||
self.headers = headers
|
||
# 抖音边缘网关的 ArgusSecurityPlugin 会对这批接口做业务前置校验，缺少
|
||
# x-tt-argus 头时直接 403，响应体为
|
||
# "Blocked by ArgusSecurityPlugin Uifid Not Found"（补了 uifid 但没这个头则是
|
||
# "... Signature Not Found"）。当前网关尚未校验该头的值，可传任意字符串；
|
||
# 一旦升级到真校验，需要改为 WebView 内注入 JS 让页面自带 SDK 补齐。
|
||
self.headers.setdefault("x-tt-argus", DOUYIN_ARGUS_HEADER_VALUE)
|
||
uifid = cookie_dict.get("UIFID") or cookie_dict.get("UIFID_TEMP", "")
|
||
if uifid:
|
||
self.headers.setdefault("uifid", uifid)
|
||
self._host = "https://www.douyin.com"
|
||
self.cookie_urls = [
|
||
"https://douyin.com",
|
||
self._host,
|
||
"https://creator.douyin.com",
|
||
"https://douhot.douyin.com",
|
||
"https://live.douyin.com",
|
||
]
|
||
self.playwright_page = playwright_page
|
||
self.cookie_dict = cookie_dict
|
||
# Initialize proxy pool (from ProxyRefreshMixin)
|
||
self.init_proxy_pool(proxy_ip_pool)
|
||
|
||
async def __process_req_params(
|
||
self,
|
||
uri: str,
|
||
params: Optional[Dict] = None,
|
||
headers: Optional[Dict] = None,
|
||
request_method="GET",
|
||
):
|
||
|
||
if not params:
|
||
return
|
||
headers = headers or self.headers
|
||
local_storage: Dict = await self.playwright_page.evaluate("() => window.localStorage") # type: ignore
|
||
common_params = {
|
||
"device_platform": "webapp",
|
||
"aid": "6383",
|
||
"channel": "channel_pc_web",
|
||
"version_code": "190600",
|
||
"version_name": "19.6.0",
|
||
"update_version_code": "170400",
|
||
"pc_client_type": "1",
|
||
"cookie_enabled": "true",
|
||
"browser_language": "zh-CN",
|
||
"browser_platform": "MacIntel",
|
||
"browser_name": "Chrome",
|
||
"browser_version": "125.0.0.0",
|
||
"browser_online": "true",
|
||
"engine_name": "Blink",
|
||
"os_name": "Mac OS",
|
||
"os_version": "10.15.7",
|
||
"cpu_core_num": "8",
|
||
"device_memory": "8",
|
||
"engine_version": "109.0",
|
||
"platform": "PC",
|
||
"screen_width": "2560",
|
||
"screen_height": "1440",
|
||
'effective_type': '4g',
|
||
"round_trip_time": "50",
|
||
"webid": get_web_id(),
|
||
"msToken": local_storage.get("xmst"),
|
||
}
|
||
params.update(common_params)
|
||
query_string = urllib.parse.urlencode(params)
|
||
|
||
# 20240927 a-bogus update (JS version)
|
||
post_data = {}
|
||
if request_method == "POST":
|
||
post_data = params
|
||
|
||
if "/v1/web/general/search" not in uri:
|
||
a_bogus = await get_a_bogus(uri, query_string, post_data, headers["User-Agent"], self.playwright_page)
|
||
params["a_bogus"] = a_bogus
|
||
|
||
async def request(self, method, url, **kwargs):
|
||
# Check whether the proxy has expired before each request
|
||
await self._refresh_proxy_if_expired()
|
||
|
||
async with make_async_client(proxy=self.proxy) as client:
|
||
response = await client.request(method, url, timeout=self.timeout, **kwargs)
|
||
try:
|
||
if response.text == "" or response.text == "blocked":
|
||
utils.logger.error(f"request params incrr, response.text: {response.text}")
|
||
raise Exception("account blocked")
|
||
return response.json()
|
||
except Exception as e:
|
||
raise DataFetchError(f"{e}, {response.text}")
|
||
|
||
async def get(self, uri: str, params: Optional[Dict] = None, headers: Optional[Dict] = None):
|
||
"""
|
||
GET请求
|
||
"""
|
||
await self.__process_req_params(uri, params, headers)
|
||
headers = headers or self.headers
|
||
return await self.request(method="GET", url=f"{self._host}{uri}", params=params, headers=headers)
|
||
|
||
async def post(self, uri: str, data: dict, headers: Optional[Dict] = None):
|
||
await self.__process_req_params(uri, data, headers)
|
||
headers = headers or self.headers
|
||
return await self.request(method="POST", url=f"{self._host}{uri}", data=data, headers=headers)
|
||
|
||
async def pong(self, browser_context: BrowserContext) -> bool:
|
||
local_storage = await self.playwright_page.evaluate("() => window.localStorage")
|
||
if local_storage.get("HasUserLogin", "") == "1":
|
||
return True
|
||
|
||
_, cookie_dict = await utils.convert_browser_context_cookies(
|
||
browser_context,
|
||
urls=self.cookie_urls,
|
||
)
|
||
return cookie_dict.get("LOGIN_STATUS") == "1"
|
||
|
||
async def update_cookies(self, browser_context: BrowserContext, urls: Optional[list[str]] = None):
|
||
cookie_str, cookie_dict = await utils.convert_browser_context_cookies(
|
||
browser_context,
|
||
urls=urls or self.cookie_urls,
|
||
)
|
||
self.headers["Cookie"] = cookie_str
|
||
self.cookie_dict = cookie_dict
|
||
|
||
async def search_info_by_keyword(
|
||
self,
|
||
keyword: str,
|
||
offset: int = 0,
|
||
search_channel: SearchChannelType = SearchChannelType.GENERAL,
|
||
sort_type: SearchSortType = SearchSortType.GENERAL,
|
||
publish_time: PublishTimeType = PublishTimeType.UNLIMITED,
|
||
search_id: str = "",
|
||
):
|
||
"""
|
||
DouYin Web Search API
|
||
:param keyword:
|
||
:param offset:
|
||
:param search_channel:
|
||
:param sort_type:
|
||
:param publish_time: ·
|
||
:param search_id: ·
|
||
:return:
|
||
"""
|
||
query_params = {
|
||
'search_channel': search_channel.value,
|
||
'enable_history': '1',
|
||
'keyword': keyword,
|
||
'search_source': 'tab_search',
|
||
'query_correct_type': '1',
|
||
'is_filter_search': '0',
|
||
'from_group_id': '7378810571505847586',
|
||
'offset': offset,
|
||
'count': '15',
|
||
'need_filter_settings': '1',
|
||
'list_type': 'multi',
|
||
'search_id': search_id,
|
||
}
|
||
if sort_type.value != SearchSortType.GENERAL.value or publish_time.value != PublishTimeType.UNLIMITED.value:
|
||
query_params["filter_selected"] = json.dumps({"sort_type": str(sort_type.value), "publish_time": str(publish_time.value)})
|
||
query_params["is_filter_search"] = 1
|
||
query_params["search_source"] = "tab_search"
|
||
referer_url = f"https://www.douyin.com/search/{keyword}?aid=f594bbd9-a0e2-4651-9319-ebe3cb6298c1&type=general"
|
||
headers = copy.copy(self.headers)
|
||
headers["Referer"] = urllib.parse.quote(referer_url, safe=':/')
|
||
return await self.get("/aweme/v1/web/general/search/single/", query_params, headers=headers)
|
||
|
||
async def get_video_by_id(self, aweme_id: str) -> Any:
|
||
"""
|
||
DouYin Video Detail API
|
||
:param aweme_id:
|
||
:return:
|
||
"""
|
||
# 抖音 detail 接口的 Argus 风控要求这两个参数成套出现，缺一则直接 403
|
||
# （响应体为 "Blocked by ArgusSecurityPlugin Uifid Not Found"）：
|
||
# uifid = UIFID cookie，没有时退到 UIFID_TEMP
|
||
# verifyFp / fp = s_v_web_id cookie
|
||
# 必须用 cookie 里的 s_v_web_id：实测 uifid 搭配自生成的 verifyFp 会被判成
|
||
# "Signature Not Found"，两者同源才能通过。
|
||
s_v_web_id = self.cookie_dict.get("s_v_web_id", "")
|
||
params = {
|
||
"aweme_id": aweme_id,
|
||
"uifid": self.cookie_dict.get("UIFID") or self.cookie_dict.get("UIFID_TEMP", ""),
|
||
"verifyFp": s_v_web_id,
|
||
"fp": s_v_web_id,
|
||
}
|
||
headers = copy.copy(self.headers)
|
||
del headers["Origin"]
|
||
res = await self.get("/aweme/v1/web/aweme/detail/", params, headers)
|
||
return res.get("aweme_detail", {})
|
||
|
||
async def get_aweme_comments(self, aweme_id: str, cursor: int = 0):
|
||
"""get note comments
|
||
|
||
"""
|
||
uri = "/aweme/v1/web/comment/list/"
|
||
params = {"aweme_id": aweme_id, "cursor": cursor, "count": 20, "item_type": 0}
|
||
keywords = request_keyword_var.get()
|
||
referer_url = "https://www.douyin.com/search/" + keywords + '?aid=3a3cec5a-9e27-4040-b6aa-ef548c2c1138&publish_time=0&sort_type=0&source=search_history&type=general'
|
||
headers = copy.copy(self.headers)
|
||
headers["Referer"] = urllib.parse.quote(referer_url, safe=':/')
|
||
return await self.get(uri, params)
|
||
|
||
async def get_sub_comments(self, aweme_id: str, comment_id: str, cursor: int = 0):
|
||
"""
|
||
获取子评论
|
||
"""
|
||
uri = "/aweme/v1/web/comment/list/reply/"
|
||
params = {
|
||
'comment_id': comment_id,
|
||
"cursor": cursor,
|
||
"count": 20,
|
||
"item_type": 0,
|
||
"item_id": aweme_id,
|
||
}
|
||
keywords = request_keyword_var.get()
|
||
referer_url = "https://www.douyin.com/search/" + keywords + '?aid=3a3cec5a-9e27-4040-b6aa-ef548c2c1138&publish_time=0&sort_type=0&source=search_history&type=general'
|
||
headers = copy.copy(self.headers)
|
||
headers["Referer"] = urllib.parse.quote(referer_url, safe=':/')
|
||
return await self.get(uri, params)
|
||
|
||
async def get_aweme_all_comments(
|
||
self,
|
||
aweme_id: str,
|
||
crawl_interval: float = 1.0,
|
||
is_fetch_sub_comments=False,
|
||
callback: Optional[Callable] = None,
|
||
max_count: int = 10,
|
||
):
|
||
"""
|
||
获取帖子的所有评论，包括子评论
|
||
:param aweme_id: 帖子ID
|
||
:param crawl_interval: 抓取间隔
|
||
:param is_fetch_sub_comments: 是否抓取子评论
|
||
:param callback: 回调函数，用于处理抓取到的评论
|
||
:param max_count: 一次帖子爬取的最大评论数量
|
||
:return: 评论列表
|
||
"""
|
||
result = []
|
||
comments_has_more = 1
|
||
comments_cursor = 0
|
||
while comments_has_more and len(result) < max_count:
|
||
comments_res = await self.get_aweme_comments(aweme_id, comments_cursor)
|
||
comments_has_more = comments_res.get("has_more", 0)
|
||
comments_cursor = comments_res.get("cursor", 0)
|
||
comments = comments_res.get("comments", [])
|
||
if not comments:
|
||
continue
|
||
if len(result) + len(comments) > max_count:
|
||
comments = comments[:max_count - len(result)]
|
||
result.extend(comments)
|
||
if callback: # If there is a callback function, execute the callback function
|
||
await callback(aweme_id, comments)
|
||
|
||
await asyncio.sleep(crawl_interval)
|
||
if not is_fetch_sub_comments:
|
||
continue
|
||
# Get secondary reviews
|
||
for comment in comments:
|
||
reply_comment_total = comment.get("reply_comment_total")
|
||
|
||
if reply_comment_total > 0:
|
||
comment_id = comment.get("cid")
|
||
sub_comments_has_more = 1
|
||
sub_comments_cursor = 0
|
||
|
||
while sub_comments_has_more:
|
||
sub_comments_res = await self.get_sub_comments(aweme_id, comment_id, sub_comments_cursor)
|
||
sub_comments_has_more = sub_comments_res.get("has_more", 0)
|
||
sub_comments_cursor = sub_comments_res.get("cursor", 0)
|
||
sub_comments = sub_comments_res.get("comments", [])
|
||
|
||
if not sub_comments:
|
||
continue
|
||
result.extend(sub_comments)
|
||
if callback: # If there is a callback function, execute the callback function
|
||
await callback(aweme_id, sub_comments)
|
||
await asyncio.sleep(crawl_interval)
|
||
return result
|
||
|
||
async def get_user_info(self, sec_user_id: str):
|
||
uri = "/aweme/v1/web/user/profile/other/"
|
||
params = {
|
||
"sec_user_id": sec_user_id,
|
||
"publish_video_strategy_type": 2,
|
||
"personal_center_strategy": 1,
|
||
}
|
||
return await self.get(uri, params)
|
||
|
||
async def get_user_aweme_posts(self, sec_user_id: str, max_cursor: str = "") -> Dict:
|
||
uri = "/aweme/v1/web/aweme/post/"
|
||
params = {
|
||
"sec_user_id": sec_user_id,
|
||
"count": 18,
|
||
"max_cursor": max_cursor,
|
||
"locate_query": "false",
|
||
"publish_video_strategy_type": 2,
|
||
}
|
||
return await self.get(uri, params)
|
||
|
||
async def get_all_user_aweme_posts(self, sec_user_id: str, callback: Optional[Callable] = None):
|
||
posts_has_more = 1
|
||
max_cursor = ""
|
||
result = []
|
||
while posts_has_more == 1:
|
||
aweme_post_res = await self.get_user_aweme_posts(sec_user_id, max_cursor)
|
||
posts_has_more = aweme_post_res.get("has_more", 0)
|
||
max_cursor = aweme_post_res.get("max_cursor")
|
||
aweme_list = aweme_post_res.get("aweme_list") if aweme_post_res.get("aweme_list") else []
|
||
utils.logger.info(f"[DouYinClient.get_all_user_aweme_posts] get sec_user_id:{sec_user_id} video len : {len(aweme_list)}")
|
||
if callback:
|
||
await callback(aweme_list)
|
||
result.extend(aweme_list)
|
||
return result
|
||
|
||
async def resolve_short_url(self, short_url: str) -> str:
|
||
"""
|
||
解析抖音短链接,获取重定向后的真实URL
|
||
Args:
|
||
short_url: 短链接,如 https://v.douyin.com/iF12345ABC/
|
||
Returns:
|
||
重定向后的完整URL
|
||
"""
|
||
async with make_async_client(proxy=self.proxy, follow_redirects=False) as client:
|
||
try:
|
||
utils.logger.info(f"[DouYinClient.resolve_short_url] Resolving short URL: {short_url}")
|
||
response = await client.get(short_url, timeout=10)
|
||
|
||
# Short links usually return a 302 redirect
|
||
if response.status_code in [301, 302, 303, 307, 308]:
|
||
redirect_url = response.headers.get("Location", "")
|
||
utils.logger.info(f"[DouYinClient.resolve_short_url] Resolved to: {redirect_url}")
|
||
return redirect_url
|
||
else:
|
||
utils.logger.warning(f"[DouYinClient.resolve_short_url] Unexpected status code: {response.status_code}")
|
||
return ""
|
||
except Exception as e:
|
||
utils.logger.error(f"[DouYinClient.resolve_short_url] Failed to resolve short URL: {e}")
|
||
return ""
|