Compare commits
No commits in common. "26559dd2707d92fc8f1740befb97ed0c32555e65" and "5c0cec74d046f66d31a06852b87c1b50ef709fe9" have entirely different histories.
26559dd270
...
5c0cec74d0
1 changed files with 41 additions and 41 deletions
|
|
@ -1,10 +1,11 @@
|
||||||
import aiohttp
|
import requests
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Union
|
from typing import Union
|
||||||
|
|
||||||
|
|
||||||
class ZeroXZero:
|
class ZeroXZero:
|
||||||
"""
|
"""
|
||||||
An async utility class for interacting with the 0x0.st file hosting service.
|
A static utility class for interacting with the 0x0.st file hosting service.
|
||||||
|
|
||||||
Features:
|
Features:
|
||||||
- Upload files from disk
|
- Upload files from disk
|
||||||
|
|
@ -13,55 +14,54 @@ class ZeroXZero:
|
||||||
- Check service availability
|
- Check service availability
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
ENDPOINT_URL = "https://0x0.st"
|
||||||
HEADERS = {
|
HEADERS = {
|
||||||
"User-Agent": "pyzxz-uploader/1.0 (https://github.com/yourrepo)"
|
"User-Agent": "pyzxz-uploader/1.0 (https://github.com/yourrepo)"
|
||||||
}
|
}
|
||||||
|
|
||||||
def __init__(self, endpoint_url: str = "https://0x0.st"):
|
@staticmethod
|
||||||
self.endpoint_url = endpoint_url
|
def upload(file_path: Union[str, Path]) -> str:
|
||||||
|
|
||||||
async def upload(self, file_path: Union[str, Path]) -> str:
|
|
||||||
file_path = Path(file_path)
|
file_path = Path(file_path)
|
||||||
|
|
||||||
if not file_path.exists():
|
if not file_path.exists():
|
||||||
raise FileNotFoundError(f"No such file: {file_path}")
|
raise FileNotFoundError(f"No such file: {file_path}")
|
||||||
|
|
||||||
async with aiohttp.ClientSession(headers=self.HEADERS) as session:
|
with file_path.open("rb") as f:
|
||||||
with file_path.open("rb") as f:
|
response = requests.post(
|
||||||
data = aiohttp.FormData()
|
ZeroXZero.ENDPOINT_URL,
|
||||||
data.add_field("file", f, filename=file_path.name)
|
files={"file": f},
|
||||||
async with session.post(
|
headers=ZeroXZero.HEADERS
|
||||||
self.endpoint_url,
|
)
|
||||||
data=data
|
|
||||||
) as response:
|
|
||||||
text = await response.text()
|
|
||||||
if response.status == 200 and text.startswith("https://"):
|
|
||||||
return text.strip()
|
|
||||||
raise ValueError(f"Upload failed: {text.strip()}")
|
|
||||||
|
|
||||||
async def upload_from_bytes(self, data: bytes, filename: str) -> str:
|
if response.ok and response.text.startswith("https://"):
|
||||||
async with aiohttp.ClientSession(headers=self.HEADERS) as session:
|
return response.text.strip()
|
||||||
form = aiohttp.FormData()
|
raise ValueError(f"Upload failed: {response.text.strip()}")
|
||||||
form.add_field("file", data, filename=filename)
|
|
||||||
async with session.post(
|
|
||||||
self.endpoint_url,
|
|
||||||
data=form
|
|
||||||
) as response:
|
|
||||||
text = await response.text()
|
|
||||||
if response.status == 200 and text.startswith("https://"):
|
|
||||||
return text.strip()
|
|
||||||
raise ValueError(f"Upload failed: {text.strip()}")
|
|
||||||
|
|
||||||
async def upload_text(self, text: str, filename: str = "text.txt") -> str:
|
@staticmethod
|
||||||
return await self.upload_from_bytes(text.encode("utf-8"), filename)
|
def upload_from_bytes(data: bytes, filename: str) -> str:
|
||||||
|
files = {"file": (filename, data)}
|
||||||
|
response = requests.post(
|
||||||
|
ZeroXZero.ENDPOINT_URL,
|
||||||
|
files=files,
|
||||||
|
headers=ZeroXZero.HEADERS
|
||||||
|
)
|
||||||
|
|
||||||
async def is_available(self) -> bool:
|
if response.ok and response.text.startswith("https://"):
|
||||||
|
return response.text.strip()
|
||||||
|
raise ValueError(f"Upload failed: {response.text.strip()}")
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def upload_text(text: str, filename: str = "text.txt") -> str:
|
||||||
|
return ZeroXZero.upload_from_bytes(text.encode("utf-8"), filename)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def is_available() -> bool:
|
||||||
try:
|
try:
|
||||||
async with aiohttp.ClientSession(headers=self.HEADERS) as session:
|
response = requests.head(
|
||||||
async with session.head(
|
ZeroXZero.ENDPOINT_URL,
|
||||||
self.endpoint_url,
|
timeout=3,
|
||||||
timeout=aiohttp.ClientTimeout(total=3)
|
headers=ZeroXZero.HEADERS
|
||||||
) as response:
|
)
|
||||||
return response.status == 200
|
return response.status_code == 200
|
||||||
except Exception:
|
except requests.RequestException:
|
||||||
return False
|
return False
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue