Panniantong
1b8b202e1e
Major install flow polish: - Channels organized by setup difficulty: 🟢 Zero-config (Web, GitHub, RSS) — works immediately 🔵 Cookie-based (Twitter, YouTube, Bilibili, XiaoHongShu) — free, ~2min 🟡 Free API key (Exa Search) — one key, 30 seconds 🟠 Proxy-based (Reddit, Bilibili on server) — $1/month - Every channel explains: what it does, what's needed, what you miss without it - Server vs local affects which channels need proxy New: XiaoHongShu channel (cookie-based, falls back to Jina Reader) New configure commands: twitter-cookies, xhs-cookie, youtube-cookies Each command auto-tests after saving.
111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""XiaoHongShu (小红书) — via cookie-based API access.
|
|
|
|
Backend: XHS web API + cookies
|
|
Swap to: any XHS access method
|
|
"""
|
|
|
|
import re
|
|
import json
|
|
import requests
|
|
from urllib.parse import urlparse
|
|
from .base import Channel, ReadResult
|
|
|
|
|
|
class XiaoHongShuChannel(Channel):
|
|
name = "xiaohongshu"
|
|
description = "XiaoHongShu (小红书) notes"
|
|
backends = ["XHS Web API"]
|
|
requires_config = ["xhs_cookie"]
|
|
tier = 2
|
|
|
|
def can_handle(self, url: str) -> bool:
|
|
domain = urlparse(url).netloc.lower()
|
|
return "xiaohongshu.com" in domain or "xhslink.com" in domain
|
|
|
|
async def read(self, url: str, config=None) -> ReadResult:
|
|
cookie = config.get("xhs_cookie") if config else None
|
|
|
|
if not cookie:
|
|
# Fallback to Jina Reader (works for some public notes)
|
|
from agent_eyes.channels.web import WebChannel
|
|
return await WebChannel().read(url, config)
|
|
|
|
# Extract note ID from URL
|
|
note_id = self._extract_note_id(url)
|
|
if not note_id:
|
|
from agent_eyes.channels.web import WebChannel
|
|
return await WebChannel().read(url, config)
|
|
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
|
|
"Cookie": cookie,
|
|
"Referer": "https://www.xiaohongshu.com/",
|
|
}
|
|
|
|
# Fetch note page
|
|
resp = requests.get(
|
|
f"https://www.xiaohongshu.com/explore/{note_id}",
|
|
headers=headers,
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
html = resp.text
|
|
|
|
# Extract note data from HTML
|
|
title, content, author = self._parse_html(html)
|
|
|
|
return ReadResult(
|
|
title=title or f"XHS Note {note_id}",
|
|
content=content or "Could not extract content. Cookie may be expired.",
|
|
url=url,
|
|
author=author,
|
|
platform="xiaohongshu",
|
|
)
|
|
|
|
def _extract_note_id(self, url: str) -> str:
|
|
"""Extract note ID from various XHS URL formats."""
|
|
# https://www.xiaohongshu.com/explore/xxxxx
|
|
# https://www.xiaohongshu.com/discovery/item/xxxxx
|
|
# https://xhslink.com/xxxxx
|
|
path = urlparse(url).path
|
|
parts = path.strip("/").split("/")
|
|
if parts:
|
|
return parts[-1]
|
|
return ""
|
|
|
|
def _parse_html(self, html: str):
|
|
"""Extract title, content, author from XHS HTML."""
|
|
title = ""
|
|
content = ""
|
|
author = ""
|
|
|
|
# Try to find JSON data in page
|
|
match = re.search(r'window\.__INITIAL_STATE__\s*=\s*({.*?})\s*</script>', html, re.DOTALL)
|
|
if match:
|
|
try:
|
|
# XHS embeds note data in initial state
|
|
state = json.loads(match.group(1).replace('undefined', 'null'))
|
|
note_data = state.get("note", {}).get("noteDetailMap", {})
|
|
if note_data:
|
|
first_note = list(note_data.values())[0]
|
|
note = first_note.get("note", {})
|
|
title = note.get("title", "")
|
|
content = note.get("desc", "")
|
|
author = note.get("user", {}).get("nickname", "")
|
|
except (json.JSONDecodeError, KeyError, IndexError):
|
|
pass
|
|
|
|
# Fallback: extract from meta tags
|
|
if not title:
|
|
m = re.search(r'<title>(.*?)</title>', html)
|
|
if m:
|
|
title = m.group(1)
|
|
|
|
if not content:
|
|
m = re.search(r'<meta name="description" content="(.*?)"', html)
|
|
if m:
|
|
content = m.group(1)
|
|
|
|
return title, content, author
|