front/xhs-agent-desktop/engine/scripts/xhs/my_account.py

"""当前登录账号主页：加载笔记列表（含图文与视频）。"""

from __future__ import annotations

import json
import logging
import re
import time
from typing import Callable

from .browse_feeds import _emit_progress
from .cdp import Page
from .errors import is_cancelled
from .human import get_scroll_interval, human_scroll_page, sleep_random
from .login import check_login_status
from .selectors import USER_PROFILE_NAV_LINK
from .types import Feed, UserBasicInfo, UserInteraction, UserProfileResponse
from .urls import EXPLORE_URL
from .user_profile import (
    _EXTRACT_USER_DATA_JS,
    _EXTRACT_USER_NOTES_JS,
    _wait_for_initial_state,
)

logger = logging.getLogger(__name__)

ProgressFn = Callable[[str], None]

_MIN_SCROLL_DELTA = 80
_STAGNANT_LIMIT = 3


def resolve_current_profile_url(page: Page) -> str:
    """从导航栏「我」解析个人主页完整 URL。"""
    page.navigate(EXPLORE_URL)
    page.wait_for_load()
    if not check_login_status(page):
        raise RuntimeError("未登录，请先在 Chrome 中登录小红书")
    href = page.evaluate(
        f"document.querySelector({json.dumps(USER_PROFILE_NAV_LINK)})"
        f"?.getAttribute('href') || ''"
    )
    if not href or "/user/profile/" not in str(href):
        raise RuntimeError("未找到个人主页链接，请确认已登录")
    href = str(href).strip()
    if href.startswith("http"):
        return href
    return f"https://www.xiaohongshu.com{href}"


def _parse_notes_payload(notes_result: str) -> list[Feed]:
    notes_feeds_raw = json.loads(notes_result)
    feeds: list[Feed] = []
    for feed_group in notes_feeds_raw:
        if isinstance(feed_group, list):
            for f in feed_group:
                if isinstance(f, dict):
                    feeds.append(Feed.from_dict(f))
        elif isinstance(feed_group, dict):
            feeds.append(Feed.from_dict(feed_group))
    return feeds


def _snapshot_profile(page: Page) -> UserProfileResponse:
    user_data_result = page.evaluate(_EXTRACT_USER_DATA_JS)
    if not user_data_result:
        raise RuntimeError("无法读取个人主页资料，请确认页面已加载")

    notes_result = page.evaluate(_EXTRACT_USER_NOTES_JS)
    if not notes_result:
        raise RuntimeError("无法读取个人笔记列表，请确认主页笔记 Tab 已展示")

    user_page_data = json.loads(user_data_result)
    basic_info = UserBasicInfo.from_dict(user_page_data.get("basicInfo", {}))
    interactions = [
        UserInteraction.from_dict(i) for i in user_page_data.get("interactions", [])
    ]
    feeds = _parse_notes_payload(notes_result)
    return UserProfileResponse(
        user_basic_info=basic_info,
        interactions=interactions,
        feeds=feeds,
    )


def _merge_feeds(into: dict[str, Feed], feeds: list[Feed]) -> int:
    added = 0
    for feed in feeds:
        fid = str(feed.id or "").strip()
        if not fid or fid in into:
            continue
        into[fid] = feed
        added += 1
    return added


def list_my_account_feeds(
    page: Page,
    *,
    scroll_speed: str = "normal",
    max_scroll_rounds: int = 12,
    on_progress: ProgressFn = None,
) -> tuple[UserProfileResponse, bool, str]:
    """打开当前账号主页，滚动加载笔记（图文+视频）。

    Returns:
        (profile, cancelled, profile_url)
    """
    profile_url = resolve_current_profile_url(page)
    _emit_progress(on_progress, "正在打开我的主页…")
    page.navigate(profile_url)
    page.wait_for_load()
    page.wait_dom_stable()
    _wait_for_initial_state(page)

    by_id: dict[str, Feed] = {}
    profile = _snapshot_profile(page)
    _merge_feeds(by_id, profile.feeds)
    _emit_progress(on_progress, f"我的主页：已读取 {len(by_id)} 篇笔记")

    stagnant = 0
    last_top = page.get_scroll_top()
    for attempt in range(max_scroll_rounds):
        if is_cancelled():
            _emit_progress(on_progress, "我的主页：已停止")
            profile.feeds = list(by_id.values())
            return profile, True, profile_url

        _emit_progress(on_progress, f"我的主页：滚动加载 ({attempt + 1})…")
        actual_delta, current_top = human_scroll_page(
            page,
            scroll_speed,
            False,
            1,
        )
        sleep_random(0.3, 0.8)
        time.sleep(get_scroll_interval(scroll_speed))

        snap = _snapshot_profile(page)
        added = _merge_feeds(by_id, snap.feeds)
        if added:
            stagnant = 0
            _emit_progress(on_progress, f"我的主页：已读取 {len(by_id)} 篇笔记")
        else:
            stagnant += 1

        if actual_delta < _MIN_SCROLL_DELTA or current_top == last_top:
            stagnant += 1
        else:
            stagnant = max(0, stagnant - 1)
        last_top = current_top

        if stagnant >= _STAGNANT_LIMIT:
            _emit_progress(on_progress, "我的主页：已滑到底部")
            break

    profile.feeds = list(by_id.values())
    return profile, False, profile_url


def profile_url_user_id(url: str) -> str:
    m = re.search(r"/user/profile/([a-fA-F0-9]+)", url)
    return m.group(1) if m else ""