fix: avoid xreach false alarm and xiaoyuzhou install crash

This commit is contained in:
robert.chen 2026-03-12 19:26:58 +08:00 committed by GitHub
parent 31f00b8d78
commit 3416c0f6d3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 112 additions and 6 deletions

View file

@ -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})。"

View file

@ -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")

View file

@ -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

View file

@ -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