front/xhs-agent-desktop/engine/xhs_bridge/feed_detail_parser.py

from __future__ import annotations

import json
from datetime import datetime
from typing import Any


def parse_feed_detail_payload(data: Any) -> dict[str, Any]:
    """将 Skills CLI 详情响应解析为 UI 可用的结构。"""
    if data is None:
        return {"ok": False, "error": "空响应"}

    if isinstance(data, str):
        text = data.strip()
        if not text:
            return {"ok": False, "error": "空响应"}
        try:
            data = json.loads(text)
        except json.JSONDecodeError:
            return {"ok": False, "error": "无法解析详情 JSON"}

    if not isinstance(data, dict):
        return {"ok": False, "error": "响应格式无效"}

    if data.get("success") is False:
        return {"ok": False, "error": str(data.get("message") or data.get("error") or "请求失败")}

    note, comments = _extract_note_and_comments(data)
    if note is None:
        return {"ok": False, "error": "未找到笔记详情"}

    return {
        "ok": True,
        "note": _normalize_note(note),
        "comments": _normalize_comments(comments),
    }


def _extract_note_and_comments(data: dict[str, Any]) -> tuple[dict[str, Any] | None, Any]:
    if "note" in data and isinstance(data["note"], dict):
        comments = data.get("comments")
        return data["note"], comments

    inner = data.get("data")
    if isinstance(inner, dict):
        if "note" in inner and isinstance(inner["note"], dict):
            return inner["note"], inner.get("comments")

        nested = inner.get("data")
        if isinstance(nested, dict):
            note = nested.get("note")
            if isinstance(note, dict):
                return note, nested.get("comments")

    return None, None


def _normalize_note(note: dict[str, Any]) -> dict[str, Any]:
    user = note.get("user") or {}
    interact = note.get("interactInfo") or note.get("interact_info") or {}
    note_type = str(note.get("type") or "").lower()
    images = _normalize_images(note)
    video = note.get("video") or {}
    duration = 0
    if isinstance(video, dict):
        capa = video.get("capa") or {}
        if isinstance(capa, dict):
            duration = int(capa.get("duration") or 0)

    return {
        "note_id": str(note.get("noteId") or note.get("note_id") or ""),
        "title": str(note.get("title") or "（无标题）"),
        "desc": str(note.get("desc") or ""),
        "type": note_type,
        "is_video": note_type == "video" or bool(video),
        "video_duration": duration,
        "time": _format_time(note.get("time")),
        "ip_location": str(note.get("ipLocation") or note.get("ip_location") or ""),
        "user": {
            "user_id": str(user.get("userId") or user.get("user_id") or ""),
            "nickname": str(
                user.get("nickname") or user.get("nickName") or user.get("nick_name") or "—"
            ),
            "avatar": str(user.get("avatar") or ""),
        },
        "interact": {
            "liked": bool(interact.get("liked")),
            "liked_count": str(interact.get("likedCount") or interact.get("liked_count") or "0"),
            "collected_count": str(
                interact.get("collectedCount") or interact.get("collected_count") or "0"
            ),
            "comment_count": str(
                interact.get("commentCount") or interact.get("comment_count") or "0"
            ),
            "shared_count": str(interact.get("sharedCount") or interact.get("shared_count") or "0"),
        },
        "images": images,
    }


def _normalize_images(note: dict[str, Any]) -> list[dict[str, Any]]:
    result: list[dict[str, Any]] = []
    image_list = note.get("imageList") or note.get("image_list") or []
    if isinstance(image_list, list):
        for img in image_list:
            if not isinstance(img, dict):
                continue
            url = (
                img.get("urlDefault")
                or img.get("url_default")
                or img.get("urlPre")
                or img.get("url_pre")
                or img.get("url")
                or ""
            )
            if url:
                result.append(
                    {
                        "url": str(url),
                        "width": int(img.get("width") or 0),
                        "height": int(img.get("height") or 0),
                        "live_photo": bool(img.get("livePhoto") or img.get("live_photo")),
                    }
                )
    return result


def _normalize_comments(comments: Any) -> list[dict[str, Any]]:
    if comments is None:
        return []
    if isinstance(comments, dict):
        items = comments.get("list") or []
    elif isinstance(comments, list):
        items = comments
    else:
        return []

    if not isinstance(items, list):
        return []

    return [_normalize_comment(c) for c in items if isinstance(c, dict)]


def _normalize_comment(raw: dict[str, Any]) -> dict[str, Any]:
    user = raw.get("userInfo") or raw.get("user_info") or {}
    sub_raw = raw.get("subComments") or raw.get("sub_comments") or []
    sub_comments: list[dict[str, Any]] = []
    if isinstance(sub_raw, list):
        sub_comments = [_normalize_comment(s) for s in sub_raw if isinstance(s, dict)]

    tags = raw.get("showTags") or raw.get("show_tags") or []
    if not isinstance(tags, list):
        tags = []

    return {
        "id": str(raw.get("id") or ""),
        "content": str(raw.get("content") or ""),
        "like_count": str(raw.get("likeCount") or raw.get("like_count") or "0"),
        "time": _format_time(raw.get("createTime") or raw.get("create_time")),
        "ip_location": str(raw.get("ipLocation") or raw.get("ip_location") or ""),
        "user": {
            "nickname": str(
                user.get("nickname") or user.get("nickName") or user.get("nick_name") or "匿名"
            ),
            "avatar": str(user.get("avatar") or ""),
        },
        "sub_comment_count": str(
            raw.get("subCommentCount") or raw.get("sub_comment_count") or len(sub_comments)
        ),
        "tags": [str(t) for t in tags],
        "sub_comments": sub_comments,
    }


def _format_time(value: Any) -> str:
    if value is None:
        return ""
    try:
        ts = int(value)
    except (TypeError, ValueError):
        return str(value)
    if ts > 1_000_000_000_000:
        ts //= 1000
    try:
        return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M")
    except (OSError, ValueError, OverflowError):
        return ""