Hoan/xiaozhi-esp32-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3520 from xinnan-tech/py-test-weather

Browse Source 
fix: harden weather and news plugins (#3465)
This commit is contained in:
wengzh authored and GitHub committed 2026-10-29 10:10:50 +08:00
2 parents 8c5330f742 15d9b0fba1
commit 4dbed0ab1d
3 files changed
+504 -45
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

main/xiaozhi-server/plugins_func/functions/get_news_from_newsnow.py
+51 -17
View File
@@ -1,6 +1,7 @@
import random
import httpx
from io import BytesIO
from urllib.parse import quote, urlsplit, urlunsplit, parse_qsl
from markitdown import MarkItDown, StreamInfo
from config.logger import setup_logging
from plugins_func.register import register_function, ToolType, ActionResponse, Action
@@ -54,6 +55,27 @@ CHANNEL_MAP = {
# 默认新闻来源字典，当配置中没有指定时使用
DEFAULT_NEWS_SOURCES = "澎湃新闻;百度热搜;财联社"
DEFAULT_NEWS_API_URL = "https://newsnow.busiyi.world/api/s?id="
def _build_news_api_url(source, configured_url=None):
"""拼出新闻接口的请求地址和查询参数。
配置里带 {source} 占位符就替换占位符；否则把 id 设成新闻源 id，
这样配置里已经写了 id 时也不会拼出脏串。
"""
api_url = str(configured_url).strip() if configured_url else DEFAULT_NEWS_API_URL
if "{source}" in api_url:
return api_url.replace("{source}", quote(source, safe="")), None
parts = urlsplit(api_url)
query = dict(parse_qsl(parts.query, keep_blank_values=True))
query["id"] = source
return (
urlunsplit((parts.scheme, parts.netloc, parts.path, "", parts.fragment)),
query,
)
def _get_newsnow_config(conn):
# 从连接配置获取
plugins = conn.config.get("plugins", {})
@@ -61,6 +83,8 @@ def _get_newsnow_config(conn):
sources = newsnow.get("news_sources", "")
if isinstance(sources, str) and sources.strip():
return sources
if isinstance(sources, (list, tuple)):
return ";".join(str(item).strip() for item in sources if str(item).strip())
return ""
@@ -113,23 +137,26 @@ GET_NEWS_FROM_NEWSNOW_FUNCTION_DESC = {
async def fetch_news_from_api(conn: "ConnectionHandler", source="thepaper"):
"""从API获取新闻列表"""
try:
api_url = f"https://newsnow.busiyi.world/api/s?id={source}"
news_config = conn.config.get("plugins", {}).get("get_news_from_newsnow", {})
if news_config.get("url"):
api_url = news_config["url"] + source
configured_url = news_config.get("url") if isinstance(news_config, dict) else None
api_url, request_params = _build_news_api_url(source, configured_url)
headers = {"User-Agent": "Mozilla/5.0"}
async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
response = await client.get(api_url, headers=headers)
response = await client.get(api_url, params=request_params, headers=headers)
if response.status_code != 200:
logger.bind(tag=TAG).error(
f"获取新闻API失败：HTTP {response.status_code}，source={source}"
)
return []
data = response.json()
if "items" in data:
return data["items"]
else:
items = data.get("items") if isinstance(data, dict) else None
if not isinstance(items, list):
logger.bind(tag=TAG).error(f"获取新闻API响应格式错误: {data}")
return []
return [item for item in items if isinstance(item, dict) and item.get("title")]
except Exception as e:
logger.bind(tag=TAG).error(f"获取新闻API失败: {e}")
@@ -239,18 +266,24 @@ async def get_news_from_newsnow(
english_source_id = None
# 检查输入的中文名称是否在配置的新闻源中
news_sources_list = [
name.strip() for name in news_sources.split(";") if name.strip()
]
news_sources_list = [name.strip() for name in news_sources.split(";") if name.strip()]
if source in news_sources_list:
# 如果输入的中文名称在配置的新闻源中，在 CHANNEL_MAP 中查找对应的英文ID
english_source_id = CHANNEL_MAP.get(source)
elif not source and news_sources_list:
source = news_sources_list[0]
english_source_id = CHANNEL_MAP.get(source)
# 如果找不到对应的英文ID，使用默认源
if not english_source_id:
logger.bind(tag=TAG).warning(f"无效的新闻源: {source}，使用默认源澎湃新闻")
english_source_id = "thepaper"
source = "澎湃新闻"
fallback_source = next(
(name for name in news_sources_list if CHANNEL_MAP.get(name)),
"澎湃新闻",
)
logger.bind(tag=TAG).warning(
f"无效的新闻源: {source}，使用配置中的默认源{fallback_source}"
)
english_source_id = CHANNEL_MAP[fallback_source]
source = fallback_source
logger.bind(tag=TAG).info(f"获取新闻: 新闻源={source}({english_source_id})")
@@ -266,20 +299,21 @@ async def get_news_from_newsnow(
# 随机选择一条新闻
selected_news = random.choice(news_items)
title = selected_news.get("title", "未知标题")
# 保存当前新闻链接到连接对象，以便后续查询详情
if not hasattr(conn, "last_newsnow_link"):
conn.last_newsnow_link = {}
conn.last_newsnow_link = {
"url": selected_news.get("url", "#"),
"title": selected_news.get("title", "未知标题"),
"title": title,
"source_id": english_source_id,
}
# 构建新闻报告
news_report = (
f"根据下列数据，用{lang}回应用户的新闻查询请求：\n\n"
f"新闻标题: {selected_news['title']}\n"
f"新闻标题: {title}\n"
# f"新闻来源: {source}\n"
f"(请以自然、流畅的方式向用户播报这条新闻标题，"
f"提示用户可以要求获取详细内容，此时会获取新闻的详细内容。)"
main/xiaozhi-server/plugins_func/functions/get_weather.py
+117 -28
View File
@@ -1,3 +1,4 @@
import asyncio
import httpx
from bs4 import BeautifulSoup
from config.logger import setup_logging
@@ -112,20 +113,106 @@ WEATHER_CODE_MAP = {
async def fetch_city_info(location, api_key, api_host):
url = f"https://{api_host}/geo/v2/city/lookup?key={api_key}&location={location}&lang=zh"
url = f"https://{api_host}/geo/v2/city/lookup"
async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0)) as client:
response = await client.get(url, headers=HEADERS)
data = response.json()
if data.get("error") is not None:
response = await client.get(
url,
params={
"key": api_key,
"location": location,
"lang": "zh",
"number": 1,
},
headers=HEADERS,
)
try:
data = response.json()
except ValueError:
logger.bind(tag=TAG).error(
f"获取天气失败，原因：{data.get('error', {}).get('detail')}"
f"获取天气城市失败：HTTP {response.status_code} 返回了非 JSON 内容"
)
return None
if not isinstance(data, dict):
logger.bind(tag=TAG).error("获取天气城市失败：响应不是 JSON 对象")
return None
if response.status_code != 200 or data.get("code") != "200":
logger.bind(tag=TAG).error(
f"获取天气城市失败：HTTP {response.status_code}，代码 {data.get('code')}"
)
return None
return data.get("location", [])[0] if data.get("location") else None
async def fetch_weather_data(city_info, api_key, api_host):
location_id = city_info.get("id")
if not location_id:
return None
base_url = f"https://{api_host}/v7/weather"
params = {"key": api_key, "location": location_id, "lang": "zh"}
async def request(client, endpoint):
response = await client.get(f"{base_url}/{endpoint}", params=params, headers=HEADERS)
try:
data = response.json()
except ValueError:
return None
if response.status_code != 200 or data.get("code") != "200":
logger.bind(tag=TAG).error(
f"获取天气数据失败：接口={endpoint} HTTP {response.status_code}，代码 {data.get('code')}"
)
return None
return data
try:
async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
now, daily = await asyncio.gather(
request(client, "now"), request(client, "7d")
)
except Exception as e:
logger.bind(tag=TAG).warning(f"获取天气 JSON 接口失败，将尝试网页接口: {e}")
return None
if now is None or daily is None:
return None
return now, daily
def build_weather_report(city_info, now_data, daily_data):
city_name = city_info.get("name") or city_info.get("adm2") or "未知位置"
now = now_data.get("now", {})
weather_report = f"您查询的位置是：{city_name}\n\n"
weather_report += f"当前天气: {now.get('text', '未知')}\n"
current_basic = {
"温度": f"{now['temp']}℃" if now.get("temp") is not None else None,
"体感温度": f"{now['feelsLike']}℃" if now.get("feelsLike") is not None else None,
"湿度": f"{now['humidity']}%" if now.get("humidity") is not None else None,
"风向": now.get("windDir"),
"风力": now.get("windScale"),
"能见度": f"{now['vis']}公里" if now.get("vis") is not None else None,
"气压": f"{now['pressure']}百帕" if now.get("pressure") is not None else None,
"降水量": f"{now['precip']}毫米" if now.get("precip") is not None else None,
}
current_basic = {key: value for key, value in current_basic.items() if value}
if current_basic:
weather_report += "详细参数：\n"
weather_report += "".join(f" · {key}: {value}\n" for key, value in current_basic.items())
weather_report += "\n未来7天预报：\n"
for item in daily_data.get("daily", [])[:7]:
date = item.get("fxDate", "未知日期")
weather = item.get("textDay") or WEATHER_CODE_MAP.get(item.get("iconDay"), "未知")
high = item.get("tempMax", "未知")
low = item.get("tempMin", "未知")
weather_report += f"{date}: {weather}，气温 {low}~{high}℃\n"
weather_report += "\n（如需某一天的具体天气，请告诉我日期）"
return weather_report
async def fetch_weather_page(url):
async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
async with httpx.AsyncClient(
timeout=httpx.Timeout(10.0, connect=3.0), follow_redirects=True
) as client:
response = await client.get(url, headers=HEADERS)
return BeautifulSoup(response.text, "html.parser") if response.status_code == 200 else None
@@ -201,29 +288,31 @@ async def get_weather(conn: "ConnectionHandler", location: str = None, lang: str
city_info = await fetch_city_info(location, api_key, api_host)
if not city_info:
return ActionResponse(
Action.REQLLM, f"未找到相关的城市: {location}，请确认地点是否正确", None
Action.REQLLM,
f"暂时查不到 {location} 的天气（城市未找到或天气服务不可用），请稍后再试",
None,
)
soup = await fetch_weather_page(city_info["fxLink"])
if not soup:
return ActionResponse(Action.REQLLM, None, "请求失败")
city_name, current_abstract, current_basic, temps_list = parse_weather_info(soup)
weather_report = f"您查询的位置是：{city_name}\n\n当前天气: {current_abstract}\n"
# 添加有效的当前天气参数
if current_basic:
weather_report += "详细参数：\n"
for key, value in current_basic.items():
if value != "0": # 过滤无效值
weather_report += f" · {key}: {value}\n"
# 添加7天预报
weather_report += "\n未来7天预报：\n"
for date, weather, high, low in temps_list:
weather_report += f"{date}: {weather}，气温 {low}~{high}\n"
# 提示语
weather_report += "\n（如需某一天的具体天气，请告诉我日期）"
weather_data = await fetch_weather_data(city_info, api_key, api_host)
if weather_data:
weather_report = build_weather_report(city_info, *weather_data)
else:
fx_link = city_info.get("fxLink")
soup = await fetch_weather_page(fx_link) if fx_link else None
if not soup:
return ActionResponse(Action.REQLLM, None, "请求天气数据失败")
city_name, current_abstract, current_basic, temps_list = parse_weather_info(soup)
weather_report = f"您查询的位置是：{city_name}\n\n当前天气: {current_abstract}\n"
if current_basic:
weather_report += "详细参数：\n"
weather_report += "".join(
f" · {key}: {value}\n" for key, value in current_basic.items() if value != "0"
)
weather_report += "\n未来7天预报：\n"
weather_report += "".join(
f"{date}: {weather}，气温 {low}~{high}\n"
for date, weather, high, low in temps_list
)
weather_report += "\n（如需某一天的具体天气，请告诉我日期）"
# 缓存完整的天气报告
cache_manager.set(CacheType.WEATHER, weather_cache_key, weather_report)
main/xiaozhi-server/tests/plugins_func/test_weather_news_plugins.py
+336
View File
@@ -0,0 +1,336 @@
"""Tests for the weather and newsnow plugin helpers.
The two plugins talk to external APIs, so the HTTP layer is replaced with a
fake client and every request is recorded. Nothing here touches the network.
"""
import httpx
import pytest
from plugins_func.functions import get_news_from_newsnow as news_module
from plugins_func.functions import get_weather as weather_module
class FakeResponse:
def __init__(self, status_code=200, payload=None, text="", json_error=None):
self.status_code = status_code
self.payload = payload
self.text = text
self.json_error = json_error
def json(self):
if self.json_error is not None:
raise self.json_error
return self.payload
class FakeAsyncClient:
"""Stands in for httpx.AsyncClient and records what was requested."""
requests = []
responder = None
init_kwargs = {}
def __init__(self, *args, **kwargs):
FakeAsyncClient.init_kwargs = kwargs
async def __aenter__(self):
return self
async def __aexit__(self, *exc_info):
return False
async def get(self, url, params=None, headers=None):
FakeAsyncClient.requests.append({"url": url, "params": params})
return FakeAsyncClient.responder(url, params)
@pytest.fixture
def fake_http(monkeypatch):
"""Replace httpx.AsyncClient with the recording fake."""
FakeAsyncClient.requests = []
FakeAsyncClient.responder = lambda url, params: FakeResponse(payload={"code": "200"})
FakeAsyncClient.init_kwargs = {}
monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
return FakeAsyncClient
class FakeConn:
def __init__(self, plugins=None):
self.config = {"plugins": plugins or {}}
self.client_ip = None
# ---------------------------------------------------------------------------
# 城市查询
# ---------------------------------------------------------------------------
async def test_city_lookup_returns_none_on_error_code(fake_http):
fake_http.responder = lambda url, params: FakeResponse(payload={"code": "402"})
result = await weather_module.fetch_city_info("杭州", "key", "host")
assert result is None
async def test_city_lookup_returns_none_on_non_json_body(fake_http):
fake_http.responder = lambda url, params: FakeResponse(
text="<html>oops</html>", json_error=ValueError("not json")
)
result = await weather_module.fetch_city_info("杭州", "key", "host")
assert result is None
async def test_city_lookup_returns_none_on_non_dict_body(fake_http):
fake_http.responder = lambda url, params: FakeResponse(payload=["not", "an", "object"])
result = await weather_module.fetch_city_info("杭州", "key", "host")
assert result is None
async def test_city_lookup_returns_none_on_empty_location(fake_http):
fake_http.responder = lambda url, params: FakeResponse(
payload={"code": "200", "location": []}
)
result = await weather_module.fetch_city_info("杭州", "key", "host")
assert result is None
async def test_city_lookup_searches_all_cities(fake_http):
"""不带 range 参数就是全球检索，写死 cn 会查不到境外城市。"""
fake_http.responder = lambda url, params: FakeResponse(
payload={"code": "200", "location": [{"id": "101210101", "name": "杭州"}]}
)
result = await weather_module.fetch_city_info("杭州", "key", "host")
assert result == {"id": "101210101", "name": "杭州"}
sent = fake_http.requests[0]["params"]
assert "range" not in sent
assert sent["location"] == "杭州"
assert sent["number"] == 1
# ---------------------------------------------------------------------------
# 天气数据
# ---------------------------------------------------------------------------
def _weather_responder(now_payload, daily_payload):
def responder(url, params):
if url.endswith("/now"):
return now_payload if isinstance(now_payload, FakeResponse) else FakeResponse(
payload=now_payload
)
return daily_payload if isinstance(daily_payload, FakeResponse) else FakeResponse(
payload=daily_payload
)
return responder
async def test_weather_data_needs_both_now_and_daily(fake_http):
fake_http.responder = _weather_responder(
{"code": "200", "now": {"text": "晴"}},
FakeResponse(status_code=401, payload={"code": "401"}),
)
assert await weather_module.fetch_weather_data({"id": "1"}, "key", "host") is None
async def test_weather_data_needs_daily_when_now_failed(fake_http):
fake_http.responder = _weather_responder(
FakeResponse(status_code=401, payload={"code": "401"}),
{"code": "200", "daily": [{"fxDate": "2026-10-24"}]},
)
assert await weather_module.fetch_weather_data({"id": "1"}, "key", "host") is None
async def test_weather_data_returns_both_when_all_succeed(fake_http):
fake_http.responder = _weather_responder(
{"code": "200", "now": {"text": "晴"}},
{"code": "200", "daily": [{"fxDate": "2026-10-24"}]},
)
now, daily = await weather_module.fetch_weather_data({"id": "1"}, "key", "host")
assert now["now"]["text"] == "晴"
assert daily["daily"][0]["fxDate"] == "2026-10-24"
async def test_weather_page_follows_redirects(fake_http):
"""旧形态 fxLink 会 302，不跟随重定向就拿不到内容。"""
fake_http.responder = lambda url, params: FakeResponse(text="<html></html>")
await weather_module.fetch_weather_page("https://www.qweather.com/weather/101210101.html")
assert fake_http.init_kwargs.get("follow_redirects") is True
def test_weather_report_keeps_pressure_and_precipitation():
report = weather_module.build_weather_report(
{"name": "杭州"},
{"now": {"text": "多云", "temp": "24", "pressure": "1003", "precip": "0.0"}},
{"daily": []},
)
assert "气压: 1003百帕" in report
assert "降水量: 0.0毫米" in report
async def test_weather_report_keeps_pressure_and_precipitation_in_flow(monkeypatch):
"""走完整条链路时报告里也要有气压和降水量。"""
monkeypatch.setattr(
weather_module,
"fetch_city_info",
_async_value({"id": "1", "name": "杭州"}),
)
monkeypatch.setattr(
weather_module,
"fetch_weather_data",
_async_value(
({"now": {"text": "多云", "pressure": "1003", "precip": "0.0"}}, {"daily": []})
),
)
response = await weather_module.get_weather(
FakeConn({"get_weather": {"api_key": "k", "api_host": "h"}}), location="气压测试市"
)
assert "气压: 1003百帕" in response.result
assert "降水量: 0.0毫米" in response.result
async def test_missing_fxlink_does_not_call_web_fallback(monkeypatch):
"""城市信息里没有 fxLink 时不该去抓网页，直接报失败。"""
called = []
async def fake_city_info(*args, **kwargs):
return {"id": "1", "name": "杭州"}
async def fake_weather_data(*args, **kwargs):
return None
async def fake_page(url):
called.append(url)
return None
monkeypatch.setattr(weather_module, "fetch_city_info", fake_city_info)
monkeypatch.setattr(weather_module, "fetch_weather_data", fake_weather_data)
monkeypatch.setattr(weather_module, "fetch_weather_page", fake_page)
response = await weather_module.get_weather(
FakeConn({"get_weather": {"api_key": "k", "api_host": "h"}}), location="无链接市"
)
assert called == []
assert response.response == "请求天气数据失败"
async def test_city_not_found_message_does_not_blame_the_place_name(monkeypatch):
"""key 失效和城市找不到都落到这句话，不能让用户去核对地名。"""
async def fake_city_info(*args, **kwargs):
return None
monkeypatch.setattr(weather_module, "fetch_city_info", fake_city_info)
response = await weather_module.get_weather(
FakeConn({"get_weather": {"api_key": "k", "api_host": "h"}}), location="随便什么市"
)
assert "请确认地点是否正确" not in response.result
assert "天气服务不可用" in response.result
# ---------------------------------------------------------------------------
# 新闻接口
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"configured_url, expected_url, expected_params",
[
(None, "https://newsnow.busiyi.world/api/s", {"id": "thepaper"}),
(
"https://newsnow.busiyi.world/api/s?id=thepaper",
"https://newsnow.busiyi.world/api/s",
{"id": "thepaper"},
),
(
"https://newsnow.busiyi.world/api/s?id={source}",
"https://newsnow.busiyi.world/api/s?id=thepaper",
None,
),
(
"https://newsnow.busiyi.world/api/s",
"https://newsnow.busiyi.world/api/s",
{"id": "thepaper"},
),
(
"https://newsnow.busiyi.world/api/s?id=",
"https://newsnow.busiyi.world/api/s",
{"id": "thepaper"},
),
(
"https://newsnow.busiyi.world/api/s?id=thepaper&limit=5",
"https://newsnow.busiyi.world/api/s",
{"id": "thepaper", "limit": "5"},
),
],
)
def test_news_api_url_forms(configured_url, expected_url, expected_params):
url, params = news_module._build_news_api_url("thepaper", configured_url)
assert url == expected_url
assert params == expected_params
async def test_news_api_url_forms_are_actually_requested(fake_http):
"""配置里写了 id 也不能拼出脏串。"""
fake_http.responder = lambda url, params: FakeResponse(
payload={"items": [{"id": "1", "title": "标题"}]}
)
conn = FakeConn({"get_news_from_newsnow": {"url": "https://newsnow.busiyi.world/api/s?id=thepaper"}})
await news_module.fetch_news_from_api(conn, "thepaper")
assert fake_http.requests[0]["url"] == "https://newsnow.busiyi.world/api/s"
assert fake_http.requests[0]["params"] == {"id": "thepaper"}
async def test_news_items_must_be_a_list(fake_http):
fake_http.responder = lambda url, params: FakeResponse(payload={"items": "oops"})
assert await news_module.fetch_news_from_api(FakeConn(), "thepaper") == []
async def test_news_items_without_title_are_filtered_out(fake_http):
fake_http.responder = lambda url, params: FakeResponse(
payload={"items": [{"id": "1"}, {"id": "2", "title": ""}]}
)
assert await news_module.fetch_news_from_api(FakeConn(), "thepaper") == []
async def test_news_items_keep_valid_entries(fake_http):
fake_http.responder = lambda url, params: FakeResponse(
payload={"items": [{"id": "1", "title": ""}, {"id": "2", "title": "有效标题"}]}
)
items = await news_module.fetch_news_from_api(FakeConn(), "thepaper")
assert items == [{"id": "2", "title": "有效标题"}]
def _async_value(value):
async def _inner(*args, **kwargs):
return value
return _inner