diff --git a/.claude/settings.local.json b/.claude/settings.local.json
new file mode 100644
index 0000000..cb53030
--- /dev/null
+++ b/.claude/settings.local.json
@@ -0,0 +1,7 @@
+{
+ "permissions": {
+ "allow": [
+ "WebFetch(domain:community.groq.com)"
+ ]
+ }
+}
diff --git a/agent_reach/channels/__init__.py b/agent_reach/channels/__init__.py
index 5f8d5d8..f367f2a 100644
--- a/agent_reach/channels/__init__.py
+++ b/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(),
diff --git a/agent_reach/channels/discord.py b/agent_reach/channels/discord.py
deleted file mode 100644
index 35cb83a..0000000
--- a/agent_reach/channels/discord.py
+++ /dev/null
@@ -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
diff --git a/agent_reach/channels/toutiao.py b/agent_reach/channels/toutiao.py
deleted file mode 100644
index f91d0df..0000000
--- a/agent_reach/channels/toutiao.py
+++ /dev/null
@@ -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 ", 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]
diff --git a/tests/test_discord_channel.py b/tests/test_discord_channel.py
deleted file mode 100644
index 18fa939..0000000
--- a/tests/test_discord_channel.py
+++ /dev/null
@@ -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
diff --git a/tests/test_toutiao_channel.py b/tests/test_toutiao_channel.py
deleted file mode 100644
index 07c2ebf..0000000
--- a/tests/test_toutiao_channel.py
+++ /dev/null
@@ -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''
-
- 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 = ''
- assert _parse_search_results(html) == []