zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
292 lines
9.1 KiB
Python
292 lines
9.1 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any
|
||
|
||
# 从 API/CLI 载荷中提取笔记列表时尝试的键（首页、搜索通用）
|
||
_FEED_LIST_KEYS = ("feeds", "items", "list", "results", "notes", "records")
|
||
|
||
|
||
def _normalize_profile_feed_item(item: dict[str, Any]) -> dict[str, Any]:
|
||
"""将 user-profile / my-account 的扁平 feed 转为列表行。"""
|
||
if item.get("noteCard") or item.get("note_card") or item.get("note"):
|
||
return _normalize_feed_item(item)
|
||
note_card: dict[str, Any] = {
|
||
"displayTitle": item.get("displayTitle")
|
||
or item.get("display_title")
|
||
or item.get("title"),
|
||
"type": item.get("type"),
|
||
"user": item.get("user") if isinstance(item.get("user"), dict) else {},
|
||
"interactInfo": item.get("interactInfo")
|
||
if isinstance(item.get("interactInfo"), dict)
|
||
else item.get("interact_info")
|
||
if isinstance(item.get("interact_info"), dict)
|
||
else {},
|
||
}
|
||
if item.get("video"):
|
||
note_card["video"] = item.get("video")
|
||
return _normalize_feed_item(
|
||
{
|
||
"id": item.get("id") or item.get("feed_id"),
|
||
"xsecToken": item.get("xsecToken") or item.get("xsec_token"),
|
||
"modelType": item.get("modelType") or item.get("model_type") or "note",
|
||
"noteCard": note_card,
|
||
}
|
||
)
|
||
|
||
|
||
def parse_user_profile_feeds(data: Any) -> list[dict[str, Any]]:
|
||
"""解析用户主页 / 我的账号接口返回的 feeds 数组。"""
|
||
if data is None:
|
||
return []
|
||
if isinstance(data, dict):
|
||
feeds = data.get("feeds")
|
||
if isinstance(feeds, list):
|
||
return [
|
||
_normalize_profile_feed_item(x) for x in feeds if isinstance(x, dict)
|
||
]
|
||
if isinstance(data, list):
|
||
return [_normalize_profile_feed_item(x) for x in data if isinstance(x, dict)]
|
||
return []
|
||
|
||
|
||
def parse_feeds_payload(data: Any) -> list[dict[str, Any]]:
|
||
"""将 Skills CLI 返回统一解析为 Feed 行列表（首页推荐、搜索内容共用）。"""
|
||
if data is None:
|
||
return []
|
||
|
||
if isinstance(data, str):
|
||
text = data.strip()
|
||
if not text:
|
||
return []
|
||
try:
|
||
data = json.loads(text)
|
||
except json.JSONDecodeError:
|
||
return []
|
||
|
||
if isinstance(data, list):
|
||
return [_normalize_feed_item(x) for x in data if isinstance(x, dict)]
|
||
|
||
if isinstance(data, dict):
|
||
feeds = _extract_feeds_list(data)
|
||
if not feeds:
|
||
val = data.get("feeds")
|
||
if isinstance(val, list):
|
||
feeds = val
|
||
return [_normalize_profile_feed_item(x) for x in feeds if isinstance(x, dict)]
|
||
|
||
return []
|
||
|
||
|
||
def _extract_feeds_list(data: dict[str, Any]) -> list[Any]:
|
||
"""从 dict 载荷中取出笔记数组（兼容搜索/首页不同包裹层）。"""
|
||
for key in _FEED_LIST_KEYS:
|
||
val = data.get(key)
|
||
if isinstance(val, list):
|
||
return val
|
||
|
||
inner = data.get("data")
|
||
if isinstance(inner, dict):
|
||
for key in _FEED_LIST_KEYS:
|
||
val = inner.get(key)
|
||
if isinstance(val, list):
|
||
return val
|
||
|
||
search = data.get("search")
|
||
if isinstance(search, dict):
|
||
for key in _FEED_LIST_KEYS:
|
||
val = search.get(key)
|
||
if isinstance(val, list):
|
||
return val
|
||
feeds_wrap = search.get("feeds")
|
||
if isinstance(feeds_wrap, dict):
|
||
val = feeds_wrap.get("value") or feeds_wrap.get("_value")
|
||
if isinstance(val, list):
|
||
return val
|
||
|
||
return []
|
||
|
||
|
||
def _note_dict(item: dict[str, Any]) -> dict[str, Any]:
|
||
"""合并 noteCard / note_card / note（搜索与首页字段名不一致）。"""
|
||
merged: dict[str, Any] = {}
|
||
for key in ("noteCard", "note_card", "note"):
|
||
block = item.get(key)
|
||
if isinstance(block, dict):
|
||
merged.update(block)
|
||
return merged
|
||
|
||
|
||
def _merge_interact(note: dict[str, Any], item: dict[str, Any]) -> dict[str, Any]:
|
||
"""合并各层 interactInfo（noteCard 与 CLI to_dict 顶层可能各有一部分）。"""
|
||
merged: dict[str, Any] = {}
|
||
for src in (
|
||
note.get("interactInfo"),
|
||
note.get("interact_info"),
|
||
item.get("interactInfo"),
|
||
item.get("interact_info"),
|
||
item.get("interact"),
|
||
):
|
||
if isinstance(src, dict):
|
||
merged.update(src)
|
||
return merged
|
||
|
||
|
||
def _interact_count(interact: dict[str, Any], *keys: str) -> str:
|
||
for key in keys:
|
||
val = interact.get(key)
|
||
if val is None:
|
||
continue
|
||
text = str(val).strip()
|
||
if text:
|
||
return text
|
||
return "0"
|
||
|
||
|
||
def _comment_count(note: dict[str, Any], item: dict[str, Any], interact: dict[str, Any]) -> str:
|
||
count = _interact_count(
|
||
interact,
|
||
"commentCount",
|
||
"comment_count",
|
||
"commentsCount",
|
||
"comments_count",
|
||
)
|
||
if count != "0":
|
||
return count
|
||
count = _interact_count(
|
||
note,
|
||
"commentCount",
|
||
"comment_count",
|
||
"commentsCount",
|
||
"comments_count",
|
||
)
|
||
if count != "0":
|
||
return count
|
||
return _interact_count(
|
||
item,
|
||
"commentCount",
|
||
"comment_count",
|
||
"commentsCount",
|
||
"comments_count",
|
||
)
|
||
|
||
|
||
def _liked_count(note: dict[str, Any], item: dict[str, Any], interact: dict[str, Any]) -> str:
|
||
count = _interact_count(interact, "likedCount", "liked_count", "liked")
|
||
if count != "0":
|
||
return count
|
||
count = _interact_count(note, "likedCount", "liked_count", "liked")
|
||
if count != "0":
|
||
return count
|
||
return _interact_count(item, "likedCount", "liked_count", "liked")
|
||
|
||
|
||
NOTE_TYPE_VIDEO = "视频"
|
||
NOTE_TYPE_IMAGE = "图文"
|
||
|
||
|
||
def format_note_type_label(raw: Any) -> str:
|
||
"""将笔记类型转为中文：图文 / 视频。"""
|
||
text = str(raw or "").strip()
|
||
if not text:
|
||
return ""
|
||
lower = text.lower()
|
||
if lower in ("video", "videonote", "video_note", "videonotev2"):
|
||
return NOTE_TYPE_VIDEO
|
||
if lower in ("normal", "image", "pic", "photo", "图文"):
|
||
return NOTE_TYPE_IMAGE
|
||
# modelType 常为 note，不能当作图文/视频
|
||
if lower == "note":
|
||
return ""
|
||
if "video" in lower:
|
||
return NOTE_TYPE_VIDEO
|
||
if lower in ("图文",):
|
||
return NOTE_TYPE_IMAGE
|
||
return ""
|
||
|
||
|
||
def _has_video_payload(note: dict[str, Any], item: dict[str, Any]) -> bool:
|
||
if item.get("video") or note.get("video"):
|
||
return True
|
||
for key in ("videoDuration", "video_duration", "videoDurationText"):
|
||
if note.get(key) or item.get(key):
|
||
return True
|
||
cover = note.get("cover") or note.get("imageList") or note.get("image_list")
|
||
if isinstance(cover, dict) and cover.get("video"):
|
||
return True
|
||
if isinstance(cover, list):
|
||
for img in cover:
|
||
if isinstance(img, dict) and (
|
||
img.get("video") or str(img.get("type") or "").lower() == "video"
|
||
):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _resolve_note_type(note: dict[str, Any], item: dict[str, Any]) -> str:
|
||
if _has_video_payload(note, item):
|
||
return NOTE_TYPE_VIDEO
|
||
# noteCard.type 才是列表项图文/视频的主要依据（优先于 modelType）
|
||
candidates = (
|
||
note.get("type"),
|
||
note.get("noteType"),
|
||
note.get("note_type"),
|
||
item.get("type"),
|
||
item.get("noteType"),
|
||
item.get("note_type"),
|
||
item.get("modelType"),
|
||
item.get("model_type"),
|
||
note.get("modelType"),
|
||
note.get("model_type"),
|
||
)
|
||
for raw in candidates:
|
||
label = format_note_type_label(raw)
|
||
if label in (NOTE_TYPE_VIDEO, NOTE_TYPE_IMAGE):
|
||
return label
|
||
return "—"
|
||
|
||
|
||
def _normalize_feed_item(item: dict[str, Any]) -> dict[str, Any]:
|
||
note = _note_dict(item)
|
||
user = note.get("user") if isinstance(note.get("user"), dict) else {}
|
||
interact = _merge_interact(note, item)
|
||
|
||
title = (
|
||
note.get("displayTitle")
|
||
or note.get("display_title")
|
||
or note.get("title")
|
||
or item.get("displayTitle")
|
||
or item.get("display_title")
|
||
or item.get("title")
|
||
or "（无标题）"
|
||
)
|
||
if not user:
|
||
user = item.get("user") if isinstance(item.get("user"), dict) else {}
|
||
author = (
|
||
user.get("nickname")
|
||
or user.get("nickName")
|
||
or user.get("nick_name")
|
||
or user.get("name")
|
||
or item.get("nickname")
|
||
or item.get("nickName")
|
||
or item.get("author")
|
||
or "—"
|
||
)
|
||
|
||
return {
|
||
"feed_id": str(item.get("id") or item.get("feed_id") or ""),
|
||
"xsec_token": str(item.get("xsecToken") or item.get("xsec_token") or ""),
|
||
"title": str(title),
|
||
"author": str(author),
|
||
"liked": _liked_count(note, item, interact),
|
||
"comments": _comment_count(note, item, interact),
|
||
"collected": _interact_count(
|
||
interact,
|
||
"collectedCount",
|
||
"collected_count",
|
||
)
|
||
or _interact_count(note, "collectedCount", "collected_count")
|
||
or _interact_count(item, "collectedCount", "collected_count"),
|
||
"note_type": _resolve_note_type(note, item),
|
||
}
|