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

"""从页面 __INITIAL_STATE__ 或 DOM 提取笔记列表（首页 / 搜索共用）。"""

from __future__ import annotations

import json
import logging
import re

from .cdp import Page

logger = logging.getLogger(__name__)

_EXTRACT_STATE_FEEDS_JS = """
(() => {
  function unwrap(v) {
    if (v == null) return null;
    if (Array.isArray(v)) return v;
    if (typeof v !== "object") return null;
    if (Array.isArray(v.value)) return v.value;
    if (Array.isArray(v._value)) return v._value;
    if (v.value !== undefined) return unwrap(v.value);
    if (v._value !== undefined) return unwrap(v._value);
    return null;
  }
  function str(v) {
    if (v == null) return "";
    return String(v).trim();
  }
  function compact(item) {
    if (!item || typeof item !== "object") return null;
    let nc = item.noteCard || item.note_card || item.note;
    if (!nc || typeof nc !== "object") nc = {};
    if (!nc.displayTitle && !nc.title && (item.displayTitle || item.title)) {
      nc = Object.assign({}, item, nc);
    }
    const user = Object.assign({}, nc.user || {}, item.user || {});
    const interact = Object.assign(
      {},
      nc.interactInfo || nc.interact_info || {},
      item.interactInfo || item.interact_info || item.interact || {},
    );
    const id = str(item.id || nc.id || nc.noteId || item.note_id);
    if (!id) return null;
    return {
      id,
      xsecToken: str(item.xsecToken || item.xsec_token || nc.xsecToken || nc.xsec_token),
      modelType: str(item.modelType || item.model_type || nc.type),
      noteCard: {
        displayTitle: str(nc.displayTitle || nc.display_title || nc.title || item.title),
        type: str(nc.type || item.type),
        user: {
          nickname: str(user.nickname || user.nickName || user.nick_name || user.name),
          userId: str(user.userId || user.user_id || user.id),
        },
        interactInfo: {
          likedCount: str(
            interact.likedCount || interact.liked_count || interact.liked || item.liked_count || "",
          ),
          commentCount: str(
            interact.commentCount || interact.comment_count || interact.comments || "",
          ),
          collectedCount: str(interact.collectedCount || interact.collected_count || ""),
        },
      },
    };
  }
  const st = window.__INITIAL_STATE__;
  if (!st) return "";
  const candidates = [
    () => st.search && st.search.feeds,
    () => st.search && st.search.notes,
    () => st.search && st.search.items,
    () => st.feed && st.feed.feeds,
    () => st.feeds,
    () => st.homefeed && st.homefeed.feeds,
    () => st.homeFeed && st.homeFeed.feeds,
  ];
  for (const pick of candidates) {
    try {
      const data = unwrap(pick());
      if (data && data.length) {
        const out = data.map(compact).filter(Boolean);
        if (out.length) return JSON.stringify(out);
      }
    } catch (e) {}
  }
  return "";
})()
"""

