opensource/bisheng
3
0
Fork 0
mirror of https://github.com/dataelement/bisheng.git synced 2026-08-30 17:58:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(channel): log information subscription failures

Browse Source
This commit is contained in:
GuoQing Zhang
2026-08-06 16:06:10 +08:00
parent 008994987f
commit 2a820648b9
2 changed files with 98 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/bisheng/core/external/bisheng_information_client/client.py
Vendored
+70 -49
View File
@@ -3,42 +3,54 @@ from typing import Callable, Optional
import httpx
from aiohttp import ClientTimeout
from loguru import logger
from bisheng.common.errcode.channel import (
BishengInformationServiceError,
BishengInformationUnAuthorizedError,
InformationSourceAuthError,
InformationSourceCrawlLimitError,
InformationSourcePageError,
InformationSourceParseError,
InformationSourceSubscriptionLimitError,
InformationSourceWechatSearchLimitError,
)
from bisheng.core.config.settings import IntelligenceCenterConf
from bisheng.common.errcode.channel import BishengInformationUnAuthorizedError, BishengInformationServiceError, \
InformationSourceParseError, InformationSourceAuthError, InformationSourcePageError, \
InformationSourceCrawlLimitError, InformationSourceSubscriptionLimitError, \
InformationSourceWechatSearchLimitError
from bisheng.core.external.bisheng_information_client.response_schema import InformationSourceResponse, \
CrawlWebsiteResponse, InformationArticlesResponse
from bisheng.core.external.bisheng_information_client.response_schema import (
CrawlWebsiteResponse,
InformationArticlesResponse,
InformationSourceResponse,
)
from bisheng.core.external.http_client.client import AsyncHttpClient
class BusinessType(str, Enum):
"""Business types for information sources."""
WECHAT = 'wechat'
WEBSITE = 'website'
WECHAT = "wechat"
WEBSITE = "website"
class InformationSourceAddError(Exception):
"""Exception raised when adding an information source fails."""
pass
class InformationSourceListError(Exception):
"""Exception raised when listing information sources fails."""
pass
class InformationSourceSubscribeError(Exception):
"""Exception raised when subscribing to an information source fails."""
pass
class BishengInformationClient(object):
def __init__(self, http_client: Optional[AsyncHttpClient],
get_conf: Callable[[], IntelligenceCenterConf]):
def __init__(self, http_client: Optional[AsyncHttpClient], get_conf: Callable[[], IntelligenceCenterConf]):
self.http_client = http_client
self._get_conf = get_conf
@@ -76,12 +88,12 @@ class BishengInformationClient(object):
raise InformationSourcePageError()
def _handle_common_response_code(
self,
response_body: dict,
default_error_message: str,
*,
include_parse_errors: bool = False,
unknown_error_handler: Optional[Callable[[dict], None]] = None,
self,
response_body: dict,
default_error_message: str,
*,
include_parse_errors: bool = False,
unknown_error_handler: Optional[Callable[[dict], None]] = None,
) -> None:
code = response_body.get("code", -1)
if code == 200:
@@ -104,12 +116,12 @@ class BishengInformationClient(object):
return response.json()
def _handle_response(
self,
response,
default_error_message: str,
*,
include_parse_errors: bool = False,
unknown_error_handler: Optional[Callable[[dict], None]] = None,
self,
response,
default_error_message: str,
*,
include_parse_errors: bool = False,
unknown_error_handler: Optional[Callable[[dict], None]] = None,
) -> dict:
if response.status_code != 200:
raise BishengInformationServiceError(
@@ -166,17 +178,14 @@ class BishengInformationClient(object):
return information_source_data
async def search_information_sources(self, query: str, business_type: Optional[BusinessType] = None, page: int = 1,
page_size: int = 20) -> tuple[list[InformationSourceResponse], int]:
async def search_information_sources(
self, query: str, business_type: Optional[BusinessType] = None, page: int = 1, page_size: int = 20
) -> tuple[list[InformationSourceResponse], int]:
"""Search information sources by keyword."""
base_url, headers, timeout = self._build_request_options()
endpoint = f"{base_url}/information/search"
params = {
"keyword": query,
"page": page,
"page_size": page_size
}
params = {"keyword": query, "page": page, "page_size": page_size}
if business_type:
params["business_type"] = business_type.value
@@ -200,17 +209,14 @@ class BishengInformationClient(object):
information_sources_data = response_body.get("data", [])
return [InformationSourceResponse.model_validate(item) for item in information_sources_data]
async def list_information_sources(self, business_type: BusinessType, page: int = 1, page_size: int = 20) -> tuple[
list[InformationSourceResponse], int]:
async def list_information_sources(
self, business_type: BusinessType, page: int = 1, page_size: int = 20
) -> tuple[list[InformationSourceResponse], int]:
"""List all information sources."""
base_url, headers, timeout = self._build_request_options()
endpoint = f"{base_url}/information/list"
params = {
"business_type": business_type.value,
"page": page,
"page_size": page_size
}
params = {"business_type": business_type.value, "page": page, "page_size": page_size}
response = await self.http_client.get(endpoint, headers=headers, params=params, timeout=timeout)
response_body = self._handle_response(response, "Failed to list information sources")
@@ -226,12 +232,20 @@ class BishengInformationClient(object):
base_url, headers, timeout = self._build_request_options()
endpoint = f"{base_url}/information/subscribe"
data = {"information_ids": source_ids}
response = await self.http_client.post(endpoint, body=data, headers=headers, timeout=timeout)
self._handle_response(
response,
"Failed to subscribe to information source",
unknown_error_handler=lambda _: self._raise_subscribe_error(response, "subscribe to"),
)
try:
response = await self.http_client.post(endpoint, body=data, headers=headers, timeout=timeout)
self._handle_response(
response,
"Failed to subscribe to information source",
unknown_error_handler=lambda _: self._raise_subscribe_error(response, "subscribe to"),
)
except Exception:
logger.exception(
"BISHENG_INFORMATION_SUBSCRIPTION_REQUEST_FAILED endpoint={} source_count={}",
endpoint,
len(source_ids),
)
raise
async def unsubscribe_information_source(self, source_ids: list[str]) -> None:
"""Unsubscribe from an information source by source_id."""
@@ -261,9 +275,14 @@ class BishengInformationClient(object):
return CrawlWebsiteResponse.model_validate(result)
def get_information_articles(self, information_id: str, return_information: bool = False,
min_create_time: int = None, page: int = 1, page_size: int = 20) \
-> InformationArticlesResponse:
def get_information_articles(
self,
information_id: str,
return_information: bool = False,
min_create_time: int = None,
page: int = 1,
page_size: int = 20,
) -> InformationArticlesResponse:
"""Get articles of an information source by source_id."""
base_url, headers, timeout = self._build_request_options()
endpoint = f"{base_url}/information/articles/{information_id}"
@@ -279,6 +298,8 @@ class BishengInformationClient(object):
with httpx.Client() as client:
response = client.get(endpoint, headers=headers, params=params, timeout=timeout_value)
response_body = self._handle_response(response, "Failed to get information articles")
return InformationArticlesResponse(information=response_body.get("data", {}).get("information"),
articles=response_body.get("data", {}).get("articles", []),
total=response_body.get("totalCount", 0))
return InformationArticlesResponse(
information=response_body.get("data", {}).get("information"),
articles=response_body.get("data", {}).get("articles", []),
total=response_body.get("totalCount", 0),
)
src/backend/test/channel/test_bisheng_information_client.py
+28
View File
@@ -0,0 +1,28 @@
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pytest
from bisheng.common.errcode.channel import BishengInformationServiceError
from bisheng.core.config.settings import IntelligenceCenterConf
from bisheng.core.external.bisheng_information_client.client import BishengInformationClient
async def test_subscribe_failure_emits_collectable_error_log_and_reraises():
http_client = SimpleNamespace(
post=AsyncMock(return_value=SimpleNamespace(status_code=503, body={"message": "unavailable"}))
)
client = BishengInformationClient(
http_client=http_client,
get_conf=lambda: IntelligenceCenterConf(base_url="http://information.test", api_key="secret"),
)
with patch("bisheng.core.external.bisheng_information_client.client.logger.exception") as log_exception:
with pytest.raises(BishengInformationServiceError):
await client.subscribe_information_source(["source-1", "source-2"])
log_exception.assert_called_once_with(
"BISHENG_INFORMATION_SUBSCRIPTION_REQUEST_FAILED endpoint={} source_count={}",
"http://information.test/information/subscribe",
2,
)