Spaces:
Runtime error
Runtime error
File size: 4,104 Bytes
0a1b571 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
import random
from enum import IntEnum
from io import BytesIO
from typing import Any, Optional, overload
from httpx import HTTPError
from hibiapi.api.sauce.constants import SauceConstants
from hibiapi.utils.decorators import enum_auto_doc
from hibiapi.utils.exceptions import ClientSideException
from hibiapi.utils.net import catch_network_error
from hibiapi.utils.routing import BaseEndpoint, BaseHostUrl
class UnavailableSourceException(ClientSideException):
code = 422
detail = "given image is not avaliable to fetch"
class ImageSourceOversizedException(UnavailableSourceException):
code = 413
detail = (
"given image size is rather than maximum limit "
f"{SauceConstants.IMAGE_MAXIMUM_SIZE} bytes"
)
class HostUrl(BaseHostUrl):
allowed_hosts = SauceConstants.IMAGE_ALLOWED_HOST
class UploadFileIO(BytesIO):
@classmethod
def __get_validators__(cls):
yield cls.validate
@classmethod
def validate(cls, v: Any) -> BytesIO:
if not isinstance(v, BytesIO):
raise ValueError(f"Expected UploadFile, received: {type(v)}")
return v
@enum_auto_doc
class DeduplicateType(IntEnum):
DISABLED = 0
"""no result deduplicating"""
IDENTIFIER = 1
"""consolidate search results and deduplicate by item identifier"""
ALL = 2
"""all implemented deduplicate methods such as by series name"""
class SauceEndpoint(BaseEndpoint, cache_endpoints=False):
base = "https://saucenao.com"
async def fetch(self, host: HostUrl) -> UploadFileIO:
try:
response = await self.client.get(
url=host,
headers=SauceConstants.IMAGE_HEADERS,
timeout=SauceConstants.IMAGE_TIMEOUT,
)
response.raise_for_status()
if len(response.content) > SauceConstants.IMAGE_MAXIMUM_SIZE:
raise ImageSourceOversizedException
return UploadFileIO(response.content)
except HTTPError as e:
raise UnavailableSourceException(detail=str(e)) from e
@catch_network_error
async def request(
self, *, file: UploadFileIO, params: dict[str, Any]
) -> dict[str, Any]:
response = await self.client.post(
url=self._join(
self.base,
"search.php",
params={
**params,
"api_key": random.choice(SauceConstants.API_KEY),
"output_type": 2,
},
),
files={"file": file},
)
if response.status_code >= 500:
response.raise_for_status()
return response.json()
@overload
async def search(
self,
*,
url: HostUrl,
size: int = 30,
deduplicate: DeduplicateType = DeduplicateType.ALL,
database: Optional[int] = None,
enabled_mask: Optional[int] = None,
disabled_mask: Optional[int] = None,
) -> dict[str, Any]:
...
@overload
async def search(
self,
*,
file: UploadFileIO,
size: int = 30,
deduplicate: DeduplicateType = DeduplicateType.ALL,
database: Optional[int] = None,
enabled_mask: Optional[int] = None,
disabled_mask: Optional[int] = None,
) -> dict[str, Any]:
...
async def search(
self,
*,
url: Optional[HostUrl] = None,
file: Optional[UploadFileIO] = None,
size: int = 30,
deduplicate: DeduplicateType = DeduplicateType.ALL,
database: Optional[int] = None,
enabled_mask: Optional[int] = None,
disabled_mask: Optional[int] = None,
):
if url is not None:
file = await self.fetch(url)
assert file is not None
return await self.request(
file=file,
params={
"dbmask": enabled_mask,
"dbmaski": disabled_mask,
"db": database,
"numres": size,
"dedupe": deduplicate,
},
)
|