From d594c20c13f46c368bfec133de3a4f1a97cd4d79 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E5=91=98=E9=98=BF=E6=B1=9F=28Relakkes?= =?UTF-8?q?=29?= Date: Tue, 11 Aug 2026 18:10:16 +0800 Subject: [PATCH] =?UTF-8?q?fix(xhs,bilibili):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E8=AE=BF=E9=97=AE=E5=8F=97=E9=99=90=E5=BC=82=E5=B8=B8=E5=87=BB?= =?UTF-8?q?=E7=A9=BF=E4=B8=8E=E8=AF=84=E8=AE=BA=E9=87=87=E9=9B=86=E8=BE=B9?= =?UTF-8?q?=E7=95=8C=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit xhs: PR #958 把 IPBlockError / PlatformAccessError 加入 request() 的 retry_if_not_exception_type 后，tenacity 会直接重抛原异常而不再包装成 RetryError，core 层的 except 分支接不住，单条笔记被限流会让整批 asyncio.gather 抛出，同批已抓取但未入库的数据全部丢失。 - get_note_detail_async_task / get_creators_and_notes 捕获访问受限异常， 记录明确日志后跳过当前条目，恢复原有的"跳过并继续"语义 - 移除 request() 中已不可达的 IP_ERROR_CODE 分支 bilibili: 修复 get_video_all_comments 的两处翻页边界问题 - is_first_page 改为独立标志，接口返回 next=0 且 is_end=False 时 不再把后续页误判为首页而重复注入置顶评论 - result 无条件累加，否则开启楼中楼抓取时循环守卫永不推进， max_count 完全失效并可能死循环；截断提前到抓取楼中楼之前， 避免为已被丢弃的评论抓子评论 --- media_platform/bilibili/client.py | 15 +-- media_platform/xhs/client.py | 3 +- media_platform/xhs/core.py | 22 ++++- tests/test_bilibili_client_comments.py | 76 +++++++++++++++ tests/test_xhs_core_access_error.py | 124 +++++++++++++++++++++++++ 5 files changed, 231 insertions(+), 9 deletions(-) create mode 100644 tests/test_xhs_core_access_error.py diff --git a/media_platform/bilibili/client.py b/media_platform/bilibili/client.py index 7f46096..6d46e42 100644 --- a/media_platform/bilibili/client.py +++ b/media_platform/bilibili/client.py @@ -289,8 +289,8 @@ class BilibiliClient(AbstractApiClient, ProxyRefreshMixin): is_end = False next_page = 0 max_retries = 3 + is_first_page = True while not is_end and len(result) < max_count: - is_first_page = next_page == 0 comments_res = None for attempt in range(max_retries): try: @@ -332,6 +332,7 @@ class BilibiliClient(AbstractApiClient, ProxyRefreshMixin): for comment in comment_list if str(comment.get("rpid")) not in pinned_ids ] + is_first_page = False # Check if is_end and next exist if "is_end" not in cursor_info or "next" not in cursor_info: @@ -344,6 +345,10 @@ class BilibiliClient(AbstractApiClient, ProxyRefreshMixin): if not isinstance(is_end, bool): utils.logger.warning(f"[BilibiliClient.get_video_all_comments] 'is_end' is not a boolean for video_id: {video_id}. Assuming end of comments.") is_end = True + # Truncate before fetching sub-comments, otherwise max_count neither caps + # the stored comments nor stops us from crawling sub-comments we discard. + if len(result) + len(comment_list) > max_count: + comment_list = comment_list[:max_count - len(result)] if is_fetch_sub_comments: for comment in comment_list: comment_id = comment['rpid'] @@ -356,14 +361,12 @@ class BilibiliClient(AbstractApiClient, ProxyRefreshMixin): crawl_interval, callback, ) - if len(result) + len(comment_list) > max_count: - comment_list = comment_list[:max_count - len(result)] if callback: # If there is a callback function, execute it await callback(video_id, comment_list) await asyncio.sleep(crawl_interval) - if not is_fetch_sub_comments: - result.extend(comment_list) - continue + # Always accumulate, otherwise the `len(result) < max_count` loop guard + # never advances when sub-comment crawling is enabled. + result.extend(comment_list) return result async def get_video_all_level_two_comments( diff --git a/media_platform/xhs/client.py b/media_platform/xhs/client.py index 39bb663..4627cd2 100644 --- a/media_platform/xhs/client.py +++ b/media_platform/xhs/client.py @@ -188,8 +188,7 @@ class XiaoHongShuClient(AbstractApiClient, ProxyRefreshMixin): data: Dict = response_data if response_data is not None else response.json() if data["success"]: return data.get("data", data.get("success", {})) - elif data["code"] == self.IP_ERROR_CODE: - raise IPBlockError(self.IP_ERROR_STR) + # IP_ERROR_CODE / SECURITY_LIMIT_CODE are already handled above, before return_response. elif data["code"] in (self.NOTE_NOT_FOUND_CODE, self.NOTE_ABNORMAL_CODE): raise NoteNotFoundError(f"Note not found or abnormal, code: {data['code']}") else: diff --git a/media_platform/xhs/core.py b/media_platform/xhs/core.py index 334fef3..2dc64e9 100644 --- a/media_platform/xhs/core.py +++ b/media_platform/xhs/core.py @@ -42,7 +42,12 @@ from tools.cdp_browser import CDPBrowserManager from var import crawler_type_var, source_keyword_var from .client import XiaoHongShuClient -from .exception import DataFetchError, NoteNotFoundError +from .exception import ( + DataFetchError, + IPBlockError, + NoteNotFoundError, + PlatformAccessError, +) from .field import SearchSortType from .help import parse_note_info_from_note_url, parse_creator_info_from_url, get_search_id from .login import XiaoHongShuLogin @@ -206,6 +211,13 @@ class XiaoHongShuCrawler(AbstractCrawler): except ValueError as e: utils.logger.error(f"[XiaoHongShuCrawler.get_creators_and_notes] Failed to parse creator URL: {e}") continue + except (IPBlockError, PlatformAccessError) as e: + # Access restricted on the creator homepage, skip this creator instead of crashing the run. + utils.logger.error( + f"[XiaoHongShuCrawler.get_creators_and_notes] Access restricted for creator {creator_url}: {e}. " + f"建议降低采集频率、更换 IP 或检查账号状态" + ) + continue # Use fixed crawling interval crawl_interval = config.CRAWLER_MAX_SLEEP_SEC @@ -316,6 +328,14 @@ class XiaoHongShuCrawler(AbstractCrawler): except NoteNotFoundError as ex: utils.logger.warning(f"[XiaoHongShuCrawler.get_note_detail_async_task] Note not found: {note_id}, {ex}") return None + except (IPBlockError, PlatformAccessError) as ex: + # Access restricted (IP block / rate limit / account security). + # Skip this note instead of aborting the whole asyncio.gather batch. + utils.logger.error( + f"[XiaoHongShuCrawler.get_note_detail_async_task] Access restricted while getting note {note_id}: {ex}. " + f"建议降低采集频率、更换 IP 或检查账号状态" + ) + return None except DataFetchError as ex: utils.logger.error(f"[XiaoHongShuCrawler.get_note_detail_async_task] Get note detail error: {ex}") return None diff --git a/tests/test_bilibili_client_comments.py b/tests/test_bilibili_client_comments.py index 48bb66f..e283651 100644 --- a/tests/test_bilibili_client_comments.py +++ b/tests/test_bilibili_client_comments.py @@ -73,3 +73,79 @@ async def test_video_comments_accept_direct_top_comment(): ) assert [comment["rpid"] for comment in callbacks] == [1] + + +@pytest.mark.asyncio +async def test_pinned_comment_not_reinjected_when_cursor_next_stays_zero(): + """cursor.next 仍为 0 时继续翻页,置顶评论不能被重复注入。""" + client = object.__new__(BilibiliClient) + collected = [] + page_calls = 0 + + async def get_video_comments(video_id, order_mode, next_page): + nonlocal page_calls + page_calls += 1 + # 异常/兜底场景:接口一直回 next=0 且 is_end=False + return { + "cursor": {"is_end": False, "next": 0}, + "replies": [{"rpid": 2, "rcount": 0}], + "top": {"upper": {"rpid": 1, "rcount": 0}}, + } + + async def callback(video_id, comments): + collected.extend(comment["rpid"] for comment in comments) + + client.get_video_comments = get_video_comments + + await client.get_video_all_comments( + video_id="323173868", crawl_interval=0, callback=callback, max_count=5 + ) + + assert page_calls > 1, "该用例需要真正翻过页才有意义" + assert collected.count(1) == 1, f"置顶评论被重复采集: {collected}" + + +@pytest.mark.asyncio +async def test_max_count_applies_when_fetching_sub_comments(): + """开启楼中楼抓取时 max_count 依然生效,且不为被截断的评论抓子评论。""" + client = object.__new__(BilibiliClient) + collected = [] + sub_comment_calls = [] + page_calls = 0 + + async def get_video_comments(video_id, order_mode, next_page): + nonlocal page_calls + page_calls += 1 + base = page_calls * 10 + return { + "cursor": {"is_end": False, "next": page_calls}, + "replies": [ + {"rpid": base + 1, "rcount": 1}, + {"rpid": base + 2, "rcount": 1}, + ], + } + + async def get_video_all_level_two_comments( + video_id, comment_id, order_mode, ps, crawl_interval, callback + ): + sub_comment_calls.append(comment_id) + + async def callback(video_id, comments): + collected.extend(comment["rpid"] for comment in comments) + + client.get_video_comments = get_video_comments + client.get_video_all_level_two_comments = get_video_all_level_two_comments + + result = await client.get_video_all_comments( + video_id="323173868", + crawl_interval=0, + is_fetch_sub_comments=True, + callback=callback, + max_count=3, + ) + + # is_end 永远是 False,只能靠 max_count 收敛 + assert len(result) == 3 + assert collected == [11, 12, 21] + # 第 2 页的 rpid=22 被截断,不应该再去抓它的楼中楼 + assert sub_comment_calls == [11, 12, 21] diff --git a/tests/test_xhs_core_access_error.py b/tests/test_xhs_core_access_error.py new file mode 100644 index 0000000..0b8485b --- /dev/null +++ b/tests/test_xhs_core_access_error.py @@ -0,0 +1,124 @@ +# -*- coding: utf-8 -*- +""" +回归测试(小红书 xhs):访问受限异常不能击穿到 asyncio.gather。 + +背景: +IPBlockError / PlatformAccessError 被加入 request() 的 retry_if_not_exception_type 之后, +tenacity 会直接重抛原异常而不再包装成 RetryError。core 层原先只 catch +RetryError / NoteNotFoundError / DataFetchError / KeyError,导致单条笔记被限流 +就会让整批 gather 抛出,同批已抓到但未入库的数据全部丢失。 + +覆盖: +1. 笔记详情任务遇到 IPBlockError / PlatformAccessError 时返回 None 并跳过。 +2. 批量 gather 不会因为其中一条被限流而整体失败。 +3. 创作者主页被限流时跳过该创作者,而不是中断整个任务。 +""" + +import asyncio +from unittest.mock import AsyncMock + +import pytest + +import config +from media_platform.xhs.core import XiaoHongShuCrawler +from media_platform.xhs.exception import IPBlockError, PlatformAccessError + + +def make_crawler(xhs_client): + crawler = XiaoHongShuCrawler.__new__(XiaoHongShuCrawler) + crawler.xhs_client = xhs_client + return crawler + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + IPBlockError("Network connection error"), + PlatformAccessError("XHS request blocked with HTTP 403"), + ], +) +async def test_note_detail_task_skips_access_error(error): + xhs_client = AsyncMock() + xhs_client.get_note_by_id.side_effect = error + xhs_client.get_note_by_id_from_html.side_effect = AssertionError( + "被限流后不应再走 HTML 兜底" + ) + + result = await make_crawler(xhs_client).get_note_detail_async_task( + note_id="n1", + xsec_source="pc_search", + xsec_token="token", + semaphore=asyncio.Semaphore(1), + ) + + assert result is None + + +@pytest.mark.asyncio +async def test_gather_survives_single_blocked_note(): + """一条笔记被限流不能让同批次其他笔记的结果一起丢掉。""" + xhs_client = AsyncMock() + + async def get_note_by_id(note_id, xsec_source, xsec_token): + if note_id == "blocked": + raise PlatformAccessError("XHS account security restriction, code: 300011") + return {"note_id": note_id} + + xhs_client.get_note_by_id.side_effect = get_note_by_id + crawler = make_crawler(xhs_client) + + semaphore = asyncio.Semaphore(2) + results = await asyncio.gather( + *[ + crawler.get_note_detail_async_task( + note_id=note_id, + xsec_source="pc_search", + xsec_token="token", + semaphore=semaphore, + ) + for note_id in ("ok1", "blocked", "ok2") + ] + ) + + assert [r.get("note_id") if r else None for r in results] == ["ok1", None, "ok2"] + + +@pytest.mark.asyncio +async def test_creator_flow_skips_blocked_creator(monkeypatch): + """创作者主页 403 时跳过该创作者,不中断整个采集任务。""" + monkeypatch.setattr( + config, + "XHS_CREATOR_ID_LIST", + [ + "https://www.xiaohongshu.com/user/profile/blocked?xsec_token=a&xsec_source=pc_feed", + "https://www.xiaohongshu.com/user/profile/ok?xsec_token=b&xsec_source=pc_feed", + ], + raising=False, + ) + + crawled_user_ids = [] + xhs_client = AsyncMock() + + async def get_creator_info(user_id, xsec_token, xsec_source): + if user_id == "blocked": + raise PlatformAccessError("XHS request blocked with HTTP 403") + return {"user_id": user_id} + + async def get_all_notes_by_creator(user_id, **kwargs): + crawled_user_ids.append(user_id) + return [] + + xhs_client.get_creator_info.side_effect = get_creator_info + xhs_client.get_all_notes_by_creator.side_effect = get_all_notes_by_creator + + crawler = make_crawler(xhs_client) + crawler.batch_get_note_comments = AsyncMock() + monkeypatch.setattr( + "media_platform.xhs.core.xhs_store.save_creator", AsyncMock(), raising=False + ) + + await crawler.get_creators_and_notes() + + # 被限流的创作者整体跳过,后面的创作者照常采集 + assert crawled_user_ids == ["ok"]