| 1 | """评论采集：针对单个作品拉取全部评论（可选含二级回复），导出为 JSON。 |
| 2 | |
| 3 | 设计要点： |
| 4 | - 复用 DouyinAPIClient 的分页请求与签名 |
| 5 | - 与下载流程解耦：作为独立的 helper，由 BaseDownloader 在保存媒体后按需调用 |
| 6 | - 输出位置：与媒体同目录，文件名 `{file_stem}_comments.json` |
| 7 | - 支持上限 max_comments（默认 0 = 不限）和 include_replies |
| 8 | """ |
| 9 | |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | import asyncio |
| 13 | from pathlib import Path |
| 14 | from typing import TYPE_CHECKING, Any, Dict, List, Optional |
| 15 | |
| 16 | from utils.logger import setup_logger |
| 17 | |
| 18 | if TYPE_CHECKING: # pragma: no cover |
| 19 | from core.api_client import DouyinAPIClient |
| 20 | from storage.metadata_handler import MetadataHandler |
| 21 | |
| 22 | logger = setup_logger("CommentsCollector") |
| 23 | |
| 24 | |
| 25 | class CommentsCollector: |
| 26 | def __init__( |
| 27 | self, |
| 28 | api_client: "DouyinAPIClient", |
| 29 | metadata_handler: "MetadataHandler", |
| 30 | *, |
| 31 | include_replies: bool = False, |
| 32 | max_comments: int = 0, |
| 33 | page_size: int = 20, |
| 34 | retry_delay_seconds: float = 1.0, |
| 35 | ): |
| 36 | self.api_client = api_client |
| 37 | self.metadata_handler = metadata_handler |
| 38 | self.include_replies = include_replies |
| 39 | self.max_comments = int(max_comments or 0) |
| 40 | self.page_size = max(1, int(page_size or 20)) |
| 41 | self.retry_delay_seconds = float(retry_delay_seconds or 1.0) |
| 42 | |
| 43 | async def collect_and_save(self, aweme_id: str, output_path: Path) -> Optional[Dict[str, Any]]: |
| 44 | """抓取评论并写入 output_path，失败时返回 None。""" |
| 45 | comments = await self.collect(aweme_id) |
| 46 | if comments is None: |
| 47 | return None |
| 48 | |
| 49 | payload = { |
| 50 | "aweme_id": aweme_id, |
| 51 | "count": len(comments), |
| 52 | "include_replies": self.include_replies, |
| 53 | "comments": comments, |
| 54 | } |
| 55 | # MetadataHandler.save_metadata 内部已吞异常并返回 bool |
| 56 | saved = await self.metadata_handler.save_metadata(payload, output_path) |
| 57 | if not saved: |
| 58 | logger.warning("Failed to save comments for %s to %s", aweme_id, output_path) |
| 59 | return None |
| 60 | return payload |
| 61 | |
| 62 | async def collect(self, aweme_id: str) -> Optional[List[Dict[str, Any]]]: |
| 63 | """抓取评论列表（不写盘），失败返回 None。""" |
| 64 | all_comments: List[Dict[str, Any]] = [] |
| 65 | cursor = 0 |
| 66 | seen_ids: set = set() |
| 67 | |
| 68 | while True: |
| 69 | try: |
| 70 | page = await self.api_client.get_aweme_comments( |
| 71 | aweme_id, |
| 72 | cursor=cursor, |
| 73 | count=self.page_size, |
| 74 | include_replies=self.include_replies, |
| 75 | ) |
| 76 | except Exception as exc: |
| 77 | logger.warning( |
| 78 | "Comments fetch error for %s cursor=%s: %s", |
| 79 | aweme_id, |
| 80 | cursor, |
| 81 | exc, |
| 82 | ) |
| 83 | return None |
| 84 | |
| 85 | items = page.get("items") or [] |
| 86 | if not items: |
| 87 | break |
| 88 | |
| 89 | for item in items: |
| 90 | if not isinstance(item, dict): |
| 91 | continue |
| 92 | cid = item.get("cid") or item.get("comment_id") |
| 93 | key = str(cid) if cid else None |
| 94 | if key and key in seen_ids: |
| 95 | continue |
| 96 | if key: |
| 97 | seen_ids.add(key) |
| 98 | all_comments.append(item) |
| 99 | if 0 < self.max_comments <= len(all_comments): |
| 100 | return all_comments[: self.max_comments] |
| 101 | |
| 102 | if not page.get("has_more"): |
| 103 | break |
| 104 | next_cursor = page.get("max_cursor") or 0 |
| 105 | if next_cursor == cursor: |
| 106 | # cursor 未推进但服务器称 has_more=True：可能是接口变更或异常返回， |
| 107 | # 升级为 warning 便于线上观察。 |
| 108 | logger.warning( |
| 109 | "Comments cursor stuck (aweme=%s, cursor=%s, has_more=True); " |
| 110 | "stopping to avoid infinite loop.", |
| 111 | aweme_id, |
| 112 | cursor, |
| 113 | ) |
| 114 | break |
| 115 | cursor = next_cursor |
| 116 | await asyncio.sleep(self.retry_delay_seconds * 0.1) # 轻度节流 |
| 117 | |
| 118 | return all_comments |
| 119 |