diff --git a/agent_reach/channels/twitter.py b/agent_reach/channels/twitter.py index 56f4d54..6963cae 100644 --- a/agent_reach/channels/twitter.py +++ b/agent_reach/channels/twitter.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- """Twitter/X — check if xreach CLI is available.""" +import json import shutil import subprocess from .base import Channel @@ -19,6 +20,53 @@ def _parse_version(ver_str: str) -> tuple[int, ...]: return (0, 0, 0) +def _detect_xreach_version(xreach_path: str) -> str: + """Best-effort xreach version detection. + + Some xreach-cli releases ship package.json@0.3.2 while `xreach --version` + still prints 0.3.0 because the embedded dist version file was not updated. + Prefer the newer of: + 1) `xreach --version` + 2) `npm list -g xreach-cli --json --depth=0` + """ + versions: list[str] = [] + + try: + ver_result = subprocess.run( + [xreach_path, "--version"], capture_output=True, + encoding="utf-8", errors="replace", timeout=5 + ) + version_str = (ver_result.stdout or ver_result.stderr).strip() + if version_str: + versions.append(version_str) + except Exception: + pass + + npm = shutil.which("npm") + if npm: + try: + npm_result = subprocess.run( + [npm, "list", "-g", "xreach-cli", "--json", "--depth=0"], + capture_output=True, encoding="utf-8", errors="replace", timeout=10, + ) + if npm_result.returncode == 0 and npm_result.stdout: + data = json.loads(npm_result.stdout) + npm_ver = ( + data.get("dependencies", {}) + .get("xreach-cli", {}) + .get("version", "") + .strip() + ) + if npm_ver: + versions.append(npm_ver) + except Exception: + pass + + if not versions: + return "" + return max(versions, key=_parse_version) + + class TwitterChannel(Channel): name = "twitter" description = "Twitter/X 推文" @@ -39,13 +87,9 @@ class TwitterChannel(Channel): ) # Check version — longform tweet support requires >= 0.3.2 try: - ver_result = subprocess.run( - [xreach, "--version"], capture_output=True, - encoding="utf-8", errors="replace", timeout=5 - ) - version_str = (ver_result.stdout or ver_result.stderr).strip() + version_str = _detect_xreach_version(xreach) version_tuple = _parse_version(version_str) - if version_tuple < _MIN_XREACH_VERSION: + if version_str and version_tuple < _MIN_XREACH_VERSION: min_str = ".".join(str(x) for x in _MIN_XREACH_VERSION) return "warn", ( f"xreach CLI 版本过旧(当前 {version_str},需 >= {min_str})。" diff --git a/agent_reach/cli.py b/agent_reach/cli.py index d4370e2..9b1b1ab 100644 --- a/agent_reach/cli.py +++ b/agent_reach/cli.py @@ -460,7 +460,9 @@ def _install_system_deps(): def _install_xiaoyuzhou_deps(): """Install Xiaoyuzhou podcast transcription script.""" import shutil + from agent_reach.config import Config + config = Config() print("Setting up Xiaoyuzhou podcast transcription...") tools_dir = os.path.expanduser("~/.agent-reach/tools/xiaoyuzhou") diff --git a/tests/test_twitter_channel.py b/tests/test_twitter_channel.py new file mode 100644 index 0000000..5ccc000 --- /dev/null +++ b/tests/test_twitter_channel.py @@ -0,0 +1,39 @@ +# -*- coding: utf-8 -*- + +from unittest.mock import patch, Mock + +from agent_reach.channels.twitter import _detect_xreach_version, TwitterChannel + + +def _cp(stdout="", stderr="", returncode=0): + m = Mock() + m.stdout = stdout + m.stderr = stderr + m.returncode = returncode + return m + + +def test_detect_xreach_version_prefers_npm_when_cli_version_is_stale(): + with patch("shutil.which", return_value="/opt/homebrew/bin/npm"), patch( + "subprocess.run", + side_effect=[ + _cp(stdout="0.3.0\n"), + _cp(stdout='{"dependencies":{"xreach-cli":{"version":"0.3.2"}}}'), + ], + ): + assert _detect_xreach_version("/opt/homebrew/bin/xreach") == "0.3.2" + + +def test_twitter_channel_does_not_false_warn_when_npm_has_newer_xreach(): + channel = TwitterChannel() + with patch("shutil.which", side_effect=lambda name: "/opt/homebrew/bin/xreach" if name == "xreach" else "/opt/homebrew/bin/npm"), patch( + "subprocess.run", + side_effect=[ + _cp(stdout="0.3.0\n"), + _cp(stdout='{"dependencies":{"xreach-cli":{"version":"0.3.2"}}}'), + _cp(stdout="authenticated\n", returncode=0), + ], + ): + status, message = channel.check() + assert status == "ok" + assert "完整可用" in message diff --git a/tests/test_xiaoyuzhou_install.py b/tests/test_xiaoyuzhou_install.py new file mode 100644 index 0000000..f79c594 --- /dev/null +++ b/tests/test_xiaoyuzhou_install.py @@ -0,0 +1,21 @@ +# -*- coding: utf-8 -*- + +from unittest.mock import patch + +import agent_reach.cli as cli + + +class _DummyConfig: + def get(self, _key): + return None + + +def test_install_xiaoyuzhou_deps_does_not_raise_when_no_groq_key(capsys): + with patch("agent_reach.config.Config", return_value=_DummyConfig()), \ + patch("os.path.isfile", side_effect=lambda p: True if str(p).endswith("transcribe.sh") else False), \ + patch("shutil.which", return_value=None): + cli._install_xiaoyuzhou_deps() + + out = capsys.readouterr().out + assert "Xiaoyuzhou" in out + assert "Groq API key not set" in out