返回 douyin-downloader
api_client.py
根目录 / core / api_client.py
1 from __future__ import annotations
2
3 import asyncio
4 import random
5 from typing import Any, Dict, List, Optional, Tuple
6 from urllib.parse import urlencode
7
8 import aiohttp
9
10 from auth import MsTokenManager
11 from utils.cookie_utils import sanitize_cookies
12 from utils.logger import setup_logger
13 from utils.xbogus import XBogus
14
15 try:
16 from utils.abogus import ABogus, BrowserFingerprintGenerator
17 except Exception: # pragma: no cover - optional dependency
18 ABogus = None
19 BrowserFingerprintGenerator = None
20
21 logger = setup_logger("APIClient")
22
23 _USER_AGENT_POOL = [
24 (
25 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
26 "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
27 ),
28 (
29 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
30 "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
31 ),
32 ]
33
34
35 class DouyinAPIClient:
36 BASE_URL = "https://www.douyin.com"
37 _BROWSER_COOKIE_BLOCKLIST = {
38 "sessionid",
39 "sessionid_ss",
40 "sid_tt",
41 "sid_guard",
42 "uid_tt",
43 "uid_tt_ss",
44 "passport_auth_status",
45 "passport_auth_status_ss",
46 "passport_assist_user",
47 "passport_auth_mix_state",
48 "passport_mfa_token",
49 "login_time",
50 }
51
52 def __init__(self, cookies: Dict[str, str], proxy: Optional[str] = None):
53 self.cookies = sanitize_cookies(cookies or {})
54 self.proxy = str(proxy or "").strip()
55 self._session: Optional[aiohttp.ClientSession] = None
56 self._browser_post_aweme_items: Dict[str, Dict[str, Any]] = {}
57 self._browser_post_stats: Dict[str, int] = {}
58 selected_ua = random.choice(_USER_AGENT_POOL)
59 self.headers = {
60 "User-Agent": selected_ua,
61 "Referer": "https://www.douyin.com/?recommend=1",
62 "Accept": "*/*",
63 "Accept-Encoding": "gzip, deflate",
64 "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
65 }
66 self._signer = XBogus(self.headers["User-Agent"])
67 self._ms_token_manager = MsTokenManager(user_agent=self.headers["User-Agent"])
68 self._ms_token = (self.cookies.get("msToken") or "").strip()
69 self._abogus_enabled = ABogus is not None and BrowserFingerprintGenerator is not None
70
71 async def __aenter__(self) -> "DouyinAPIClient":
72 await self._ensure_session()
73 return self
74
75 async def __aexit__(self, exc_type, exc, tb):
76 await self.close()
77
78 async def _ensure_session(self):
79 if self._session is None or self._session.closed:
80 self._session = aiohttp.ClientSession(
81 headers=self.headers,
82 cookies=self.cookies,
83 timeout=aiohttp.ClientTimeout(total=30),
84 raise_for_status=False,
85 )
86
87 async def close(self):
88 if self._session and not self._session.closed:
89 await self._session.close()
90
91 async def get_session(self) -> aiohttp.ClientSession:
92 await self._ensure_session()
93 if self._session is None:
94 raise RuntimeError("Failed to create aiohttp session")
95 return self._session
96
97 async def _ensure_ms_token(self) -> str:
98 if self._ms_token:
99 return self._ms_token
100
101 token = await asyncio.to_thread(
102 self._ms_token_manager.ensure_ms_token,
103 self.cookies,
104 )
105 self._ms_token = token.strip()
106 if self._ms_token:
107 self.cookies["msToken"] = self._ms_token
108 if self._session and not self._session.closed:
109 self._session.cookie_jar.update_cookies({"msToken": self._ms_token})
110 return self._ms_token
111
112 async def _default_query(self) -> Dict[str, Any]:
113 ms_token = await self._ensure_ms_token()
114 return {
115 "device_platform": "webapp",
116 "aid": "6383",
117 "channel": "channel_pc_web",
118 "update_version_code": "170400",
119 "pc_client_type": "1",
120 "pc_libra_divert": "Windows",
121 "version_code": "290100",
122 "version_name": "29.1.0",
123 "cookie_enabled": "true",
124 "screen_width": "1536",
125 "screen_height": "864",
126 "browser_language": "zh-CN",
127 "browser_platform": "Win32",
128 "browser_name": "Chrome",
129 "browser_version": "139.0.0.0",
130 "browser_online": "true",
131 "engine_name": "Blink",
132 "engine_version": "139.0.0.0",
133 "os_name": "Windows",
134 "os_version": "10",
135 "cpu_core_num": "16",
136 "device_memory": "8",
137 "platform": "PC",
138 "downlink": "10",
139 "effective_type": "4g",
140 "round_trip_time": "200",
141 "support_h265": "1",
142 "support_dash": "1",
143 "uifid": "",
144 "msToken": ms_token,
145 }
146
147 def sign_url(self, url: str) -> Tuple[str, str]:
148 signed_url, _xbogus, ua = self._signer.build(url)
149 return signed_url, ua
150
151 def build_signed_path(self, path: str, params: Dict[str, Any]) -> Tuple[str, str]:
152 query = urlencode(params)
153 base_url = f"{self.BASE_URL}{path}"
154 ab_signed = self._build_abogus_url(base_url, query)
155 if ab_signed:
156 return ab_signed
157 return self.sign_url(f"{base_url}?{query}")
158
159 def _build_abogus_url(self, base_url: str, query: str) -> Optional[Tuple[str, str]]:
160 if not self._abogus_enabled:
161 return None
162
163 try:
164 browser_fp = BrowserFingerprintGenerator.generate_fingerprint("Chrome")
165 signer = ABogus(fp=browser_fp, user_agent=self.headers["User-Agent"])
166 params_with_ab, _ab, ua, _body = signer.generate_abogus(query, "")
167 return f"{base_url}?{params_with_ab}", ua
168 except Exception as exc:
169 logger.warning("Failed to generate a_bogus, fallback to X-Bogus: %s", exc)
170 return None
171
172 async def _request_json(
173 self,
174 path: str,
175 params: Dict[str, Any],
176 *,
177 suppress_error: bool = False,
178 max_retries: int = 3,
179 ) -> Dict[str, Any]:
180 await self._ensure_session()
181 delays = [1, 2, 5]
182 last_exc: Optional[Exception] = None
183
184 for attempt in range(max_retries):
185 signed_url, ua = self.build_signed_path(path, params)
186 try:
187 async with self._session.get(
188 signed_url,
189 headers={**self.headers, "User-Agent": ua},
190 proxy=self.proxy or None,
191 ) as response:
192 if response.status == 200:
193 body = await response.read()
194 if not body:
195 # Empty 200 response is a common anti-bot signal
196 # from Douyin. Retry with a fresh signature.
197 logger.warning(
198 "Empty 200 response for %s (attempt %d/%d), "
199 "likely anti-bot; will retry",
200 path,
201 attempt + 1,
202 max_retries,
203 )
204 last_exc = RuntimeError(f"Empty 200 response for {path} (anti-bot)")
205 if attempt < max_retries - 1:
206 delay = delays[min(attempt, len(delays) - 1)]
207 await asyncio.sleep(delay)
208 continue
209 try:
210 data = await response.json(content_type=None)
211 except Exception:
212 import json as _json
213
214 try:
215 data = _json.loads(body)
216 except Exception:
217 logger.warning(
218 "Non-JSON 200 response for %s, length=%d",
219 path,
220 len(body),
221 )
222 return {}
223 return data if isinstance(data, dict) else {}
224 if response.status < 500 and response.status != 429:
225 log_fn = logger.debug if suppress_error else logger.error
226 log_fn(
227 "Request failed: path=%s, status=%s",
228 path,
229 response.status,
230 )
231 return {}
232 last_exc = RuntimeError(f"HTTP {response.status} for {path}")
233 except Exception as exc:
234 last_exc = exc
235
236 if attempt < max_retries - 1:
237 delay = delays[min(attempt, len(delays) - 1)]
238 logger.debug(
239 "Request retry %d/%d for %s in %ds",
240 attempt + 1,
241 max_retries,
242 path,
243 delay,
244 )
245 await asyncio.sleep(delay)
246
247 log_fn = logger.debug if suppress_error else logger.error
248 log_fn("Request failed after %d attempts: path=%s, error=%s", max_retries, path, last_exc)
249 return {}
250
251 @staticmethod
252 def _normalize_paged_response(
253 raw_data: Any,
254 *,
255 item_keys: Optional[List[str]] = None,
256 source: str = "api",
257 ) -> Dict[str, Any]:
258 raw = raw_data if isinstance(raw_data, dict) else {}
259 keys = item_keys or []
260 keys = ["items", *keys, "aweme_list", "mix_list", "music_list"]
261
262 items: List[Dict[str, Any]] = []
263 for key in keys:
264 value = raw.get(key)
265 if isinstance(value, list):
266 items = value
267 break
268
269 has_more_value = raw.get("has_more", False)
270 try:
271 has_more = bool(int(has_more_value))
272 except (TypeError, ValueError):
273 has_more = bool(has_more_value)
274
275 max_cursor_value = raw.get("max_cursor")
276 if max_cursor_value is None:
277 max_cursor_value = raw.get("cursor", 0)
278 try:
279 max_cursor = int(max_cursor_value or 0)
280 except (TypeError, ValueError):
281 max_cursor = 0
282
283 status_code_value = raw.get("status_code", 0)
284 try:
285 status_code = int(status_code_value or 0)
286 except (TypeError, ValueError):
287 status_code = 0
288
289 risk_flags = {
290 "login_tip": bool(
291 ((raw.get("not_login_module") or {}).get("guide_login_tip_exist"))
292 if isinstance(raw.get("not_login_module"), dict)
293 else False
294 ),
295 "verify_page": bool(raw.get("verify_ticket")),
296 }
297
298 normalized = {
299 "items": items,
300 "aweme_list": items, # 兼容旧调用方
301 "has_more": has_more,
302 "max_cursor": max_cursor,
303 "status_code": status_code,
304 "source": source,
305 "risk_flags": risk_flags,
306 "raw": raw,
307 }
308 for key, value in raw.items():
309 if key not in normalized:
310 normalized[key] = value
311 return normalized
312
313 async def _build_user_page_params(
314 self, sec_uid: str, max_cursor: int, count: int
315 ) -> Dict[str, Any]:
316 params = await self._default_query()
317 params.update(
318 {
319 "sec_user_id": sec_uid,
320 "max_cursor": max_cursor,
321 "count": count,
322 "locate_query": "false",
323 }
324 )
325 return params
326
327 # aid=1128 works for videos but filters out image/note content;
328 # aid=6383 works for notes/gallery but may miss some video content.
329 _DETAIL_AID_CANDIDATES = ("6383", "1128")
330
331 async def get_video_detail(
332 self, aweme_id: str, *, suppress_error: bool = False
333 ) -> Optional[Dict[str, Any]]:
334 for aid in self._DETAIL_AID_CANDIDATES:
335 params = await self._default_query()
336 params.update(
337 {
338 "aweme_id": aweme_id,
339 "aid": aid,
340 }
341 )
342
343 data = await self._request_json(
344 "/aweme/v1/web/aweme/detail/",
345 params,
346 suppress_error=(suppress_error or aid != self._DETAIL_AID_CANDIDATES[-1]),
347 )
348 if not data:
349 continue
350
351 detail = data.get("aweme_detail")
352 if detail:
353 return detail
354
355 # API returned data but aweme_detail is null — check if content was
356 # filtered (e.g. filter_reason="images_base" for note/gallery).
357 filter_info = data.get("filter_detail")
358 if isinstance(filter_info, dict) and filter_info.get("filter_reason"):
359 logger.info(
360 "Aweme %s filtered with aid=%s (reason=%s), retrying",
361 aweme_id,
362 aid,
363 filter_info["filter_reason"],
364 )
365 continue
366
367 # aweme_detail is null without a filter reason — no retry needed
368 break
369
370 return None
371
372 async def get_user_post(
373 self, sec_uid: str, max_cursor: int = 0, count: int = 18
374 ) -> Dict[str, Any]:
375 params = await self._build_user_page_params(sec_uid, max_cursor, count)
376 params.update(
377 {
378 "show_live_replay_strategy": "1",
379 "need_time_list": "1",
380 "time_list_query": "0",
381 "whale_cut_token": "",
382 "cut_version": "1",
383 "publish_video_strategy_type": "2",
384 }
385 )
386 raw = await self._request_json("/aweme/v1/web/aweme/post/", params)
387 return self._normalize_paged_response(raw, item_keys=["aweme_list"])
388
389 async def get_user_like(
390 self, sec_uid: str, max_cursor: int = 0, count: int = 20
391 ) -> Dict[str, Any]:
392 params = await self._build_user_page_params(sec_uid, max_cursor, count)
393 raw = await self._request_json("/aweme/v1/web/aweme/favorite/", params)
394 return self._normalize_paged_response(raw, item_keys=["aweme_list"])
395
396 async def get_user_mix(
397 self, sec_uid: str, max_cursor: int = 0, count: int = 20
398 ) -> Dict[str, Any]:
399 params = await self._build_user_page_params(sec_uid, max_cursor, count)
400 raw = await self._request_json("/aweme/v1/web/mix/list/", params)
401 return self._normalize_paged_response(raw, item_keys=["mix_list"])
402
403 async def get_user_music(
404 self, sec_uid: str, max_cursor: int = 0, count: int = 20
405 ) -> Dict[str, Any]:
406 params = await self._build_user_page_params(sec_uid, max_cursor, count)
407 raw = await self._request_json("/aweme/v1/web/music/list/", params)
408 return self._normalize_paged_response(raw, item_keys=["music_list"])
409
410 async def get_following_page(
411 self,
412 sec_uid: str,
413 *,
414 max_time: int = 0,
415 count: int = 20,
416 ) -> Dict[str, Any]:
417 """Fetch a single page of the logged-in account's following list.
418
419 Desktop-only: used by ``core/following.FollowingService`` to sync the
420 "My Following" tab. Douyin's web endpoint paginates via time-based
421 cursoring: the response contains ``min_time`` which must be passed as
422 ``max_time`` in the next request to get the next page. The ``count``
423 parameter is capped at 20 by the server regardless of what we send.
424
425 Returns a normalized dict with ``items``, ``has_more``, ``min_time``,
426 ``max_time``, ``status_code``, and ``raw`` (the full response).
427 """
428 params = await self._default_query()
429 params.update(
430 {
431 "user_id": sec_uid,
432 "sec_user_id": sec_uid,
433 "offset": 0,
434 "count": count,
435 "source_type": "1",
436 "gps_access": "0",
437 "address_book_access": "0",
438 "min_change": "0",
439 }
440 )
441 if max_time > 0:
442 params["max_time"] = max_time
443 raw = await self._request_json("/aweme/v1/web/user/following/list/", params)
444 normalized = self._normalize_paged_response(
445 raw,
446 item_keys=["followings", "follow_list", "user_list"],
447 )
448 # Expose the time-based pagination fields for the sync loop.
449 normalized["min_time"] = int(raw.get("min_time") or 0) if isinstance(raw, dict) else 0
450 normalized["max_time_resp"] = int(raw.get("max_time") or 0) if isinstance(raw, dict) else 0
451 return normalized
452
453 async def _build_collect_page_params(self, max_cursor: int, count: int) -> Dict[str, Any]:
454 params = await self._default_query()
455 params.update(
456 {
457 "cursor": max_cursor,
458 "count": count,
459 "version_code": "170400",
460 "version_name": "17.4.0",
461 }
462 )
463 return params
464
465 async def get_user_collects(
466 self, sec_uid: str, max_cursor: int = 0, count: int = 10
467 ) -> Dict[str, Any]:
468 if sec_uid and sec_uid != "self":
469 logger.warning("Collect folders currently require self sec_uid, got=%s", sec_uid)
470 return self._normalize_paged_response({}, item_keys=["collects_list"], source="api")
471
472 params = await self._build_collect_page_params(max_cursor, count)
473 raw = await self._request_json("/aweme/v1/web/collects/list/", params)
474 return self._normalize_paged_response(raw, item_keys=["collects_list"])
475
476 async def get_collect_aweme(
477 self, collects_id: str, max_cursor: int = 0, count: int = 10
478 ) -> Dict[str, Any]:
479 params = await self._build_collect_page_params(max_cursor, count)
480 params.update({"collects_id": collects_id})
481 raw = await self._request_json("/aweme/v1/web/collects/video/list/", params)
482 return self._normalize_paged_response(raw, item_keys=["aweme_list"])
483
484 async def get_user_collect_mix(
485 self, sec_uid: str, max_cursor: int = 0, count: int = 12
486 ) -> Dict[str, Any]:
487 if sec_uid and sec_uid != "self":
488 logger.warning("Collect mix currently require self sec_uid, got=%s", sec_uid)
489 return self._normalize_paged_response({}, item_keys=["mix_infos"], source="api")
490
491 params = await self._build_collect_page_params(max_cursor, count)
492 raw = await self._request_json("/aweme/v1/web/mix/listcollection/", params)
493 return self._normalize_paged_response(raw, item_keys=["mix_infos"])
494
495 async def get_user_info(self, sec_uid: str) -> Optional[Dict[str, Any]]:
496 params = await self._default_query()
497 params.update({"sec_user_id": sec_uid})
498
499 data = await self._request_json("/aweme/v1/web/user/profile/other/", params)
500 if data:
501 return data.get("user")
502 return None
503
504 async def get_self_info(self) -> Optional[Dict[str, Any]]:
505 """Fetch the logged-in user's own profile.
506
507 Uses the ``/aweme/v1/web/user/profile/self/`` endpoint which
508 identifies the user from the session cookies — no ``sec_uid``
509 parameter needed. Returns the ``user`` dict (containing
510 ``sec_uid``, ``uid``, ``nickname``, etc.) or ``None`` on failure.
511
512 Desktop-only: used by the Following sync to resolve the
513 logged-in user's ``sec_uid`` before calling
514 ``get_following_page``.
515 """
516 params = await self._default_query()
517 data = await self._request_json(
518 "/aweme/v1/web/user/profile/self/", params
519 )
520 if data:
521 return data.get("user")
522 return None
523
524 async def get_mix_detail(self, mix_id: str) -> Optional[Dict[str, Any]]:
525 params = await self._default_query()
526 params.update({"mix_id": mix_id})
527 data = await self._request_json("/aweme/v1/web/mix/detail/", params)
528 if not data:
529 return None
530 return data.get("mix_info") or data.get("mix_detail") or data
531
532 async def get_mix_aweme(self, mix_id: str, cursor: int = 0, count: int = 20) -> Dict[str, Any]:
533 params = await self._default_query()
534 params.update({"mix_id": mix_id, "cursor": cursor, "count": count})
535 raw = await self._request_json("/aweme/v1/web/mix/aweme/", params)
536 return self._normalize_paged_response(raw, item_keys=["aweme_list"])
537
538 async def get_music_detail(self, music_id: str) -> Optional[Dict[str, Any]]:
539 params = await self._default_query()
540 params.update({"music_id": music_id})
541 data = await self._request_json("/aweme/v1/web/music/detail/", params)
542 if not data:
543 return None
544 return data.get("music_info") or data.get("music_detail") or data
545
546 async def get_music_aweme(
547 self, music_id: str, cursor: int = 0, count: int = 20
548 ) -> Dict[str, Any]:
549 params = await self._default_query()
550 params.update({"music_id": music_id, "cursor": cursor, "count": count})
551 raw = await self._request_json("/aweme/v1/web/music/aweme/", params)
552 return self._normalize_paged_response(raw, item_keys=["aweme_list"])
553
554 async def get_live_room_info(
555 self, room_id: str, *, sec_user_id: str = ""
556 ) -> Optional[Dict[str, Any]]:
557 """通过房间号(web_rid)拉取直播间信息。
558
559 返回包含 room_info + stream_url 的 dict;若房间不在直播中或接口失败返回 None。
560 """
561 params = await self._default_query()
562 params.update(
563 {
564 "web_rid": room_id,
565 "room_id_str": room_id,
566 "enter_source": "",
567 "is_need_double_stream": "false",
568 "cookie_enabled": "true",
569 }
570 )
571 if sec_user_id:
572 params["sec_user_id"] = sec_user_id
573
574 raw = await self._request_json(
575 "/webcast/room/web/enter/",
576 params,
577 suppress_error=True,
578 )
579 if not raw:
580 return None
581
582 data_section = raw.get("data") if isinstance(raw.get("data"), dict) else raw
583 if not isinstance(data_section, dict):
584 return None
585
586 room_list = data_section.get("data")
587 room = None
588 if isinstance(room_list, list) and room_list:
589 first = room_list[0]
590 if isinstance(first, dict):
591 room = first
592 elif isinstance(data_section.get("room"), dict):
593 room = data_section.get("room")
594 elif isinstance(raw.get("room"), dict):
595 room = raw.get("room")
596
597 if not isinstance(room, dict):
598 return None
599
600 user = data_section.get("user") if isinstance(data_section, dict) else None
601 return {
602 "room": room,
603 "user": user if isinstance(user, dict) else {},
604 "raw": raw,
605 }
606
607 async def get_hot_search_board(self) -> Dict[str, Any]:
608 """获取抖音热搜榜。返回归一化 dict,items 为热搜词条列表。"""
609 params = await self._default_query()
610 params.update({"detail_list": "1", "source": "6"})
611 raw = await self._request_json(
612 "/aweme/v1/web/hot/search/list/", params, suppress_error=True
613 )
614 # 热榜返回结构中数据在 data.word_list 或 word_list
615 data_root = raw.get("data") if isinstance(raw.get("data"), dict) else raw
616 word_list = data_root.get("word_list") if isinstance(data_root, dict) else None
617 status_code = int(raw.get("status_code") or 0)
618 items = word_list if isinstance(word_list, list) else []
619 # 响应为空 + 非正常状态码时显式告警,方便排查 cookie 失效/签名失败
620 if not items and (status_code or not raw):
621 logger.warning(
622 "Hot search board returned no items (status_code=%s). "
623 "Check cookies / signature; Douyin may be rejecting the request.",
624 status_code,
625 )
626 return {
627 "items": items,
628 "has_more": False,
629 "max_cursor": 0,
630 "status_code": status_code,
631 "raw": raw,
632 }
633
634 async def search_aweme(
635 self,
636 keyword: str,
637 *,
638 offset: int = 0,
639 count: int = 10,
640 sort_type: int = 0,
641 publish_time: int = 0,
642 ) -> Dict[str, Any]:
643 """搜索作品。
644
645 Args:
646 sort_type: 0 综合 / 1 最多点赞 / 2 最新发布
647 publish_time: 0 不限 / 1 一天内 / 7 一周内 / 182 半年内
648 """
649 params = await self._default_query()
650 params.update(
651 {
652 "keyword": keyword,
653 "search_channel": "aweme_video_web",
654 "sort_type": sort_type,
655 "publish_time": publish_time,
656 "search_source": "normal_search",
657 "query_correct_type": "1",
658 "is_filter_search": 1 if (sort_type or publish_time) else 0,
659 "offset": offset,
660 "count": count,
661 }
662 )
663 raw = await self._request_json(
664 "/aweme/v1/web/general/search/single/", params, suppress_error=True
665 )
666 # 搜索结果每条在 data[].aweme_info;需要拍平
667 data_list = raw.get("data") if isinstance(raw.get("data"), list) else []
668 items: List[Dict[str, Any]] = []
669 for entry in data_list:
670 if not isinstance(entry, dict):
671 continue
672 aweme_info = entry.get("aweme_info")
673 if isinstance(aweme_info, dict):
674 items.append(aweme_info)
675
676 has_more_value = raw.get("has_more", 0)
677 try:
678 has_more = bool(int(has_more_value))
679 except (TypeError, ValueError):
680 has_more = bool(has_more_value)
681
682 cursor_value = raw.get("cursor") or raw.get("offset") or 0
683 try:
684 next_offset = int(cursor_value)
685 except (TypeError, ValueError):
686 next_offset = 0
687
688 status_code = int(raw.get("status_code") or 0)
689 if not items and (status_code or not raw):
690 logger.warning(
691 "Search returned no items for keyword=%r (status_code=%s, offset=%s). "
692 "Possible causes: cookies expired, signature rejected, or query blocked.",
693 keyword,
694 status_code,
695 offset,
696 )
697
698 return {
699 "items": items,
700 "has_more": has_more,
701 "max_cursor": next_offset,
702 "status_code": status_code,
703 "raw": raw,
704 }
705
706 async def get_aweme_comments(
707 self,
708 aweme_id: str,
709 *,
710 cursor: int = 0,
711 count: int = 20,
712 include_replies: bool = False,
713 ) -> Dict[str, Any]:
714 """获取作品评论列表(一页)。
715
716 Args:
717 aweme_id: 作品 ID
718 cursor: 分页游标(首次传 0)
719 count: 每页数量(抖音上限一般为 20)
720 include_replies: 是否拉取每条评论的二级回复(额外请求)
721 Returns:
722 归一化后的分页响应 dict,items 为评论列表。
723 """
724 params = await self._default_query()
725 params.update(
726 {
727 "aweme_id": aweme_id,
728 "cursor": cursor,
729 "count": count,
730 "item_type": "0",
731 "insert_ids": "",
732 "whale_cut_token": "",
733 "cut_version": "1",
734 "rcFT": "",
735 }
736 )
737 raw = await self._request_json("/aweme/v1/web/comment/list/", params)
738 normalized = self._normalize_paged_response(raw, item_keys=["comments"])
739
740 if include_replies:
741 comments = normalized.get("items") or []
742 for comment in comments:
743 if not isinstance(comment, dict):
744 continue
745 comment_id = comment.get("cid") or comment.get("comment_id")
746 if not comment_id or int(comment.get("reply_comment_total") or 0) <= 0:
747 continue
748 try:
749 reply_page = await self.get_aweme_comment_replies(
750 aweme_id=aweme_id, comment_id=str(comment_id), count=count
751 )
752 comment["_replies"] = reply_page.get("items") or []
753 except Exception as exc: # noqa: BLE001
754 logger.debug("Fetch reply for comment %s failed: %s", comment_id, exc)
755 return normalized
756
757 async def get_aweme_comment_replies(
758 self,
759 *,
760 aweme_id: str,
761 comment_id: str,
762 cursor: int = 0,
763 count: int = 20,
764 ) -> Dict[str, Any]:
765 """获取某条评论的二级回复列表。"""
766 params = await self._default_query()
767 params.update(
768 {
769 "item_id": aweme_id,
770 "comment_id": comment_id,
771 "cursor": cursor,
772 "count": count,
773 }
774 )
775 raw = await self._request_json("/aweme/v1/web/comment/list/reply/", params)
776 return self._normalize_paged_response(raw, item_keys=["comments"])
777
778 async def resolve_short_url(
779 self, short_url: str, *, timeout_seconds: float = 10.0
780 ) -> Optional[str]:
781 """跟随短链 302,返回最终 URL。失败时返回 None。
782
783 单独设置较短超时(默认 10s),避免被目标站挂死后拖慢整轮下载。
784 HTTP 状态码 ≥ 400 时视为解析失败,返回 None 以避免把错误页 URL
785 继续喂给下游 parser,从而在下游触发更隐晦的 "Unsupported URL" 噪声。
786 """
787 try:
788 await self._ensure_session()
789 async with self._session.get(
790 short_url,
791 allow_redirects=True,
792 timeout=aiohttp.ClientTimeout(total=timeout_seconds),
793 proxy=self.proxy or None,
794 ) as response:
795 final_url = str(response.url)
796 if response.status >= 400:
797 logger.warning(
798 "Short URL resolved with HTTP %s (treated as failure): %s -> %s",
799 response.status,
800 short_url,
801 final_url,
802 )
803 return None
804 return final_url
805 except asyncio.TimeoutError:
806 logger.error(
807 "Timeout resolving short URL after %.1fs: %s",
808 timeout_seconds,
809 short_url,
810 )
811 return None
812 except Exception as e:
813 logger.error("Failed to resolve short URL: %s, error: %s", short_url, e)
814 return None
815
816 async def collect_user_post_ids_via_browser(
817 self,
818 sec_uid: str,
819 *,
820 expected_count: int = 0,
821 headless: bool = False,
822 max_scrolls: int = 240,
823 idle_rounds: int = 8,
824 wait_timeout_seconds: int = 600,
825 ) -> List[str]:
826 try:
827 from playwright.async_api import async_playwright
828 except Exception as exc:
829 logger.warning("Playwright not available, browser fallback disabled: %s", exc)
830 return []
831
832 target_url = f"{self.BASE_URL}/user/{sec_uid}"
833 timeout_ms = max(30, int(wait_timeout_seconds)) * 1000
834 ids: List[str] = []
835 seen: set[str] = set()
836 post_api_ids: List[str] = []
837 post_api_seen: set[str] = set()
838 post_api_aweme_items: Dict[str, Dict[str, Any]] = {}
839 post_api_page_hits = 0
840 self._browser_post_aweme_items = {}
841 self._browser_post_stats = {}
842
843 def _merge(new_ids: List[str]):
844 for aweme_id in new_ids:
845 if aweme_id and aweme_id not in seen:
846 seen.add(aweme_id)
847 ids.append(aweme_id)
848
849 logger.warning(
850 "API翻页受限,启动浏览器兜底采集(可在弹出页面手动通过验证码/登录):%s",
851 target_url,
852 )
853
854 async with async_playwright() as playwright:
855 browser = await playwright.chromium.launch(
856 headless=headless,
857 args=[
858 "--disable-blink-features=AutomationControlled",
859 "--disable-dev-shm-usage",
860 "--no-sandbox",
861 ],
862 )
863 context = await browser.new_context(
864 user_agent=self.headers.get("User-Agent", ""),
865 locale="zh-CN",
866 viewport={"width": 1600, "height": 900},
867 )
868 cookies = self._browser_cookie_payload()
869 if cookies:
870 await context.add_cookies(cookies)
871
872 page = await context.new_page()
873 pending_response_tasks: List[asyncio.Task] = []
874
875 async def _handle_response(response):
876 nonlocal post_api_page_hits
877 url = response.url or ""
878 if "/aweme/v1/web/aweme/post/" not in url:
879 return
880 try:
881 data = await response.json()
882 except Exception:
883 return
884 aweme_items = data.get("aweme_list") if isinstance(data, dict) else None
885 if isinstance(aweme_items, list):
886 post_api_page_hits += 1
887 extracted: List[str] = []
888 for item in aweme_items:
889 if not isinstance(item, dict):
890 continue
891 aweme_id = item.get("aweme_id")
892 if not aweme_id:
893 continue
894 aweme_id_str = str(aweme_id)
895 extracted.append(aweme_id_str)
896 if aweme_id_str not in post_api_aweme_items:
897 post_api_aweme_items[aweme_id_str] = item
898 _merge(extracted)
899 for aweme_id in extracted:
900 if aweme_id not in post_api_seen:
901 post_api_seen.add(aweme_id)
902 post_api_ids.append(aweme_id)
903
904 def _on_response(response):
905 pending_response_tasks.append(asyncio.create_task(_handle_response(response)))
906
907 page.on("response", _on_response)
908
909 try:
910 try:
911 await page.goto(target_url, wait_until="domcontentloaded", timeout=timeout_ms)
912 except Exception as exc:
913 logger.warning(
914 "Browser goto timeout or error, continue with current page state: %s",
915 exc,
916 )
917
918 title = ""
919 try:
920 title = await page.title()
921 except Exception:
922 pass
923 if "验证码" in title:
924 if headless:
925 logger.warning(
926 "检测到验证码页面且当前为 headless 模式,无法人工验证。"
927 "请将 browser_fallback.headless 设为 false。"
928 )
929 return []
930 logger.warning("检测到验证码页面,请在浏览器中完成验证,程序会自动继续采集。")
931 await self._wait_for_manual_verification(
932 page, wait_timeout_seconds=wait_timeout_seconds
933 )
934 if not page.is_closed():
935 try:
936 await page.goto(
937 target_url,
938 wait_until="domcontentloaded",
939 timeout=timeout_ms,
940 )
941 except Exception as exc:
942 logger.warning("Reload user page after verification failed: %s", exc)
943
944 try:
945 warmup_seconds = min(20, max(3, int(wait_timeout_seconds)))
946 for _ in range(warmup_seconds):
947 if page.is_closed():
948 logger.warning("Browser page closed during warmup")
949 break
950 _merge(await self._extract_aweme_ids_from_page(page))
951 if ids:
952 break
953 await page.wait_for_timeout(1000)
954
955 stable_rounds = 0
956 max_scroll_rounds = max(1, int(max_scrolls))
957 idle_stop_rounds = max(1, int(idle_rounds))
958
959 for _ in range(max_scroll_rounds):
960 if page.is_closed():
961 logger.warning("Browser page closed during scrolling")
962 break
963 await page.mouse.wheel(0, 3800)
964 await page.wait_for_timeout(1200)
965
966 before = len(ids)
967 _merge(await self._extract_aweme_ids_from_page(page))
968 if len(ids) == before:
969 stable_rounds += 1
970 else:
971 stable_rounds = 0
972
973 if expected_count > 0 and len(ids) >= expected_count:
974 break
975 if expected_count <= 0 and stable_rounds >= idle_stop_rounds:
976 break
977 except Exception as exc:
978 logger.warning(
979 "Browser collection interrupted, use collected ids so far: %s",
980 exc,
981 )
982 finally:
983 if pending_response_tasks:
984 await asyncio.gather(*pending_response_tasks, return_exceptions=True)
985 try:
986 browser_cookies = await context.cookies(self.BASE_URL)
987 self._sync_browser_cookies(browser_cookies)
988 except Exception as exc:
989 logger.debug("Sync browser cookies skipped: %s", exc)
990 await context.close()
991 await browser.close()
992
993 selected_ids: List[str] = []
994 selected_seen: set[str] = set()
995 for aweme_id in post_api_ids + ids:
996 if aweme_id and aweme_id not in selected_seen:
997 selected_seen.add(aweme_id)
998 selected_ids.append(aweme_id)
999 self._browser_post_aweme_items = post_api_aweme_items
1000 self._browser_post_stats = {
1001 "merged_ids": len(ids),
1002 "post_api_ids": len(post_api_ids),
1003 "selected_ids": len(selected_ids),
1004 "post_items": len(post_api_aweme_items),
1005 "post_pages": post_api_page_hits,
1006 }
1007 logger.warning(
1008 "浏览器兜底采集 aweme_id: merged=%s, from_post_api=%s, selected=%s, post_items=%s",
1009 len(ids),
1010 len(post_api_ids),
1011 len(selected_ids),
1012 len(post_api_aweme_items),
1013 )
1014 return selected_ids
1015
1016 def pop_browser_post_aweme_items(self) -> Dict[str, Dict[str, Any]]:
1017 items = self._browser_post_aweme_items
1018 self._browser_post_aweme_items = {}
1019 return items
1020
1021 def pop_browser_post_stats(self) -> Dict[str, int]:
1022 stats = self._browser_post_stats
1023 self._browser_post_stats = {}
1024 return stats
1025
1026 def _browser_cookie_payload(self) -> List[Dict[str, str]]:
1027 payload: List[Dict[str, str]] = []
1028 for name, value in self.cookies.items():
1029 if not name:
1030 continue
1031 if name in self._BROWSER_COOKIE_BLOCKLIST:
1032 continue
1033 payload.append(
1034 {
1035 "name": str(name),
1036 "value": str(value or ""),
1037 "url": f"{self.BASE_URL}/",
1038 }
1039 )
1040 return payload
1041
1042 async def _extract_aweme_ids_from_page(self, page) -> List[str]:
1043 script = """
1044 () => {
1045 const result = [];
1046 const seen = new Set();
1047 const push = (id) => {
1048 if (!id || seen.has(id)) return;
1049 seen.add(id);
1050 result.push(id);
1051 };
1052
1053 const collectFrom = (text, pattern) => {
1054 if (!text) return;
1055 let match;
1056 while ((match = pattern.exec(text)) !== null) {
1057 push(match[1]);
1058 }
1059 };
1060
1061 const links = document.querySelectorAll("a[href]");
1062 for (const node of links) {
1063 const href = node.getAttribute("href") || "";
1064 collectFrom(href, /\\/video\\/(\\d{15,20})/g);
1065 collectFrom(href, /\\/note\\/(\\d{15,20})/g);
1066 }
1067
1068 const html = document.documentElement ? document.documentElement.innerHTML : "";
1069 collectFrom(html, /"aweme_id":"(\\d{15,20})"/g);
1070 collectFrom(html, /"group_id":"(\\d{15,20})"/g);
1071
1072 return result;
1073 }
1074 """
1075 try:
1076 data = await page.evaluate(script)
1077 if isinstance(data, list):
1078 return [str(x) for x in data if x]
1079 except Exception as exc:
1080 logger.debug("Extract aweme_id from page failed: %s", exc)
1081 return []
1082
1083 async def _wait_for_manual_verification(self, page, *, wait_timeout_seconds: int) -> None:
1084 deadline = asyncio.get_running_loop().time() + max(30, int(wait_timeout_seconds))
1085 while asyncio.get_running_loop().time() < deadline:
1086 if page.is_closed():
1087 logger.warning("Browser page closed while waiting manual verification")
1088 return
1089 title = ""
1090 try:
1091 title = await page.title()
1092 except Exception:
1093 pass
1094 if "验证码" not in title:
1095 logger.warning("验证码页面已退出,继续采集。")
1096 return
1097 await page.wait_for_timeout(1000)
1098
1099 logger.warning("等待手动验证超时(%ss),继续按当前页面状态采集。", wait_timeout_seconds)
1100
1101 def _sync_browser_cookies(self, browser_cookies: List[Dict[str, Any]]) -> None:
1102 merged: Dict[str, str] = {}
1103 for cookie in browser_cookies or []:
1104 if not isinstance(cookie, dict):
1105 continue
1106 name = str(cookie.get("name") or "").strip()
1107 value = str(cookie.get("value") or "").strip()
1108 domain = str(cookie.get("domain") or "")
1109 if not name or not value:
1110 continue
1111 if "douyin.com" not in domain:
1112 continue
1113 merged[name] = value
1114
1115 if not merged:
1116 return
1117
1118 self.cookies.update(merged)
1119 if self._session and not self._session.closed:
1120 self._session.cookie_jar.update_cookies(merged)
1121 logger.warning("Synced %s browser cookie(s) back to API client", len(merged))
1122
1122 lines PYTHON