61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
"""媒体文件 → base64 data URI 转换工具。"""
|
|
import base64
|
|
import mimetypes
|
|
import os
|
|
|
|
import httpx
|
|
from sqlalchemy import select
|
|
|
|
from app.config import settings
|
|
from app.models.system_config import SystemConfig
|
|
|
|
|
|
async def get_llm_media_as_base64(db=None) -> bool:
|
|
"""读取 SystemConfig 中的 llm_media_as_base64 设置。无 DB 连接时回退到 env。"""
|
|
if db is not None:
|
|
try:
|
|
result = await db.execute(
|
|
select(SystemConfig).where(SystemConfig.key == "llm_media_as_base64").limit(1)
|
|
)
|
|
config = result.scalar_one_or_none()
|
|
if config:
|
|
return config.value.lower() in ("true", "1", "yes")
|
|
except Exception:
|
|
pass
|
|
return settings.LLM_MEDIA_AS_BASE64
|
|
|
|
|
|
async def media_to_base64(url: str, fallback_mime: str = "image/png", max_mb: int = 20) -> str:
|
|
"""将任意媒体 URL/路径转为 base64 data URI。
|
|
|
|
- data: URI → 原样返回
|
|
- http(s):// → 下载后编码（受 max_mb 限制）
|
|
- /uploads/xxx 或本地路径 → 读盘编码
|
|
"""
|
|
if url.startswith("data:"):
|
|
return url
|
|
|
|
if url.startswith(("http://", "https://")):
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
resp = await client.get(url)
|
|
size_mb = len(resp.content) / (1024 * 1024)
|
|
if size_mb > max_mb:
|
|
raise ValueError(f"媒体文件过大 ({size_mb:.1f}MB > {max_mb}MB)")
|
|
b64 = base64.b64encode(resp.content).decode()
|
|
mime = resp.headers.get("content-type") or fallback_mime
|
|
return f"data:{mime};base64,{b64}"
|
|
|
|
# 本地路径
|
|
if url.startswith("/generate/"):
|
|
type_pre = '/generate/videos/' if url.startswith("/generate/videos/") else '/generate/images/'
|
|
type_path = settings.STORAGE_LOCAL_PATH if url.startswith("/generate/videos/") else settings.STORAGE_IMAGE_LOCAL_PATH
|
|
relative_path = url.replace(type_pre, "", 1).lstrip("/")
|
|
file_path = os.path.join(type_path, relative_path)
|
|
else:
|
|
relative_path = url.replace("/uploads/", "", 1).lstrip("/")
|
|
file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, relative_path)
|
|
mime = mimetypes.guess_type(file_path)[0] or fallback_mime
|
|
with open(file_path, "rb") as f:
|
|
b64 = base64.b64encode(f.read()).decode()
|
|
return f"data:{mime};base64,{b64}"
|