feat: restore Xiaohongshu read-only collection

2026-09-14 20:05:32 +08:00
parent 44a28954cf
commit 4c37d0c9bf
13 changed files with 1539 additions and 39 deletions
@@ -46,6 +46,7 @@ from .douyin import (
    DouyinBrowser,
    DouyinError,
    SubscriptionManager,
+    XiaohongshuBrowser,
 )
 from .proxy import ProxyExit, ProxyRegistry

@@ -64,6 +65,12 @@ DOUYIN_IDENTITY_PATH = "/aweme/v1/web/user/profile/self/"
 DOUYIN_IDENTITY_URL = IDENTITY_URL
 DOUYIN_WORKS_PATH = WORKS_PATH
 DOUYIN_COMMENTS_PATH = COMMENTS_PATH
+XHS_ACCOUNT_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$")
+XHS_IDENTITY_PATH = "/api/sns/web/v2/user/me"
+XHS_USER_POSTED_PATH = "/api/sns/web/v1/user_posted"
+XHS_COMMENTS_PATH = "/api/sns/web/v2/comment/page"
+XHS_SEARCH_PATH = "/api/sns/web/v2/search/notes"
+XHS_FEED_PATH = "/api/sns/web/v1/feed"


 def _noop() -> None:
@@ -85,12 +92,14 @@ class Gateway:
        token: str,
        self_name: str,
        browser: DouyinBrowser | None = None,
+        xiaohongshu_browser: XiaohongshuBrowser | None = None,
    ) -> None:
        self.docker = docker
        self.network = network
        self.token = token
        self.self_name = self_name
        self.browser = browser or DouyinBrowser(self._browser_endpoint)
+        self.xiaohongshu_browser = xiaohongshu_browser or XiaohongshuBrowser(self._browser_endpoint)
        self.proxies = ProxyRegistry()
        self.reservations = AliasReservationManager(docker, self_name)
        self.subscriptions = SubscriptionManager(self.browser)
@@ -666,6 +675,76 @@ class Gateway:
                )
        return identity

+    def get_xiaohongshu(self, alias: str, input: dict) -> dict:
+        target = input.get("url", "")
+        if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_url(target):
+            raise RequestError("invalid restricted Xiaohongshu request", 400)
+        with self._alias_lock(alias):
+            self._require_douyin_generation(alias, input)
+            try:
+                response = self.xiaohongshu_browser.get(alias, target)
+                self._require_douyin_generation(alias, input)
+            except DouyinError as exc:
+                LOG.warning("Xiaohongshu GET failed alias=%s reason=%s", alias, str(exc))
+                raise RequestError("restricted Xiaohongshu operation failed") from exc
+        return {"status": response.status, "body": response.body, "challenge": response.challenge}
+
+    def post_xiaohongshu(self, alias: str, input: dict) -> dict:
+        target = input.get("url", "")
+        body = input.get("body")
+        if (
+            not valid_xiaohongshu_generation(input)
+            or not valid_xhs_post_url(target)
+            or not isinstance(body, dict)
+        ):
+            raise RequestError("invalid restricted Xiaohongshu POST request", 400)
+        try:
+            encoded = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode()
+        except (TypeError, ValueError) as exc:
+            raise RequestError("invalid restricted Xiaohongshu POST body", 400) from exc
+        with self._alias_lock(alias):
+            self._require_douyin_generation(alias, input)
+            try:
+                response = self.xiaohongshu_browser.post(alias, target, encoded)
+                self._require_douyin_generation(alias, input)
+            except DouyinError as exc:
+                LOG.warning("Xiaohongshu POST failed alias=%s reason=%s", alias, str(exc))
+                raise RequestError("restricted Xiaohongshu operation failed") from exc
+        return {"status": response.status, "body": response.body, "challenge": response.challenge}
+
+    def get_xiaohongshu_media(self, alias: str, input: dict) -> dict:
+        target = input.get("url", "")
+        if not valid_xiaohongshu_generation(input) or not valid_xiaohongshu_media_url(target):
+            raise RequestError("invalid restricted Xiaohongshu media request", 400)
+        with self._alias_lock(alias):
+            self._require_douyin_generation(alias, input)
+            try:
+                response = self.xiaohongshu_browser.get_media(alias, target)
+                self._require_douyin_generation(alias, input)
+            except DouyinError as exc:
+                LOG.warning("Xiaohongshu media download failed alias=%s reason=%s", alias, str(exc))
+                raise RequestError("restricted Xiaohongshu media download failed") from exc
+        return {"status": response.status, "content_type": response.content_type, "body_base64": response.body_base64}
+
+    def xiaohongshu_identity(self, alias: str, input: dict) -> dict:
+        expected_account_key = input.get("expected_account_key", "")
+        if (
+            not valid_xiaohongshu_generation(input)
+            or not isinstance(expected_account_key, str)
+            or not XHS_ACCOUNT_KEY_RE.fullmatch(expected_account_key)
+        ):
+            raise RequestError("invalid Xiaohongshu identity request", 400)
+        with self._alias_lock(alias):
+            self._require_douyin_generation(alias, input)
+            try:
+                identity = self.xiaohongshu_browser.identity(alias)
+            except DouyinError as exc:
+                LOG.warning("Xiaohongshu identity verification failed alias=%s reason=%s", alias, str(exc))
+                raise RequestError("Xiaohongshu login identity could not be verified") from exc
+            if identity.get("uid") != expected_account_key:
+                raise RequestError("Xiaohongshu identity does not match the expected account", 409)
+        return identity
+
    def douyin_action(self, alias: str, input: dict) -> dict:
        expected_uid = input.get("expected_uid", "")
        action = input.get("action", "")
