跳转至

工具与辅助

日志管理器 - 单例模式

提供统一的日志记录接口,支持控制台和文件双输出。 自动处理 Windows UTF-8 编码问题。

属性:

名称 类型 描述
_instance Optional[Logger]

单例实例

_logger Optional[Logger]

日志记录器实例

示例:

>>> from ai_service.utils.logger import logger
>>> logger.info("这是一条信息日志")
>>> logger.error("这是一条错误日志")

__new__

__new__()

单例模式实现 - 确保只有一个实例

返回:

名称 类型 描述
Logger Logger

日志管理器单例实例

__init__

__init__()

初始化日志管理器

get_logger

get_logger()

获取日志器实例

返回:

类型 描述
Logger

logging.Logger: 底层的 logging.Logger 实例

__getattr__

__getattr__(name)

委托属性访问到底层的 logger 实例

允许直接调用 logger.info()、logger.error() 等方法。

参数:

名称 类型 描述 默认
name str

属性名称

必需

返回:

类型 描述
Any

底层 logger 实例的对应属性

通用辅助函数模块

提供常用的纯函数工具,保持无状态、可复用。

parse_datetime

parse_datetime(date_str, fmt='%Y/%m/%d %H:%M')

解析日期时间字符串为 datetime 对象

参数:

名称 类型 描述 默认
date_str str

日期时间字符串

必需
fmt str

日期时间格式,默认为 '%Y/%m/%d %H:%M'

'%Y/%m/%d %H:%M'

返回:

类型 描述
Optional[datetime]

Optional[datetime]: 解析后的 datetime 对象,解析失败返回 None

示例:

>>> from ai_service.utils.helpers import parse_datetime
>>> dt = parse_datetime("2025/12/01 16:00")
>>> print(dt)
2025-12-01 16:00:00
>>> # 自定义格式
>>> dt = parse_datetime("2025-12-01", fmt='%Y-%m-%d')

safe_json_loads

safe_json_loads(json_str, default=None)

安全地解析 JSON 字符串

参数:

名称 类型 描述 默认
json_str str

JSON 字符串

必需
default Any

解析失败时返回的默认值,默认为 None

None

返回:

名称 类型 描述
Any Any

解析后的 Python 对象,失败返回 default

示例:

>>> from ai_service.utils.helpers import safe_json_loads
>>> data = safe_json_loads('{"key": "value"}')
>>> print(data)
{'key': 'value'}
>>> # 解析失败返回默认值
>>> data = safe_json_loads('invalid json', default={})
>>> print(data)
{}

safe_get_nested

safe_get_nested(data, keys, default=None)

安全地获取嵌套字典中的值

参数:

名称 类型 描述 默认
data Dict

源字典

必需
keys List[str]

键路径列表

必需
default Any

获取失败时返回的默认值

None

返回:

名称 类型 描述
Any Any

获取到的值,失败返回 default

示例:

>>> from ai_service.utils.helpers import safe_get_nested
>>> data = {"a": {"b": {"c": 123}}}
>>> value = safe_get_nested(data, ["a", "b", "c"])
>>> print(value)
123
>>> # 键不存在时返回默认值
>>> value = safe_get_nested(data, ["a", "x", "y"], default=0)
>>> print(value)
0

truncate_string

truncate_string(text, max_length=100, suffix='...')

截断字符串到指定长度

参数:

名称 类型 描述 默认
text str

原始字符串

必需
max_length int

最大长度,默认 100

100
suffix str

截断后的后缀,默认 "..."

'...'

返回:

名称 类型 描述
str str

截断后的字符串

示例:

>>> from ai_service.utils.helpers import truncate_string
>>> text = "This is a very long string that needs to be truncated"
>>> short = truncate_string(text, max_length=20)
>>> print(short)
This is a very lo...

normalize_whitespace

normalize_whitespace(text)

标准化字符串中的空白字符

将多个连续空白字符替换为单个空格,并去除首尾空白。

参数:

名称 类型 描述 默认
text str

原始字符串

必需

返回:

名称 类型 描述
str str

标准化后的字符串

示例:

>>> from ai_service.utils.helpers import normalize_whitespace
>>> text = "  hello   world  \n  test  "
>>> normalized = normalize_whitespace(text)
>>> print(normalized)
hello world test

chunks

chunks(lst, n)

将列表分割为固定大小的块

参数:

名称 类型 描述 默认
lst List

原始列表

必需
n int

每块的大小

必需

返回:

类型 描述
List[List]

List[List]: 分割后的列表

示例:

>>> from ai_service.utils.helpers import chunks
>>> data = [1, 2, 3, 4, 5, 6, 7, 8, 9]
>>> result = chunks(data, 3)
>>> print(result)
[[1, 2, 3], [4, 5, 6], [7, 8, 9]]

retry_on_exception

retry_on_exception(
    func,
    max_retries=3,
    delay=1.0,
    exceptions=(Exception,),
    default=None,
)

对函数进行重试装饰

参数:

名称 类型 描述 默认
func Callable[[], T]

要执行的函数

必需
max_retries int

最大重试次数

3
delay float

重试间隔(秒)

1.0
exceptions tuple

需要重试的异常类型

(Exception,)
default T | None

失败后返回的默认值

None

返回:

类型 描述
T | None

函数执行结果或默认值

示例:

>>> from ai_service.utils.helpers import retry_on_exception
>>> import requests
>>>
>>> def fetch_data():
>>>     return requests.get("https://api.example.com/data")
>>>
>>> result = retry_on_exception(fetch_data, max_retries=3)