diff --git a/hamsterimg.py b/hamsterimg.py new file mode 100755 index 0000000..132bd57 --- /dev/null +++ b/hamsterimg.py @@ -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()) diff --git a/hamsterimg_test.py b/hamsterimg_test.py new file mode 100644 index 0000000..f94d815 --- /dev/null +++ b/hamsterimg_test.py @@ -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:]]))