"""媒体文件 → 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}"