File size: 3,118 Bytes
d0c18f0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """
URL输入处理器 - 供所有插件复用
支持 Base64 和 URL 两种输入方式
"""
import base64
import httpx
from typing import Optional, Tuple
from urllib.parse import urlparse
from .http_client import get_http_client
MAX_FILE_SIZE = 50 * 1024 * 1024 # 50MB
async def resolve_content(
content_base64: Optional[str],
content_url: Optional[str],
) -> Tuple[bytes, str]:
"""
统一解析输入内容
Args:
content_base64: Base64编码的内容（优先使用）
content_url: 文件URL地址（备选方式）
Returns:
(原始bytes, 来源说明)
Raises:
ValueError: 输入无效或下载失败
"""
if content_base64:
try:
return base64.b64decode(content_base64), "base64"
except Exception as e:
raise ValueError(f"Base64解码失败: {e}")
if content_url:
parsed = urlparse(content_url)
if parsed.scheme not in ("http", "https"):
raise ValueError(f"不支持的URL协议: {parsed.scheme}")
if not parsed.netloc:
raise ValueError(f"无效的URL: {content_url}")
try:
client = get_http_client()
resp = await client.get(content_url)
if resp.status_code != 200:
raise ValueError(f"下载失败: HTTP {resp.status_code}")
content_length = len(resp.content)
if content_length > MAX_FILE_SIZE:
raise ValueError(f"文件过大: {content_length} bytes (上限 {MAX_FILE_SIZE})")
return resp.content, f"url:{content_url}"
except httpx.TimeoutException:
raise ValueError(f"下载超时")
except httpx.RequestError as e:
raise ValueError(f"网络请求失败: {e}")
raise ValueError("必须提供 content_base64 或 content_url")
async def resolve_text_content(
content_base64: Optional[str],
content_url: Optional[str],
encoding: str = "utf-8",
) -> Tuple[str, str]:
"""
解析文本内容（自动解码）
Args:
content_base64: Base64编码的内容
content_url: 文件URL地址
encoding: 文本编码（默认utf-8）
Returns:
(解码后的文本, 来源说明)
"""
content, source = await resolve_content(content_base64, content_url)
try:
return content.decode(encoding), source
except UnicodeDecodeError as e:
raise ValueError(f"文本解码失败 ({encoding}): {e}")
def encode_bytes_to_base64(data: bytes) -> str:
"""将字节数据编码为 Base64"""
return base64.b64encode(data).decode("utf-8")
def encode_text_to_base64(text: str, encoding: str = "utf-8") -> str:
"""将文本编码为 Base64"""
return base64.b64encode(text.encode(encoding)).decode("utf-8")
def decode_base64_to_bytes(base64_str: str) -> bytes:
"""将 Base64 解码为字节"""
return base64.b64decode(base64_str)
def decode_base64_to_text(base64_str: str, encoding: str = "utf-8") -> str:
"""将 Base64 解码为文本"""
return base64.b64decode(base64_str).decode(encoding) |