import io import mimetypes import socket import threading import uuid from pathlib import Path from typing import Callable, Dict, Optional import requests from requests.adapters import HTTPAdapter class UploadCancelledException(Exception): """Raised when an ongoing video upload is cancelled by the user.""" pass class FastHTTPAdapter(HTTPAdapter): """ High-throughput HTTPAdapter for multi-gigabit uploads (1-2 Gbps). - Sets socket SO_SNDBUF to 4 MB (instead of default 64 KB) to allow large TCP window scaling. - Enables TCP_NODELAY (disables Nagle's algorithm). - Sets HTTPConnection.blocksize to 1 MB (instead of default 16 KB) to reduce syscalls and GIL contention by 64x. """ def __init__( self, blocksize: int = 1024 * 1024, # 1 MB read blocks sndbuf: int = 4 * 1024 * 1024, # 4 MB socket send buffer **kwargs, ): self.custom_blocksize = blocksize self.custom_sndbuf = sndbuf super().__init__(**kwargs) def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs): sock_opts = [ (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1), (socket.SOL_SOCKET, socket.SO_SNDBUF, self.custom_sndbuf), ] pool_kwargs["blocksize"] = self.custom_blocksize pool_kwargs["socket_options"] = sock_opts return super().init_poolmanager(connections, maxsize, block=block, **pool_kwargs) def create_fast_session(blocksize_mb: int = 1, sndbuf_mb: int = 4) -> requests.Session: """Creates a requests.Session configured with high-performance TCP buffers.""" session = requests.Session() adapter = FastHTTPAdapter( blocksize=max(1, blocksize_mb) * 1024 * 1024, sndbuf=max(1, sndbuf_mb) * 1024 * 1024, ) session.mount("https://", adapter) session.mount("http://", adapter) return session class DiskStreamingMultipart: """ Zero-memory streaming multipart/form-data request body. Streams files of any size (even 10GB - 50GB+) directly from disk in large chunks without copying or buffering the file in RAM. Supports instant cancellation via cancel_event. """ def __init__( self, fields: Dict[str, str], file_field_name: str, file_path: Path, filename: Optional[str] = None, mime_type: Optional[str] = None, chunk_size: int = 16 * 1024 * 1024, # 16 MB chunks progress_callback: Optional[Callable[[float], None]] = None, cancel_event: Optional[threading.Event] = None, ): self.boundary = f"----WebKitFormBoundary{uuid.uuid4().hex}" self.chunk_size = chunk_size self.file_path = Path(file_path) if not self.file_path.is_file(): raise FileNotFoundError(f"File not found: {self.file_path}") self.file_size = self.file_path.stat().st_size self.filename = filename or self.file_path.name if not mime_type: guessed, _ = mimetypes.guess_type(self.filename) mime_type = guessed or "application/octet-stream" self.mime_type = mime_type self.progress_callback = progress_callback self.cancel_event = cancel_event self.bytes_read = 0 # Optimal disk buffer size capped by configured max RAM (default: 128 MB) max_ram_bytes = 128 * 1024 * 1024 try: from database import get_setting from config import DEFAULT_MAX_DISK_BUFFER_MB ram_setting = get_setting("max_disk_buffer_mb") mb = int(ram_setting) if ram_setting and ram_setting.isdigit() else DEFAULT_MAX_DISK_BUFFER_MB max_ram_bytes = max(8, mb) * 1024 * 1024 except Exception: pass # Clamp chunk_size so single or concurrent stream memory never exceeds RAM cap self.chunk_size = min(chunk_size, max_ram_bytes) # Build prefix bytes (all metadata fields + file header) prefix_parts = [] for name, value in fields.items(): prefix_parts.append( f"--{self.boundary}\r\n" f'Content-Disposition: form-data; name="{name}"\r\n\r\n' f"{value}\r\n" ) prefix_parts.append( f"--{self.boundary}\r\n" f'Content-Disposition: form-data; name="{file_field_name}"; filename="{self.filename}"\r\n' f"Content-Type: {self.mime_type}\r\n\r\n" ) self.prefix_bytes = "".join(prefix_parts).encode("utf-8") self.suffix_bytes = f"\r\n--{self.boundary}--\r\n".encode("utf-8") self.total_size = len(self.prefix_bytes) + self.file_size + len(self.suffix_bytes) # File handle & stream buffers (optimized I/O buffering capped to 8 MB or chunk size) io_buffering = max(64 * 1024, min(self.chunk_size, 8 * 1024 * 1024)) self._prefix_io: Optional[io.BytesIO] = io.BytesIO(self.prefix_bytes) self._file_handle = open(self.file_path, "rb", buffering=io_buffering) self._suffix_io: Optional[io.BytesIO] = io.BytesIO(self.suffix_bytes) @property def content_type(self) -> str: return f"multipart/form-data; boundary={self.boundary}" @property def headers(self) -> Dict[str, str]: return { "Content-Type": self.content_type, "Content-Length": str(self.total_size), } def __len__(self) -> int: return self.total_size def read(self, size: int = -1) -> bytes: # Check cancellation before reading if self.cancel_event and self.cancel_event.is_set(): self.close() raise UploadCancelledException("A videó feltöltése megszakítva a felhasználó által.") if size is None or size < 0: size = self.chunk_size # 1. Read header prefix if self._prefix_io is not None: chunk = self._prefix_io.read(size) if chunk: return chunk self._prefix_io = None # 2. Read directly from disk file in chunks if self._file_handle is not None and not self._file_handle.closed: chunk = self._file_handle.read(size) if chunk: self.bytes_read += len(chunk) if self.progress_callback and self.file_size > 0: pct = 5.0 + (self.bytes_read / self.file_size) * 85.0 self.progress_callback(min(pct, 90.0)) return chunk else: self._file_handle.close() # 3. Read footer suffix if self._suffix_io is not None: chunk = self._suffix_io.read(size) if chunk: return chunk self._suffix_io = None return b"" def close(self): if self._file_handle is not None and not self._file_handle.closed: try: self._file_handle.close() except Exception: pass