| 5 | |
| 6 | |
| 7 | class WebFile: |
| 8 | def __init__(self, url: str, session: Optional[requests.Session] = None, headers: Optional[Dict[str, str]] = None, size: Optional[int] = None): |
| 9 | self.url = url |
| 10 | self.session = session or requests.Session() |
| 11 | self.session.headers.update(headers or {}) |
| 12 | self._offset = 0 |
| 13 | self.size = size if size is not None else self._fetch_size() |
| 14 | |
| 15 | def _fetch_size(self): |
| 16 | with self.session.get(self.url, stream=True) as response: |
| 17 | response.raise_for_status() |
| 18 | content_length = response.headers.get("Content-Length") |
| 19 | if content_length is None: |
| 20 | raise ValueError("Missing Content-Length in header") |
| 21 | return int(content_length) |
| 22 | |
| 23 | def _fetch_data(self, offset: int, n: int) -> bytes: |
| 24 | headers = {"Range": f"bytes={offset}-{min(offset + n - 1, self.size)}"} |
| 25 | response = self.session.get(self.url, headers=headers) |
| 26 | response.raise_for_status() |
| 27 | return response.content |
| 28 | |
| 29 | def seekable(self) -> bool: |
| 30 | return True |
| 31 | |
| 32 | def tell(self) -> int: |
| 33 | return self._offset |
| 34 | |
| 35 | def available(self) -> int: |
| 36 | return self.size - self._offset |
| 37 | |
| 38 | def seek(self, offset: int, whence: int = 0) -> None: |
| 39 | if whence == 0: |
| 40 | new_offset = offset |
| 41 | elif whence == 1: |
| 42 | new_offset = self._offset + offset |
| 43 | elif whence == 2: |
| 44 | new_offset = self.size + offset |
| 45 | else: |
| 46 | raise ValueError("Invalid value for whence") |
| 47 | |
| 48 | self._offset = max(0, min(new_offset, self.size)) |
| 49 | |
| 50 | def read(self, n: Optional[int] = None) -> bytes: |
| 51 | if n is None or n < 0: |
| 52 | n = self.available() |
| 53 | else: |
| 54 | n = min(n, self.available()) |
| 55 | |
| 56 | if n == 0: |
| 57 | return b'' |
| 58 | |
| 59 | data = self._fetch_data(self._offset, n) |
| 60 | self._offset += len(data) |
| 61 | |
| 62 | return data |
| 63 | |
| 64 | def close(self) -> None: |