diff --git a/agent_reach/channels/__init__.py b/agent_reach/channels/__init__.py index f2a0149..4893590 100644 --- a/agent_reach/channels/__init__.py +++ b/agent_reach/channels/__init__.py @@ -23,7 +23,7 @@ from .weibo import WeiboChannel from .xiaoyuzhou import XiaoyuzhouChannel from .v2ex import V2EXChannel from .xueqiu import XueqiuChannel - +from .toutiao import ToutiaoChannel ALL_CHANNELS: List[Channel] = [ @@ -40,6 +40,7 @@ ALL_CHANNELS: List[Channel] = [ XiaoyuzhouChannel(), V2EXChannel(), XueqiuChannel(), + ToutiaoChannel(), RSSChannel(), ExaSearchChannel(), WebChannel(), diff --git a/agent_reach/channels/toutiao.py b/agent_reach/channels/toutiao.py new file mode 100644 index 0000000..f91d0df --- /dev/null +++ b/agent_reach/channels/toutiao.py @@ -0,0 +1,128 @@ +# -*- 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_toutiao_channel.py b/tests/test_toutiao_channel.py new file mode 100644 index 0000000..07c2ebf --- /dev/null +++ b/tests/test_toutiao_channel.py @@ -0,0 +1,117 @@ +# -*- 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) == []