177 lines
8.0 KiB
Python
177 lines
8.0 KiB
Python
import base64
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from urllib.parse import quote
|
||
|
||
from fastapi import APIRouter, HTTPException, status
|
||
from PIL import Image, UnidentifiedImageError
|
||
from pydantic import BaseModel, SecretStr
|
||
from starlette.concurrency import run_in_threadpool
|
||
|
||
from app import f95
|
||
from app.blobs import put_bytes
|
||
from app.deps import DB, CurrentUser, has_perm
|
||
from app.f95_session import open_session, seal
|
||
from app.routers.games import encode_cover
|
||
|
||
router = APIRouter(prefix="/api/metadata", tags=["metadata"])
|
||
MAX_GALLERY_IMAGES = 24
|
||
|
||
|
||
class F95In(BaseModel):
|
||
url: str
|
||
|
||
|
||
class F95SessionIn(BaseModel):
|
||
cookie: SecretStr = SecretStr("") # Compatibility with previously saved clients.
|
||
xf_user: SecretStr = SecretStr("")
|
||
xf_session: SecretStr = SecretStr("")
|
||
cf_clearance: SecretStr = SecretStr("")
|
||
user_agent: str = ""
|
||
|
||
|
||
class GalleryImage(BaseModel):
|
||
hash: str
|
||
preview: str
|
||
|
||
|
||
class MetaOut(BaseModel):
|
||
url: str
|
||
title: str = ""
|
||
description: str = ""
|
||
tags: list[str] = []
|
||
cover_hash: str | None = None
|
||
cover_preview: str | None = None
|
||
cover_warning: str | None = None
|
||
gallery: list[GalleryImage] = []
|
||
gallery_warning: str | None = None
|
||
auth_required: bool = False
|
||
metadata_available: bool = True
|
||
|
||
|
||
def import_cover(url: str, thread_url: str, cookie: str, user_agent: str,
|
||
max_size: tuple[int, int] = (640, 640)) -> tuple[str, str]:
|
||
data = f95.fetch_image(url, thread_url, cookie, user_agent)
|
||
try:
|
||
encoded = encode_cover(data) if max_size == (640, 640) else encode_cover(data, max_size)
|
||
preview = encoded if max_size == (640, 640) else encode_cover(encoded)
|
||
except UnidentifiedImageError as e:
|
||
if data[4:8] == b"ftyp" and (b"avif" in data[8:32] or b"avis" in data[8:32]):
|
||
raise f95.F95Error("F95zone вернул AVIF, который сервер не смог прочитать. Обновите зависимости backend") from e
|
||
raise f95.F95Error("Ответ F95zone не удалось распознать как изображение") from e
|
||
except Image.DecompressionBombError as e:
|
||
raise f95.F95Error("У превью F95zone слишком большое разрешение") from e
|
||
except (OSError, ValueError) as e:
|
||
raise f95.F95Error("Не удалось преобразовать изображение F95zone в обложку") from e
|
||
try:
|
||
cover_hash = put_bytes(encoded, ".webp")[0]
|
||
except OSError as e:
|
||
raise f95.F95Error("Не удалось сохранить обложку на сервере. Проверьте доступ к хранилищу") from e
|
||
# Unsaved covers cannot be read through the protected /blobs route yet.
|
||
return cover_hash, "data:image/webp;base64," + base64.b64encode(preview).decode("ascii")
|
||
|
||
|
||
def load_metadata(url: str, cookie: str = "", user_agent: str = "") -> dict:
|
||
url = f95.validate_url(url)
|
||
try:
|
||
metadata = f95.load(url, cookie, user_agent)
|
||
except f95.F95AuthError as e:
|
||
return {"url": url, "auth_required": True, "metadata_available": False, "cover_warning": str(e)}
|
||
logged_in = metadata.pop("logged_in", None)
|
||
gallery_urls = metadata.pop("gallery_urls", [])
|
||
metadata["cover_hash"] = None
|
||
if image := metadata.pop("image_url", None):
|
||
try:
|
||
metadata["cover_hash"], metadata["cover_preview"] = import_cover(image, metadata["url"], cookie, user_agent)
|
||
except f95.F95AuthError as e:
|
||
metadata["auth_required"] = True
|
||
metadata["cover_warning"] = f"{e}. Остальные данные загружены."
|
||
except f95.F95Error as e:
|
||
metadata["cover_warning"] = f"{e}. Остальные данные загружены."
|
||
else:
|
||
metadata["auth_required"] = not cookie or logged_in in ("false", "0")
|
||
metadata["cover_warning"] = "В теме не найдено доступное превью. Остальные данные загружены."
|
||
metadata["gallery"] = []
|
||
|
||
def load_screenshot(image_url: str):
|
||
try:
|
||
return import_cover(image_url, metadata["url"], cookie, user_agent, (1920, 1920))
|
||
except f95.F95Error as error:
|
||
return error
|
||
|
||
if gallery_urls:
|
||
failed = 0
|
||
first_error = ""
|
||
seen = {metadata["cover_hash"]}
|
||
with ThreadPoolExecutor(max_workers=4) as executor:
|
||
for result in executor.map(load_screenshot, gallery_urls[:MAX_GALLERY_IMAGES]):
|
||
if isinstance(result, f95.F95Error):
|
||
failed += 1
|
||
first_error = first_error or str(result)
|
||
if isinstance(result, f95.F95AuthError):
|
||
metadata["auth_required"] = True
|
||
else:
|
||
image_hash, preview = result
|
||
if image_hash not in seen:
|
||
seen.add(image_hash)
|
||
metadata["gallery"].append({"hash": image_hash, "preview": preview})
|
||
warnings = []
|
||
if failed:
|
||
warnings.append(f"Не удалось загрузить изображений галереи: {failed}. {first_error}.")
|
||
if len(gallery_urls) > MAX_GALLERY_IMAGES:
|
||
warnings.append(f"За один импорт загружается до {MAX_GALLERY_IMAGES} изображений галереи.")
|
||
if warnings:
|
||
metadata["gallery_warning"] = " ".join(warnings)
|
||
return metadata
|
||
|
||
|
||
def require_import(user: CurrentUser) -> None:
|
||
if not (has_perm(user, "games.upload") or has_perm(user, "games.edit")):
|
||
raise HTTPException(status.HTTP_403_FORBIDDEN, "Нет прав")
|
||
|
||
|
||
@router.put("/f95/session", status_code=204)
|
||
async def save_f95_session(body: F95SessionIn, user: CurrentUser, db: DB) -> None:
|
||
require_import(user)
|
||
values = {key: getattr(body, key).get_secret_value().strip()
|
||
for key in ("xf_user", "xf_session", "cf_clearance")}
|
||
if any(values.values()):
|
||
if not (values["xf_user"] or values["xf_session"]):
|
||
raise HTTPException(422, "Добавьте значение xf_user или xf_session для входа в F95zone")
|
||
if any(len(value) > 8192 or any(ord(c) < 33 or ord(c) > 126 or c == ";" for c in value)
|
||
for value in values.values()):
|
||
raise HTTPException(422, "Скопируйте только значение cookie, без имени и дополнительных строк")
|
||
# DevTools can display decoded values. Preserve existing escapes and encode delimiters.
|
||
cookie = "; ".join(f"{key}={quote(value, safe='%') if key.startswith('xf_') else value}"
|
||
for key, value in values.items() if value)
|
||
else:
|
||
cookie = body.cookie.get_secret_value().strip()
|
||
if cookie.lower().startswith("cookie:"):
|
||
cookie = cookie[7:].strip()
|
||
agent = body.user_agent.strip()
|
||
if (not cookie or len(cookie) > 16384 or len(agent) > 2048
|
||
or any(ord(c) < 32 or ord(c) > 126 for c in cookie + agent)
|
||
or not all("=" in part and part.strip().split("=", 1)[0] for part in cookie.split(";"))):
|
||
raise HTTPException(422, "Вставьте значение заголовка Cookie из запроса F95zone одной строкой")
|
||
user.f95_session = seal(user.id, cookie, agent)
|
||
await db.commit()
|
||
|
||
|
||
@router.delete("/f95/session", status_code=204)
|
||
async def delete_f95_session(user: CurrentUser, db: DB) -> None:
|
||
user.f95_session = None
|
||
await db.commit()
|
||
|
||
|
||
@router.post("/f95", response_model=MetaOut)
|
||
async def from_f95(body: F95In, user: CurrentUser, db: DB) -> dict:
|
||
require_import(user)
|
||
cookie, agent = open_session(user.id, user.f95_session)
|
||
try:
|
||
result = await run_in_threadpool(load_metadata, body.url, cookie, agent)
|
||
if result.get("auth_required") and user.f95_session:
|
||
user.f95_session = None
|
||
await db.commit()
|
||
return result
|
||
except f95.F95Error as e:
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, str(e))
|