Skip to content

Commit 3e0455c

Browse files
authored
添加公共工具模块
1 parent f5fff64 commit 3e0455c

1 file changed

Lines changed: 106 additions & 0 deletions

File tree

astrbot/core/utils/image_utils.py

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
"""图片处理工具函数。
2+
提供 MIME 类型检测与 base64 Data URL 编码等公共能力,
3+
供 ProviderRequest 及各 Provider 适配器复用,避免重复实现。
4+
"""
5+
6+
from __future__ import annotations
7+
8+
import base64
9+
10+
11+
def detect_image_mime_type(header_bytes: bytes) -> str:
12+
"""根据文件头magic bytes检测图片的实际 MIME 类型。
13+
14+
依次匹配常见图片格式的文件头特征,均不匹配时回退到 image/jpeg
15+
以保持向后兼容。支持的格式:JPEG、PNG、GIF、WebP、BMP、TIFF、
16+
ICO、SVG、AVIF、HEIF/HEIC。
17+
18+
Args:
19+
header_bytes: 文件头原始字节。SVG检测需要至少 256 字节;
20+
其他二进制格式最多需要 16 字节。
21+
22+
Returns:
23+
对应的 MIME 类型字符串,例如 "image/png"。
24+
"""
25+
if len(header_bytes) >= 3 and header_bytes[:3] == b'\xff\xd8\xff':
26+
return "image/jpeg"
27+
if len(header_bytes) >= 8 and header_bytes[:8] == b'\x89PNG\r\n\x1a\n':
28+
return "image/png"
29+
if len(header_bytes) >= 4 and header_bytes[:4] == b'GIF8':
30+
return "image/gif"
31+
# WebP: RIFF????WEBP
32+
if len(header_bytes) >= 12 and header_bytes[:4] == b'RIFF' and header_bytes[8:12] == b'WEBP':
33+
return "image/webp"
34+
if len(header_bytes) >= 2 and header_bytes[:2] == b'BM':
35+
return "image/bmp"
36+
# TIFF: 小端 (II) 或大端 (MM)
37+
if len(header_bytes) >= 4 and header_bytes[:4] in (b'II\x2a\x00', b'MM\x00\x2a'):
38+
return "image/tiff"
39+
if len(header_bytes) >= 4 and header_bytes[:4] == b'\x00\x00\x01\x00':
40+
return "image/x-icon"
41+
# SVG 为文本格式,检测头部是否含有 <svg 标签。
42+
# 调用方需传入至少 256 字节以覆盖带有 XML 声明的 SVG 文件。
43+
if b'<svg' in header_bytes[:256].lower():
44+
return "image/svg+xml"
45+
# AVIF: ftyp box brand = avif
46+
if len(header_bytes) >= 12 and header_bytes[4:12] == b'ftypavif':
47+
return "image/avif"
48+
# HEIF/HEIC: ftyp box,brand 为 heic/heix/hevc/hevx/mif1
49+
if len(header_bytes) >= 12 and header_bytes[4:8] == b'ftyp':
50+
brand = header_bytes[8:12]
51+
if brand in (b'heic', b'heix', b'hevc', b'hevx', b'mif1'):
52+
return "image/heif"
53+
# 无法识别,回退到 image/jpeg
54+
return "image/jpeg"
55+
56+
57+
# SVG 检测所需的最小头部字节数。
58+
# 其他二进制格式的魔术字节最多 16 字节,SVG 需要更多以跳过可能的 XML 声明。
59+
_HEADER_READ_SIZE = 256
60+
61+
# 对应 _HEADER_READ_SIZE 字节所需的 base64 字符数。
62+
# base64 每 3 字节编码为 4 字符,向上取整后再补充少量余量保证填充对齐。
63+
# ceil(256 / 3) * 4 = 344
64+
_BASE64_SAMPLE_CHARS = 344
65+
66+
67+
async def encode_image_to_base64_url(image_url: str) -> str:
68+
"""将图片转换为 base64 Data URL,自动检测实际 MIME 类型。
69+
70+
原实现硬编码 image/jpeg,会导致 PNG 等格式在严格校验的接口上报错。
71+
现通过读取文件头魔术字节来推断正确的 MIME 类型。
72+
73+
对于 SVG 文件,需要读取至少 256 字节的头部才能可靠检测,
74+
因此统一将读取/解码量提升到 256 字节,消除之前字节数不足的问题。
75+
76+
Args:
77+
image_url: 本地文件路径,或以 "base64://" 开头的 base64 字符串。
78+
79+
Returns:
80+
形如 "data:image/png;base64,..." 的 Data URL 字符串。
81+
"""
82+
if image_url.startswith("base64://"):
83+
raw_b64 = image_url[len("base64://"):]
84+
# 从 base64 数据中解码足量字节以检测实际格式。
85+
# 取前 344 个 base64 字符可解码出约 258 字节,足以覆盖 SVG 检测所需的 256 字节。
86+
try:
87+
sample = raw_b64[:_BASE64_SAMPLE_CHARS]
88+
# 确保 base64 填充正确,避免解码报错
89+
missing_padding = len(sample) % 4
90+
if missing_padding:
91+
sample += '=' * (4 - missing_padding)
92+
header_bytes = base64.b64decode(sample)
93+
mime_type = detect_image_mime_type(header_bytes)
94+
except Exception:
95+
# 解码失败时安全回退
96+
mime_type = "image/jpeg"
97+
return f"data:{mime_type};base64,{raw_b64}"
98+
99+
with open(image_url, "rb") as f:
100+
# 读取 256 字节用于格式检测,以支持需要较多头部数据的 SVG 等格式,
101+
# 再 seek 回起点读取完整内容进行 base64 编码。
102+
header_bytes = f.read(_HEADER_READ_SIZE)
103+
mime_type = detect_image_mime_type(header_bytes)
104+
f.seek(0)
105+
image_bs64 = base64.b64encode(f.read()).decode("utf-8")
106+
return f"data:{mime_type};base64,{image_bs64}"

0 commit comments

Comments
 (0)