| """ |
| 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 |
|
|
|
|
| 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) |
