feat(xhs): add "collect" crawler type to fetch collected/收藏 notes

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_*`）。
+
 <details>
 <summary>🖥️ <strong>WebUI 可视化操作界面</strong></summary>
 
@@ -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扫二维码登录
 
 # 其他平台爬虫使用示例，执行下面的命令查看

cmd_arg/arg.py

@@ -63,6 +63,7 @@ class CrawlerTypeEnum(str, Enum):
     SEARCH = "search"
     DETAIL = "detail"
     CREATOR = "creator"
+    COLLECT = "collect"
 
 
 class SaveDataOptionEnum(str, Enum):

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

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",
+]

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

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)

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