mirror of
				https://github.com/zulip/zulip.git
				synced 2025-11-04 05:53:43 +00:00 
			
		
		
		
	
		
			
				
	
	
		
			137 lines
		
	
	
		
			3.9 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			137 lines
		
	
	
		
			3.9 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
import os
 | 
						|
from collections.abc import Callable, Iterator
 | 
						|
from datetime import datetime
 | 
						|
from typing import IO, Any, BinaryIO
 | 
						|
 | 
						|
from zerver.models import Realm, UserProfile
 | 
						|
 | 
						|
INLINE_MIME_TYPES = [
 | 
						|
    "application/pdf",
 | 
						|
    "audio/aac",
 | 
						|
    "audio/flac",
 | 
						|
    "audio/mp4",
 | 
						|
    "audio/mpeg",
 | 
						|
    "audio/wav",
 | 
						|
    "audio/webm",
 | 
						|
    "image/apng",
 | 
						|
    "image/avif",
 | 
						|
    "image/gif",
 | 
						|
    "image/jpeg",
 | 
						|
    "image/png",
 | 
						|
    "image/webp",
 | 
						|
    "video/mp4",
 | 
						|
    "video/webm",
 | 
						|
    # To avoid cross-site scripting attacks, DO NOT add types such
 | 
						|
    # as application/xhtml+xml, application/x-shockwave-flash,
 | 
						|
    # image/svg+xml, text/html, or text/xml.
 | 
						|
]
 | 
						|
 | 
						|
 | 
						|
class ZulipUploadBackend:
 | 
						|
    # Message attachment uploads
 | 
						|
    def get_public_upload_root_url(self) -> str:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def generate_message_upload_path(self, realm_id: str, uploaded_file_name: str) -> str:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def upload_message_attachment(
 | 
						|
        self,
 | 
						|
        path_id: str,
 | 
						|
        filename: str,
 | 
						|
        content_type: str,
 | 
						|
        file_data: bytes,
 | 
						|
        user_profile: UserProfile | None,
 | 
						|
    ) -> None:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def save_attachment_contents(self, path_id: str, filehandle: BinaryIO) -> None:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def delete_message_attachment(self, path_id: str) -> bool:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def delete_message_attachments(self, path_ids: list[str]) -> None:
 | 
						|
        for path_id in path_ids:
 | 
						|
            self.delete_message_attachment(path_id)
 | 
						|
 | 
						|
    def all_message_attachments(
 | 
						|
        self, include_thumbnails: bool = False
 | 
						|
    ) -> Iterator[tuple[str, datetime]]:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    # Avatar image uploads
 | 
						|
    def get_avatar_url(self, hash_key: str, medium: bool = False) -> str:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def get_avatar_contents(self, file_path: str) -> tuple[bytes, str]:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def get_avatar_path(self, hash_key: str, medium: bool = False) -> str:
 | 
						|
        if medium:
 | 
						|
            return f"{hash_key}-medium.png"
 | 
						|
        else:
 | 
						|
            return f"{hash_key}.png"
 | 
						|
 | 
						|
    def upload_single_avatar_image(
 | 
						|
        self,
 | 
						|
        file_path: str,
 | 
						|
        *,
 | 
						|
        user_profile: UserProfile,
 | 
						|
        image_data: bytes,
 | 
						|
        content_type: str | None,
 | 
						|
        future: bool = True,
 | 
						|
    ) -> None:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def delete_avatar_image(self, path_id: str) -> None:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    # Realm icon and logo uploads
 | 
						|
    def realm_avatar_and_logo_path(self, realm: Realm) -> str:
 | 
						|
        return os.path.join(str(realm.id), "realm")
 | 
						|
 | 
						|
    def get_realm_icon_url(self, realm_id: int, version: int) -> str:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def upload_realm_icon_image(
 | 
						|
        self, icon_file: IO[bytes], user_profile: UserProfile, content_type: str
 | 
						|
    ) -> None:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def get_realm_logo_url(self, realm_id: int, version: int, night: bool) -> str:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def upload_realm_logo_image(
 | 
						|
        self, logo_file: IO[bytes], user_profile: UserProfile, night: bool, content_type: str
 | 
						|
    ) -> None:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    # Realm emoji uploads
 | 
						|
    def get_emoji_url(self, emoji_file_name: str, realm_id: int, still: bool = False) -> str:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def upload_single_emoji_image(
 | 
						|
        self,
 | 
						|
        path: str,
 | 
						|
        content_type: str | None,
 | 
						|
        user_profile: UserProfile,
 | 
						|
        image_data: bytes,
 | 
						|
    ) -> None:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    # Export tarballs
 | 
						|
    def get_export_tarball_url(self, realm: Realm, export_path: str) -> str:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def upload_export_tarball(
 | 
						|
        self,
 | 
						|
        realm: Realm,
 | 
						|
        tarball_path: str,
 | 
						|
        percent_callback: Callable[[Any], None] | None = None,
 | 
						|
    ) -> str:
 | 
						|
        raise NotImplementedError
 | 
						|
 | 
						|
    def delete_export_tarball(self, export_path: str) -> str | None:
 | 
						|
        raise NotImplementedError
 |