feat(hamsterimg): add image uploader
This commit is contained in:
abdus committed
1 parent
6eb0ab7d5a
commit
0d97094092
2 files changed
+307
No files matched your search
Executable
+112
@@ -0,0 +1,112 @@
|
||||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# dependencies = ["httpx"]
|
||||
# ///
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
@dataclass
|
||||
class UploadedImage:
|
||||
url: str
|
||||
thumb_url: str
|
||||
delete_url: str
|
||||
|
||||
|
||||
def upload_image_to_hamsterimg(api_key: str, image_paths: list[Path]) -> list[UploadedImage]:
|
||||
if not image_paths:
|
||||
return []
|
||||
if not api_key:
|
||||
raise ValueError("HamsterImg API key is required")
|
||||
for image_path in image_paths:
|
||||
if not image_path.is_file():
|
||||
raise FileNotFoundError(f"Image file not found: {image_path}")
|
||||
with httpx.Client(headers={"X-API-Key": api_key}, timeout=60) as client:
|
||||
|
||||
def upload(image_path: Path) -> UploadedImage:
|
||||
with image_path.open(mode="rb") as image_file:
|
||||
response = client.post(
|
||||
url="https://hamsterimg.net/api/1/upload",
|
||||
data={"format": "json"},
|
||||
files={"source": (image_path.name, image_file)},
|
||||
)
|
||||
try:
|
||||
result = response.json()
|
||||
except ValueError:
|
||||
result = {}
|
||||
if response.is_error or not isinstance(result, dict) or result.get("status_code") != 200:
|
||||
error = result.get("error", {}) if isinstance(result, dict) else {}
|
||||
message = error.get("message") if isinstance(error, dict) else str(error)
|
||||
status = result.get("status_txt") if isinstance(result, dict) else None
|
||||
reason = message or status or response.reason_phrase or "Unknown error"
|
||||
raise RuntimeError(f"HamsterImg upload failed for {image_path}: {reason}")
|
||||
try:
|
||||
image = result["image"]
|
||||
url = image["url"]
|
||||
thumb_url = image["thumb"]["url"]
|
||||
delete_url = image["delete_url"]
|
||||
except (KeyError, TypeError) as error:
|
||||
raise ValueError(f"HamsterImg returned incomplete image details for {image_path}") from error
|
||||
if not all(isinstance(value, str) and value for value in (url, thumb_url, delete_url)):
|
||||
raise ValueError(f"HamsterImg returned missing image URLs for {image_path}")
|
||||
return UploadedImage(url=url, thumb_url=thumb_url, delete_url=delete_url)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=min(8, len(image_paths))) as executor:
|
||||
futures = [executor.submit(upload, image_path=image_path) for image_path in image_paths]
|
||||
return [future.result() for future in futures]
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Upload images to HamsterImg and print their URLs.",
|
||||
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
||||
)
|
||||
parser.add_argument("image_paths", nargs="+", type=Path, help="Image files to upload")
|
||||
parser.add_argument("--api-key", help="HamsterImg API key; also reads HAMSTERIMG_API_KEY")
|
||||
parser.add_argument(
|
||||
"--format",
|
||||
dest="output_format",
|
||||
choices=["text", "json", "bbcode"],
|
||||
default="text",
|
||||
help="Output format; BBCode shows thumbnails linked to full images",
|
||||
)
|
||||
arguments = parser.parse_args()
|
||||
arguments.api_key = arguments.api_key or os.getenv("HAMSTERIMG_API_KEY")
|
||||
if not arguments.api_key:
|
||||
parser.error("Provide --api-key or set HAMSTERIMG_API_KEY")
|
||||
for image_path in arguments.image_paths:
|
||||
if not image_path.is_file():
|
||||
parser.error(f"Image file not found: {image_path}")
|
||||
return arguments
|
||||
|
||||
|
||||
def main() -> int:
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
|
||||
arguments = parse_args()
|
||||
try:
|
||||
images = upload_image_to_hamsterimg(api_key=arguments.api_key, image_paths=arguments.image_paths)
|
||||
except (OSError, ValueError, RuntimeError, httpx.HTTPError) as error:
|
||||
logging.error(f"{error}")
|
||||
return 1
|
||||
if arguments.output_format == "json":
|
||||
print(json.dumps([asdict(obj=image) for image in images], indent=2, ensure_ascii=False))
|
||||
return 0
|
||||
for image in images:
|
||||
if arguments.output_format == "bbcode":
|
||||
print(f"[url={image.url}][img]{image.thumb_url}[/img][/url]")
|
||||
else:
|
||||
print(image.url)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,195 @@
|
||||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# dependencies = ["httpx", "pytest"]
|
||||
# ///
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from threading import Barrier
|
||||
from unittest.mock import patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import hamsterimg as script
|
||||
|
||||
|
||||
def test_hamsterimg_uploads_in_parallel_and_preserves_order(tmp_path: Path) -> None:
|
||||
image_paths = [tmp_path / "first.jpg", tmp_path / "second.png"]
|
||||
for image_path in image_paths:
|
||||
image_path.write_bytes(image_path.name.encode())
|
||||
barrier = Barrier(parties=2)
|
||||
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
assert str(request.url) == "https://hamsterimg.net/api/1/upload"
|
||||
assert request.method == "POST"
|
||||
assert request.headers["X-API-Key"] == "test-key"
|
||||
assert request.headers["Content-Type"].startswith("multipart/form-data;")
|
||||
body = request.read()
|
||||
assert b'name="source"' in body
|
||||
assert b'name="format"' in body
|
||||
filename = "first.jpg" if b'filename="first.jpg"' in body else "second.png"
|
||||
barrier.wait(timeout=5)
|
||||
return httpx.Response(status_code=200, json={
|
||||
"status_code": 200,
|
||||
"image": {
|
||||
"url": f"https://images.example/{filename}",
|
||||
"thumb": {"url": f"https://images.example/thumb/{filename}"},
|
||||
"delete_url": f"https://hamsterimg.net/{filename}/delete/token",
|
||||
},
|
||||
})
|
||||
|
||||
client = httpx.Client(transport=httpx.MockTransport(handler=respond), headers={"X-API-Key": "test-key"})
|
||||
with patch.object(script.httpx, "Client", return_value=client) as factory:
|
||||
images = script.upload_image_to_hamsterimg(api_key="test-key", image_paths=image_paths)
|
||||
factory.assert_called_once_with(headers={"X-API-Key": "test-key"}, timeout=60)
|
||||
assert images == [
|
||||
script.UploadedImage(
|
||||
url="https://images.example/first.jpg",
|
||||
thumb_url="https://images.example/thumb/first.jpg",
|
||||
delete_url="https://hamsterimg.net/first.jpg/delete/token",
|
||||
),
|
||||
script.UploadedImage(
|
||||
url="https://images.example/second.png",
|
||||
thumb_url="https://images.example/thumb/second.png",
|
||||
delete_url="https://hamsterimg.net/second.png/delete/token",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status, body, error", [
|
||||
(401, {"error": {"message": "Invalid API key"}}, RuntimeError),
|
||||
(200, {"status_code": 400, "error": {"message": "Invalid image"}}, RuntimeError),
|
||||
(200, {"status_code": 200, "image": {}}, ValueError),
|
||||
(200, {"status_code": 200, "image": {"url": "https://images.example/full.jpg", "thumb": {"url": "https://images.example/thumb.jpg"}}}, ValueError),
|
||||
(200, {"status_code": 200, "image": {"url": "https://images.example/full.jpg", "thumb": {"url": None}, "delete_url": "https://hamsterimg.net/delete/token"}}, ValueError),
|
||||
])
|
||||
def test_hamsterimg_reports_upload_errors(tmp_path: Path, status: int, body: dict, error: type[Exception]) -> None:
|
||||
image_path = tmp_path / "image.jpg"
|
||||
image_path.write_bytes(b"image")
|
||||
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(status_code=status, json=body)
|
||||
|
||||
client = httpx.Client(transport=httpx.MockTransport(handler=respond))
|
||||
with patch.object(script.httpx, "Client", return_value=client):
|
||||
with pytest.raises(error):
|
||||
script.upload_image_to_hamsterimg(api_key="test-key", image_paths=[image_path])
|
||||
|
||||
|
||||
def test_hamsterimg_upload_error_includes_api_reason(tmp_path: Path) -> None:
|
||||
image_path = tmp_path / "image.jpg"
|
||||
image_path.write_bytes(b"image")
|
||||
|
||||
def respond(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
status_code=400,
|
||||
json={"status_code": 400, "error": {"message": "Unable to read image"}},
|
||||
)
|
||||
|
||||
client = httpx.Client(transport=httpx.MockTransport(handler=respond))
|
||||
with patch.object(script.httpx, "Client", return_value=client):
|
||||
with pytest.raises(RuntimeError, match="Unable to read image"):
|
||||
script.upload_image_to_hamsterimg(api_key="test-key", image_paths=[image_path])
|
||||
|
||||
|
||||
def test_hamsterimg_validates_inputs_before_uploading(tmp_path: Path) -> None:
|
||||
image_path = tmp_path / "image.jpg"
|
||||
image_path.write_bytes(b"image")
|
||||
with patch.object(script.httpx, "Client") as client:
|
||||
assert script.upload_image_to_hamsterimg(api_key="test-key", image_paths=[]) == []
|
||||
with pytest.raises(ValueError, match="API key"):
|
||||
script.upload_image_to_hamsterimg(api_key="", image_paths=[image_path])
|
||||
with pytest.raises(FileNotFoundError):
|
||||
script.upload_image_to_hamsterimg(api_key="test-key", image_paths=[image_path, tmp_path / "missing.jpg"])
|
||||
client.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("output_format", [None, "text", "json", "bbcode"])
|
||||
def test_upload_image_command_output(
|
||||
tmp_path: Path, output_format: str | None, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
image_paths = [tmp_path / "first.jpg", tmp_path / "second.png"]
|
||||
for image_path in image_paths:
|
||||
image_path.write_bytes(b"image")
|
||||
images = [
|
||||
script.UploadedImage(
|
||||
url="https://images.example/first.jpg",
|
||||
thumb_url="https://images.example/first.th.jpg",
|
||||
delete_url="https://hamsterimg.net/first/delete/token",
|
||||
),
|
||||
script.UploadedImage(
|
||||
url="https://images.example/second.png",
|
||||
thumb_url="https://images.example/second.th.png",
|
||||
delete_url="https://hamsterimg.net/second/delete/token",
|
||||
),
|
||||
]
|
||||
arguments = ["hamsterimg.py", *map(str, image_paths)]
|
||||
if output_format is not None:
|
||||
arguments.extend(["--format", output_format])
|
||||
monkeypatch.setattr(sys, "argv", arguments)
|
||||
monkeypatch.setenv("HAMSTERIMG_API_KEY", "test-key")
|
||||
with patch.object(script, "upload_image_to_hamsterimg", return_value=images) as upload:
|
||||
assert script.main() == 0
|
||||
result = capsys.readouterr()
|
||||
assert result.err == ""
|
||||
upload.assert_called_once_with(api_key="test-key", image_paths=image_paths)
|
||||
if output_format == "json":
|
||||
assert json.loads(result.out) == [
|
||||
{
|
||||
"url": "https://images.example/first.jpg",
|
||||
"thumb_url": "https://images.example/first.th.jpg",
|
||||
"delete_url": "https://hamsterimg.net/first/delete/token",
|
||||
},
|
||||
{
|
||||
"url": "https://images.example/second.png",
|
||||
"thumb_url": "https://images.example/second.th.png",
|
||||
"delete_url": "https://hamsterimg.net/second/delete/token",
|
||||
},
|
||||
]
|
||||
elif output_format == "bbcode":
|
||||
assert result.out == (
|
||||
"[url=https://images.example/first.jpg][img]https://images.example/first.th.jpg[/img][/url]\n"
|
||||
"[url=https://images.example/second.png][img]https://images.example/second.th.png[/img][/url]\n"
|
||||
)
|
||||
else:
|
||||
assert result.out == "https://images.example/first.jpg\nhttps://images.example/second.png\n"
|
||||
|
||||
|
||||
def test_upload_image_command_failure(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
image_path = tmp_path / "image.jpg"
|
||||
image_path.write_bytes(b"image")
|
||||
monkeypatch.setattr(sys, "argv", ["hamsterimg.py", str(image_path), "--api-key", "explicit-key", "--format", "json"])
|
||||
with patch.object(script, "upload_image_to_hamsterimg", side_effect=RuntimeError("Invalid image")) as upload:
|
||||
assert script.main() == 1
|
||||
assert capsys.readouterr().out == ""
|
||||
assert "Invalid image" in caplog.text
|
||||
upload.assert_called_once_with(api_key="explicit-key", image_paths=[image_path])
|
||||
|
||||
|
||||
def test_upload_image_command_requires_paths(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sys, "argv", ["hamsterimg.py", "--api-key", "test-key"])
|
||||
with patch.object(script, "upload_image_to_hamsterimg") as upload:
|
||||
with pytest.raises(SystemExit) as error:
|
||||
script.main()
|
||||
assert error.value.code == 2
|
||||
upload.assert_not_called()
|
||||
|
||||
|
||||
def test_upload_image_command_requires_key(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
image_path = tmp_path / "image.jpg"
|
||||
image_path.write_bytes(b"image")
|
||||
monkeypatch.setattr(sys, "argv", ["hamsterimg.py", str(image_path)])
|
||||
monkeypatch.delenv("HAMSTERIMG_API_KEY", raising=False)
|
||||
with patch.object(script, "upload_image_to_hamsterimg") as upload:
|
||||
with pytest.raises(SystemExit) as error:
|
||||
script.main()
|
||||
assert error.value.code == 2
|
||||
upload.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))
|
||||
Reference in new issue
Block a user