88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
"""Incoming / outgoing file helpers for MAX (aligned with Matrix workspace layout)."""
|
|
from __future__ import annotations
|
|
|
|
import mimetypes
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
|
|
def guess_upload_type(mime_type: str | None, *, attachment_type: str) -> str:
|
|
if attachment_type == "image":
|
|
return "image"
|
|
if attachment_type == "video":
|
|
return "video"
|
|
if attachment_type == "audio":
|
|
return "audio"
|
|
mime = mime_type or ""
|
|
if mime.startswith("image/"):
|
|
return "image"
|
|
if mime.startswith("video/"):
|
|
return "video"
|
|
if mime.startswith("audio/"):
|
|
return "audio"
|
|
return "file"
|
|
|
|
|
|
async def save_incoming_from_url(
|
|
*,
|
|
api: MaxBotApi,
|
|
workspace_root: Path,
|
|
filename: str,
|
|
url: str,
|
|
) -> str:
|
|
data = await api.download_file(url)
|
|
workspace_root.mkdir(parents=True, exist_ok=True)
|
|
relative_path, absolute_path = build_agent_workspace_path(
|
|
workspace_root=workspace_root,
|
|
filename=filename,
|
|
)
|
|
absolute_path.parent.mkdir(parents=True, exist_ok=True)
|
|
absolute_path.write_bytes(data)
|
|
return relative_path
|
|
|
|
|
|
async def upload_file_as_attachment(
|
|
api: MaxBotApi,
|
|
*,
|
|
filename: str,
|
|
content: bytes,
|
|
upload_type: str,
|
|
) -> dict:
|
|
meta = await api.get_upload_url(upload_type)
|
|
upload_url = meta.get("url")
|
|
token = meta.get("token")
|
|
if not isinstance(upload_url, str) or not upload_url:
|
|
raise RuntimeError("MAX uploads response missing url")
|
|
|
|
async with httpx.AsyncClient(timeout=httpx.Timeout(120.0)) as client:
|
|
response = await client.post(
|
|
upload_url,
|
|
files={"data": (filename, content, guess_mimetype(filename))},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
payload: dict = {}
|
|
if token:
|
|
payload["token"] = token
|
|
if upload_type == "image":
|
|
return {"type": "image", "payload": payload}
|
|
|
|
type_map = {
|
|
"file": "file",
|
|
"video": "video",
|
|
"audio": "audio",
|
|
}
|
|
mapped = type_map.get(upload_type, "file")
|
|
return {"type": mapped, "payload": payload}
|
|
|
|
|
|
def guess_mimetype(filename: str) -> str:
|
|
mime, _ = mimetypes.guess_type(filename)
|
|
return mime or "application/octet-stream"
|
|
|
|
|
|
def read_workspace_bytes(workspace_path: str | Path, *, agent_workspace: str) -> bytes:
|
|
root = Path(agent_workspace)
|
|
resolved = resolve_workspace_attachment_path(root, str(workspace_path))
|
|
return resolved.read_bytes()
|