237 lines
8.8 KiB
Python
237 lines
8.8 KiB
Python
"""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()
|
|
)
|