sawz/Agent-Reach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #210 from MatrixA/fix/xueqiu-cookie-auth

Browse Source 
Tests passed locally (75/75). Fixes xueqiu 400 errors with browser cookie auth, correct UA/Referer, and updated hot posts endpoint.
This commit is contained in:
Pnant
2026-03-31 11:15:17 +08:00
committed by GitHub
parent e51bd91e82 d793afaba6
commit 526616e13f
9 changed files with 293 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent_reach/channels/xueqiu.py
+134 -24
View File
@@ -4,12 +4,18 @@
import http.cookiejar
import json
import re
import urllib.parse
import urllib.request
from typing import Any
from .base import Channel
_UA = "agent-reach/1.0"
_UA = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
)
_REFERER = "https://xueqiu.com/"
_TIMEOUT = 10
_XUEQIU_HOME = "https://xueqiu.com"
@@ -22,11 +28,90 @@ _opener = urllib.request.build_opener(
_cookies_initialized = False
def _inject_cookie_string(cookie_str: str) -> None:
"""Parse a 'name=value; name2=value2' string and inject into the cookie jar."""
for pair in cookie_str.split(";"):
pair = pair.strip()
if "=" not in pair:
continue
name, _, value = pair.partition("=")
cookie = http.cookiejar.Cookie(
version=0,
name=name.strip(),
value=value.strip(),
port=None,
port_specified=False,
domain=".xueqiu.com",
domain_specified=True,
domain_initial_dot=True,
path="/",
path_specified=True,
secure=True,
expires=None,
discard=True,
comment=None,
comment_url=None,
rest={},
)
_cookie_jar.set_cookie(cookie)
def _load_cookies_from_config() -> bool:
"""Try to load Xueqiu cookies from agent-reach config file (xueqiu_cookie key)."""
try:
from ..config import Config
cfg = Config()
cookie_str = cfg.get("xueqiu_cookie")
if not cookie_str:
return False
_inject_cookie_string(cookie_str)
return True
except Exception:
return False
def _load_cookies_from_browser() -> bool:
"""Try to silently load Xueqiu cookies from the local Chrome browser.
Only succeeds when browser_cookie3 is installed AND the user is logged in
(xq_a_token present). Failures are silently ignored so that agents without
a local browser keep working.
"""
try:
import browser_cookie3
cookies = list(browser_cookie3.chrome(domain_name=".xueqiu.com"))
if not any(c.name == "xq_a_token" for c in cookies):
return False
for c in cookies:
_cookie_jar.set_cookie(c)
return True
except Exception:
return False
def _ensure_cookies() -> None:
"""Visit xueqiu.com homepage once to obtain session cookies."""
"""Populate session cookies using the best available source.
Priority order:
1. Saved cookie string in ~/.agent-reach/config.yaml (set by configure --from-browser)
2. Live Chrome browser cookies via browser_cookie3 (if installed + logged in)
3. Homepage visit fallback (only yields anti-DDoS acw_tc,
not enough for stock APIs)
"""
global _cookies_initialized
if _cookies_initialized:
return
if _load_cookies_from_config():
_cookies_initialized = True
return
if _load_cookies_from_browser():
_cookies_initialized = True
return
# Fallback: visit homepage to pick up acw_tc anti-DDoS cookie.
# This is not sufficient for authenticated APIs but avoids hard failures
# on public endpoints that only need the session cookie.
req = urllib.request.Request(_XUEQIU_HOME, headers={"User-Agent": _UA})
_opener.open(req, timeout=_TIMEOUT)
_cookies_initialized = True
@@ -35,7 +120,9 @@ def _ensure_cookies() -> None:
def _get_json(url: str) -> Any:
"""Fetch *url* with Xueqiu session cookies and return parsed JSON."""
_ensure_cookies()
req = urllib.request.Request(url, headers={"User-Agent": _UA})
req = urllib.request.Request(
url, headers={"User-Agent": _UA, "Referer": _REFERER}
)
with _opener.open(req, timeout=_TIMEOUT) as resp:
return json.loads(resp.read().decode("utf-8"))
@@ -51,17 +138,15 @@ def _strip_html(text: str) -> str:
class XueqiuChannel(Channel):
name = "xueqiu"
description = "雪球股票行情与社区动态"
backends = ["Xueqiu API (public)"]
tier = 0
backends = ["Xueqiu API (需要登录 Cookie)"]
tier = 1
# ------------------------------------------------------------------ #
# URL routing
# ------------------------------------------------------------------ #
def can_handle(self, url: str) -> bool:
from urllib.parse import urlparse
d = urlparse(url).netloc.lower()
d = urllib.parse.urlparse(url).netloc.lower()
return "xueqiu.com" in d
# ------------------------------------------------------------------ #
@@ -70,13 +155,18 @@ class XueqiuChannel(Channel):
def check(self, config=None):
try:
data = _get_json("https://stock.xueqiu.com/v5/stock/batch/quote.json?symbol=SH000001")
data = _get_json(
"https://stock.xueqiu.com/v5/stock/batch/quote.json?symbol=SH000001"
)
items = (data.get("data") or {}).get("items") or []
if items:
return "ok", "公开 API 可用（行情、搜索、热帖、热股）"
return "warn", "API 响应异常（返回数据为空）"
except Exception as e:
return "warn", f"Xueqiu API 连接失败（可能需要代理）：{e}"
return "warn", (
f"Xueqiu API 连接失败：{e}。"
"请先登录雪球后运行：agent-reach configure --from-browser chrome"
)
# ------------------------------------------------------------------ #
# Data-fetching methods
@@ -92,7 +182,9 @@ class XueqiuChannel(Channel):
symbol, name, current, percent, chg, high, low, open, last_close,
volume, amount, market_capital, turnover_rate, pe_ttm, timestamp
"""
data = _get_json(f"https://stock.xueqiu.com/v5/stock/batch/quote.json?symbol={symbol}")
data = _get_json(
f"https://stock.xueqiu.com/v5/stock/batch/quote.json?symbol={symbol}"
)
items = (data.get("data") or {}).get("items") or []
q = (items[0].get("quote") or {}) if items else {}
return {
@@ -124,7 +216,8 @@ class XueqiuChannel(Channel):
symbol, name, exchange
"""
data = _get_json(
f"https://xueqiu.com/stock/search.json?code={urllib.request.quote(query)}&size={limit}"
f"https://xueqiu.com/stock/search.json"
f"?code={urllib.parse.quote(query)}&size={limit}"
)
stocks = data.get("stocks") or []
results = []
@@ -141,29 +234,45 @@ class XueqiuChannel(Channel):
def get_hot_posts(self, limit: int = 20) -> list:
"""获取雪球热门帖子。
Uses the v4 public timeline endpoint which returns posts in a `list`
array. Each item carries a JSON-encoded `data` field containing the
actual post payload (title, description, user, like_count, target).
Args:
limit: 最多返回条数（上限 50）
Returns a list of dicts with keys:
id, title, text, author, likes, url
"""
data = _get_json("https://xueqiu.com/statuses/hot/listV3.json?source=hot&page=1")
items = (data.get("data") or {}).get("items") or []
data = _get_json(
"https://xueqiu.com/v4/statuses/public_timeline_by_category.json"
"?since_id=-1&max_id=-1&count=20&category=-1"
)
items = data.get("list") or []
results = []
for item in items[:limit]:
original = item.get("original_status") or item
text = _strip_html(original.get("text") or original.get("description") or "")
user = original.get("user") or {}
# Each item.data is a JSON string containing the real post payload
try:
post = (
json.loads(item["data"])
if isinstance(item.get("data"), str)
else {}
)
except (json.JSONDecodeError, KeyError):
post = {}
user = post.get("user") or {}
text = _strip_html(
post.get("text") or post.get("description") or ""
)
target = post.get("target", "")
results.append(
{
"id": original.get("id", 0),
"title": original.get("title") or "",
"id": post.get("id", 0),
"title": post.get("title") or "",
"text": text[:200],
"author": user.get("screen_name", ""),
"likes": original.get("like_count", 0),
"url": f"https://xueqiu.com{original['target']}"
if original.get("target")
else "",
"likes": post.get("like_count", 0),
"url": f"https://xueqiu.com{target}" if target else "",
}
)
return results
@@ -179,7 +288,8 @@ class XueqiuChannel(Channel):
symbol, name, current, percent, rank
"""
data = _get_json(
f"https://stock.xueqiu.com/v5/stock/hot_stock/list.json?size={limit}&type={stock_type}"
f"https://stock.xueqiu.com/v5/stock/hot_stock/list.json"
f"?size={limit}&type={stock_type}"
)
items = (data.get("data") or {}).get("items") or []
results = []
agent_reach/cookie_extract.py
+18
View File
@@ -32,6 +32,12 @@ PLATFORM_SPECS = [
"cookies": ["SESSDATA", "bili_jct"],
"config_key": "bilibili",
},
{
"name": "Xueqiu",
"domains": [".xueqiu.com", "xueqiu.com"],
"cookies": None, # grab all — xq_a_token + session cookies required
"config_key": "xueqiu",
},
]
@@ -217,4 +223,16 @@ def configure_from_browser(browser: str, config) -> List[Tuple[str, bool, str]]:
results_list.append(("Bilibili", False,
f"No SESSDATA found. Make sure you're logged into bilibili.com in {browser}."))
if "xueqiu" in extracted:
cookie_str = extracted["xueqiu"].get("cookie_string", "")
# Only save if xq_a_token is present — anonymous cookies are useless
if cookie_str and "xq_a_token" in cookie_str:
config.set("xueqiu_cookie", cookie_str)
n_cookies = len(cookie_str.split(";"))
results_list.append(("Xueqiu", True, f"{n_cookies} cookies (含 xq_a_token)"))
elif cookie_str:
results_list.append(("Xueqiu", False,
f"找到 {len(cookie_str.split(';'))} 个 Cookie 但缺少 xq_a_token，"
f"请先在 {browser} 中登录 xueqiu.com"))
return results_list
agent_reach/skill/SKILL.md
+3 -3
View File
@@ -255,7 +255,7 @@ print(result[0]["error"]) # 提示使用站内搜索或 Exa channel
> No auth required. Results are public JSON. V2EX 节点名见 https://www.v2ex.com/planes
## 雪球 / Xueqiu (public API)
## 雪球 / Xueqiu (需要登录 Cookie)
```python
from agent_reach.channels.xueqiu import XueqiuChannel
@@ -274,7 +274,7 @@ stocks = ch.search_stock("茅台", limit=5)
for s in stocks:
print(f"{s['name']} ({s['symbol']}) - {s['exchange']}")
# 热门帖子
# 热门帖子（v4 timeline，含作者、点赞数、正文）
# 返回字段：id, title, text(前200字), author, likes, url
posts = ch.get_hot_posts(limit=10)
for p in posts:
@@ -287,7 +287,7 @@ for s in hot:
print(f"#{s['rank']} {s['name']} ({s['symbol']}): {s['current']} ({s['percent']}%)")
```
> 无需登录。自动获取会话 Cookie，所有公开 API 均可直接使用。
> 需要登录 Cookie。先在 Chrome 登录 xueqiu.com，再运行 `agent-reach configure --from-browser chrome`。
## RSS (feedparser)