工具与辅助
日志管理器 - 单例模式
提供统一的日志记录接口,支持控制台和文件双输出。 自动处理 Windows UTF-8 编码问题。
属性:
| 名称 | 类型 | 描述 |
|---|---|---|
_instance |
Optional[Logger]
|
单例实例 |
_logger |
Optional[Logger]
|
日志记录器实例 |
示例:
>>> from ai_service.utils.logger import logger
>>> logger.info("这是一条信息日志")
>>> logger.error("这是一条错误日志")
__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)