"""小红书创作者中心 + 前台评论抓取。

登录一次后，Cookie 持久化在 .auth-xhs/，之后直接复用。
一次抓取共享一个 Chromium 会话，稳定性优于每步一个进程。

设计原则（和 douyin-session 一致）：
- 不逆向 x-s / x-t 签名、不伪造请求——用登录态浏览器，让页面自己发带签名的请求，
  我们只被动拦截返回的 JSON。
- 创作者**自己的**笔记数据走 galaxy 接口（不需要 xsec_token），是最稳的主路。
- 评论走前台 web API（需要 xsec_token），让页面自己导航触发带 token 的请求；
  拿不到就优雅降级（report.md 标 comments_unavailable，cheat-retro 回落到 manual 粘评论）。

本文件新增的能力（来自 xhs-analytics 的公开页解析）：
- fetch_public_note: 无登录解析 explore 页面 __INITIAL_STATE__，拿正文 / 图片 / 标签。
- fetch_public_comments: 公开页兜底 top 评论。
- download_image: 下载笔记图片到本地。
"""
from __future__ import annotations

import asyncio
import json
import re
import urllib.parse
from pathlib import Path
from typing import Any

import requests
from playwright.async_api import BrowserContext, Page, Response, async_playwright
from paths import auth_dir, debug_dir

CREATOR_HOME = "https://creator.xiaohongshu.com/new/home"
CREATOR_NOTE_MANAGER = "https://creator.xiaohongshu.com/new/note-manager"
# galaxy 接口路径片段——宽松匹配，接口偶有版本号变化
GALAXY_NOTE_LIST_KEYS = (
    # 松匹配后缀，兼容 /api/galaxy/creator/... 与 /api/galaxy/v2/creator/...（实测是 v2）
    "/creator/note/user/posted",
)
GALAXY_NOTE_STATS_KEYS = (
    "/api/galaxy/creator/data/note_stats",
    "/api/galaxy/creator/data/note_detail",
)
# 前台 web API
FEED_KEY = "/api/sns/web/v1/feed"
COMMENT_KEY = "/api/sns/web/v2/comment/page"

DEFAULT_USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"
)


class Session:
    """单浏览器会话，按顺序跑多步抓取。"""

    def __init__(self, ctx: BrowserContext, pw: Any) -> None:
        self.ctx = ctx
        self.pw = pw

    @classmethod
    async def open(cls, headless: bool = False) -> "Session":
        pw = await async_playwright().start()
        auth_path = auth_dir()
        auth_path.mkdir(parents=True, exist_ok=True)
        common_kwargs = {
            "user_data_dir": str(auth_path),
            "headless": headless,
            "viewport": {"width": 1440, "height": 900},
            "args": ["--disable-blink-features=AutomationControlled"],
        }
        try:
            ctx = await pw.chromium.launch_persistent_context(**common_kwargs)
        except Exception as exc:
            # 部分环境只装了系统 Chrome，没下载 Playwright Chromium，fallback 到 channel=chrome
            try:
                ctx = await pw.chromium.launch_persistent_context(
                    **common_kwargs, channel="chrome"
                )
            except Exception:
                raise exc
        return cls(ctx, pw)

    async def close(self) -> None:
        try:
            await self.ctx.close()
        finally:
            await self.pw.stop()


# 创作者中心登录凭证——扫码登 creator.xiaohongshu.com 产生的就是这些（galaxy 主路只需它们）。
# 注意：创作者中心登录 *不* 产生 web_session（那是主站 www 前台 cookie），早期版本只认
# web_session 是个 bug，会导致登录成功却一直检测不到、白等超时。
CREATOR_LOGIN_COOKIES = (
    "access-token-creator.xiaohongshu.com",
    "galaxy_creator_session_id",
    "customer-sso-sid",
)
# 主站前台凭证——feed / 评论 web API 需要；只在登录后访问过 www 才会下发。
WEB_LOGIN_COOKIE = "web_session"