_EXTRACT_DOM_FEEDS_JS = """
(() => {
  function clean(t) {
    return (t || "").replace(/\\s+/g, " ").trim();
  }
  function parseCount(text) {
    const t = clean(text);
    if (!t) return "";
    const m = t.match(/^([\\d.]+\\s*[万亿]?)\\s*$/);
    return m ? m[1] : t;
  }
  function pickAuthor(card) {
    const sels = [
      '[class*="author"] [class*="name"]',
      '[class*="user-name"]',
      '[class*="nickname"]',
      '[class*="name"][class*="user"]',
      'a[href*="/user/"] span',
      '.author-wrapper .name',
      '.user .name',
    ];
    for (const sel of sels) {
      const el = card.querySelector(sel);
      const t = clean(el && el.textContent);
      if (t && t.length < 40) return t;
    }
    const userLink = card.querySelector('a[href*="/user/"]');
    if (userLink) {
      const t = clean(userLink.textContent);
      if (t && t.length < 40) return t;
    }
    return "";
  }
  function pickLike(card) {
    const sels = [
      '[class*="like-wrapper"] span',
      '[class*="liked"] span',
      '[class*="like"] span.count',
      '[class*="engage"] [class*="like"]',
      'span[class*="count"]',
    ];
    for (const sel of sels) {
      const nodes = card.querySelectorAll(sel);
      for (const el of nodes) {
        const t = clean(el.textContent);
        if (/^[\\d.]+[万亿]?$/.test(t)) return parseCount(t);
      }
    }
    const text = clean(card.innerText || "");
    const m = text.match(/([\\d.]+[万亿]?)\\s*赞/);
    if (m) return m[1];
    return "";
  }
  const seen = new Set();
  const out = [];
  const anchors = document.querySelectorAll('a[href*="/explore/"]');
  for (const a of anchors) {
    const href = (a.getAttribute("href") || a.href || "");
    const m = href.match(/\\/explore\\/([a-f0-9]{16,24})/i);
    if (!m) continue;
    const id = m[1];
    if (seen.has(id)) continue;
    seen.add(id);
    let token = "";
    const tok = href.match(/xsec_token=([^&]+)/i);
    if (tok) {
      try { token = decodeURIComponent(tok[1]); } catch (e) { token = tok[1]; }
    }
    const card =
      a.closest("section, article, li, [class*='note'], [class*='feed'], [class*='card']") ||
      a.parentElement ||
      a;
    const titleEl = card.querySelector(
      "[class*='title'], .title, h3, h4, [class*='desc']",
    );
    const title = clean((titleEl && titleEl.textContent) || a.textContent)
      .split("\\n")
      .map(s => s.trim())
      .filter(Boolean)[0] || "（无标题）";
    const author = pickAuthor(card);
    const liked = pickLike(card);
    out.push({
      id,
      xsecToken: token,
      modelType: "normal",
      noteCard: {
        displayTitle: title.slice(0, 200),
        user: { nickname: author },
        interactInfo: {
          likedCount: liked,
          commentCount: "",
          collectedCount: "",
        },
      },
    });
  }
  return JSON.stringify(out);
})()
"""

_FEED_ID_RE = re.compile(r"^[a-f0-9]{16,24}$", re.I)


def _note_block(item: dict) -> dict:
    for key in ("noteCard", "note_card", "note"):
        block = item.get(key)
        if isinstance(block, dict):
            return block
    return {}


def _feed_richness(item: dict) -> int:
    note = _note_block(item)
    user = note.get("user") if isinstance(note.get("user"), dict) else {}
    interact = note.get("interactInfo") if isinstance(note.get("interactInfo"), dict) else {}
    if not interact and isinstance(note.get("interact_info"), dict):
        interact = note["interact_info"]
    score = 0
    nick = str(user.get("nickname") or user.get("nickName") or user.get("nick_name") or "").strip()
    if nick and nick != "—":
        score += 3
    liked = str(interact.get("likedCount") or interact.get("liked_count") or "").strip()
    if liked and liked != "0":
        score += 2
    title = str(note.get("displayTitle") or note.get("display_title") or note.get("title") or "").strip()
    if title and title != "（无标题）":
        score += 1
    token = str(item.get("xsecToken") or item.get("xsec_token") or "").strip()
    if token:
        score += 1
    return score


