Compare commits
2 commits
5c0cec74d0
...
26559dd270
| Author | SHA1 | Date | |
|---|---|---|---|
| 26559dd270 | |||
| d751c5ab45 |
1 changed files with 41 additions and 41 deletions
|
|
@ -1,11 +1,10 @@
|
||||||
import requests
|
import aiohttp
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Union
|
from typing import Union
|
||||||
|
|
||||||
|
|
||||||
class ZeroXZero:
|
class ZeroXZero:
|
||||||
"""
|
"""
|
||||||
A static utility class for interacting with the 0x0.st file hosting service.
|
An async utility class for interacting with the 0x0.st file hosting service.
|
||||||
|
|
||||||
Features:
|
Features:
|
||||||
- Upload files from disk
|
- Upload files from disk
|
||||||
|
|
@ -14,54 +13,55 @@ 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)"
|
||||||
}
|
}
|
||||||
|
|
||||||
@staticmethod
|
def __init__(self, endpoint_url: str = "https://0x0.st"):
|
||||||
def upload(file_path: Union[str, Path]) -> str:
|
self.endpoint_url = endpoint_url
|
||||||
|
|
||||||
|
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}")
|
||||||
|
|
||||||
with file_path.open("rb") as f:
|
async with aiohttp.ClientSession(headers=self.HEADERS) as session:
|
||||||
response = requests.post(
|
with file_path.open("rb") as f:
|
||||||
ZeroXZero.ENDPOINT_URL,
|
data = aiohttp.FormData()
|
||||||
files={"file": f},
|
data.add_field("file", f, filename=file_path.name)
|
||||||
headers=ZeroXZero.HEADERS
|
async with session.post(
|
||||||
)
|
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()}")
|
||||||
|
|
||||||
if response.ok and response.text.startswith("https://"):
|
async def upload_from_bytes(self, data: bytes, filename: str) -> str:
|
||||||
return response.text.strip()
|
async with aiohttp.ClientSession(headers=self.HEADERS) as session:
|
||||||
raise ValueError(f"Upload failed: {response.text.strip()}")
|
form = aiohttp.FormData()
|
||||||
|
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()}")
|
||||||
|
|
||||||
@staticmethod
|
async def upload_text(self, text: str, filename: str = "text.txt") -> str:
|
||||||
def upload_from_bytes(data: bytes, filename: str) -> str:
|
return await self.upload_from_bytes(text.encode("utf-8"), filename)
|
||||||
files = {"file": (filename, data)}
|
|
||||||
response = requests.post(
|
|
||||||
ZeroXZero.ENDPOINT_URL,
|
|
||||||
files=files,
|
|
||||||
headers=ZeroXZero.HEADERS
|
|
||||||
)
|
|
||||||
|
|
||||||
if response.ok and response.text.startswith("https://"):
|
async def is_available(self) -> bool:
|
||||||
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:
|
||||||
response = requests.head(
|
async with aiohttp.ClientSession(headers=self.HEADERS) as session:
|
||||||
ZeroXZero.ENDPOINT_URL,
|
async with session.head(
|
||||||
timeout=3,
|
self.endpoint_url,
|
||||||
headers=ZeroXZero.HEADERS
|
timeout=aiohttp.ClientTimeout(total=3)
|
||||||
)
|
) as response:
|
||||||
return response.status_code == 200
|
return response.status == 200
|
||||||
except requests.RequestException:
|
except Exception:
|
||||||
return False
|
return False
|
||||||
Loading…
Add table
Reference in a new issue