mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-06 01:18:45 +00:00
2a98360105
Maintainer edit for PR 4115: rebase onto origin/main and split gateway HTTP routing from token, media, and workspace services so WebSocketChannel depends on explicit gateway services instead of GatewayHTTPHandler internals. Preserve file edit channel capabilities and restore tools.restrict_to_workspace wiring through ChannelManager.
93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
"""Media gateway services shared by WebUI HTTP routes and WebSocket frames."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from websockets.http11 import Request as WsRequest
|
|
from websockets.http11 import Response
|
|
|
|
from nanobot.config.paths import get_media_dir
|
|
from nanobot.webui.media_api import (
|
|
attach_signed_media_urls,
|
|
serve_signed_media,
|
|
sign_media_path,
|
|
sign_or_stage_media_path,
|
|
signed_media_attachments,
|
|
)
|
|
from nanobot.webui.transcript import rewrite_local_markdown_images
|
|
|
|
|
|
class WebUIMediaGateway:
|
|
"""Own media URL signing and WebUI markdown/media augmentation."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
workspace_path: Path,
|
|
logger: Any,
|
|
media_dir: Callable[[str | None], Path] | None = None,
|
|
secret: bytes | None = None,
|
|
) -> None:
|
|
self.workspace_path = workspace_path
|
|
self.logger = logger
|
|
self._media_dir = media_dir or (lambda channel=None: get_media_dir(channel))
|
|
self.secret = secret or secrets.token_bytes(32)
|
|
|
|
def serve_signed_media(
|
|
self,
|
|
sig: str,
|
|
payload: str,
|
|
*,
|
|
request: WsRequest | None = None,
|
|
) -> Response:
|
|
return serve_signed_media(
|
|
sig,
|
|
payload,
|
|
secret=self.secret,
|
|
request=request,
|
|
media_dir=self._media_dir,
|
|
)
|
|
|
|
def sign_media_path(self, abs_path: Path) -> str | None:
|
|
return sign_media_path(
|
|
abs_path,
|
|
secret=self.secret,
|
|
media_dir=self._media_dir,
|
|
)
|
|
|
|
def sign_or_stage_media_path(self, path: Path) -> dict[str, str] | None:
|
|
return sign_or_stage_media_path(
|
|
path,
|
|
secret=self.secret,
|
|
media_dir=self._media_dir,
|
|
logger=self.logger,
|
|
)
|
|
|
|
def rewrite_local_markdown_images(
|
|
self,
|
|
text: str,
|
|
*,
|
|
workspace_path: Path | None = None,
|
|
) -> str:
|
|
return rewrite_local_markdown_images(
|
|
text,
|
|
workspace_path=workspace_path or self.workspace_path,
|
|
sign_path=self.sign_or_stage_media_path,
|
|
)
|
|
|
|
def augment_media_urls(self, payload: dict[str, Any]) -> None:
|
|
attach_signed_media_urls(payload, sign_path=self.sign_media_path)
|
|
|
|
def augment_transcript_media(self, paths: list[str]) -> list[dict[str, Any]]:
|
|
return signed_media_attachments(
|
|
paths,
|
|
sign_path=self.sign_or_stage_media_path,
|
|
)
|
|
|
|
def augment_transcript_user_media(self, paths: list[str]) -> list[dict[str, Any]]:
|
|
return self.augment_transcript_media(paths)
|