def _merge_feed_records(base: dict, incoming: dict) -> dict:
    """合并两条同源笔记记录，保留更完整的作者/互动数据。"""
    if _feed_richness(incoming) > _feed_richness(base):
        rich, poor = incoming, base
    else:
        rich, poor = base, incoming

    out = json.loads(json.dumps(rich, ensure_ascii=False))
    poor_note = _note_block(poor)
    out_note = _note_block(out)
    if not out_note:
        out["noteCard"] = json.loads(json.dumps(poor_note, ensure_ascii=False))
        out_note = out["noteCard"]

    poor_user = poor_note.get("user") if isinstance(poor_note.get("user"), dict) else {}
    out_user = out_note.get("user") if isinstance(out_note.get("user"), dict) else {}
    if not out_user:
        out_note["user"] = {}
        out_user = out_note["user"]
    for key in ("nickname", "nickName", "nick_name"):
        if not str(out_user.get(key) or "").strip() and poor_user.get(key):
            out_user["nickname"] = poor_user[key]
            break

    poor_inter = poor_note.get("interactInfo") or poor_note.get("interact_info") or {}
    if not isinstance(poor_inter, dict):
        poor_inter = {}
    out_inter = out_note.get("interactInfo") or out_note.get("interact_info") or {}
    if not isinstance(out_inter, dict):
        out_inter = {}
        out_note["interactInfo"] = out_inter
    for key in ("likedCount", "liked_count", "commentCount", "comment_count", "collectedCount"):
        if not str(out_inter.get(key) or "").strip() and poor_inter.get(key):
            canon = "likedCount" if "like" in key else (
                "commentCount" if "comment" in key else "collectedCount"
            )
            out_inter[canon] = poor_inter[key]

    if not str(out.get("xsecToken") or out.get("xsec_token") or "").strip():
        tok = poor.get("xsecToken") or poor.get("xsec_token")
        if tok:
            out["xsecToken"] = tok
    if not str(out.get("id") or "").strip() and poor.get("id"):
        out["id"] = poor["id"]
    title = str(out_note.get("displayTitle") or out_note.get("title") or "").strip()
    if not title or title == "（无标题）":
        pt = poor_note.get("displayTitle") or poor_note.get("title")
        if pt:
            out_note["displayTitle"] = pt
    return out


def _feed_key(item: dict) -> str | None:
    fid = str(item.get("id") or item.get("feed_id") or "").strip()
    if fid and _FEED_ID_RE.match(fid):
        return fid
    note = _note_block(item)
    nid = str(note.get("id") or note.get("noteId") or "").strip()
    if nid and _FEED_ID_RE.match(nid):
        return nid
    return None


def extract_feeds_list(page: Page) -> list[dict]:
    """先 __INITIAL_STATE__（含作者/点赞），再用 DOM 补全缺失条目。"""
    merged: dict[str, dict] = {}
    state_count = 0
    dom_count = 0

    raw = page.evaluate(_EXTRACT_STATE_FEEDS_JS)
    if isinstance(raw, str) and raw.strip():
        try:
            data = json.loads(raw)
            if isinstance(data, list):
                for item in data:
                    if isinstance(item, dict):
                        key = _feed_key(item)
                        if key:
                            merged[key] = item
                            state_count += 1
        except json.JSONDecodeError as exc:
            logger.warning("解析 __INITIAL_STATE__ feeds 失败: %s", exc)

    dom_raw = page.evaluate(_EXTRACT_DOM_FEEDS_JS)
    if isinstance(dom_raw, str) and dom_raw.strip():
        try:
            data = json.loads(dom_raw)
            if isinstance(data, list):
                for item in data:
                    if isinstance(item, dict):
                        key = _feed_key(item)
                        if not key:
                            continue
                        if key in merged:
                            merged[key] = _merge_feed_records(merged[key], item)
                        else:
                            merged[key] = item
                            dom_count += 1
        except json.JSONDecodeError as exc:
            logger.warning("解析 DOM feeds 失败: %s", exc)

    if merged:
        logger.info(
            "提取笔记 %d 条（state=%d, dom新增/合并=%d）",
            len(merged),
            state_count,
            dom_count,
        )
    return list(merged.values())


def merge_feed_dicts(target: dict[str, dict], items: list[dict]) -> int:
    added = 0
    for item in items:
        key = _feed_key(item)
        if not key:
            continue
        if key not in target:
            added += 1
            target[key] = item
        else:
            target[key] = _merge_feed_records(target[key], item)
    return added