Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 55 additions & 10 deletions agent_reach/channels/reddit.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,19 @@
# -*- coding: utf-8 -*-
"""Reddit — search and read via rdt-cli (public-clis/rdt-cli)."""
"""Reddit — search and read via rdt-cli (public-clis/rdt-cli).

NOTE: Reddit requires authentication since 2024. All API requests
(including public subreddit reads) return HTTP 403 without a valid
session cookie. Run `rdt login` after installation to authenticate.
"""

import json
import shutil
import subprocess

from .base import Channel

_CREDENTIAL_FILE = "~/.config/rdt-cli/credential.json"


class RedditChannel(Channel):
name = "reddit"
Expand All @@ -14,18 +23,54 @@ class RedditChannel(Channel):

def can_handle(self, url: str) -> bool:
from urllib.parse import urlparse

d = urlparse(url).netloc.lower()
return "reddit.com" in d or "redd.it" in d

def check(self, config=None):
rdt = shutil.which("rdt")
if rdt:
return "ok", (
"rdt-cli 可用(搜索帖子、阅读全文、查看评论,无需登录)"
if not rdt:
return "off", (
"需要安装 rdt-cli(推荐使用最新版 v0.4.2+):\n"
" pip install 'rdt-cli>=0.4.2'\n"
"或:\n"
" uv tool install rdt-cli\n"
"最新源码:https://github.com/public-clis/rdt-cli\n"
"安装后运行 `rdt login` 登录(需先在浏览器登录 reddit.com)"
)

try:
r = subprocess.run(
[rdt, "status", "--json"],
capture_output=True,
encoding="utf-8",
errors="replace",
timeout=10,
)
return "off", (
"需要安装 rdt-cli:\n"
" pipx install rdt-cli\n"
"或:\n"
" uv tool install rdt-cli"
)
data = json.loads(r.stdout or "{}")
authenticated = data.get("data", {}).get("authenticated", False)
username = data.get("data", {}).get("username") or ""

if authenticated:
suffix = f"(已登录:{username})" if username else ""
return "ok", (f"rdt-cli 可用{suffix}(搜索帖子、阅读全文、查看评论)")

return "warn", (
"rdt-cli 已安装但未登录。Reddit 自 2024 年起要求认证,"
"未登录时所有请求均返回 403。\n\n"
"方法一(自动):运行 `rdt login`\n"
" 先在浏览器登录 reddit.com,再运行此命令自动提取 Cookie。\n\n"
"方法二(手动,适用于 Chrome/Edge 127+ 无法自动提取时):\n"
" 1. Chrome 应用商店安装 Cookie-Editor 扩展:\n"
" https://chromewebstore.google.com/detail/cookie-editor/hlkenndednhfkekhgcdicdfddnkalmdm\n"
" 2. 在浏览器打开 reddit.com(确保已登录)\n"
" 3. 点击 Cookie-Editor 图标,找到 `reddit_session`,复制其 Value\n"
f" 4. 将以下内容写入 {_CREDENTIAL_FILE}:\n"
' {"cookies": {"reddit_session": "<粘贴 Value>"}, '
'"source": "manual", "username": "<你的用户名>", '
'"modhash": null, "saved_at": 0, "last_verified_at": null}\n\n'
"验证:`rdt status --json` 确认 authenticated: true"
)

except (json.JSONDecodeError, FileNotFoundError, subprocess.TimeoutExpired):
return "warn", "rdt-cli 已安装但状态检查失败,运行 `rdt status` 查看详情"
70 changes: 67 additions & 3 deletions tests/test_channels.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,9 @@
from urllib.error import URLError

from agent_reach.channels import get_all_channels, get_channel
from agent_reach.channels.v2ex import V2EXChannel
from agent_reach.channels.xiaohongshu import XiaoHongShuChannel
from agent_reach.channels.xueqiu import XueqiuChannel
from agent_reach.channels.v2ex import V2EXChannel


class TestChannelRegistry:
Expand Down Expand Up @@ -223,8 +223,6 @@ def test_get_topic_returns_detail_and_replies(self, monkeypatch):
},
]

call_count = {"n": 0}

class FakeResponse:
def __init__(self, payload):
self._payload = payload
Expand Down Expand Up @@ -648,6 +646,72 @@ def fake_open(req, timeout=None):
assert "agent-reach" not in captured["ua"]


class TestRedditChannel:
def test_reports_off_when_not_installed(self, monkeypatch):
monkeypatch.setattr(shutil, "which", lambda _: None)
from agent_reach.channels.reddit import RedditChannel
status, msg = RedditChannel().check()
assert status == "off"
assert "rdt-cli" in msg
assert "public-clis/rdt-cli" in msg

def test_reports_ok_when_authenticated(self, monkeypatch):
monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/rdt")
fake_output = json.dumps({
"ok": True,
"schema_version": "1",
"data": {"authenticated": True, "username": "testuser", "cookie_count": 1},
})

def fake_run(cmd, **kwargs):
return subprocess.CompletedProcess(cmd, 0, fake_output, "")

monkeypatch.setattr(subprocess, "run", fake_run)
from agent_reach.channels.reddit import RedditChannel
status, msg = RedditChannel().check()
assert status == "ok"
assert "testuser" in msg

def test_reports_warn_when_not_authenticated(self, monkeypatch):
monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/rdt")
fake_output = json.dumps({
"ok": True,
"schema_version": "1",
"data": {"authenticated": False, "username": None, "cookie_count": 0},
})

def fake_run(cmd, **kwargs):
return subprocess.CompletedProcess(cmd, 0, fake_output, "")

monkeypatch.setattr(subprocess, "run", fake_run)
from agent_reach.channels.reddit import RedditChannel
status, msg = RedditChannel().check()
assert status == "warn"
assert "403" in msg
assert "rdt login" in msg
assert "Cookie-Editor" in msg
assert "chromewebstore.google.com" in msg

def test_reports_warn_when_status_check_fails(self, monkeypatch):
monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/rdt")

def fake_run(cmd, **kwargs):
return subprocess.CompletedProcess(cmd, 1, "not valid json{{{", "")

monkeypatch.setattr(subprocess, "run", fake_run)
from agent_reach.channels.reddit import RedditChannel
status, msg = RedditChannel().check()
assert status == "warn"

def test_can_handle_reddit_urls(self):
from agent_reach.channels.reddit import RedditChannel
ch = RedditChannel()
assert ch.can_handle("https://www.reddit.com/r/python/comments/abc123/")
assert ch.can_handle("https://redd.it/abc123")
assert not ch.can_handle("https://github.com/user/repo")
assert not ch.can_handle("https://v2ex.com/t/123")


class TestXiaoHongShuChannel:
def test_reports_ok_when_cli_authenticated(self, monkeypatch):
monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs")
Expand Down