"""Xiaohongshu browser implementation.""" from __future__ import annotations import json import time from urllib.parse import parse_qs, urlsplit import websocket from ..browser.cdp import RESOLVE_TIMEOUT, BrowserError from ..browser.response import ( BrowserResponse, detect_challenge, ) from .douyin import ( ACCOUNT_KEY_RE, RESPONSE_LIMIT, DouyinBrowser, DouyinError, ) XHS_ORIGIN = "https://www.xiaohongshu.com" XHS_API_ORIGIN = "https://edith.xiaohongshu.com" XHS_SEARCH_ORIGIN = "https://so.xiaohongshu.com" XHS_IDENTITY_URL = XHS_API_ORIGIN + "/api/sns/web/v2/user/me" XHS_ALLOWED_HOSTS = frozenset( {"www.xiaohongshu.com", "edith.xiaohongshu.com", "so.xiaohongshu.com"} ) class XiaohongshuBrowser(DouyinBrowser): def __init__(self, endpoint=None, *, target_id: str = "") -> None: super().__init__( endpoint, origin=XHS_ORIGIN, url_validator=is_xiaohongshu_url, media_validator=is_xiaohongshu_media_url, media_selector="video, img.note-slider-img", target_id=target_id, ) def post(self, alias: str, target: str, body: bytes) -> BrowserResponse: if not is_xiaohongshu_url(target) or len(body) > RESPONSE_LIMIT: raise DouyinError("restricted Xiaohongshu POST request is invalid") try: body_text = body.decode("utf-8") except UnicodeDecodeError as exc: raise DouyinError("restricted Xiaohongshu POST body is not UTF-8") from exc with self.connection(alias) as cdp: if cdp.evaluate("location.origin") != self.origin: raise DouyinError("restricted browser origin changed") expression = f"""(async()=>{{ const r=await fetch({json.dumps(target)},{{method:'POST',headers:{{'content-type':'application/json'}},body:{json.dumps(body_text)},credentials:'include',redirect:'error'}}); if(!r.body)return {{status:r.status,body:'',too_large:false}}; const reader=r.body.getReader(), decoder=new TextDecoder(); let size=0, responseBody=''; for(;;){{const item=await reader.read();if(item.done)break; if(size+item.value.byteLength>={RESPONSE_LIMIT}){{await reader.cancel();return {{too_large:true}};}} size+=item.value.byteLength;responseBody+=decoder.decode(item.value,{{stream:true}}); }} responseBody+=decoder.decode();return {{status:r.status,body:responseBody,too_large:false}}; }})()""" result = cdp.evaluate(expression) if ( not isinstance(result, dict) or result.get("too_large") or not isinstance(result.get("status"), int) ): raise DouyinError("restricted Xiaohongshu POST failed") status = result["status"] if 300 <= status < 400: raise DouyinError("restricted Xiaohongshu POST redirected") response_body = result.get("body") if not isinstance(response_body, str): raise DouyinError("restricted Xiaohongshu POST returned invalid body") return BrowserResponse( status, response_body, detect_challenge(status, response_body) ) def resolve(self, alias: str, target: str) -> str: if not is_xiaohongshu_share_url(target): raise DouyinError("restricted Xiaohongshu share URL is invalid") with self.connection(alias) as cdp: if cdp.evaluate("location.origin") not in {"null", self.origin}: raise DouyinError("restricted browser origin changed") try: cdp.command("Page.enable") cdp.notify("Page.navigate", {"url": target}) except (OSError, websocket.WebSocketException) as exc: raise DouyinError("Xiaohongshu share URL navigation failed") from exc deadline = time.monotonic() + RESOLVE_TIMEOUT final_url = "" while time.monotonic() < deadline: try: event = cdp.wait_event( "Page.frameNavigated", lambda params: isinstance(params.get("frame"), dict) and not params["frame"].get("parentId"), timeout=max(0.01, deadline - time.monotonic()), ) except BrowserError: break frame = event.get("params", {}).get("frame", {}) current_url = frame.get("url") if isinstance(frame, dict) else None if isinstance(current_url, str) and _is_xiaohongshu_page_url(current_url): final_url = current_url break if not final_url: raise DouyinError( "Xiaohongshu share URL did not resolve to a supported page" ) return final_url def identity(self, alias: str, expected_uid: str | None = None) -> dict: response = self.get(alias, XHS_IDENTITY_URL) try: payload = json.loads(response.body) except json.JSONDecodeError as exc: raise DouyinError("Xiaohongshu identity response is invalid") from exc data = payload.get("data") if isinstance(payload, dict) else None user_info = data.get("user_info") if isinstance(data, dict) else None user_id = data.get("user_id", "") if isinstance(data, dict) else "" nickname = data.get("nickname", "") if isinstance(data, dict) else "" if isinstance(user_info, dict): user_id = user_id or user_info.get("user_id", "") nickname = nickname or user_info.get("nickname", "") success = payload.get("success") if isinstance(payload, dict) else None if ( response.status != 200 or not isinstance(payload, dict) or not isinstance(success, bool) or not success or not isinstance(user_id, str) or not ACCOUNT_KEY_RE.fullmatch(user_id) or nickname is not None and not isinstance(nickname, str) ): raise DouyinError("Xiaohongshu login is not valid") if expected_uid and user_id != expected_uid: raise DouyinError( "Xiaohongshu identity does not match the expected account" ) return {"uid": user_id, "user_id": user_id, "nickname": nickname or ""} def is_xiaohongshu_media_url(value: object) -> bool: if not isinstance(value, str): return False try: parsed = urlsplit(value) port = parsed.port except (TypeError, ValueError): return False return ( parsed.scheme == "https" and parsed.hostname == "www.xiaohongshu.com" and port is None and parsed.username is None and parsed.password is None and parsed.fragment == "" and parsed.path.startswith("/explore/") ) def is_xiaohongshu_url(value: object) -> bool: if not isinstance(value, str): return False try: parsed = urlsplit(value) port = parsed.port except (TypeError, ValueError): return False return ( parsed.scheme == "https" and parsed.hostname in XHS_ALLOWED_HOSTS and port is None and parsed.username is None and parsed.password is None and parsed.fragment == "" ) def is_xiaohongshu_share_url(value: object) -> bool: if not isinstance(value, str): return False try: parsed = urlsplit(value) port = parsed.port except (TypeError, ValueError): return False path = parsed.path.strip("/") return ( parsed.scheme == "https" and parsed.hostname in {"xhslink.com", "www.xhslink.com"} and port is None and parsed.username is None and parsed.password is None and parsed.fragment == "" and bool(path) and len(path) <= 256 and not parsed.query ) def _is_xiaohongshu_page_url(value: object) -> bool: if not isinstance(value, str): return False try: parsed = urlsplit(value) port = parsed.port query = parse_qs(parsed.query, keep_blank_values=True) except (TypeError, ValueError): return False parts = parsed.path.strip("/").split("/") if not ( parsed.scheme == "https" and parsed.hostname == "www.xiaohongshu.com" and port is None and parsed.username is None and parsed.password is None and parsed.fragment == "" and ( len(parts) == 2 and parts[0] == "explore" or len(parts) == 3 and parts[:2] == ["user", "profile"] ) ): return False return all( key in {"xsec_token", "xsec_source"} and len(values) == 1 and len(values[0]) <= 2048 and not any(char in values[0] for char in "\r\n") for key, values in query.items() )