| import requests | |
| from typing import * | |
| __all__ = ["WebFile"] | |
| class WebFile: | |
| def __init__(self, url: str, session: Optional[requests.Session] = None, headers: Optional[Dict[str, str]] = None, size: Optional[int] = None): | |
| self.url = url | |
| self.session = session or requests.Session() | |
| self.session.headers.update(headers or {}) | |
| self._offset = 0 | |
| self.size = size if size is not None else self._fetch_size() | |
| def _fetch_size(self): | |
| with self.session.get(self.url, stream=True) as response: | |
| response.raise_for_status() | |
| content_length = response.headers.get("Content-Length") | |
| if content_length is None: | |
| raise ValueError("Missing Content-Length in header") | |
| return int(content_length) | |
| def _fetch_data(self, offset: int, n: int) -> bytes: | |
| headers = {"Range": f"bytes={offset}-{min(offset + n - 1, self.size)}"} | |
| response = self.session.get(self.url, headers=headers) | |
| response.raise_for_status() | |
| return response.content | |
| def seekable(self) -> bool: | |
| return True | |
| def tell(self) -> int: | |
| return self._offset | |
| def available(self) -> int: | |
| return self.size - self._offset | |
| def seek(self, offset: int, whence: int = 0) -> None: | |
| if whence == 0: | |
| new_offset = offset | |
| elif whence == 1: | |
| new_offset = self._offset + offset | |
| elif whence == 2: | |
| new_offset = self.size + offset | |
| else: | |
| raise ValueError("Invalid value for whence") | |
| self._offset = max(0, min(new_offset, self.size)) | |
| def read(self, n: Optional[int] = None) -> bytes: | |
| if n is None or n < 0: | |
| n = self.available() | |
| else: | |
| n = min(n, self.available()) | |
| if n == 0: | |
| return b'' | |
| data = self._fetch_data(self._offset, n) | |
| self._offset += len(data) | |
| return data | |
| def close(self) -> None: | |
| pass | |
| def __enter__(self): | |
| return self | |
| def __exit__(self, exc_type, exc_value, traceback): | |
| pass | |