209 lines
7.3 KiB
Python
209 lines
7.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
|
|
class UploadError(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class UploadResult:
|
|
host: str
|
|
file_path: Path
|
|
download_url: str | None
|
|
embed_url: str | None
|
|
file_code: str | None
|
|
raw: dict[str, Any]
|
|
|
|
|
|
class BaseUploader:
|
|
host_name = "base"
|
|
|
|
def upload_file(self, file_path: Path, credential: str) -> UploadResult:
|
|
raise NotImplementedError
|
|
|
|
@staticmethod
|
|
def _get_json(session: requests.Session, url: str, timeout: int = 45) -> dict[str, Any]:
|
|
response = session.get(url, timeout=timeout)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
if isinstance(data, dict) and data.get("status") in (401, 403, 429, 500):
|
|
raise UploadError(str(data.get("msg") or data.get("message") or data))
|
|
return data
|
|
|
|
|
|
class DoodstreamUploader(BaseUploader):
|
|
host_name = "doodstream.com"
|
|
|
|
def __init__(self, api_base: str = "https://doodapi.co") -> None:
|
|
self.api_base = api_base.rstrip("/")
|
|
|
|
def upload_file(self, file_path: Path, credential: str) -> UploadResult:
|
|
key = credential.strip()
|
|
if not key:
|
|
raise UploadError("Doodstream API Key fehlt.")
|
|
|
|
with requests.Session() as session:
|
|
server_data = self._get_json(session, f"{self.api_base}/api/upload/server?key={key}")
|
|
upload_url = (server_data.get("result") or "").strip()
|
|
if not upload_url:
|
|
raise UploadError("Kein Upload-Server von Doodstream erhalten.")
|
|
|
|
with file_path.open("rb") as stream:
|
|
files = {"file": (file_path.name, stream)}
|
|
form = {"api_key": key}
|
|
target = f"{upload_url}?{key}"
|
|
response = session.post(target, data=form, files=files, timeout=1800)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
|
|
item = None
|
|
result = payload.get("result")
|
|
if isinstance(result, list) and result:
|
|
item = result[0]
|
|
elif isinstance(result, dict):
|
|
item = result
|
|
else:
|
|
item = {}
|
|
|
|
return UploadResult(
|
|
host=self.host_name,
|
|
file_path=file_path,
|
|
download_url=item.get("download_url") or item.get("protected_dl"),
|
|
embed_url=item.get("protected_embed"),
|
|
file_code=item.get("filecode") or item.get("file_code"),
|
|
raw=payload,
|
|
)
|
|
|
|
|
|
class VoeUploader(BaseUploader):
|
|
host_name = "voe.sx"
|
|
|
|
def __init__(self, api_base: str = "https://voe.sx") -> None:
|
|
self.api_base = api_base.rstrip("/")
|
|
|
|
def upload_file(self, file_path: Path, credential: str) -> UploadResult:
|
|
key = credential.strip()
|
|
if not key:
|
|
raise UploadError("VOE API Key fehlt.")
|
|
|
|
with requests.Session() as session:
|
|
server_data = self._get_json(session, f"{self.api_base}/api/upload/server?key={key}")
|
|
upload_url = (server_data.get("result") or "").strip()
|
|
if not upload_url:
|
|
raise UploadError("Kein Upload-Server von VOE erhalten.")
|
|
|
|
with file_path.open("rb") as stream:
|
|
files = {"file": (file_path.name, stream)}
|
|
target = f"{upload_url}?key={key}"
|
|
response = session.post(target, files=files, timeout=1800)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
|
|
file_code = (
|
|
payload.get("file", {}).get("file_code")
|
|
if isinstance(payload.get("file"), dict)
|
|
else None
|
|
)
|
|
|
|
download = f"https://voe.sx/{file_code}" if file_code else None
|
|
embed = f"https://voe.sx/e/{file_code}" if file_code else None
|
|
|
|
return UploadResult(
|
|
host=self.host_name,
|
|
file_path=file_path,
|
|
download_url=download,
|
|
embed_url=embed,
|
|
file_code=file_code,
|
|
raw=payload,
|
|
)
|
|
|
|
|
|
class GenericApiUploader(BaseUploader):
|
|
"""Tries API style used by Dood/VOE clones."""
|
|
|
|
def __init__(self, host_name: str, base_url: str) -> None:
|
|
self.host_name = host_name
|
|
self.base_url = base_url.rstrip("/")
|
|
|
|
def _build_links(self, payload: dict[str, Any], file_code: str | None) -> tuple[str | None, str | None]:
|
|
if isinstance(payload.get("result"), dict):
|
|
result = payload["result"]
|
|
return (
|
|
result.get("download_url") or result.get("url") or result.get("protected_download"),
|
|
result.get("embed_url") or result.get("protected_embed"),
|
|
)
|
|
if isinstance(payload.get("result"), list) and payload["result"]:
|
|
item = payload["result"][0]
|
|
if isinstance(item, dict):
|
|
return (
|
|
item.get("download_url") or item.get("url") or item.get("protected_download"),
|
|
item.get("embed_url") or item.get("protected_embed"),
|
|
)
|
|
if file_code:
|
|
return (f"{self.base_url}/{file_code}", f"{self.base_url}/e/{file_code}")
|
|
return (None, None)
|
|
|
|
def upload_file(self, file_path: Path, credential: str) -> UploadResult:
|
|
key = credential.strip()
|
|
if not key:
|
|
raise UploadError(f"{self.host_name}: API Key fehlt.")
|
|
|
|
with requests.Session() as session:
|
|
candidates = [
|
|
f"{self.base_url}/api/upload/server?key={key}",
|
|
f"{self.base_url}/api/v1/upload/server?key={key}",
|
|
]
|
|
server_data: dict[str, Any] | None = None
|
|
for url in candidates:
|
|
try:
|
|
server_data = self._get_json(session, url)
|
|
if server_data.get("result"):
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
if not server_data or not server_data.get("result"):
|
|
raise UploadError(
|
|
f"{self.host_name}: Kein kompatibler API-Endpunkt gefunden. "
|
|
"Bitte API-Doku/Key pruefen."
|
|
)
|
|
|
|
upload_url = str(server_data["result"]).strip()
|
|
with file_path.open("rb") as stream:
|
|
files = {"file": (file_path.name, stream)}
|
|
data = {"api_key": key, "key": key}
|
|
response = session.post(f"{upload_url}?key={key}", data=data, files=files, timeout=1800)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
|
|
file_code = None
|
|
if isinstance(payload.get("file"), dict):
|
|
file_code = payload["file"].get("file_code")
|
|
if not file_code and isinstance(payload.get("result"), dict):
|
|
file_code = payload["result"].get("filecode") or payload["result"].get("file_code")
|
|
|
|
download, embed = self._build_links(payload, file_code)
|
|
return UploadResult(
|
|
host=self.host_name,
|
|
file_path=file_path,
|
|
download_url=download,
|
|
embed_url=embed,
|
|
file_code=file_code,
|
|
raw=payload,
|
|
)
|
|
|
|
|
|
def build_uploaders() -> dict[str, BaseUploader]:
|
|
return {
|
|
"doodstream.com": DoodstreamUploader(),
|
|
"voe.sx": VoeUploader(),
|
|
"vidmoly.me": GenericApiUploader("vidmoly.me", "https://vidmoly.me"),
|
|
"byse.sx": GenericApiUploader("byse.sx", "https://byse.sx"),
|
|
}
|