Blame for scraper.py

"""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("&lt;", "<")
        .replace("&gt;", ">")
    )


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("&amp;", "&")
    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