119 lines
5.2 KiB
Python
119 lines
5.2 KiB
Python
import base64
|
|
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 read_bytes
|
|
from app.deps import DB, CurrentUser, has_perm
|
|
from app.f95_session import open_session, seal
|
|
from app.routers.games import make_cover
|
|
|
|
router = APIRouter(prefix="/api/metadata", tags=["metadata"])
|
|
|
|
|
|
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 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
|
|
auth_required: bool = False
|
|
metadata_available: bool = True
|
|
|
|
|
|
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)
|
|
metadata["cover_hash"] = None
|
|
if image := metadata.pop("image_url", None):
|
|
try:
|
|
metadata["cover_hash"] = make_cover(f95.fetch_image(image, metadata["url"], cookie, user_agent))
|
|
# Unsaved covers cannot be read through the protected /blobs route yet.
|
|
metadata["cover_preview"] = "data:image/webp;base64," + base64.b64encode(
|
|
read_bytes(metadata["cover_hash"])).decode("ascii")
|
|
except f95.F95AuthError as e:
|
|
metadata["auth_required"] = True
|
|
metadata["cover_warning"] = f"{e}. Остальные данные загружены."
|
|
except f95.F95Error as e:
|
|
metadata["cover_warning"] = f"{e}. Остальные данные загружены."
|
|
except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError):
|
|
metadata["cover_warning"] = "Не удалось загрузить превью F95zone. Остальные данные загружены."
|
|
else:
|
|
metadata["auth_required"] = not cookie or logged_in in ("false", "0")
|
|
metadata["cover_warning"] = "В теме не найдено доступное превью. Остальные данные загружены."
|
|
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))
|