async def _cookie_map(ctx: BrowserContext, host: str) -> dict[str, str]:
    try:
        return {c["name"]: c.get("value", "") for c in await ctx.cookies(host)}
    except Exception:
        return {}


async def _creator_logged_in(ctx: BrowserContext) -> bool:
    names = await _cookie_map(ctx, "https://creator.xiaohongshu.com")
    return any(names.get(n) for n in CREATOR_LOGIN_COOKIES)


async def _has_web_session(ctx: BrowserContext) -> bool:
    for host in ("https://www.xiaohongshu.com", "https://creator.xiaohongshu.com"):
        if (await _cookie_map(ctx, host)).get(WEB_LOGIN_COOKIE):
            return True
    return False


async def _acquire_web_session(page: Page) -> None:
    """创作者中心登录后，访问主站让 SSO 下发 web_session（前台 feed/评论需要）。best-effort。"""
    try:
        await page.goto("https://www.xiaohongshu.com/explore",
                        wait_until="domcontentloaded", timeout=30000)
        await asyncio.sleep(3)
        if await _has_web_session(page.context):
            print("[登录] ✓ 已获取主站 web_session（前台评论/互动可用）")
        else:
            print("[登录] 注意：未拿到 web_session，前台评论可能要 manual；galaxy 主路不受影响。")
    except Exception:
        pass


async def ensure_login(timeout_s: int = 180, max_refresh: int = 5) -> bool:
    """扫码登录创作者中心；检测到创作者登录态后顺便换取 web_session，然后自动关闭。

    二维码本身会过期，所以用 timeout_s 作为单次等待上限，超时后自动刷新页面重新出码，
    最多刷新 max_refresh 次。用户有充足时间扫码。
    """
    sess = await Session.open()
    try:
        page = await sess.ctx.new_page()
        await page.goto(CREATOR_HOME)
        print(f"[登录] 在弹出的 Chromium 窗口里扫码登录小红书创作者中心。每次二维码有效期约 {timeout_s} 秒，超时自动刷新。")

        for refresh in range(max_refresh + 1):
            for i in range(timeout_s):
                try:
                    if await _creator_logged_in(sess.ctx) and "login" not in page.url:
                        print(f"[登录] ✓ 创作者中心登录态已确认（总用时 {refresh * timeout_s + i}s）")
                        await _acquire_web_session(page)
                        await asyncio.sleep(1)
                        return True
                except Exception:
                    pass

                # 每 30 秒提醒一次，避免用户以为卡死
                if i > 0 and i % 30 == 0:
                    print(f"[登录] 已等待 {i} 秒，请用小红书 App 扫码（或等待自动刷新二维码）……")

                await asyncio.sleep(1)

            if refresh < max_refresh:
                print(f"[登录] 本次二维码未扫码或已过期，正在刷新页面重新出码（第 {refresh + 1}/{max_refresh} 次刷新）……")
                try:
                    await page.reload(wait_until="domcontentloaded", timeout=30000)
                except Exception:
                    await page.goto(CREATOR_HOME)

        print("[登录] 超过最大刷新次数仍未检测到登录态，已停止。如需继续请重新运行本命令。")
        return False
    finally:
        await sess.close()


async def fetch_recent_notes(sess: Session, limit: int = 50) -> list[dict]:
    """创作者中心笔记管理页 → 拦截 galaxy 笔记列表 + 单篇运营数据（含曝光/浏览）。"""
    captured: list[dict] = []
    all_urls: list[str] = []

    page = await sess.ctx.new_page()

    async def on_response(resp: Response) -> None:
        all_urls.append(resp.url)
        if any(k in resp.url for k in GALAXY_NOTE_LIST_KEYS + GALAXY_NOTE_STATS_KEYS):
            try:
                data = await resp.json()
                captured.append({"url": resp.url, "data": data})
                if len(captured) == 1 and isinstance(data, dict):
                    print(f"[诊断] galaxy 接口 keys: {list(data.keys())[:8]}")
            except Exception:
                pass

    page.on("response", on_response)
    try:
        await page.goto(CREATOR_NOTE_MANAGER, wait_until="domcontentloaded", timeout=60000)
        await asyncio.sleep(8)
        # cookie 过期时会被 302 到登录页
        if "login" in page.url or "redirectReason=401" in page.url:
            print("[登录] 创作者中心已跳转登录页，cookie 可能已过期。请运行：python crawler.py login")
            return []
        for _ in range(4):
            await page.evaluate("window.scrollBy(0, 1200)")
            await asyncio.sleep(1.5)
        notes = _parse_note_list(captured, limit)
        if not notes:
            _dump(all_urls, "creator_urls.txt", captured, "creator_captured.json")
            print(f"[诊断] 笔记列表为空，{len(all_urls)} 个请求已 dump 到 .cheat-cache/xhs-explore-debug/。")
        return notes
    finally:
        await page.close()


def _iter_candidates(data: Any) -> list:
    """从任意 galaxy response 里挖出"笔记数组"。结构多变，宽松找。"""
    out: list = []
    if isinstance(data, dict):
        # 常见包装：{data: {...}} / {data: [...]} / 顶层直接 list 字段
        inner = data.get("data") if isinstance(data.get("data"), (dict, list)) else data
        targets = [inner] if not isinstance(inner, list) else []
        if isinstance(inner, list):
            out.extend(inner)
        for t in targets:
            if isinstance(t, dict):
                for key in ("notes", "note_list", "list", "items", "note_stats", "result"):
                    val = t.get(key)
                    if isinstance(val, list):
                        out.extend(val)
    return out


def _parse_note_list(captured: list[dict], limit: int) -> list[dict]:
    by_id: dict[str, dict] = {}
    for item in captured:
        for raw in _iter_candidates(item["data"]):
            if not isinstance(raw, dict):
                continue
            note = _normalize_note(raw)
            if not note["note_id"]:
                continue
            # 同一 note 可能在 list 接口和 stats 接口各出现一次——合并，非空字段优先
            existing = by_id.get(note["note_id"])
            if existing:
                for k, v in note.items():
                    if v and not existing.get(k):
                        existing[k] = v
            else:
                by_id[note["note_id"]] = note
    return list(by_id.values())[:limit]


def _first(d: dict, *keys: str) -> Any:
    for k in keys:
        if k in d and d[k] is not None and d[k] != "":
            return d[k]
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    return 0


def _normalize_note(v: dict) -> dict:
    note_id = v.get("note_id") or v.get("id") or v.get("noteId") or ""
    # 字段名已用真实返回校准（2026-05 /api/galaxy/v2/creator/note/user/posted）：
    #   观看 view_count | 点赞 likes | 收藏 collected_count | 评论 comments_count
    #   分享 shared_count | 发布时间 visible_time(unix秒) | 单篇 token xsec_token
    # 确认名放首位，旧候选留作兜底以防接口再次改版。
    return {
        "note_id": str(note_id),
        "title": v.get("display_title") or v.get("title") or v.get("desc") or v.get("name") or "",
        "create_time": _to_int(_first(v, "visible_time", "create_time", "post_time", "publish_time")),
        "view_count": _to_int(_first(v, "view_count", "view", "imp", "impression", "read_count", "pv")),
        "like_count": _to_int(_first(v, "likes", "like_count", "liked_count", "like")),
        "collect_count": _to_int(_first(v, "collected_count", "collect_count", "collect", "fav_count")),
        "comment_count": _to_int(_first(v, "comments_count", "comment_count", "comment", "cmt_count")),
        "share_count": _to_int(_first(v, "shared_count", "share_count", "share")),
        "fans_inc": _to_int(_first(v, "fans", "fans_inc", "new_fans", "follow_count")),
        "post_time_str": v.get("time") or "",  # galaxy 自带本地时间串，比 epoch 省去时区换算
        "xsec_token": v.get("xsec_token") or "",
        "note_type": v.get("type") or "",
        "raw": v,
    }


def _to_int(x: Any) -> int:
    try:
        if isinstance(x, str):
            x = x.replace(",", "").strip()
        return int(float(x))
    except (ValueError, TypeError):
        return 0


# ---------------------------------------------------------------------------
# 公开页解析（来自 xhs-analytics 的核心能力）：无登录拿正文、图片、标签、评论兜底
# ---------------------------------------------------------------------------

def _extract_initial_state(html: str) -> dict | None:
    """从小红书 explore 页面 HTML 中解析 window.__INITIAL_STATE__。"""
    marker = "window.__INITIAL_STATE__="
    start = html.find(marker)
    if start == -1:
        return None
    script_start = html.rfind("<script", 0, start)
    script_end = html.find("</script>", start)
    if script_start == -1 or script_end == -1:
        return None
    script = html[script_start:script_end]
    assign_start = script.find(marker) + len(marker)
    json_str = script[assign_start:]
    json_str = re.sub(r":\s*undefined\s*([,}\]])", r":null\1", json_str)
    json_str = json_str.rstrip().rstrip(';').rstrip()
    try:
        return json.loads(json_str)
    except Exception:
        return None


def _image_url(img: dict) -> str:
    """从 imageList 元素中提取可用的图片 URL。"""
    if not isinstance(img, dict):
        return ""
    for key in ("urlDefault", "url"):
        if img.get(key):
            return img[key]
    for info in img.get("infoList", []) or []:
        if isinstance(info, dict) and info.get("imageScene") == "WB_DFT" and info.get("url"):
            return info["url"]
    for info in img.get("infoList", []) or []:
        if isinstance(info, dict) and info.get("url"):
            return info["url"]
    return ""


def _xsec_token_from_url(url: str | None) -> str:
    """从笔记 URL 的 query 中拆出 xsec_token（保留原编码）。"""
    if not url:
        return ""
    parsed = urllib.parse.urlparse(url)
    return urllib.parse.unquote(urllib.parse.parse_qs(parsed.query).get("xsec_token", [""])[0])


def fetch_public_note(note_id: str, xsec_token: str) -> dict:
    """无登录抓取 explore 公开页，解析 __INITIAL_STATE__。

    返回 dict：success, note_id, title, desc/body, images, tags, time, counts, raw。
    """
    token = urllib.parse.unquote(xsec_token)
    url = (
        f"https://www.xiaohongshu.com/explore/{note_id}"
        f"?xsec_token={urllib.parse.quote(token)}"
        f"&xsec_source=pc_creatormng"
    )
    headers = {
        "User-Agent": DEFAULT_USER_AGENT,
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    }
    r = requests.get(url, headers=headers, timeout=20)
    r.raise_for_status()

    state = _extract_initial_state(r.text)
    if not state:
        raise ValueError("无法解析页面初始状态")

    note_detail_map = state.get("note", {}).get("noteDetailMap", {})
    if note_id not in note_detail_map:
        raise ValueError("页面未返回该笔记数据")

    raw = note_detail_map[note_id]
    note = raw.get("note", {})
    interact = note.get("interactInfo", {})

    return {
        "success": True,
        "note_id": note_id,
        "title": note.get("title", ""),
        "desc": note.get("desc", ""),
        "body": note.get("desc", ""),
        "type": note.get("type", ""),
        "time": note.get("time", 0),
        "images": [_image_url(img) for img in note.get("imageList", [])],
        "tags": [tag.get("name", "") for tag in note.get("tagList", []) if tag.get("name")],
        "counts": {
            "liked": _to_int(interact.get("likedCount")),
            "collected": _to_int(interact.get("collectedCount")),
            "comment": _to_int(interact.get("commentCount")),
            "shared": _to_int(interact.get("shareCount")),
        },
        "raw": raw,
    }


