refactor: remove Discord and Toutiao channels (#234)

Discord only returns server metadata (no messages without Bot Token),
not useful enough. Toutiao search was fragile (HTML scraping). Both
can be revisited when better upstream tools appear.

75 tests passing.

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

.claude/settings.local.json

@@ -0,0 +1,7 @@
+{
+  "permissions": {
+    "allow": [
+      "WebFetch(domain:community.groq.com)"
+    ]
+  }
+}

agent_reach/channels/__init__.py

@@ -23,8 +23,6 @@ from .weibo import WeiboChannel
 from .xiaoyuzhou import XiaoyuzhouChannel
 from .v2ex import V2EXChannel
 from .xueqiu import XueqiuChannel
-from .toutiao import ToutiaoChannel
-from .discord import DiscordChannel
 
 
 ALL_CHANNELS: List[Channel] = [
@@ -41,8 +39,6 @@ ALL_CHANNELS: List[Channel] = [
     XiaoyuzhouChannel(),
     V2EXChannel(),
     XueqiuChannel(),
-    ToutiaoChannel(),
-    DiscordChannel(),
     RSSChannel(),
     ExaSearchChannel(),
     WebChannel(),

agent_reach/channels/discord.py

@@ -1,97 +0,0 @@
-# -*- coding: utf-8 -*-
-"""Discord channel — search via Exa, read public server info via Invite API."""
-
-import json
-import shutil
-import subprocess
-import urllib.request
-from typing import Tuple
-
-from .base import Channel
-
-_UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
-_INVITE_API = "https://discord.com/api/v10/invites/{code}?with_counts=true"
-
-
-def _exa_available() -> bool:
-    mcporter = shutil.which("mcporter")
-    if not mcporter:
-        return False
-    try:
-        r = subprocess.run(
-            [mcporter, "config", "list"],
-            capture_output=True, text=True, timeout=5
-        )
-        return "exa" in r.stdout.lower()
-    except Exception:
-        return False
-
-
-def _get_invite_code(url: str) -> str:
-    """Extract invite code from discord.gg/xxx or discord.com/invite/xxx URLs."""
-    url = url.rstrip("/")
-    for prefix in ("discord.gg/", "discord.com/invite/"):
-        if prefix in url:
-            return url.split(prefix)[-1].split("/")[0].split("?")[0]
-    return ""
-
-
-class DiscordChannel(Channel):
-    name = "discord"
-    description = "Discord 服务器信息与内容搜索"
-    backends = ["Exa via mcporter (搜索)", "Discord Invite API (服务器信息)"]
-    tier = 0
-
-    def can_handle(self, url: str) -> bool:
-        return "discord.gg" in url or "discord.com" in url
-
-    def check(self, config=None) -> Tuple[str, str]:
-        if _exa_available():
-            return "ok", "Discord 可用：Exa 搜索内容，Invite API 读取公开服务器信息"
-        return "warn", (
-            "Discord Invite API 可用（无需配置），但搜索功能需要 mcporter + Exa MCP。"
-            "运行 `agent-reach install --env=auto` 安装 Exa。"
-        )
-
-    def read(self, url: str) -> str:
-        """读取 Discord 服务器公开信息（Invite API）或通过 Jina 读取页面。"""
-        code = _get_invite_code(url)
-        if code:
-            api_url = _INVITE_API.format(code=code)
-            req = urllib.request.Request(api_url, headers={"User-Agent": _UA})
-            with urllib.request.urlopen(req, timeout=10) as resp:
-                data = json.loads(resp.read().decode("utf-8"))
-            guild = data.get("guild", {})
-            channel = data.get("channel", {})
-            members = data.get("approximate_member_count", "未知")
-            online = data.get("approximate_presence_count", "未知")
-            return (
-                f"# {guild.get('name', '未知服务器')}\n\n"
-                f"**描述**: {guild.get('description') or '无'}\n"
-                f"**成员**: {members:,} 人（在线 {online:,}）\n"
-                f"**频道**: #{channel.get('name', '未知')}\n"
-                f"**邀请链接**: {url}\n"
-            )
-        # Fallback: Jina Reader
-        if not url.startswith(("http://", "https://")):
-            url = "https://" + url
-        jina_url = f"https://r.jina.ai/{url}"
-        req = urllib.request.Request(
-            jina_url, headers={"User-Agent": _UA, "Accept": "text/plain"}
-        )
-        with urllib.request.urlopen(req, timeout=15) as resp:
-            return resp.read().decode("utf-8")
-
-    def search(self, query: str, limit: int = 5) -> str:
-        """通过 Exa 搜索 Discord 内容（服务器、帖子、讨论）。"""
-        mcporter = shutil.which("mcporter")
-        if not mcporter:
-            return "搜索需要 mcporter，请运行 `npm install -g mcporter` 安装。"
-        cmd = (
-            f"mcporter call 'exa.web_search_exa("
-            f"query: \"{query} site:discord.com OR site:discord.gg\", "
-            f"numResults: {limit}, "
-            f"includeDomains: [\"discord.com\", \"discord.gg\"])'"
-        )
-        r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
-        return r.stdout or r.stderr

agent_reach/channels/toutiao.py

@@ -1,128 +0,0 @@
-# -*- coding: utf-8 -*-
-"""Toutiao (今日头条) — search articles and trending content."""
-
-import json
-import re
-import urllib.parse
-import urllib.request
-from typing import List, Tuple
-
-from .base import Channel
-
-_UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
-_TIMEOUT = 15
-_SEARCH_URL = (
-    "https://so.toutiao.com/search"
-    "?dvpf=pc&source=input&keyword={keyword}&enable_druid_v2=1"
-)
-
-_SKIP_TEMPLATES = ("Search", "Bottom", "76-", "20-", "26-", "67-baike")
-
-
-def _fetch_html(url: str) -> str:
-    """Fetch URL and return HTML string."""
-    req = urllib.request.Request(url, headers={"User-Agent": _UA})
-    with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
-        return resp.read().decode("utf-8")
-
-
-def _parse_search_results(html: str) -> list:
-    """Extract article results from Toutiao search page HTML.
-
-    The page embeds each search result as a JSON object inside a <script> tag.
-    """
-    scripts = re.findall(r"<script[^>]*>(.*?)</script>", html, re.DOTALL)
-    articles = []
-    for s in scripts:
-        if len(s) < 1000:
-            continue
-        if not s.strip().startswith("{"):
-            continue
-        try:
-            data = json.loads(s).get("data", {})
-        except (json.JSONDecodeError, ValueError):
-            continue
-        if not isinstance(data, dict):
-            continue
-
-        title = data.get("title", "")
-        tpl = data.get("template_key", "")
-        if not title:
-            continue
-        if any(tpl.startswith(p) for p in _SKIP_TEMPLATES):
-            continue
-
-        # article_url is the primary field; display paths are fallbacks
-        article_url = (
-            data.get("article_url", "")
-            or data.get("source_url", "")
-            or (data.get("display") or {}).get("info", {}).get("url", "")
-        )
-
-        if not article_url:
-            continue
-
-        articles.append({
-            "title": title,
-            "url": article_url,
-            "source": data.get("media_name", "") or data.get("source", ""),
-            "abstract": (data.get("abstract", "") or "")[:300],
-            "publish_time": data.get("publish_time"),
-            "read_count": data.get("read_count"),
-            "comment_count": data.get("comment_count"),
-        })
-    return articles
-
-
-class ToutiaoChannel(Channel):
-    name = "toutiao"
-    description = "今日头条搜索与资讯"
-    backends = ["Toutiao Web (public)"]
-    tier = 0
-
-    def can_handle(self, url: str) -> bool:
-        from urllib.parse import urlparse
-        d = urlparse(url).netloc.lower()
-        return "toutiao.com" in d
-
-    def check(self, config=None) -> Tuple[str, str]:
-        try:
-            test_url = _SEARCH_URL.format(keyword="test")
-            req = urllib.request.Request(test_url, headers={"User-Agent": _UA})
-            with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
-                if resp.status == 200:
-                    content = resp.read().decode("utf-8")
-                    if "data" in content or "title" in content:
-                        return "ok", "头条搜索可用（搜索文章、视频、资讯）"
-                    return "warn", "头条搜索返回非预期内容"
-            return "warn", "头条搜索返回非 200 状态"
-        except Exception as e:
-            return "warn", f"头条搜索连接失败（可能需要代理）：{e}"
-
-    def read(self, url: str) -> str:
-        """通过 Jina Reader 读取头条文章正文。"""
-        if not url.startswith(("http://", "https://")):
-            url = "https://" + url
-        jina_url = f"https://r.jina.ai/{url}"
-        req = urllib.request.Request(
-            jina_url,
-            headers={"User-Agent": _UA, "Accept": "text/plain"},
-        )
-        with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
-            return resp.read().decode("utf-8")
-
-    def search(self, keyword: str, limit: int = 10) -> list:
-        """搜索头条文章。
-
-        Args:
-            keyword: 搜索关键词
-            limit:   最多返回条数
-
-        Returns:
-            list of dicts with keys:
-              title, url, source, abstract, publish_time, read_count, comment_count
-        """
-        encoded = urllib.parse.quote(keyword)
-        url = _SEARCH_URL.format(keyword=encoded)
-        html = _fetch_html(url)
-        return _parse_search_results(html)[:limit]

tests/test_discord_channel.py

@@ -1,87 +0,0 @@
-# -*- coding: utf-8 -*-
-"""Tests for DiscordChannel."""
-
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-from agent_reach.channels.discord import DiscordChannel, _get_invite_code
-
-
-@pytest.fixture
-def ch():
-    return DiscordChannel()
-
-
-class TestDiscordChannelAttributes:
-    def test_name(self, ch):
-        assert ch.name == "discord"
-
-    def test_tier(self, ch):
-        assert ch.tier == 0
-
-    def test_backends(self, ch):
-        assert ch.backends
-
-
-class TestDiscordCanHandle:
-    def test_discord_gg(self, ch):
-        assert ch.can_handle("https://discord.gg/python")
-
-    def test_discord_com(self, ch):
-        assert ch.can_handle("https://discord.com/invite/rust")
-
-    def test_discord_channel_url(self, ch):
-        assert ch.can_handle("https://discord.com/channels/123/456")
-
-    def test_rejects_other(self, ch):
-        assert not ch.can_handle("https://www.slack.com")
-
-
-class TestGetInviteCode:
-    def test_discord_gg(self):
-        assert _get_invite_code("https://discord.gg/python") == "python"
-
-    def test_discord_com_invite(self):
-        assert _get_invite_code("https://discord.com/invite/rust-lang") == "rust-lang"
-
-    def test_trailing_slash(self):
-        assert _get_invite_code("https://discord.gg/python/") == "python"
-
-    def test_no_invite_code(self):
-        assert _get_invite_code("https://discord.com/channels/123/456") == ""
-
-
-class TestDiscordCheck:
-    def test_check_ok_with_exa(self, ch):
-        with patch("agent_reach.channels.discord._exa_available", return_value=True):
-            status, msg = ch.check()
-        assert status == "ok"
-        assert "Exa" in msg
-
-    def test_check_warn_without_exa(self, ch):
-        with patch("agent_reach.channels.discord._exa_available", return_value=False):
-            status, msg = ch.check()
-        assert status == "warn"
-        assert "Invite API" in msg
-
-
-class TestDiscordRead:
-    def test_read_invite_url(self, ch):
-        mock_data = {
-            "guild": {"name": "Python", "description": "The Python community"},
-            "channel": {"name": "general"},
-            "approximate_member_count": 400000,
-            "approximate_presence_count": 30000,
-        }
-        mock_resp = MagicMock()
-        mock_resp.__enter__ = lambda s: s
-        mock_resp.__exit__ = MagicMock(return_value=False)
-        mock_resp.read.return_value = __import__("json").dumps(mock_data).encode()
-
-        with patch("urllib.request.urlopen", return_value=mock_resp):
-            result = ch.read("https://discord.gg/python")
-
-        assert "Python" in result
-        assert "400,000" in result
-        assert "general" in result

tests/test_toutiao_channel.py

@@ -1,117 +0,0 @@
-# -*- coding: utf-8 -*-
-"""Tests for ToutiaoChannel."""
-
-import json
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-from agent_reach.channels.toutiao import ToutiaoChannel, _parse_search_results
-
-
-@pytest.fixture
-def ch():
-    return ToutiaoChannel()
-
-
-class TestToutiaoChannelAttributes:
-    def test_name(self, ch):
-        assert ch.name == "toutiao"
-
-    def test_tier(self, ch):
-        assert ch.tier == 0
-
-    def test_backends(self, ch):
-        assert ch.backends
-
-
-class TestToutiaoCanHandle:
-    def test_matches_toutiao(self, ch):
-        assert ch.can_handle("https://www.toutiao.com/article/123")
-
-    def test_matches_search(self, ch):
-        assert ch.can_handle("https://so.toutiao.com/search?keyword=ai")
-
-    def test_rejects_other(self, ch):
-        assert not ch.can_handle("https://www.baidu.com")
-
-
-class TestToutiaoCheck:
-    def test_check_ok(self, ch):
-        mock_resp = MagicMock()
-        mock_resp.__enter__ = lambda s: s
-        mock_resp.__exit__ = MagicMock(return_value=False)
-        mock_resp.status = 200
-        mock_resp.read.return_value = b'{"data": {"title": "test"}}'
-        with patch("urllib.request.urlopen", return_value=mock_resp):
-            status, msg = ch.check()
-        assert status == "ok"
-
-    def test_check_exception(self, ch):
-        with patch("urllib.request.urlopen", side_effect=Exception("timeout")):
-            status, msg = ch.check()
-        assert status == "warn"
-        assert "timeout" in msg
-
-
-class TestParseSearchResults:
-    def _make_script(self, data: dict) -> str:
-        # Production code skips scripts < 1000 chars; pad to exceed threshold
-        payload = json.dumps({"data": data})
-        padding = "x" * max(0, 1001 - len(payload))
-        padded = payload[:-1] + f',"_pad":"{padding}"' + payload[-1]
-        return f'<script type="text/javascript">{padded}</script>'
-
-    def test_extracts_article_url_first(self):
-        html = self._make_script({
-            "title": "测试文章",
-            "article_url": "https://www.toutiao.com/article/1",
-            "template_key": "undefined-default",
-        })
-        results = _parse_search_results(html)
-        assert len(results) == 1
-        assert results[0]["url"] == "https://www.toutiao.com/article/1"
-        assert results[0]["title"] == "测试文章"
-
-    def test_fallback_to_source_url(self):
-        html = self._make_script({
-            "title": "备用 URL 文章",
-            "source_url": "https://www.toutiao.com/source/1",
-            "template_key": "undefined-default",
-        })
-        results = _parse_search_results(html)
-        assert len(results) == 1
-        assert results[0]["url"] == "https://www.toutiao.com/source/1"
-
-    def test_skips_entries_without_url(self):
-        html = self._make_script({
-            "title": "无 URL 文章",
-            "template_key": "undefined-default",
-        })
-        results = _parse_search_results(html)
-        assert results == []
-
-    def test_skips_entries_without_title(self):
-        html = self._make_script({
-            "article_url": "https://www.toutiao.com/article/1",
-            "template_key": "undefined-default",
-        })
-        results = _parse_search_results(html)
-        assert results == []
-
-    def test_skips_skip_templates(self):
-        html = self._make_script({
-            "title": "广告",
-            "article_url": "https://www.toutiao.com/article/1",
-            "template_key": "Bottom-ad",
-        })
-        results = _parse_search_results(html)
-        assert results == []
-
-    def test_empty_html(self):
-        assert _parse_search_results("") == []
-
-    def test_short_scripts_ignored(self):
-        # script < 1000 chars should be skipped
-        html = '<script>{"data": {"title": "x", "article_url": "y"}}</script>'
-        assert _parse_search_results(html) == []