Files
AstrBot/astrbot/core/utils/media_utils.py
Weilong Liao 7c366a708b fix: unify media reference handling (#8764)
* fix: unify media reference handling

* fix: accept bare base64 record media refs

* chore: update agents.md

* fix: unify file URI handling across media components and utilities

* fix: unify media reference type handling with MediaRefStr alias

* Potential fix for pull request finding 'CodeQL / Incomplete URL substring sanitization'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* Update astrbot/core/platform/sources/discord/discord_platform_adapter.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* fix: unify media handling and improve base64 decoding across components

* fix: simplify client_kwargs type definition and enhance media message handling in platform adapter

* fix: unify media utility documentation and enhance function descriptions

* perf: drop "pilk" requirement, improve audio outbound for tencent-related IM apps which using silk

* fix: unify Tencent Silk audio handling and enhance media resolver functionality

---

- Centralize media reference materialization and base64 resolution for local paths, http(s), base64://, data URIs, and legacy bare base64 payloads.
- Normalize incoming Record audio to wav and Image media to temporary jpg during preprocess, with event-scoped cleanup.
- Reuse the shared media resolver across OpenAI, Gemini, Anthropic, MiMo, DeerFlow, STT, and platform media paths while sanitizing logs and cleaning temporary conversion outputs.
- Ensure generated TTS audio is tracked for cleanup after the event finishes.

fix #8676
fix #8543
fix #7588
fix #7580
fix #8030
fix #8034
fix #7461
fix #7565
fix #6509
fix #7144
fix #7795



---------

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
2026-06-14 10:37:16 +08:00

1481 lines
48 KiB
Python

"""Media file utilities.
Provides shared media reference materialization, format conversion, duration
probing, and image compression helpers.
"""
import asyncio
import base64
import binascii
import io
import mimetypes
import os
import subprocess
import uuid
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import TypeAlias
from urllib.parse import unquote, urlparse, urlsplit
from urllib.request import url2pathname
from PIL import Image as PILImage
from astrbot import logger
from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
from astrbot.core.utils.io import download_file
from astrbot.core.utils.tencent_record_helper import (
tencent_silk_to_wav,
wav_to_tencent_silk,
)
IMAGE_COMPRESS_DEFAULT_MAX_SIZE = 1280
IMAGE_COMPRESS_DEFAULT_QUALITY = 95
IMAGE_COMPRESS_DEFAULT_OPTIMIZE = True
IMAGE_COMPRESS_DEFAULT_MIN_FILE_SIZE_MB = 1.0
MEDIA_MIME_EXTENSIONS = {
"audio/wav": ".wav",
"audio/wave": ".wav",
"audio/x-wav": ".wav",
"audio/mpeg": ".mp3",
"audio/mp3": ".mp3",
"audio/ogg": ".ogg",
"audio/opus": ".opus",
"audio/flac": ".flac",
"audio/aac": ".aac",
"audio/amr": ".amr",
"audio/silk": ".silk",
"image/jpeg": ".jpg",
"image/png": ".png",
"image/gif": ".gif",
"image/webp": ".webp",
"image/bmp": ".bmp",
"image/tiff": ".tiff",
"image/avif": ".avif",
"video/mp4": ".mp4",
"video/webm": ".webm",
"video/quicktime": ".mov",
}
IMAGE_FORMAT_MIME_TYPES = {
"JPEG": "image/jpeg",
"PNG": "image/png",
"GIF": "image/gif",
"WEBP": "image/webp",
"BMP": "image/bmp",
"TIFF": "image/tiff",
"AVIF": "image/avif",
}
AUDIO_FORMAT_MIME_TYPES = {
"aac": "audio/aac",
"amr": "audio/amr",
"flac": "audio/flac",
"mp3": "audio/mp3",
"ogg": "audio/ogg",
"opus": "audio/opus",
"silk": "audio/silk",
"tencent_silk": "audio/silk",
"wav": "audio/wav",
}
DEFAULT_MEDIA_SUFFIXES = {
"audio": ".wav",
"image": ".bin",
"video": ".mp4",
"file": ".bin",
}
MediaRefStr: TypeAlias = str
"""
A media reference string accepted by MediaResolver: local path, file URI, HTTP(S),
base64://, data URI, or legacy bare base64.
Examples:
Local path: ``/tmp/image.png``
File URI: ``file:///tmp/image.png``
HTTP(S) URL: ``https://example.com/image.png``
base64:// payload: ``base64://iVBORw0KGgo...``
Data URI: ``data:image/png;base64,iVBORw0KGgo...``
Legacy bare base64: ``iVBORw0KGgo...``
"""
@dataclass(slots=True)
class ResolvedMediaData:
"""Base64 media bytes plus the metadata needed by provider payloads.
Attributes:
base64_data: Raw base64 payload without a ``data:`` URI prefix.
mime_type: MIME type to send with provider payloads.
format: Optional normalized media format, such as ``wav`` for audio.
"""
base64_data: str
mime_type: str
format: str | None = None
def to_bytes(self) -> bytes:
"""Decode the base64 payload, accepting missing padding."""
return _decode_base64_payload(
self.base64_data,
error_message="invalid resolved media base64 data",
)
def to_data_url(self) -> str:
"""Return a ``data:<mime>;base64,...`` URL for multimodal providers."""
return f"data:{self.mime_type};base64,{self.base64_data}"
@dataclass(slots=True)
class _LocalMediaFile:
path: Path
mime_type: str | None = None
cleanup_paths: list[Path] = field(default_factory=list)
@dataclass(slots=True)
class ResolvedMediaFile:
"""A media reference resolved to a local path.
``cleanup_paths`` contains temporary files owned by the resolver. Callers that
use ``MediaResolver.as_path()`` get automatic cleanup; callers that need to
keep a path after the resolver returns should use ``MediaResolver.to_path()``.
"""
source_ref: MediaRefStr
media_type: str
path: Path
mime_type: str | None = None
format: str | None = None
cleanup_paths: list[Path] = field(default_factory=list)
def read_bytes(self) -> bytes:
"""Read the resolved local file."""
return self.path.read_bytes()
def to_base64(self) -> str:
"""Read the resolved local file and return raw base64 data."""
return base64.b64encode(self.read_bytes()).decode("utf-8")
def to_data_url(self) -> str:
"""Read the resolved local file and return a data URL."""
mime_type = self.mime_type or "application/octet-stream"
return f"data:{mime_type};base64,{self.to_base64()}"
def open(self, mode: str = "rb"):
"""Open the resolved local file."""
return self.path.open(mode)
def detach(self) -> None:
"""Keep temporary files alive after resolver cleanup would normally run."""
self.cleanup_paths.clear()
def cleanup(self) -> None:
_cleanup_paths(self.cleanup_paths)
def is_file_uri(value: object) -> bool:
"""Return whether a value is a ``file:`` URI.
Args:
value: Candidate media reference or local path.
Returns:
``True`` only for string values whose parsed URI scheme is ``file``.
"""
if not isinstance(value, str):
return False
try:
return urlsplit(value).scheme.lower() == "file"
except ValueError:
return False
def file_uri_to_path(file_uri: MediaRefStr) -> str:
"""Normalize file URIs to local filesystem paths.
Args:
file_uri: A ``file:`` URI or a plain filesystem path.
Returns:
The local filesystem path decoded with standard-library URL path rules.
Non-``file:`` inputs are returned unchanged for convenience.
"""
if not is_file_uri(file_uri):
return file_uri
parsed = urlparse(file_uri)
netloc = parsed.netloc or ""
path = parsed.path or ""
if netloc and netloc.lower() != "localhost":
if len(netloc) == 2 and netloc[1] == ":" and netloc[0].isalpha():
return str(Path(url2pathname(f"{netloc}{path}")))
return str(Path(url2pathname(f"//{netloc}{path}")))
path = url2pathname(path)
if len(path) >= 4 and path[0] == "/" and path[2] == ":" and path[1].isalpha():
path = path[1:]
elif os.name != "nt" and path.startswith("//"):
# Older AstrBot builds generated file:////path for POSIX absolute paths.
path = "/" + path.lstrip("/")
return str(Path(path))
def _extension_from_mime_type(mime_type: str | None) -> str | None:
"""Return a filesystem suffix for a MIME type, if one is known."""
if not mime_type:
return None
normalized = mime_type.split(";", 1)[0].strip().lower()
if not normalized:
return None
return MEDIA_MIME_EXTENSIONS.get(normalized) or mimetypes.guess_extension(
normalized
)
def _temp_media_path(media_type: str, suffix: str) -> Path:
"""Create a unique path under AstrBot's temp directory for materialized media."""
temp_dir = Path(get_astrbot_temp_path())
temp_dir.mkdir(parents=True, exist_ok=True)
safe_media_type = "".join(
char if char.isalnum() or char in {"_", "-"} else "_" for char in media_type
)
return temp_dir / f"media_{safe_media_type}_{uuid.uuid4().hex}{suffix}"
def _parse_base64_data_uri(data_uri: str) -> tuple[str | None, bytes]:
"""Parse a base64 data URI and return ``(mime_type, decoded_bytes)``."""
header, separator, payload = data_uri.partition(",")
if not separator or not header.lower().startswith("data:"):
raise ValueError("invalid data URI")
header_body = header[5:]
header_parts = header_body.split(";") if header_body else []
mime_type = header_parts[0].strip() if header_parts and header_parts[0] else None
if not any(part.lower() == "base64" for part in header_parts[1:]):
raise ValueError("data URI is not base64 encoded")
return mime_type, _decode_base64_payload(
payload,
error_message="invalid base64 data URI payload",
)
def _decode_base64_payload(
payload: str,
*,
error_message: str,
validate: bool = False,
) -> bytes:
"""Decode a base64 payload while tolerating omitted padding.
Args:
payload: Base64 payload without a data URI header.
error_message: Message to use when decoding fails.
validate: Whether to ask ``base64.b64decode`` to reject non-base64
characters.
Returns:
Decoded bytes.
Raises:
ValueError: Raised when the payload cannot be decoded.
"""
payload = "".join(payload.split())
missing_padding = len(payload) % 4
if missing_padding:
payload += "=" * (4 - missing_padding)
try:
return base64.b64decode(payload, validate=validate)
except (binascii.Error, ValueError) as exc:
raise ValueError(error_message) from exc
def describe_media_ref(media_ref: object | None) -> str:
"""Return a log-safe description of a media reference.
Args:
media_ref: Original media reference from a platform, plugin, or provider
request. It may contain a signed URL or a large base64 payload.
Returns:
A short description that avoids logging query strings, tokens, and base64
payload contents.
"""
if not media_ref:
return "<empty media ref>"
if not isinstance(media_ref, str):
return f"media ref type={type(media_ref).__name__}"
ref_len = len(media_ref)
if media_ref.startswith("data:"):
header, _, payload = media_ref.partition(",")
mime_type = header[5:].split(";", 1)[0] or "unknown"
return f"data URI mime={mime_type!r} payload_len={len(payload)}"
if media_ref.startswith("base64://"):
return f"base64 media payload_len={len(media_ref.removeprefix('base64://'))}"
parsed = urlparse(media_ref)
if parsed.scheme in {"http", "https"}:
filename = Path(unquote(parsed.path or "")).name
suffix = f" file={filename!r}" if filename else ""
return f"{parsed.scheme} URL host={parsed.netloc!r}{suffix} len={ref_len}"
if is_file_uri(media_ref):
filename = Path(file_uri_to_path(media_ref)).name
return f"file URI name={filename!r} len={ref_len}"
media_path_exists = False
try:
media_path_exists = Path(media_ref).exists()
except OSError:
pass
if not media_path_exists:
compact = "".join(media_ref.split())
if compact:
try:
_decode_base64_payload(
compact,
error_message="invalid bare base64 media payload",
validate=True,
)
except ValueError:
pass
else:
return f"bare base64 media payload_len={len(compact)}"
return f"local media path name={Path(media_ref).name!r} len={ref_len}"
def detect_image_mime_type(
image_bytes: bytes,
*,
default_mime_type: str | None = "image/jpeg",
) -> str | None:
"""Detect an image MIME type from bytes.
Args:
image_bytes: Encoded image bytes to inspect.
default_mime_type: MIME type to return when detection fails.
Returns:
The detected MIME type, or ``default_mime_type`` when detection fails or
the format is unknown.
"""
try:
with PILImage.open(io.BytesIO(image_bytes)) as image:
image.verify()
image_format = str(image.format or "").upper()
except Exception:
return default_mime_type
return IMAGE_FORMAT_MIME_TYPES.get(image_format, default_mime_type)
def _guess_mime_type(path: Path, fallback: str | None = None) -> str | None:
"""Guess a MIME type from a filename, with an optional fallback."""
return mimetypes.guess_type(path.name)[0] or fallback
def _cleanup_paths(cleanup_paths: list[Path] | None) -> None:
"""Best-effort cleanup for temporary files created by the resolver."""
for cleanup_path in cleanup_paths or []:
try:
cleanup_path.unlink(missing_ok=True)
except Exception as exc:
logger.warning("Failed to cleanup %s: %s", cleanup_path, exc)
async def _materialize_media_ref(
media_ref: MediaRefStr,
*,
media_type: str = "file",
default_suffix: str | None = None,
) -> _LocalMediaFile:
"""Resolve a plugin-facing media reference to a local file.
Supported references: local paths, file:// URIs, http(s) URLs, base64://,
data:*;base64,... URIs, and legacy bare base64 payloads.
Args:
media_ref: Original media reference from a platform, plugin, or history.
media_type: Logical media family used for temp filenames and defaults.
default_suffix: Suffix to use when the reference does not carry one.
"""
cleanup_paths: list[Path] = []
suffix = default_suffix or DEFAULT_MEDIA_SUFFIXES.get(media_type, ".bin")
if media_ref.startswith(("http://", "https://")):
parsed = urlparse(media_ref)
target_suffix = Path(parsed.path).suffix or suffix
target_path = _temp_media_path(media_type, target_suffix)
cleanup_paths.append(target_path)
try:
await download_file(media_ref, str(target_path))
except Exception:
_cleanup_paths(cleanup_paths)
raise
return _LocalMediaFile(
path=target_path,
mime_type=_guess_mime_type(target_path),
cleanup_paths=cleanup_paths,
)
if is_file_uri(media_ref):
path = Path(file_uri_to_path(media_ref))
return _LocalMediaFile(path=path, mime_type=_guess_mime_type(path))
if media_ref.startswith("data:"):
mime_type, media_bytes = _parse_base64_data_uri(media_ref)
target_suffix = _extension_from_mime_type(mime_type) or suffix
target_path = _temp_media_path(media_type, target_suffix)
cleanup_paths.append(target_path)
try:
target_path.write_bytes(media_bytes)
except Exception:
_cleanup_paths(cleanup_paths)
raise
return _LocalMediaFile(
path=target_path,
mime_type=mime_type,
cleanup_paths=cleanup_paths,
)
if media_ref.startswith("base64://"):
media_bytes = _decode_base64_payload(
media_ref.removeprefix("base64://"),
error_message="invalid base64 media payload",
)
target_path = _temp_media_path(media_type, suffix)
cleanup_paths.append(target_path)
try:
target_path.write_bytes(media_bytes)
except Exception:
_cleanup_paths(cleanup_paths)
raise
return _LocalMediaFile(path=target_path, cleanup_paths=cleanup_paths)
path = Path(media_ref)
path_exists = False
try:
path_exists = path.exists()
except OSError:
pass
if path_exists:
return _LocalMediaFile(path=path, mime_type=_guess_mime_type(path))
compact_media_ref = "".join(media_ref.split())
if compact_media_ref:
try:
media_bytes = _decode_base64_payload(
compact_media_ref,
error_message="invalid bare base64 media payload",
validate=True,
)
except ValueError:
pass
else:
target_path = _temp_media_path(media_type, suffix)
cleanup_paths.append(target_path)
try:
target_path.write_bytes(media_bytes)
except Exception:
_cleanup_paths(cleanup_paths)
raise
return _LocalMediaFile(path=target_path, cleanup_paths=cleanup_paths)
return _LocalMediaFile(path=path, mime_type=_guess_mime_type(path))
class MediaResolver:
"""Resolve, convert, and export media references.
The resolver accepts local paths, file:// URIs, http(s) URLs, base64:// payloads,
data:*;base64,... URIs, and legacy bare base64 payloads. Temporary paths are
cleaned when using as_path(), while to_path() intentionally leaves returned
paths alive for callers that need to hand them to platform SDKs.
Args:
media_ref: Source media reference. It may be a local path, ``file://`` URI,
HTTP(S) URL, ``base64://`` payload, base64 data URI, or legacy bare
base64 payload.
media_type: Logical media family. ``audio`` enables format conversion and
defaults to WAV output; ``image`` enables image MIME detection.
default_suffix: Fallback suffix for temporary files when the source does
not expose one.
"""
def __init__(
self,
media_ref: MediaRefStr,
*,
media_type: str = "file",
default_suffix: str | None = None,
) -> None:
self.media_ref = media_ref
self.media_type = media_type
self.default_suffix = default_suffix
async def _resolve_path(
self,
*,
target_format: str | None = None,
preserve_mp3: bool = False,
) -> ResolvedMediaFile:
"""Materialize the source and apply media-type-specific conversion.
For audio, ``target_format`` controls the output format, including the
QQ / Wechat / Wecom ``tencent_silk`` upload format. When it is not set, audio
resolves to WAV unless ``preserve_mp3`` is true and the source already
appears to be MP3.
"""
local_file = await _materialize_media_ref(
self.media_ref,
media_type=self.media_type,
default_suffix=self.default_suffix,
)
cleanup_paths = list(local_file.cleanup_paths)
resolved_path = local_file.path
mime_type = local_file.mime_type or _guess_mime_type(resolved_path)
resolved_format = resolved_path.suffix.lower().lstrip(".") or None
try:
if self.media_type == "audio":
audio_format = target_format
if not audio_format:
audio_format = (
"mp3" if preserve_mp3 and resolved_format == "mp3" else "wav"
)
if audio_format == "tencent_silk":
intermediate_cleanup_paths = list(cleanup_paths)
silk_path = _temp_media_path("audio", ".silk")
try:
wav_path = Path(await ensure_wav(str(resolved_path)))
if wav_path != resolved_path:
intermediate_cleanup_paths.append(wav_path)
duration = await wav_to_tencent_silk(
str(wav_path), str(silk_path)
)
if duration <= 0:
raise ValueError(
"Tencent Silk conversion returned empty duration"
)
except Exception:
_cleanup_paths([*intermediate_cleanup_paths, silk_path])
raise
_cleanup_paths(intermediate_cleanup_paths)
cleanup_paths = [silk_path]
resolved_path = silk_path
resolved_format = audio_format
mime_type = AUDIO_FORMAT_MIME_TYPES[resolved_format]
else:
if audio_format == "wav":
converted_audio_path = Path(
await ensure_wav(str(resolved_path))
)
elif resolved_format == audio_format:
converted_audio_path = resolved_path
else:
converted_audio_path = Path(
await convert_audio_format(
str(resolved_path),
output_format=audio_format,
)
)
if converted_audio_path != resolved_path:
cleanup_paths.append(converted_audio_path)
resolved_path = converted_audio_path
resolved_format = audio_format
mime_type = AUDIO_FORMAT_MIME_TYPES.get(
resolved_format, "audio/wav"
)
except Exception:
_cleanup_paths(cleanup_paths)
raise
return ResolvedMediaFile(
source_ref=self.media_ref,
media_type=self.media_type,
path=resolved_path,
mime_type=mime_type,
format=resolved_format,
cleanup_paths=cleanup_paths,
)
@asynccontextmanager
async def as_path(
self,
*,
target_format: str | None = None,
preserve_mp3: bool = False,
) -> AsyncIterator[ResolvedMediaFile]:
"""Yield a resolved local file and clean resolver-owned temp files on exit.
Use this when the consumer only needs the file during the context manager.
For audio, pass ``target_format`` to force a format such as ``wav`` or
``tencent_silk``.
"""
resolved = await self._resolve_path(
target_format=target_format,
preserve_mp3=preserve_mp3,
)
try:
yield resolved
finally:
resolved.cleanup()
async def to_path(
self,
*,
target_format: str | None = None,
preserve_mp3: bool = False,
) -> str:
"""Return a resolved local path and keep temporary files alive.
This is for message components and platform SDK calls that need a path
after the resolver method returns. Callers or event cleanup should remove
the returned temp file later.
"""
resolved = await self._resolve_path(
target_format=target_format,
preserve_mp3=preserve_mp3,
)
resolved.detach()
return str(resolved.path.resolve())
async def to_bytes(
self,
*,
target_format: str | None = None,
preserve_mp3: bool = False,
) -> bytes:
"""Resolve media, read bytes, and clean resolver-owned temp files."""
async with self.as_path(
target_format=target_format,
preserve_mp3=preserve_mp3,
) as resolved:
return resolved.read_bytes()
async def to_base64(
self,
*,
target_format: str | None = None,
preserve_mp3: bool = False,
) -> str:
"""Resolve media to raw base64 data without a data URI prefix."""
return base64.b64encode(
await self.to_bytes(
target_format=target_format,
preserve_mp3=preserve_mp3,
)
).decode("utf-8")
async def to_base64_data(
self,
*,
strict: bool = False,
target_format: str | None = None,
preserve_mp3: bool = False,
default_mime_type: str | None = "image/jpeg",
) -> ResolvedMediaData | None:
"""Resolve media to base64 data plus MIME metadata.
Args:
strict: Raise on invalid or unreadable media instead of returning
``None`` where the resolver can safely ignore the reference.
target_format: Optional output format for audio conversion.
preserve_mp3: Keep existing MP3 audio as MP3 when no target format is
provided; otherwise audio defaults to WAV.
default_mime_type: Fallback MIME type for legacy image base64 payloads
whose bytes cannot be identified.
"""
if self.media_type == "image":
async with self.as_path(target_format=target_format) as resolved:
try:
media_bytes = resolved.read_bytes()
except OSError:
if strict:
raise
return None
mime_type = detect_image_mime_type(
media_bytes,
default_mime_type=None,
)
if (
not mime_type
and resolved.mime_type
and resolved.mime_type.startswith("image/")
):
mime_type = resolved.mime_type
is_legacy_base64_ref = self.media_ref.startswith("base64://")
is_remote_or_data_ref = self.media_ref.startswith(
("http://", "https://", "data:")
) or is_file_uri(self.media_ref)
if not is_legacy_base64_ref and not is_remote_or_data_ref:
try:
_decode_base64_payload(
"".join(self.media_ref.split()),
error_message="invalid bare base64 media payload",
validate=True,
)
except ValueError:
is_legacy_base64_ref = False
else:
is_legacy_base64_ref = True
if not mime_type and is_legacy_base64_ref:
mime_type = default_mime_type
if not mime_type:
if strict:
raise ValueError(
f"Invalid image file: {describe_media_ref(self.media_ref)}"
)
return None
return ResolvedMediaData(
base64_data=base64.b64encode(media_bytes).decode("utf-8"),
mime_type=mime_type,
)
async with self.as_path(
target_format=target_format,
preserve_mp3=preserve_mp3,
) as resolved:
try:
media_bytes = resolved.read_bytes()
except OSError:
if strict:
raise
return None
mime_type = resolved.mime_type or "application/octet-stream"
return ResolvedMediaData(
base64_data=base64.b64encode(media_bytes).decode("utf-8"),
mime_type=mime_type,
format=resolved.format,
)
async def to_data_url(
self,
*,
strict: bool = False,
target_format: str | None = None,
preserve_mp3: bool = False,
default_mime_type: str | None = "image/jpeg",
) -> str | None:
"""Resolve media directly to a provider-ready data URL."""
resolved = await self.to_base64_data(
strict=strict,
target_format=target_format,
preserve_mp3=preserve_mp3,
default_mime_type=default_mime_type,
)
return resolved.to_data_url() if resolved else None
@asynccontextmanager
async def open(
self,
mode: str = "rb",
*,
target_format: str | None = None,
preserve_mp3: bool = False,
):
"""Open resolved media as a file object inside a cleanup context."""
async with self.as_path(
target_format=target_format,
preserve_mp3=preserve_mp3,
) as resolved:
with resolved.open(mode) as file_obj:
yield file_obj
async def resolve_image_ref_to_base64_data(
image_ref: MediaRefStr,
*,
strict: bool = False,
default_mime_type: str | None = "image/jpeg",
) -> ResolvedMediaData | None:
"""Resolve an image reference to base64 data and a detected MIME type.
``strict=False`` returns ``None`` for invalid images so provider payload
assembly can skip bad image refs without failing the whole request.
"""
return await MediaResolver(
image_ref,
media_type="image",
default_suffix=".bin",
).to_base64_data(
strict=strict,
default_mime_type=default_mime_type,
)
async def resolve_audio_ref_to_base64_data(
audio_ref: MediaRefStr,
*,
preserve_mp3: bool = False,
target_format: str | None = None,
) -> ResolvedMediaData:
"""Resolve an audio reference to base64 data.
Audio is converted to WAV by default. Pass preserve_mp3=True for legacy
provider payloads that intentionally keep MP3 input unchanged.
``target_format`` overrides both defaults when provided.
"""
audio_data = await MediaResolver(
audio_ref,
media_type="audio",
default_suffix=".wav",
).to_base64_data(
target_format=target_format,
preserve_mp3=preserve_mp3,
strict=True,
)
if audio_data is None:
raise ValueError(f"Invalid audio data: {describe_media_ref(audio_ref)}")
return audio_data
async def resolve_media_ref_to_base64_data(
media_ref: MediaRefStr,
*,
media_type: str,
strict: bool = False,
) -> ResolvedMediaData | None:
"""Resolve a media reference to base64 data through one shared entrypoint.
This helper keeps provider sources from knowing whether a reference is local,
HTTP(S), ``base64://``, a data URI, or a legacy bare base64 payload.
"""
if media_type == "image":
return await resolve_image_ref_to_base64_data(media_ref, strict=strict)
if media_type == "audio":
return await resolve_audio_ref_to_base64_data(media_ref)
return await MediaResolver(
media_ref,
media_type=media_type,
).to_base64_data(
strict=strict,
)
async def get_media_duration(file_path: str) -> int | None:
"""Probe media duration with ffprobe.
Args:
file_path: Local media file path.
Returns:
Duration in milliseconds, or ``None`` when probing fails.
"""
try:
# Probe duration with ffprobe.
process = await asyncio.create_subprocess_exec(
"ffprobe",
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
file_path,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout, stderr = await process.communicate()
if process.returncode == 0 and stdout:
duration_seconds = float(stdout.decode().strip())
duration_ms = int(duration_seconds * 1000)
logger.debug("Media duration detected: %sms", duration_ms)
return duration_ms
else:
logger.warning("Failed to get media duration: %s", file_path)
return None
except FileNotFoundError:
logger.warning(
"ffprobe is not installed or not in PATH. "
"Install ffmpeg: https://ffmpeg.org/"
)
return None
except Exception as e:
logger.warning("Error while probing media duration: %s", e)
return None
async def convert_audio_to_opus(audio_path: str, output_path: str | None = None) -> str:
"""Convert an audio file to Opus format.
Args:
audio_path: Source audio file path.
output_path: Optional output file path. When omitted, a temporary path is
created under AstrBot's temp directory.
Returns:
The converted Opus file path.
"""
return await convert_audio_format(
audio_path=audio_path,
output_format="opus",
output_path=output_path,
)
async def convert_video_format(
video_path: str, output_format: str = "mp4", output_path: str | None = None
) -> str:
"""Convert a video file with ffmpeg.
Args:
video_path: Source video file path.
output_format: Target format, such as ``mp4``.
output_path: Optional output file path. When omitted, a temporary path is
created under AstrBot's temp directory.
Returns:
The converted video file path.
Raises:
Exception: Raised when ffmpeg is unavailable or conversion fails.
"""
# Return early when the source already appears to be in the target format.
if video_path.lower().endswith(f".{output_format}"):
return video_path
# Create an output path when the caller does not provide one.
if output_path is None:
temp_dir = get_astrbot_temp_path()
os.makedirs(temp_dir, exist_ok=True)
output_path = os.path.join(
temp_dir,
f"media_video_{uuid.uuid4().hex}.{output_format}",
)
try:
# Convert the video with ffmpeg.
process = await asyncio.create_subprocess_exec(
"ffmpeg",
"-y",
"-i",
video_path,
"-c:v",
"libx264",
"-c:a",
"aac",
output_path,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
# Remove a partial output file created by a failed ffmpeg run.
if output_path and os.path.exists(output_path):
try:
os.remove(output_path)
logger.debug(
"Removed failed %s output file: %s",
output_format,
output_path,
)
except OSError as e:
logger.warning(
"Failed to remove failed %s output file: %s",
output_format,
e,
)
error_msg = stderr.decode() if stderr else "unknown error"
logger.error("ffmpeg video conversion failed: %s", error_msg)
raise Exception(f"ffmpeg conversion failed: {error_msg}")
logger.debug(
"Video converted successfully: %s -> %s",
video_path,
output_path,
)
return output_path
except FileNotFoundError:
logger.error(
"ffmpeg is not installed or not in PATH. "
"Install ffmpeg: https://ffmpeg.org/"
)
raise Exception("ffmpeg not found")
except Exception as e:
logger.error("Error while converting video format: %s", e)
raise
async def convert_audio_format(
audio_path: str,
output_format: str = "amr",
output_path: str | None = None,
) -> str:
"""Convert an audio file to the requested format with ffmpeg.
Args:
audio_path: Source audio file path.
output_format: Target format, such as ``amr``, ``ogg``, ``opus``, or
``wav``.
output_path: Optional output file path. When omitted, a temporary path is
created under AstrBot's temp directory.
Returns:
The converted audio file path.
Raises:
Exception: Raised when ffmpeg is unavailable or conversion fails.
"""
if audio_path.lower().endswith(f".{output_format}"):
return audio_path
if output_path is None:
temp_dir = Path(get_astrbot_temp_path())
temp_dir.mkdir(parents=True, exist_ok=True)
output_path = str(temp_dir / f"media_audio_{uuid.uuid4().hex}.{output_format}")
args = ["ffmpeg", "-y", "-i", audio_path]
if output_format == "amr":
args.extend(
[
"-ac",
"1",
"-ar",
"8000",
"-ab",
"12.2k",
"-af",
(
"highpass=f=310:poles=2,"
"lowpass=f=3720:poles=2,"
"equalizer=f=3150:width_type=h:width=1000:g=7.5,"
"loudnorm=I=-18.5:TP=-1.5:LRA=6,"
"aresample=8000"
),
]
)
elif output_format == "ogg":
args.extend(["-acodec", "libopus", "-ac", "1", "-ar", "16000"])
elif output_format == "opus":
args.extend(["-acodec", "libopus", "-ac", "1", "-ar", "16000"])
args.append(output_path)
try:
process = await asyncio.create_subprocess_exec(
*args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
_, stderr = await process.communicate()
if process.returncode != 0:
if output_path and os.path.exists(output_path):
try:
os.remove(output_path)
except OSError as e:
logger.warning(
"Failed to remove failed audio output file: %s",
e,
)
error_msg = stderr.decode() if stderr else "unknown error"
raise Exception(f"ffmpeg conversion failed: {error_msg}")
logger.debug(
"Audio converted successfully: %s -> %s",
audio_path,
output_path,
)
return output_path
except FileNotFoundError:
raise Exception("ffmpeg not found")
async def convert_audio_to_amr(audio_path: str, output_path: str | None = None) -> str:
"""Convert an audio file to AMR format.
Args:
audio_path: Source audio file path.
output_path: Optional output file path. When omitted, a temporary path is
created under AstrBot's temp directory.
Returns:
The converted AMR file path.
"""
return await convert_audio_format(
audio_path=audio_path,
output_format="amr",
output_path=output_path,
)
async def convert_audio_to_wav(audio_path: str, output_path: str | None = None) -> str:
"""Convert an audio file to WAV format.
Args:
audio_path: Source audio file path.
output_path: Optional output file path. When omitted, a temporary path is
created under AstrBot's temp directory.
Returns:
The converted WAV file path.
"""
return await convert_audio_format(
audio_path=audio_path,
output_format="wav",
output_path=output_path,
)
async def ensure_wav(audio_path: str, output_path: str | None = None) -> str:
"""Ensure the audio path points to wav format by extension/guess and convert when needed.
If the file appears to already be WAV, return it directly to avoid extra
conversion. If the file does not exist yet, return the original path so
upstream retry logic can handle platform races.
Args:
audio_path: Local audio path to inspect and convert when needed.
output_path: Optional destination path. When omitted, conversion helpers
create a temporary file under AstrBot's temp directory.
Returns:
The original path when it is already WAV or unavailable; otherwise the
converted WAV path.
Raises:
Exception: Raised by the underlying conversion helper when conversion
fails.
"""
if not audio_path:
return audio_path
if not os.path.exists(audio_path):
# File not available yet (e.g. napcat race condition);
# return the path as-is so upstream retry logic can handle it later.
return audio_path
audio_type = _get_audio_magic_type(audio_path)
if audio_type == "wav":
return audio_path
if audio_type == "silk":
if output_path is None:
temp_dir = get_astrbot_temp_path()
os.makedirs(temp_dir, exist_ok=True)
output_path = os.path.join(temp_dir, f"media_audio_{uuid.uuid4().hex}.wav")
return await tencent_silk_to_wav(audio_path, output_path)
return await convert_audio_to_wav(audio_path, output_path)
async def ensure_jpeg(image_path: str, output_path: str | None = None) -> str:
"""Ensure the image path points to a JPEG file.
Args:
image_path: Local image path to inspect and convert when needed.
output_path: Optional destination path. When omitted, a temporary file under
AstrBot's temp directory is created for converted JPEG output.
Returns:
The original path when the source is already a JPEG file with a jpg/jpeg
suffix or cannot be found; otherwise the converted JPEG path.
Raises:
Exception: Raised by Pillow when the source file cannot be opened or saved as
an image.
"""
if not image_path:
return image_path
source_path = Path(image_path)
if not source_path.exists():
return image_path
with PILImage.open(source_path) as opened_img:
image_format = str(opened_img.format or "").upper()
if image_format == "JPEG" and source_path.suffix.lower() in {".jpg", ".jpeg"}:
return image_path
if output_path is None:
temp_dir = Path(get_astrbot_temp_path())
temp_dir.mkdir(parents=True, exist_ok=True)
output_path = str(temp_dir / f"media_image_{uuid.uuid4().hex}.jpg")
jpeg_output_path = output_path
def convert_image_to_jpeg() -> str:
converted_img: PILImage.Image | None = None
flattened_img: PILImage.Image | None = None
with PILImage.open(image_path) as opened_img:
try:
working_img: PILImage.Image = opened_img
if opened_img.mode in {"RGBA", "LA"} or (
opened_img.mode == "P" and "transparency" in opened_img.info
):
flattened = PILImage.new("RGB", opened_img.size, (255, 255, 255))
flattened_img = flattened
alpha_source = opened_img.convert("RGBA")
try:
flattened.paste(alpha_source, mask=alpha_source.getchannel("A"))
finally:
alpha_source.close()
working_img = flattened
elif opened_img.mode != "RGB":
converted_img = opened_img.convert("RGB")
working_img = converted_img
working_img.save(jpeg_output_path, "JPEG")
return jpeg_output_path
finally:
if converted_img is not None:
converted_img.close()
if flattened_img is not None:
flattened_img.close()
try:
return await asyncio.to_thread(convert_image_to_jpeg)
except Exception:
if output_path and os.path.exists(output_path):
try:
os.remove(output_path)
except OSError as e:
logger.warning(
"Failed to remove failed image output file: %s",
e,
)
raise
def _get_audio_magic_type(audio_path: str) -> str:
"""Detect common audio formats from magic bytes.
Args:
audio_path: Local audio path to inspect.
Returns:
A normalized format name such as ``wav``, ``mp3``, ``opus``, ``silk``, or
an empty string when the type cannot be detected.
"""
try:
with open(audio_path, "rb") as f:
header = f.read(64)
except FileNotFoundError:
logger.warning("WAV probe file not found: %s", audio_path)
return ""
except Exception as e:
logger.warning(
"WAV probe failed: %s, error: %s",
audio_path,
e,
)
return ""
if len(header) < 12:
return ""
if header[:4] == b"RIFF" and header[8:12] == b"WAVE":
return "wav"
if header[:4] == b"#!AM":
return "amr"
if header[:4] == b"OggS":
if b"OpusHead" in header:
return "opus"
return "ogg"
if header[:3] == b"fLa":
return "flac"
if header[:3] == b"ID3" or header[:2] == b"\xff\xfb":
return "mp3"
if header[:4] == b"ftyp" and b"mp4" in header[:8]:
return "mp4"
if header.startswith(b"#!SILK_V3"):
return "silk"
# Tencent SILK: leading \x02 byte before #!SILK_V3
if header.startswith(b"\x02#!SILK_V3"):
return "silk"
return ""
async def extract_video_cover(
video_path: str,
output_path: str | None = None,
) -> str:
"""Extract a JPEG cover frame from a video.
Args:
video_path: Source video file path.
output_path: Optional output image path. When omitted, a temporary JPEG
path is created under AstrBot's temp directory.
Returns:
The extracted JPEG cover path.
Raises:
Exception: Raised when ffmpeg is unavailable or cover extraction fails.
"""
if output_path is None:
temp_dir = Path(get_astrbot_temp_path())
temp_dir.mkdir(parents=True, exist_ok=True)
output_path = str(temp_dir / f"media_cover_{uuid.uuid4().hex}.jpg")
try:
process = await asyncio.create_subprocess_exec(
"ffmpeg",
"-y",
"-i",
video_path,
"-ss",
"00:00:00",
"-frames:v",
"1",
output_path,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
_, stderr = await process.communicate()
if process.returncode != 0:
if output_path and os.path.exists(output_path):
try:
os.remove(output_path)
except OSError as e:
logger.warning(
"Failed to remove failed video cover file: %s",
e,
)
error_msg = stderr.decode() if stderr else "unknown error"
raise Exception(f"ffmpeg extract cover failed: {error_msg}")
return output_path
except FileNotFoundError:
raise Exception("ffmpeg not found")
def _compress_image_sync(
data: bytes,
temp_dir: Path,
max_size: int,
quality: int,
optimize: bool,
) -> str:
"""Run image compression synchronously via ``asyncio.to_thread``."""
with PILImage.open(io.BytesIO(data)) as opened_img:
converted_img: PILImage.Image | None = None
try:
working_img = opened_img
if opened_img.mode != "RGB":
converted_img = opened_img.convert("RGB")
working_img = converted_img
assert working_img is not None
if max(working_img.size) > max_size:
working_img.thumbnail((max_size, max_size), PILImage.Resampling.LANCZOS)
new_uuid = uuid.uuid4().hex
save_path = temp_dir / f"compressed_{new_uuid}.jpg"
working_img.save(save_path, "JPEG", quality=quality, optimize=optimize)
logger.debug(f"Image compressed successfully: {save_path}")
return str(save_path)
finally:
if converted_img is not None:
converted_img.close()
async def compress_image(
url_or_path: str,
max_size: int = IMAGE_COMPRESS_DEFAULT_MAX_SIZE,
quality: int = IMAGE_COMPRESS_DEFAULT_QUALITY,
) -> str:
"""Compress large user-uploaded images.
Args:
url_or_path: Image path or URL.
max_size: Longest edge of the compressed image in pixels.
quality: JPEG output quality in the range 1-100.
Returns:
The compressed image path. Returns the original path if compression
fails or the source does not need compression.
"""
max_size = max(int(max_size), 1)
quality = min(max(int(quality), 1), 100)
optimize = IMAGE_COMPRESS_DEFAULT_OPTIMIZE
min_file_size_bytes = int(IMAGE_COMPRESS_DEFAULT_MIN_FILE_SIZE_MB * 1024 * 1024)
data = None
def _exceeds_max_size(source: bytes | Path) -> bool:
try:
fp = io.BytesIO(source) if isinstance(source, bytes) else source
with PILImage.open(fp) as opened_img:
return max(opened_img.size) > max_size
except Exception: # noqa: BLE001
return False
# Skip compression for remote images and return the original value.
if url_or_path.startswith("http"):
return url_or_path
elif url_or_path.startswith("data:image"):
_header, encoded = url_or_path.split(",", 1)
data = _decode_base64_payload(
encoded,
error_message="invalid image data URI payload",
)
if len(data) < min_file_size_bytes and not _exceeds_max_size(data):
return url_or_path
else:
local_path = Path(url_or_path)
if not local_path.exists():
return url_or_path
if local_path.stat().st_size < min_file_size_bytes and not _exceeds_max_size(
local_path
):
return url_or_path
with local_path.open("rb") as f:
data = f.read()
if not data:
return url_or_path
temp_dir = Path(get_astrbot_temp_path())
temp_dir.mkdir(parents=True, exist_ok=True)
# Offload the blocking image processing task to a thread.
return await asyncio.to_thread(
_compress_image_sync,
data,
temp_dir,
max_size,
quality,
optimize,
)