def _normalize_public_comment(c: dict) -> dict:
    """把 __INITIAL_STATE__ / edith 接口里的评论字段统一成 adapter 格式。"""
    user = c.get("user_info") or c.get("userInfo") or {}
    like_count = c.get("like_count") or c.get("likeCount") or 0
    sub_count = c.get("sub_comment_count") or c.get("subCommentCount") or 0
    return {
        "cid": str(c.get("id") or c.get("comment_id") or ""),
        "text": c.get("content") or "",
        "like_count": _to_int(like_count),
        "sub_comment_count": _to_int(sub_count),
        "create_time": c.get("create_time") or c.get("createTime") or 0,
        "user_name": user.get("nickname") or "",
        "ip_label": c.get("ip_location") or c.get("ipLocation") or "",
    }


def fetch_public_comments(note_id: str, xsec_token: str, max_comments: int = 20) -> list[dict]:
    """公开页 __INITIAL_STATE__ 兜底 top 评论（通常 ~10 条）。"""
    token = urllib.parse.unquote(xsec_token)
    url = (
        f"https://www.xiaohongshu.com/explore/{note_id}"
        f"?xsec_token={urllib.parse.quote(token)}"
        f"&xsec_source=pc_creatormng"
    )
    headers = {
        "User-Agent": DEFAULT_USER_AGENT,
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    }
    r = requests.get(url, headers=headers, timeout=20)
    r.raise_for_status()
    state = _extract_initial_state(r.text)
    if not state:
        return []

    raw = state.get("note", {}).get("noteDetailMap", {}).get(note_id, {})
    comments = []
    # __INITIAL_STATE__ 里的评论路径可能是 comments.list 或 commentsList
    for key in ("comments", "commentsList"):
        container = raw.get(key) if isinstance(raw, dict) else None
        if isinstance(container, dict):
            arr = container.get("list") or container.get("comments") or []
            if isinstance(arr, list):
                comments.extend([_normalize_public_comment(c) for c in arr])
        elif isinstance(container, list):
            comments.extend([_normalize_public_comment(c) for c in container])

    seen = set()
    dedup = []
    for c in comments:
        if not c["cid"] or c["cid"] in seen:
            continue
        seen.add(c["cid"])
        dedup.append(c)
    dedup.sort(key=lambda x: x["like_count"], reverse=True)
    return dedup[:max_comments]


