|
6 | 6 | from collections.abc import Iterable
|
7 | 7 | from dataclasses import asdict, dataclass, fields
|
8 | 8 | from itertools import filterfalse
|
9 |
| -from typing import Any, BinaryIO |
| 9 | +from typing import Any |
10 | 10 |
|
11 | 11 | import regex
|
12 | 12 |
|
@@ -194,20 +194,25 @@ def get_encoded(value: str) -> bytes:
|
194 | 194 | def libzim_value(self) -> bytes:
|
195 | 195 | return self.get_libzim_value()
|
196 | 196 |
|
197 |
| - def get_binary_from(self, value: bytes | BinaryIO | io.BytesIO) -> bytes: |
198 |
| - if isinstance(value, BinaryIO): |
199 |
| - last_pos = value.tell() |
200 |
| - payload = value.read() |
201 |
| - value.seek(last_pos) |
202 |
| - value = payload |
203 |
| - elif isinstance(value, io.BytesIO): |
204 |
| - value = value.getvalue() |
| 197 | + def get_binary_from(self, value: bytes | io.IOBase | io.BytesIO) -> bytes: |
| 198 | + bvalue: bytes = b"" |
| 199 | + if isinstance(value, io.BytesIO): |
| 200 | + bvalue = value.getvalue() |
| 201 | + elif isinstance(value, bytes): |
| 202 | + bvalue = value |
| 203 | + else: |
| 204 | + last_pos: int |
| 205 | + if value.seekable(): |
| 206 | + last_pos = value.tell() |
| 207 | + bvalue = value.read() |
| 208 | + if value.seekable(): |
| 209 | + value.seek(last_pos) |
205 | 210 | if not self.empty_allowed and not value:
|
206 | 211 | raise ValueError("Missing value (empty not allowed)")
|
207 |
| - return value |
| 212 | + return bvalue |
208 | 213 |
|
209 | 214 | # native type is bytes
|
210 |
| - def get_cleaned_value(self, value: bytes | BinaryIO | io.BytesIO) -> bytes: |
| 215 | + def get_cleaned_value(self, value: bytes | io.IOBase | io.BytesIO) -> bytes: |
211 | 216 | return self.get_binary_from(value)
|
212 | 217 |
|
213 | 218 | def get_libzim_value(self) -> bytes:
|
@@ -316,12 +321,12 @@ class IllustrationBasedMetadata(Metadata):
|
316 | 321 | meta_mimetype = "image/png"
|
317 | 322 |
|
318 | 323 | def __init__(
|
319 |
| - self, value: bytes | BinaryIO | io.BytesIO, name: str | None = None |
| 324 | + self, value: bytes | io.IOBase | io.BytesIO, name: str | None = None |
320 | 325 | ) -> None:
|
321 | 326 | super().__init__(value=value, name=name)
|
322 | 327 |
|
323 | 328 | # native type is PNG image buffer
|
324 |
| - def get_cleaned_value(self, value: bytes | BinaryIO | io.BytesIO) -> bytes: |
| 329 | + def get_cleaned_value(self, value: bytes | io.IOBase | io.BytesIO) -> bytes: |
325 | 330 | value = self.get_binary_from(value)
|
326 | 331 | if not is_valid_image(
|
327 | 332 | image=value,
|
@@ -377,7 +382,7 @@ class IllustrationMetadata(IllustrationBasedMetadata):
|
377 | 382 | illustration_scale: int = 1
|
378 | 383 |
|
379 | 384 | def __init__(
|
380 |
| - self, value: bytes | BinaryIO | io.BytesIO, size: int, scale: int = 1 |
| 385 | + self, value: bytes | io.IOBase | io.BytesIO, size: int, scale: int = 1 |
381 | 386 | ) -> None:
|
382 | 387 | self.illustration_scale = scale
|
383 | 388 | self.illustration_size = size
|
@@ -465,7 +470,7 @@ def get_reserved_names(cls) -> list[str]:
|
465 | 470 |
|
466 | 471 | @x_protected
|
467 | 472 | class CustomMetadata(Metadata):
|
468 |
| - def __init__(self, name: str, value: bytes | BinaryIO | io.BytesIO) -> None: |
| 473 | + def __init__(self, name: str, value: bytes | io.IOBase | io.BytesIO) -> None: |
469 | 474 | self.meta_name = name
|
470 | 475 | super().__init__(value=value)
|
471 | 476 |
|
|
0 commit comments