73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
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
|
|
|
|
|