async def download_image(url: str, dest: Path, timeout: int = 30) -> bool:
    """下载单张图片到 dest（会根据 Content-Type 修正扩展名）。异步封装。"""
    if not url:
        return False
    dest = Path(dest)
    dest.parent.mkdir(parents=True, exist_ok=True)

    def _download() -> bool:
        headers = {
            "User-Agent": DEFAULT_USER_AGENT,
            "Accept": "image/webp,image/apng,image/*,*/*;q=0.8",
            "Referer": "https://www.xiaohongshu.com/",
        }
        r = requests.get(url, headers=headers, timeout=timeout, stream=True)
        r.raise_for_status()

        parsed = urllib.parse.urlparse(url)
        path = urllib.parse.unquote(parsed.path)
        ext = Path(path).suffix.lower()
        if ext not in {".jpg", ".jpeg", ".png", ".webp", ".gif", ".avif"}:
            ct = r.headers.get("Content-Type", "").lower()
            if "webp" in ct:
                ext = ".webp"
            elif "png" in ct:
                ext = ".png"
            elif "gif" in ct:
                ext = ".gif"
            else:
                ext = ".jpg"
        dest_final = dest.with_suffix(ext)

        with open(dest_final, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                if chunk:
                    f.write(chunk)
        return True

    try:
        return await asyncio.to_thread(_download)
    except Exception as exc:
        print(f"[下载图片] 失败 {url}: {exc}")
        return False


async def fetch_note_frontend(
    sess: Session,
    note_id: str,
    note_url: str | None = None,
    xsec_token: str | None = None,
) -> dict:
    """打开前台笔记页 → 拦截 feed（interact_info 确认字段）+ comment/page。

    前台需要 xsec_token + 登录态（web_session）。
    - 若传入 note_url（含 xsec_token，如从创作者后台复制的 ?xsec_token=...&xsec_source=pc_creatormng）
      → 直接用它导航，最稳。
    - 否则退回裸 explore URL（仅对已登录账号访问自己笔记可能可行）。
    token 缺失 / 未登录 → dump 并降级（评论留给 manual）。

    拦截不到评论时，会用公开页 __INITIAL_STATE__ 里的 top 评论兜底。
    """
    feed: dict = {}
    comments: list[dict] = []
    all_urls: list[str] = []

    page = await sess.ctx.new_page()

    async def on_response(resp: Response) -> None:
        all_urls.append(resp.url)
        if FEED_KEY in resp.url:
            try:
                data = await resp.json()
                ii = _extract_interact(data)
                if ii:
                    feed.update(ii)
            except Exception:
                pass
        elif COMMENT_KEY in resp.url:
            try:
                data = await resp.json()
                for c in _extract_comments(data):
                    comments.append(c)
            except Exception:
                pass

    page.on("response", on_response)
    try:
        url = note_url or f"https://www.xiaohongshu.com/explore/{note_id}"
        try:
            await page.goto(url, wait_until="domcontentloaded", timeout=60000)
        except Exception as e:
            print(f"[警告] 笔记页加载异常：{e}")
        await asyncio.sleep(5)
        if "website-login/error" in page.url or "登录" in (await page.title()):
            print("[警告] 触发登录墙（安全限制）——cookie 未登录或已过期。先跑 crawler.py login 扫码。")

        # 滚动评论区触发分页懒加载
        last = 0
        stagnant = 0
        for _ in range(40):
            await page.evaluate("window.scrollBy(0, 1400)")
            await asyncio.sleep(1.8)
            cur = len({c["cid"] for c in comments})
            if cur == last:
                stagnant += 1
                if stagnant >= 5:
                    break
            else:
                stagnant = 0
                last = cur

        # 兜底：公开页 __INITIAL_STATE__ 里的 top 评论
        if not comments:
            xsec = xsec_token or _xsec_token_from_url(note_url)
            if xsec:
                try:
                    public_comments = await asyncio.to_thread(
                        fetch_public_comments, note_id, xsec, max_comments=20
                    )
                    if public_comments:
                        comments = public_comments
                        print(f"       公开页兜底 {len(comments)} 条评论")
                except Exception as exc:
                    print(f"[诊断] 公开页评论兜底失败：{exc}")

        if not comments:
            dbg = debug_dir()
            dbg.mkdir(parents=True, exist_ok=True)
            try:
                await page.screenshot(path=str(dbg / f"note_{note_id}.png"))
            except Exception:
                pass
            (dbg / "frontend_urls.txt").write_text("\n".join(all_urls), encoding="utf-8")
            print("[诊断] 前台未拦到评论（可能 xsec_token 缺失或评论被关），已 dump URL。")

        # 去重 + 按赞降序
        seen = set()
        dedup = []
        for c in comments:
            if c["cid"] in seen:
                continue
            seen.add(c["cid"])
            dedup.append(c)
        dedup.sort(key=lambda x: x["like_count"], reverse=True)
        print(f"       前台共 {len(dedup)} 条评论")
        return {"interact": feed, "comments": dedup}
    finally:
        await page.close()


def _extract_interact(data: Any) -> dict:
    """从 feed response 里挖 interact_info（确认字段）。"""
    if not isinstance(data, dict):
        return {}
    items = []
    d = data.get("data", data)
    if isinstance(d, dict):
        items = d.get("items") or d.get("note_list") or []
    for it in items if isinstance(items, list) else []:
        node = it.get("note_card") or it.get("note") or it
        ii = node.get("interact_info") if isinstance(node, dict) else None
        if isinstance(ii, dict):
            return {
                "like_count": _to_int(ii.get("liked_count")),
                "collect_count": _to_int(ii.get("collected_count")),
                "comment_count": _to_int(ii.get("comment_count")),
                "share_count": _to_int(ii.get("share_count")),
                "ip_location": node.get("ip_location") or "",
            }
    return {}


def _extract_comments(data: Any) -> list[dict]:
    """comment/page response → 评论列表（确认字段）。"""
    out: list[dict] = []
    if not isinstance(data, dict):
        return out
    d = data.get("data", data)
    arr = d.get("comments") if isinstance(d, dict) else None
    for c in arr or []:
        if not isinstance(c, dict):
            continue
        user = c.get("user_info") or {}
        out.append({
            "cid": str(c.get("id") or c.get("comment_id") or ""),
            "text": c.get("content") or "",
            "like_count": _to_int(c.get("like_count")),
            "sub_comment_count": _to_int(c.get("sub_comment_count")),
            "create_time": c.get("create_time") or 0,
            "user_name": user.get("nickname") or "",
            "ip_label": c.get("ip_location") or "",
        })
    return out


def _dump(urls: list[str], url_file: str, captured: list[dict], cap_file: str) -> None:
    dbg = debug_dir()
    dbg.mkdir(parents=True, exist_ok=True)
    (dbg / url_file).write_text("\n".join(urls), encoding="utf-8")
    try:
        (dbg / cap_file).write_text(
            json.dumps([c["data"] for c in captured][:5], ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
    except Exception:
        pass


def _merge_public_note(note: dict, public: dict) -> None:
    """用公开页数据补全 note（不覆盖已有的 galaxy 运营数据）。"""
    if not public.get("success"):
        return
    for key in ("title", "desc", "body", "type", "images", "tags"):
        if public.get(key) and not note.get(key):
            note[key] = public[key]
    if public.get("time") and not note.get("create_time"):
        note["create_time"] = _to_int(public["time"])
    counts = public.get("counts") or {}
    mapping = {
        "like_count": counts.get("liked"),
        "collect_count": counts.get("collected"),
        "comment_count": counts.get("comment"),
        "share_count": counts.get("shared"),
    }
    for k, v in mapping.items():
        if v and not note.get(k):
            note[k] = _to_int(v)


async def fetch_all(note_id: str, note_url: str | None = None) -> dict:
    """一个会话跑完笔记列表（含 galaxy 指标）+ 前台 interact + 评论 + 公开页正文/图片兜底。"""
    sess = await Session.open()
    try:
        print("  → 打开创作者中心，拉笔记列表 + 运营数据")
        notes = await fetch_recent_notes(sess, limit=50)
        note = next((n for n in notes if n["note_id"] == note_id), None)
        if not note:
            print(f"       未在最近 {len(notes)} 条里找到 {note_id}，用最小元数据继续。")
            note = _normalize_note({"note_id": note_id})
        else:
            print(f"       ✓ {(note.get('title') or '')[:40]}（曝光 {note.get('view_count')}）")

        # 抓自己的笔记时，galaxy 列表已带每条的 xsec_token——自动拼前台 URL，免得手动粘 token 链接
        front_url = note_url
        if not front_url and note.get("xsec_token"):
            front_url = (f"https://www.xiaohongshu.com/explore/{note_id}"
                         f"?xsec_token={note['xsec_token']}&xsec_source=pc_creatormng")

        xsec = _xsec_token_from_url(front_url) or note.get("xsec_token", "")

        # 先用公开页补正文/图片/标签（无登录、低成本），不覆盖 galaxy 已确认的计数
        if xsec:
            try:
                public = await asyncio.to_thread(fetch_public_note, note_id, xsec)
                _merge_public_note(note, public)
                print("       ✓ 公开页正文/图片/标签已补全")
            except Exception as exc:
                print(f"[诊断] 公开页兜底失败：{exc}")

        print("  → 打开前台笔记页抓 interact + 评论")
        front = await fetch_note_frontend(sess, note_id, note_url=front_url, xsec_token=xsec)
        # 前台 interact 字段是确认的——用它补全/覆盖 galaxy 里可能缺的计数
        for k in ("like_count", "collect_count", "comment_count", "share_count"):
            if front["interact"].get(k):
                note[k] = front["interact"][k]
        if front["interact"].get("ip_location"):
            note["ip_location"] = front["interact"]["ip_location"]

        return {"note": note, "comments": front["comments"]}
    finally:
        await sess.close()


if __name__ == "__main__":
    asyncio.run(ensure_login())