Spaces:
Running on Zero
Running on Zero
| """Python port of scraper.rs (Rednote / Xiaohongshu scraper).""" | |
| import json | |
| import re | |
| import httpx | |
| UA_DESKTOP = ( | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " | |
| "(KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36" | |
| ) | |
| UA_MOBILE = ( | |
| "Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) AppleWebKit/605.1.15 " | |
| "(KHTML, like Gecko) Version/17.5 Mobile/15E148 Safari/604.1" | |
| ) | |
| URL_RE = re.compile(r"(?i)^https?://(.*?)?(xhslink\.com|xiaohongshu\.com)/[^\s]+$") | |
| # ---------------------------------------------------------------- helpers | |
| def is_valid_rednote_url(url: str) -> bool: | |
| return bool(URL_RE.match(url or "")) | |
| def capture_single(html: str, pattern: str) -> str: | |
| m = re.search(pattern, html) | |
| return m.group(1).strip() if m else "" | |
| def html_unescape(t: str) -> str: | |
| return ( | |
| t.replace("&", "&") | |
| .replace(""", '"') | |
| .replace("'", "'") | |
| .replace("<", "<") | |
| .replace(">", ">") | |
| ) | |
| def meta_all(html: str, key: str) -> list: | |
| """Lấy content của <meta name|property="key" content="...">, chấp nhận cả 2 thứ tự thuộc tính.""" | |
| k = re.escape(key) | |
| p1 = rf"""(?is)<meta[^>]*?(?:name|property)\s*=\s*["']{k}["'][^>]*?content\s*=\s*["'](.*?)["']""" | |
| p2 = rf"""(?is)<meta[^>]*?content\s*=\s*["'](.*?)["'][^>]*?(?:name|property)\s*=\s*["']{k}["']""" | |
| out = [] | |
| for p in (p1, p2): | |
| for m in re.finditer(p, html): | |
| v = html_unescape(m.group(1).strip()) | |
| if v and v not in out: | |
| out.append(v) | |
| return out | |
| def meta_first(html: str, key: str) -> str: | |
| r = meta_all(html, key) | |
| return r[0] if r else "" | |
| def g(obj, *path): | |
| """Safe nested getter for dict/list.""" | |
| cur = obj | |
| for p in path: | |
| if isinstance(cur, dict): | |
| cur = cur.get(p) | |
| elif isinstance(cur, list) and isinstance(p, int) and 0 <= p < len(cur): | |
| cur = cur[p] | |
| else: | |
| return None | |
| return cur | |
| def s(v) -> str: | |
| if isinstance(v, str): | |
| return v.strip() | |
| if isinstance(v, (int, float)) and not isinstance(v, bool): | |
| return str(v) | |
| return "" | |
| def fix_url(u: str) -> str: | |
| u = u.strip().replace("\\u002F", "/").replace("&", "&") | |
| if u.startswith("http://"): | |
| return "https://" + u[len("http://"):] | |
| if u.startswith("//"): | |
| return "https:" + u | |
| return u | |
| def extract_initial_state(html: str): | |
| """Tách và parse window.__INITIAL_STATE__={...} (JS object, có thể chứa `undefined`).""" | |
| marker = "window.__INITIAL_STATE__" | |
| idx = html.find(marker) | |
| if idx < 0: | |
| return None | |
| rest = html[idx + len(marker):].lstrip() | |
| if not rest.startswith("="): | |
| return None | |
| rest = rest[1:].lstrip() | |
| end = rest.find("</script>") | |
| body = rest if end < 0 else rest[:end] | |
| body = re.sub(r"([:\[,])\s*undefined\b", r"\1null", body) | |
| try: | |
| val, _ = json.JSONDecoder().raw_decode(body.strip()) | |
| return val | |
| except ValueError: | |
| return None | |
| def find_note(state): | |
| # Web: note.noteDetailMap[id].note | |
| m = g(state, "note", "noteDetailMap") | |
| if isinstance(m, dict): | |
| first_id = s(g(state, "note", "firstNoteId")) | |
| pick = g(m, first_id, "note") | |
| if not (isinstance(pick, dict) and pick): | |
| pick = None | |
| for e in m.values(): | |
| n = g(e, "note") | |
| if isinstance(n, dict) and n: | |
| pick = n | |
| break | |
| if pick: | |
| return pick | |
| # Mobile: noteData.data.noteData | |
| n = g(state, "noteData", "data", "noteData") | |
| if isinstance(n, dict): | |
| return n | |
| return None | |
| def collect_downloads(note) -> list: | |
| out = [] | |
| def push(q, u): | |
| u = fix_url(u) | |
| if u and not any(d["url"] == u for d in out): | |
| out.append({"quality": q, "url": u}) | |
| origin = s(g(note, "video", "consumer", "originVideoKey")) | |
| if origin: | |
| push("Original", f"https://sns-video-bd.xhscdn.com/{origin}") | |
| stream = g(note, "video", "media", "stream") | |
| if isinstance(stream, dict): | |
| for codec, lst in stream.items(): | |
| if not isinstance(lst, list): | |
| continue | |
| for item in lst: | |
| h = s(g(item, "height")) | |
| label = f"{h}p {codec.upper()}" if h else codec.upper() | |
| push(label, s(g(item, "masterUrl"))) | |
| backups = g(item, "backupUrls") | |
| if isinstance(backups, list): | |
| for i, u in enumerate(backups): | |
| push(f"{label} (backup {i + 1})", s(u)) | |
| return out | |
| def collect_images(note) -> list: | |
| out = [] | |
| arr = g(note, "imageList") | |
| if isinstance(arr, list): | |
| for img in arr: | |
| u = s(g(img, "urlDefault")) | |
| if not u: | |
| u = s(g(img, "infoList", 0, "url")) | |
| if not u: | |
| u = s(g(img, "url")) | |
| u = fix_url(u) | |
| if u and u not in out: | |
| out.append(u) | |
| return out | |
| # ---------------------------------------------------------------- parsing | |
| def parse_page(html: str, final_url: str) -> dict: | |
| state = extract_initial_state(html) | |
| note = find_note(state) if state is not None else None | |
| if state is None: | |
| print(f"[scraper] __INITIAL_STATE__ not found (url: {final_url})") | |
| elif note is None: | |
| print(f"[scraper] note object not found in state (url: {final_url})") | |
| title_tag = html_unescape(capture_single(html, r"(?is)<title>(.*?)</title>")) | |
| title = s(g(note, "title")) if note else "" | |
| title = title or meta_first(html, "og:title") or title_tag | |
| desc = s(g(note, "desc")) if note else "" | |
| desc = desc or meta_first(html, "description") or meta_first(html, "og:description") | |
| keywords = meta_first(html, "keywords") | |
| nickname = s(g(note, "user", "nickname")) if note else "" | |
| if not nickname: | |
| nickname = meta_first(html, "og:title").split(" - ")[0] | |
| note_id = s(g(note, "noteId")) if note else "" | |
| if not note_id: | |
| src = meta_first(html, "og:url") or final_url | |
| note_id = src.split("?")[0].rstrip("/").rsplit("/", 1)[-1] | |
| duration = s(g(note, "video", "capa", "duration")) if note else "" | |
| duration = duration or meta_first(html, "og:videotime") | |
| if note: | |
| likes = s(g(note, "interactInfo", "likedCount")) | |
| comments = s(g(note, "interactInfo", "commentCount")) | |
| collects = s(g(note, "interactInfo", "collectedCount")) | |
| else: | |
| likes = comments = collects = "" | |
| likes = likes or meta_first(html, "og:xhs:note_like") | |
| comments = comments or meta_first(html, "og:xhs:note_comment") | |
| collects = collects or meta_first(html, "og:xhs:note_collect") | |
| images = collect_images(note) if note else [] | |
| if not images: | |
| images = [fix_url(u) for u in meta_all(html, "og:image")] | |
| downloads = collect_downloads(note) if note else [] | |
| # Fallback 1: og:video | |
| if not downloads: | |
| for i, u in enumerate(meta_all(html, "og:video")): | |
| u = fix_url(u) | |
| if not any(d["url"] == u for d in downloads): | |
| downloads.append({"quality": "Original" if i == 0 else f"Alt {i + 1}", "url": u}) | |
| # Fallback 2: quét thô masterUrl / originVideoKey | |
| if not downloads: | |
| for m in re.finditer(r'"masterUrl"\s*:\s*"(.*?)"', html): | |
| u = fix_url(m.group(1)) | |
| if u and not any(d["url"] == u for d in downloads): | |
| downloads.append({"quality": "Video", "url": u}) | |
| key = capture_single(html, r'"originVideoKey"\s*:\s*"(.*?)"') | |
| if key: | |
| downloads.insert( | |
| 0, {"quality": "Original", "url": f"https://sns-video-bd.xhscdn.com/{key}"} | |
| ) | |
| return { | |
| "noteId": note_id, | |
| "nickname": nickname, | |
| "title": title, | |
| "desc": desc, | |
| "keywords": keywords, | |
| "duration": duration, | |
| "engagement": {"likes": likes, "comments": comments, "collects": collects}, | |
| "images": images, | |
| "downloads": downloads, | |
| } | |
| # ---------------------------------------------------------------- fetching | |
| def build_client(ua: str, mobile: bool) -> httpx.Client: | |
| headers = { | |
| "User-Agent": ua, | |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", | |
| "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", | |
| "Referer": "https://www.xiaohongshu.com/", | |
| } | |
| if not mobile: | |
| headers["sec-ch-ua"] = '"Chromium";v="132", "Not_A Brand";v="99"' | |
| headers["sec-ch-ua-mobile"] = "?0" | |
| headers["sec-ch-ua-platform"] = '"Windows"' | |
| return httpx.Client(headers=headers, follow_redirects=True, timeout=30) | |
| def scrape(url: str) -> dict: | |
| """Trả về dict kết quả; raise ValueError(msg) nếu thất bại.""" | |
| best = None | |
| last_err = "Unknown error" | |
| def score(r): | |
| return len(r["images"]) + int(bool(r["desc"])) + int(bool(r["nickname"])) | |
| for ua, mobile in ((UA_DESKTOP, False), (UA_MOBILE, True)): | |
| try: | |
| with build_client(ua, mobile) as client: | |
| resp = client.get(url) | |
| res = parse_page(resp.text, str(resp.url)) | |
| except Exception as e: # noqa: BLE001 | |
| last_err = f"Request failed: {e}" | |
| continue | |
| if res["downloads"]: | |
| return res | |
| if best is None or score(res) > score(best): | |
| best = res | |
| if best is None: | |
| raise ValueError(last_err) | |
| return best | |