@@ -1130,6 +1209,20 @@ class GatewayHandler(BaseHTTPRequestHandler):
            if method == "POST" and action == "proxy":
                gateway.restore_proxy(alias, body)
                return None
+        match = re.fullmatch(
+            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/xiaohongshu/(get|post|media|identity)",
+            path,
+        )
+        if match:
+            alias, action = match.groups()
+            if action == "get" and method == "POST":
+                return gateway.get_xiaohongshu(alias, body)
+            if action == "post" and method == "POST":
+                return gateway.post_xiaohongshu(alias, body)
+            if action == "media" and method == "POST":
+                return gateway.get_xiaohongshu_media(alias, body)
+            if action == "identity" and method == "POST":
+                return gateway.xiaohongshu_identity(alias, body)
        match = re.fullmatch(
            r"/v1/browsers/([a-z0-9][a-z0-9-]{0,31})/douyin/(get|media|identity|action|events)",
            path,
@@ -1419,6 +1512,97 @@ def valid_douyin_generation(value: dict) -> bool:
    )


+def valid_xiaohongshu_generation(value: dict) -> bool:
+    return valid_douyin_generation(value)
+
+
+def valid_xhs_query(query: object, allowed: set[str], required: set[str] | None = None) -> bool:
+    if not isinstance(query, dict) or not isinstance(allowed, set):
+        return False
+    required = required or set()
+    if not required.issubset(query) or not set(query).issubset(allowed):
+        return False
+    for key, values in query.items():
+        if not isinstance(key, str) or not isinstance(values, list) or len(values) != 1:
+            return False
+        if not isinstance(values[0], str) or len(values[0]) > 2048 or "\r" in values[0] or "\n" in values[0]:
+            return False
+    return True
+
+
+def _valid_xhs_host(parsed: object, host: str) -> bool:
+    return (
+        getattr(parsed, "scheme", "") == "https"
+        and getattr(parsed, "hostname", None) == host
+        and getattr(parsed, "port", None) is None
+        and getattr(parsed, "username", None) is None
+        and getattr(parsed, "password", None) is None
+        and getattr(parsed, "fragment", "") == ""
+    )
+
+
+def valid_xhs_url(raw: object) -> bool:
+    if not isinstance(raw, str):
+        return False
+    try:
+        parsed = urlsplit(raw)
+        query = parse_qs(parsed.query, keep_blank_values=True)
+    except ValueError:
+        return False
+    if _valid_xhs_host(parsed, "edith.xiaohongshu.com") and parsed.path == XHS_IDENTITY_PATH:
+        return not query
+    if _valid_xhs_host(parsed, "edith.xiaohongshu.com") and parsed.path == XHS_USER_POSTED_PATH:
+        return valid_xhs_query(
+            query,
+            {"user_id", "cursor", "num", "image_formats", "xsec_source", "xsec_token"},
+            {"user_id", "num"},
+        ) and bool(XHS_ACCOUNT_KEY_RE.fullmatch(query["user_id"][0])) and query["num"] == ["30"]
+    if _valid_xhs_host(parsed, "edith.xiaohongshu.com") and parsed.path == XHS_COMMENTS_PATH:
+        return valid_xhs_query(
+            query,
+            {"note_id", "cursor", "top_comment_id", "image_formats", "xsec_source", "xsec_token"},
+            {"note_id", "cursor", "top_comment_id"},
+        ) and bool(XHS_ACCOUNT_KEY_RE.fullmatch(query["note_id"][0]))
+    return False
+
+
+def valid_xiaohongshu_url(raw: object) -> bool:
+    return valid_xhs_url(raw)
+
+
+def valid_xhs_post_url(raw: object) -> bool:
+    if not isinstance(raw, str):
+        return False
+    try:
+        parsed = urlsplit(raw)
+        query = parse_qs(parsed.query, keep_blank_values=True)
+    except ValueError:
+        return False
+    return (
+        _valid_xhs_host(parsed, "so.xiaohongshu.com")
+        and parsed.path == XHS_SEARCH_PATH
+        and not query
+    ) or (
+        _valid_xhs_host(parsed, "edith.xiaohongshu.com")
+        and parsed.path == XHS_FEED_PATH
+        and not query
+    )
+
+
+def valid_xiaohongshu_media_url(raw: object) -> bool:
+    if not isinstance(raw, str):
+        return False
+    try:
+        parsed = urlsplit(raw)
+        query = parse_qs(parsed.query, keep_blank_values=True)
+    except ValueError:
+        return False
+    if not _valid_xhs_host(parsed, "www.xiaohongshu.com"):
+        return False
+    parts = parsed.path.strip("/").split("/")
+    return len(parts) == 2 and parts[0] == "explore" and bool(XHS_ACCOUNT_KEY_RE.fullmatch(parts[1])) and valid_xhs_query(query, {"xsec_source", "xsec_token"})
+
+
 def valid_douyin_url(raw: object) -> bool:
    if not isinstance(raw, str):
        return False