diff --git a/README.md b/README.md index 0c11c4553..35d0f9058 100644 --- a/README.md +++ b/README.md @@ -164,12 +164,22 @@ uv run main.py --platform xhs --lt qrcode --type search # 从配置文件中读取指定的帖子ID列表获取指定帖子的信息与评论信息 uv run main.py --platform xhs --lt qrcode --type detail +# 【仅小红书】爬取“收藏”的笔记：留空 config/xhs_config.py 的 XHS_COLLECT_USER_ID_LIST 即抓取当前登录账号自己收藏的笔记 +uv run main.py --platform xhs --lt qrcode --type collect + # 打开对应APP扫二维码登录 # 其他平台爬虫使用示例，执行下面的命令查看 uv run main.py --help ``` +> 💡 **小红书收藏笔记（`--type collect`）**：抓取「收藏」列表里的笔记（含笔记详情与评论）。 +> 在 `config/xhs_config.py` 中通过 `XHS_COLLECT_USER_ID_LIST` 控制抓取目标： +> - 留空 `[]` —— 抓取**当前登录账号自己**收藏的笔记（`user_id` 会自动获取）； +> - 也可填入用户 `user_id`，或带 `xsec_token` 的完整主页 URL —— 抓取他人**公开**的收藏。 +> +> 抓取数量由 `CRAWLER_MAX_NOTES_COUNT` 控制，结果写入 `data/xhs/` 目录（`collect_contents_*` / `collect_comments_*`）。 +
🖥️ WebUI 可视化操作界面 @@ -270,6 +280,9 @@ python main.py --platform xhs --lt qrcode --type search # 从配置文件中读取指定的帖子ID列表获取指定帖子的信息与评论信息 python main.py --platform xhs --lt qrcode --type detail +# 【仅小红书】爬取“收藏”的笔记（XHS_COLLECT_USER_ID_LIST 留空即抓当前登录账号自己收藏的笔记） +python main.py --platform xhs --lt qrcode --type collect + # 打开对应APP扫二维码登录 # 其他平台爬虫使用示例，执行下面的命令查看 diff --git a/cmd_arg/arg.py b/cmd_arg/arg.py index 20d1d3976..a7d989e41 100644 --- a/cmd_arg/arg.py +++ b/cmd_arg/arg.py @@ -63,6 +63,7 @@ class CrawlerTypeEnum(str, Enum): SEARCH = "search" DETAIL = "detail" CREATOR = "creator" + COLLECT = "collect" class SaveDataOptionEnum(str, Enum): diff --git a/config/base_config.py b/config/base_config.py index 28a852e08..d13d07278 100644 --- a/config/base_config.py +++ b/config/base_config.py @@ -28,7 +28,7 @@ LOGIN_TYPE = "qrcode" # qrcode or phone or cookie COOKIES = "" CRAWLER_TYPE = ( - "search" # Crawling type, search (keyword search) | detail (post details) | creator (creator homepage data) + "search" # Crawling type, search (keyword search) | detail (post details) | creator (creator homepage data) | collect (xhs only: the logged-in user's collected/收藏 notes) ) # Whether to enable IP proxy ENABLE_IP_PROXY = False diff --git a/config/xhs_config.py b/config/xhs_config.py index 02cc96645..0e254c7eb 100644 --- a/config/xhs_config.py +++ b/config/xhs_config.py @@ -35,3 +35,11 @@ "https://www.xiaohongshu.com/user/profile/5f58bd990000000001003753?xsec_token=ABYVg1evluJZZzpMX-VWzchxQ1qSNVW3r-jOEnKqMcgZw=&xsec_source=pc_search" # ........................ ] + +# 收藏笔记抓取（CRAWLER_TYPE = "collect" 时生效）。 +# 留空则抓取【当前登录账号自己】收藏的笔记（user_id 通过 /api/sns/web/v2/user/me 自动获取）。 +# 也可显式指定：可填裸 user_id，或带 xsec_token 的完整 profile URL（抓他人公开收藏）。 +XHS_COLLECT_USER_ID_LIST = [ + # "", # 例: 抓自己 -> 直接留空整个列表即可 + # "https://www.xiaohongshu.com/user/profile/xxxx?xsec_token=...&xsec_source=pc_search", +] diff --git a/media_platform/xhs/client.py b/media_platform/xhs/client.py index 1e07b0802..13e5a2246 100644 --- a/media_platform/xhs/client.py +++ b/media_platform/xhs/client.py @@ -654,6 +654,150 @@ async def get_all_notes_by_creator( ) return result + async def get_self_user_id(self) -> str: + """ + Get the logged-in user's own user_id. Used by the "collect" crawler mode + when no user_id is configured, so we crawl the currently logged-in + account's collected notes. + + The /user/me endpoint reliably returns the numeric user_id; selfinfo does + not always include it, so we try /me first and fall back to selfinfo. + + Returns: + str: user_id of the logged-in account, or "" if it cannot be resolved + """ + def _extract_uid(payload: Optional[Dict]) -> str: + if not payload: + return "" + data = payload.get("data", payload) or {} + return ( + data.get("user_id") + or data.get("userId") + or data.get("userid") + or "" + ) + + # 1) canonical "who am I" endpoint (pass empty params so the signer runs) + try: + me = await self.get("/api/sns/web/v2/user/me", {}) + uid = _extract_uid(me) + if uid: + return uid + utils.logger.info(f"[XiaoHongShuClient.get_self_user_id] /user/me returned no user_id, raw: {me}") + except Exception as e: + utils.logger.warning(f"[XiaoHongShuClient.get_self_user_id] /user/me failed: {e}") + + # 2) fallback to selfinfo + self_info = await self.query_self() + uid = _extract_uid(self_info) + if not uid: + utils.logger.info(f"[XiaoHongShuClient.get_self_user_id] selfinfo returned no user_id, raw: {self_info}") + return uid + + async def get_notes_by_collect( + self, + user_id: str, + cursor: str = "", + page_size: int = 30, + xsec_token: str = "", + xsec_source: str = "pc_user", + ) -> Dict: + """ + Get a single page of notes collected (收藏) by a user. + + Args: + user_id: The user whose collection to read (the logged-in user for own collection) + cursor: Pagination cursor returned from the previous page + page_size: Page data length + xsec_token: Verification token (only present when navigating from a profile URL) + xsec_source: Channel source + + Returns: + Dict: raw response containing "notes", "has_more" and "cursor" + """ + uri = "/api/sns/web/v2/note/collect/page" + params = { + "num": page_size, + "cursor": cursor, + "user_id": user_id, + "image_formats": "jpg,webp,avif", + } + # xsec_token/xsec_source are only required when the request originates from a + # profile URL; for the logged-in user's own collection they are not needed. + if xsec_token: + params["xsec_token"] = xsec_token + params["xsec_source"] = xsec_source + return await self.get(uri, params) + + async def get_all_notes_by_collect( + self, + user_id: str, + crawl_interval: float = 1.0, + callback: Optional[Callable] = None, + xsec_token: str = "", + xsec_source: str = "pc_user", + ) -> List[Dict]: + """ + Get all notes collected (收藏) by the specified user, paging until exhausted + or CRAWLER_MAX_NOTES_COUNT is reached. Mirrors get_all_notes_by_creator. + + Args: + user_id: The user whose collection to read + crawl_interval: Crawl delay (seconds) + callback: Called with each page of note items after it is fetched + xsec_token: Verification token + xsec_source: Channel source + + Returns: + List[Dict]: collected note items (each carries note_id and xsec_token) + """ + result: List[Dict] = [] + notes_has_more = True + notes_cursor = "" + while notes_has_more and len(result) < config.CRAWLER_MAX_NOTES_COUNT: + notes_res = await self.get_notes_by_collect( + user_id, notes_cursor, xsec_token=xsec_token, xsec_source=xsec_source + ) + if not notes_res: + utils.logger.error( + "[XiaoHongShuClient.get_all_notes_by_collect] Empty response; the collection may be private or the account restricted." + ) + break + + notes_has_more = notes_res.get("has_more", False) + notes_cursor = notes_res.get("cursor", "") + if "notes" not in notes_res: + utils.logger.info( + f"[XiaoHongShuClient.get_all_notes_by_collect] No 'notes' key found in response: {notes_res}" + ) + break + + notes = notes_res["notes"] + utils.logger.info( + f"[XiaoHongShuClient.get_all_notes_by_collect] got user_id:{user_id} collected notes len : {len(notes)}" + ) + + remaining = config.CRAWLER_MAX_NOTES_COUNT - len(result) + if remaining <= 0: + break + + notes_to_add = notes[:remaining] + # Collected-note items don't carry xsec_source; inject it so the + # downstream note-detail fetch has a valid source. + for note in notes_to_add: + note.setdefault("xsec_source", xsec_source or "pc_user") + + if callback: + await callback(notes_to_add) + + result.extend(notes_to_add) + await asyncio.sleep(crawl_interval) + + utils.logger.info( + f"[XiaoHongShuClient.get_all_notes_by_collect] Finished getting collected notes for user {user_id}, total: {len(result)}" + ) + return result + async def get_note_short_url(self, note_id: str) -> Dict: """ Get note short URL diff --git a/media_platform/xhs/core.py b/media_platform/xhs/core.py index 334fef395..c34bf05aa 100644 --- a/media_platform/xhs/core.py +++ b/media_platform/xhs/core.py @@ -121,6 +121,9 @@ async def start(self) -> None: elif config.CRAWLER_TYPE == "creator": # Get creator's information and their notes and comments await self.get_creators_and_notes() + elif config.CRAWLER_TYPE == "collect": + # Get the logged-in user's (or configured users') collected notes + await self.get_collected_notes() else: pass @@ -225,6 +228,61 @@ async def get_creators_and_notes(self) -> None: xsec_tokens.append(note_item.get("xsec_token")) await self.batch_get_note_comments(note_ids, xsec_tokens) + async def get_collected_notes(self) -> None: + """Get collected (收藏) notes for the logged-in user or configured users. + + If config.XHS_COLLECT_USER_ID_LIST is empty, the currently logged-in + account's own collection is crawled (user_id resolved via selfinfo). + Otherwise each entry may be a bare user_id or a full profile URL. + """ + utils.logger.info("[XiaoHongShuCrawler.get_collected_notes] Begin get Xiaohongshu collected notes") + + # Build the list of (user_id, xsec_token, xsec_source) targets + targets: List[CreatorUrlInfo] = [] + configured = getattr(config, "XHS_COLLECT_USER_ID_LIST", []) or [] + if configured: + for entry in configured: + entry = entry.strip() + if not entry: + continue + if "/user/profile/" in entry: + try: + targets.append(parse_creator_info_from_url(entry)) + except ValueError as e: + utils.logger.error(f"[XiaoHongShuCrawler.get_collected_notes] Failed to parse profile URL: {e}") + else: + targets.append(CreatorUrlInfo(user_id=entry, xsec_token="", xsec_source="pc_user")) + else: + self_user_id = await self.xhs_client.get_self_user_id() + if not self_user_id: + utils.logger.error( + "[XiaoHongShuCrawler.get_collected_notes] Could not resolve the logged-in user_id from selfinfo. " + "Make sure you are logged in, or set config.XHS_COLLECT_USER_ID_LIST." + ) + return + utils.logger.info(f"[XiaoHongShuCrawler.get_collected_notes] Using logged-in user_id: {self_user_id}") + targets.append(CreatorUrlInfo(user_id=self_user_id, xsec_token="", xsec_source="pc_user")) + + crawl_interval = config.CRAWLER_MAX_SLEEP_SEC + for target in targets: + all_notes_list = await self.xhs_client.get_all_notes_by_collect( + user_id=target.user_id, + crawl_interval=crawl_interval, + callback=self.fetch_creator_notes_detail, + xsec_token=target.xsec_token, + xsec_source=target.xsec_source or "pc_user", + ) + utils.logger.info( + f"[XiaoHongShuCrawler.get_collected_notes] user_id:{target.user_id} collected notes total: {len(all_notes_list)}" + ) + + note_ids = [] + xsec_tokens = [] + for note_item in all_notes_list: + note_ids.append(note_item.get("note_id")) + xsec_tokens.append(note_item.get("xsec_token")) + await self.batch_get_note_comments(note_ids, xsec_tokens) + async def fetch_creator_notes_detail(self, note_list: List[Dict]): """Concurrently obtain the specified post list and save the data""" semaphore = asyncio.Semaphore(config.MAX_CONCURRENCY_NUM) diff --git a/tests/test_xhs_collect.py b/tests/test_xhs_collect.py new file mode 100644 index 000000000..6ab9e7ae5 --- /dev/null +++ b/tests/test_xhs_collect.py @@ -0,0 +1,90 @@ +# -*- coding: utf-8 -*- + +import config +import pytest +from cmd_arg import parse_cmd +from media_platform.xhs import XiaoHongShuCrawler + + +@pytest.mark.asyncio +async def test_collect_cli_type_is_not_downgraded(): + """`--type collect` must be accepted, not silently fall back to search.""" + await parse_cmd(["--platform", "xhs", "--type", "collect"]) + assert config.CRAWLER_TYPE == "collect" + + +@pytest.mark.asyncio +async def test_get_collected_notes_uses_logged_in_user_when_list_empty(monkeypatch): + """Empty XHS_COLLECT_USER_ID_LIST -> crawl the logged-in account's own + collection: resolve its user_id and forward the collected notes to the + detail/comment pipeline.""" + monkeypatch.setattr(config, "XHS_COLLECT_USER_ID_LIST", []) + + crawler = XiaoHongShuCrawler() + calls = {} + + class FakeClient: + async def get_self_user_id(self): + return "5f0000000000000000000abc" + + async def get_all_notes_by_collect( + self, user_id, crawl_interval, callback, xsec_token, xsec_source + ): + calls["collect_user_id"] = user_id + return [ + {"note_id": "n1", "xsec_token": "t1"}, + {"note_id": "n2", "xsec_token": "t2"}, + ] + + crawler.xhs_client = FakeClient() + + async def fake_batch(note_ids, xsec_tokens): + calls["note_ids"] = note_ids + calls["xsec_tokens"] = xsec_tokens + + monkeypatch.setattr(crawler, "batch_get_note_comments", fake_batch) + + await crawler.get_collected_notes() + + assert calls["collect_user_id"] == "5f0000000000000000000abc" + assert calls["note_ids"] == ["n1", "n2"] + assert calls["xsec_tokens"] == ["t1", "t2"] + + +@pytest.mark.asyncio +async def test_get_collected_notes_parses_configured_profile_url(monkeypatch): + """A configured profile URL is parsed into user_id + xsec_token, and the + logged-in user's own id is NOT used.""" + url = ( + "https://www.xiaohongshu.com/user/profile/5f58bd990000000001003753" + "?xsec_token=ABtoken123=&xsec_source=pc_search" + ) + monkeypatch.setattr(config, "XHS_COLLECT_USER_ID_LIST", [url]) + + crawler = XiaoHongShuCrawler() + seen = {} + + class FakeClient: + async def get_self_user_id(self): + seen["self_called"] = True + return "SHOULD_NOT_BE_USED" + + async def get_all_notes_by_collect( + self, user_id, crawl_interval, callback, xsec_token, xsec_source + ): + seen["user_id"] = user_id + seen["xsec_token"] = xsec_token + return [] + + crawler.xhs_client = FakeClient() + + async def fake_batch(note_ids, xsec_tokens): + seen["batch_called"] = True + + monkeypatch.setattr(crawler, "batch_get_note_comments", fake_batch) + + await crawler.get_collected_notes() + + assert seen["user_id"] == "5f58bd990000000001003753" + assert seen["xsec_token"] == "ABtoken123=" + assert "self_called" not in seen