change: Implement 3: Complete the native-process CLI, JSON request handling, status vocabulary and exit codes
Repository Validate / repository-checks (pull_request) Successful in 3s
Repository Validate / repository-checks (pull_request) Successful in 3s
Review fix for issue #5: aligned the public CLI's size/resize semantics with the active product obligations instead of the previous exact-match behaviour. - `tools/example.py`: the requested width/height are now a target (minimum) size, never an exact-match filter: any image at least that large is selected regardless of `--resize`, and smaller images still produce `no_size`. `--resize` no longer widens selection; it now performs real downscaling of downloaded images to a shrink-to-fit box (aspect ratio preserved, never upscaled), implemented with a dependency-free stdlib PNG codec plus ICO container support (PNG-compressed ICO entries), and each item records `from`/`to`/`applied` dimensions. Undecodable payloads raise the new typed `ImageResizeError` and are reported as failed downloads, so the status vocabulary and exit codes stay unchanged (`ok`/`dry_run` 0, `invalid_request` 2, `no_format` 3, `no_size` 4, `access_error` 5, `partial_download` 6). Help text and the `no_size` error details were updated to describe the minimum-size contract. - `tools/errors.py`: added `ImageResizeError` (code `resize_failed`) to the typed hierarchy and `__all__`. - `tests/test_cli.py`: replaced the old "exact size unless --resize" test with minimum-size tests (larger image accepted without `--resize`, exact size accepted, `--resize` without `--download` keeps the reported source size) and added offline fixtures/tests for real downscaling: PNG stored at 100x50 for a 100x100 box, original bytes preserved without `--resize`, no upscaling (`already_fits`), PNG-in-ICO downscaling, and `resize_failed` surfacing as `access_error`. - `examples/input.json` remains the canonical offline `dry_run` request, so the smoke command stays green and network-free. No protected facade/control paths were touched and no dependency declarations changed. Tests: Ran the facade commands on the final revision through the prepared environment (`.venv` on PATH): `python -m compileall tools tests` → exit 0 (tools.validate); `pytest` → 83 passed (tools.test); `python -m tools.example examples/input.json` → exit 0 with one JSON document, status `dry_run` (tools.smoke); `python -m tools.example --help` → exit 0 (help-command acceptance). `tools/envelope.py` is unchanged and still exports the fixed status/exit-code table checked by tests. Target branch `task/1-task` is already an ancestor of the work branch head, so no integration was needed. Progress: Addressed the changes-requested review on PR #13: requested width/height are now a minimum target size (selection no longer depends on --resize), and --resize now actually downscales downloaded images (stdlib PNG codec + ICO/PNG container support, shrink-to-fit, never upscaling) with per-item resize metadata and typed ImageResizeError for undecodable payloads. Updated tests/test_cli.py with minimum-size and offline resize/download contract tests (83 passed). Facade validate/test/smoke/help all green; examples/input.json stays the canonical dry_run request. Published by developer-agent through Tool Backend. Closes #5
This commit was merged in pull request #13.
This commit is contained in:
@@ -12,7 +12,8 @@
|
|||||||
},
|
},
|
||||||
"projects/product-54354138-1#5": {
|
"projects/product-54354138-1#5": {
|
||||||
"events": [
|
"events": [
|
||||||
"Implemented envelope status/exit-code table and response builder, request errors, the full flag/JSON-document CLI in tools/example.py, canonical offline examples/input.json, and CLI tests; facade validate/test/smoke and --help all pass (76 tests)."
|
"Implemented envelope status/exit-code table and response builder, request errors, the full flag/JSON-document CLI in tools/example.py, canonical offline examples/input.json, and CLI tests; facade validate/test/smoke and --help all pass (76 tests).",
|
||||||
|
"Addressed the changes-requested review on PR #13: requested width/height are now a minimum target size (selection no longer depends on --resize), and --resize now actually downscales downloaded images (stdlib PNG codec + ICO/PNG container support, shrink-to-fit, never upscaling) with per-item resize metadata and typed ImageResizeError for undecodable payloads. Updated tests/test_cli.py with minimum-size and offline resize/download contract tests (83 passed). Facade validate/test/smoke/help all green; examples/input.json stays the canonical dry_run request."
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+210
-2
@@ -3,8 +3,10 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import json
|
||||||
|
import struct
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
|
import zlib
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Sequence
|
from typing import Any, Sequence
|
||||||
|
|
||||||
@@ -286,14 +288,40 @@ def test_no_size_when_only_smaller_images_exist(monkeypatch: pytest.MonkeyPatch,
|
|||||||
assert payload["status"] == envelope.STATUS_NO_SIZE
|
assert payload["status"] == envelope.STATUS_NO_SIZE
|
||||||
|
|
||||||
|
|
||||||
def test_resize_accepts_larger_images(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
def test_requested_size_is_a_minimum_not_an_exact_match(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||||
|
) -> None:
|
||||||
|
registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 400, 300)])
|
||||||
|
|
||||||
|
code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS)
|
||||||
|
|
||||||
|
assert code == envelope.EXIT_OK
|
||||||
|
assert payload["status"] == envelope.STATUS_OK
|
||||||
|
assert payload["request"]["resize"] is False
|
||||||
|
assert [item["width"] for item in payload["items"]] == [400]
|
||||||
|
assert [item["height"] for item in payload["items"]] == [300]
|
||||||
|
|
||||||
|
|
||||||
|
def test_exact_size_is_accepted(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||||
|
registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100)])
|
||||||
|
|
||||||
|
code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS)
|
||||||
|
|
||||||
|
assert code == envelope.EXIT_OK
|
||||||
|
assert len(payload["items"]) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_resize_without_download_keeps_reported_source_size(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||||
|
) -> None:
|
||||||
registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 400, 300)])
|
registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 400, 300)])
|
||||||
|
|
||||||
code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--resize")
|
code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--resize")
|
||||||
|
|
||||||
assert code == envelope.EXIT_OK
|
assert code == envelope.EXIT_OK
|
||||||
assert payload["request"]["resize"] is True
|
assert payload["request"]["resize"] is True
|
||||||
assert len(payload["items"]) == 1
|
assert payload["items"][0]["width"] == 400
|
||||||
|
assert "download" not in payload["items"][0]
|
||||||
|
|
||||||
|
|
||||||
def test_download_success_writes_files(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None:
|
def test_download_success_writes_files(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None:
|
||||||
@@ -340,3 +368,183 @@ def test_search_access_error_maps_to_access_error(monkeypatch: pytest.MonkeyPatc
|
|||||||
assert code == envelope.EXIT_ACCESS_ERROR
|
assert code == envelope.EXIT_ACCESS_ERROR
|
||||||
assert payload["status"] == envelope.STATUS_ACCESS_ERROR
|
assert payload["status"] == envelope.STATUS_ACCESS_ERROR
|
||||||
assert payload["error"]["code"] == "source_access"
|
assert payload["error"]["code"] == "source_access"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Size filtering and real downscaling of downloaded images
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
|
||||||
|
|
||||||
|
|
||||||
|
def png_bytes(width: int, height: int, rgba: tuple[int, int, int, int] = (255, 0, 0, 255)) -> bytes:
|
||||||
|
"""Build a minimal, valid RGBA PNG without any third-party dependency."""
|
||||||
|
|
||||||
|
raw = bytearray()
|
||||||
|
for _ in range(height):
|
||||||
|
raw.append(0)
|
||||||
|
raw += bytes(rgba) * width
|
||||||
|
|
||||||
|
def chunk(kind: bytes, data: bytes) -> bytes:
|
||||||
|
crc = zlib.crc32(kind + data) & 0xFFFFFFFF
|
||||||
|
return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", crc)
|
||||||
|
|
||||||
|
header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0)
|
||||||
|
return (
|
||||||
|
PNG_SIGNATURE
|
||||||
|
+ chunk(b"IHDR", header)
|
||||||
|
+ chunk(b"IDAT", zlib.compress(bytes(raw)))
|
||||||
|
+ chunk(b"IEND", b"")
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def png_size(payload: bytes) -> tuple[int, int]:
|
||||||
|
assert payload.startswith(PNG_SIGNATURE)
|
||||||
|
return struct.unpack(">II", payload[16:24])
|
||||||
|
|
||||||
|
|
||||||
|
def ico_bytes(inner_png: bytes, width: int, height: int) -> bytes:
|
||||||
|
header = struct.pack("<HHH", 0, 1, 1)
|
||||||
|
entry = struct.pack(
|
||||||
|
"<BBBBHHII",
|
||||||
|
width % 256,
|
||||||
|
height % 256,
|
||||||
|
0,
|
||||||
|
0,
|
||||||
|
1,
|
||||||
|
32,
|
||||||
|
len(inner_png),
|
||||||
|
6 + 16,
|
||||||
|
)
|
||||||
|
return header + entry + inner_png
|
||||||
|
|
||||||
|
|
||||||
|
def ico_payload(payload: bytes) -> tuple[int, int, bytes]:
|
||||||
|
reserved, kind, count = struct.unpack("<HHH", payload[:6])
|
||||||
|
assert (reserved, kind, count) == (0, 1, 1)
|
||||||
|
width, height, _colours, _reserved, _planes, _bits, size, offset = struct.unpack(
|
||||||
|
"<BBBBHHII", payload[6:22]
|
||||||
|
)
|
||||||
|
return width or 256, height or 256, payload[offset : offset + size]
|
||||||
|
|
||||||
|
|
||||||
|
DOWNLOAD_ARGS = ("--topic", "fox", "--format", "png", "--source", "fake")
|
||||||
|
|
||||||
|
|
||||||
|
def test_download_resize_downscales_stored_image(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
payload = png_bytes(200, 100)
|
||||||
|
registry = FakeRegistry(
|
||||||
|
[FakeCandidate("fox", "https://example.test/a.png", "image/png", 200, 100, payload=payload)]
|
||||||
|
)
|
||||||
|
|
||||||
|
code, document = run_main(
|
||||||
|
monkeypatch, capsys, registry, *BASE_ARGS, "--resize", "--download", "--output-dir", str(tmp_path)
|
||||||
|
)
|
||||||
|
|
||||||
|
assert code == envelope.EXIT_OK
|
||||||
|
download = document["items"][0]["download"]
|
||||||
|
assert download["ok"] is True
|
||||||
|
assert download["resize"]["applied"] is True
|
||||||
|
assert download["resize"]["from"] == {"width": 200, "height": 100}
|
||||||
|
assert download["resize"]["to"] == {"width": 100, "height": 50}
|
||||||
|
stored = Path(download["path"]).read_bytes()
|
||||||
|
assert png_size(stored) == (100, 50)
|
||||||
|
assert download["width"] == 100
|
||||||
|
assert download["height"] == 50
|
||||||
|
|
||||||
|
|
||||||
|
def test_download_without_resize_keeps_original_payload(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
payload = png_bytes(200, 100)
|
||||||
|
registry = FakeRegistry(
|
||||||
|
[FakeCandidate("fox", "https://example.test/a.png", "image/png", 200, 100, payload=payload)]
|
||||||
|
)
|
||||||
|
|
||||||
|
code, document = run_main(
|
||||||
|
monkeypatch, capsys, registry, *BASE_ARGS, "--download", "--output-dir", str(tmp_path)
|
||||||
|
)
|
||||||
|
|
||||||
|
assert code == envelope.EXIT_OK
|
||||||
|
download = document["items"][0]["download"]
|
||||||
|
assert "resize" not in download
|
||||||
|
assert Path(download["path"]).read_bytes() == payload
|
||||||
|
|
||||||
|
|
||||||
|
def test_resize_never_scales_up(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
payload = png_bytes(50, 50)
|
||||||
|
registry = FakeRegistry(
|
||||||
|
[FakeCandidate("fox", "https://example.test/a.png", "image/png", 50, 50, payload=payload)]
|
||||||
|
)
|
||||||
|
|
||||||
|
code, document = run_main(
|
||||||
|
monkeypatch, capsys, registry, *DOWNLOAD_ARGS, "--resize", "--download", "--output-dir", str(tmp_path)
|
||||||
|
)
|
||||||
|
|
||||||
|
assert code == envelope.EXIT_OK
|
||||||
|
download = document["items"][0]["download"]
|
||||||
|
assert download["resize"]["applied"] is False
|
||||||
|
assert download["resize"]["reason"] == "already_fits"
|
||||||
|
assert Path(download["path"]).read_bytes() == payload
|
||||||
|
|
||||||
|
|
||||||
|
def test_resize_downscales_png_stored_inside_ico(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
payload = ico_bytes(png_bytes(100, 100), 100, 100)
|
||||||
|
registry = FakeRegistry(
|
||||||
|
[FakeCandidate("fox", "https://example.test/a.ico", "image/x-icon", 100, 100, payload=payload)]
|
||||||
|
)
|
||||||
|
|
||||||
|
code, document = run_main(
|
||||||
|
monkeypatch,
|
||||||
|
capsys,
|
||||||
|
registry,
|
||||||
|
"--topic",
|
||||||
|
"fox",
|
||||||
|
"--format",
|
||||||
|
"ico",
|
||||||
|
"--source",
|
||||||
|
"fake",
|
||||||
|
"--width",
|
||||||
|
"50",
|
||||||
|
"--height",
|
||||||
|
"50",
|
||||||
|
"--resize",
|
||||||
|
"--download",
|
||||||
|
"--output-dir",
|
||||||
|
str(tmp_path),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert code == envelope.EXIT_OK
|
||||||
|
download = document["items"][0]["download"]
|
||||||
|
assert download["resize"]["to"] == {"width": 50, "height": 50}
|
||||||
|
width, height, inner = ico_payload(Path(download["path"]).read_bytes())
|
||||||
|
assert (width, height) == (50, 50)
|
||||||
|
assert png_size(inner) == (50, 50)
|
||||||
|
|
||||||
|
|
||||||
|
def test_undecodable_payload_with_resize_fails_the_item(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
registry = FakeRegistry(
|
||||||
|
[
|
||||||
|
FakeCandidate(
|
||||||
|
"fox", "https://example.test/a.png", "image/png", 100, 100, payload=b"not-an-image"
|
||||||
|
)
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
code, document = run_main(
|
||||||
|
monkeypatch, capsys, registry, *BASE_ARGS, "--resize", "--download", "--output-dir", str(tmp_path)
|
||||||
|
)
|
||||||
|
|
||||||
|
assert code == envelope.EXIT_ACCESS_ERROR
|
||||||
|
assert document["status"] == envelope.STATUS_ACCESS_ERROR
|
||||||
|
download = document["items"][0]["download"]
|
||||||
|
assert download["ok"] is False
|
||||||
|
assert download["error"]["code"] == "resize_failed"
|
||||||
|
|||||||
@@ -252,8 +252,20 @@ class UnsupportedFormatError(RequestError):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class ImageResizeError(ToolkitError):
|
||||||
|
"""Raised when a downloaded image cannot be downscaled to the target size."""
|
||||||
|
|
||||||
|
code = "resize_failed"
|
||||||
|
|
||||||
|
def __init__(self, reason: str, *, details: Mapping[str, Any] | None = None) -> None:
|
||||||
|
merged: dict[str, Any] = dict(details or {})
|
||||||
|
merged["reason"] = reason
|
||||||
|
super().__init__(f"cannot resize image: {reason}", details=merged)
|
||||||
|
|
||||||
|
|
||||||
__all__ += [
|
__all__ += [
|
||||||
"ACCEPTED_FORMATS",
|
"ACCEPTED_FORMATS",
|
||||||
|
"ImageResizeError",
|
||||||
"RequestDocumentError",
|
"RequestDocumentError",
|
||||||
"RequestError",
|
"RequestError",
|
||||||
"RequestFieldError",
|
"RequestFieldError",
|
||||||
|
|||||||
+349
-28
@@ -14,7 +14,9 @@ from __future__ import annotations
|
|||||||
import argparse
|
import argparse
|
||||||
import json
|
import json
|
||||||
import re
|
import re
|
||||||
|
import struct
|
||||||
import sys
|
import sys
|
||||||
|
import zlib
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Mapping, Sequence
|
from typing import Any, Mapping, Sequence
|
||||||
|
|
||||||
@@ -30,6 +32,7 @@ from .envelope import (
|
|||||||
)
|
)
|
||||||
from .errors import (
|
from .errors import (
|
||||||
ACCEPTED_FORMATS,
|
ACCEPTED_FORMATS,
|
||||||
|
ImageResizeError,
|
||||||
RequestDocumentError,
|
RequestDocumentError,
|
||||||
RequestError,
|
RequestError,
|
||||||
RequestFieldError,
|
RequestFieldError,
|
||||||
@@ -110,13 +113,13 @@ def build_parser() -> argparse.ArgumentParser:
|
|||||||
"--width",
|
"--width",
|
||||||
type=int,
|
type=int,
|
||||||
metavar="PX",
|
metavar="PX",
|
||||||
help="desired image width in pixels",
|
help="target width in pixels; only images at least this wide are accepted",
|
||||||
)
|
)
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"--height",
|
"--height",
|
||||||
type=int,
|
type=int,
|
||||||
metavar="PX",
|
metavar="PX",
|
||||||
help="desired image height in pixels",
|
help="target height in pixels; only images at least this tall are accepted",
|
||||||
)
|
)
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"-f",
|
"-f",
|
||||||
@@ -152,8 +155,10 @@ def build_parser() -> argparse.ArgumentParser:
|
|||||||
"--resize",
|
"--resize",
|
||||||
action="store_true",
|
action="store_true",
|
||||||
help=(
|
help=(
|
||||||
"accept images larger than the requested size and downscale them; "
|
"downscale every downloaded image so it fits the requested width "
|
||||||
"without this flag the size must match exactly"
|
"and height, keeping the aspect ratio and never scaling up; "
|
||||||
|
"without it the original bytes are written and the actual size is "
|
||||||
|
"reported"
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
@@ -317,9 +322,18 @@ def _matches_format(candidate: SourceCandidate, image_format: str) -> bool:
|
|||||||
|
|
||||||
|
|
||||||
def _matches_size(candidate: SourceCandidate, request: Mapping[str, Any]) -> bool:
|
def _matches_size(candidate: SourceCandidate, request: Mapping[str, Any]) -> bool:
|
||||||
width = request.get("width")
|
"""True when ``candidate`` is usable for the requested target size.
|
||||||
height = request.get("height")
|
|
||||||
if width is None and height is None:
|
The requested width and height are a target size, not an exact match: any
|
||||||
|
image that is at least that large is usable because it can always be
|
||||||
|
downscaled, while a smaller image would have to be scaled up and is
|
||||||
|
therefore rejected. ``--resize`` only decides whether the downloaded bytes
|
||||||
|
are downscaled; it never widens or narrows this filter.
|
||||||
|
"""
|
||||||
|
|
||||||
|
wanted_width = request.get("width")
|
||||||
|
wanted_height = request.get("height")
|
||||||
|
if wanted_width is None and wanted_height is None:
|
||||||
return True
|
return True
|
||||||
actual_width = getattr(candidate, "width", None)
|
actual_width = getattr(candidate, "width", None)
|
||||||
actual_height = getattr(candidate, "height", None)
|
actual_height = getattr(candidate, "height", None)
|
||||||
@@ -327,13 +341,11 @@ def _matches_size(candidate: SourceCandidate, request: Mapping[str, Any]) -> boo
|
|||||||
return False
|
return False
|
||||||
if not isinstance(actual_width, int) or not isinstance(actual_height, int):
|
if not isinstance(actual_width, int) or not isinstance(actual_height, int):
|
||||||
return False
|
return False
|
||||||
if request.get("resize"):
|
if wanted_width is not None and actual_width < wanted_width:
|
||||||
fits_width = width is None or actual_width >= width
|
return False
|
||||||
fits_height = height is None or actual_height >= height
|
if wanted_height is not None and actual_height < wanted_height:
|
||||||
return fits_width and fits_height
|
return False
|
||||||
matches_width = width is None or actual_width == width
|
return True
|
||||||
matches_height = height is None or actual_height == height
|
|
||||||
return matches_width and matches_height
|
|
||||||
|
|
||||||
|
|
||||||
def _select(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[SourceCandidate], list[SourceCandidate]]:
|
def _select(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[SourceCandidate], list[SourceCandidate]]:
|
||||||
@@ -351,8 +363,315 @@ def _slug(candidate: SourceCandidate) -> str:
|
|||||||
return (slug or "image")[:60]
|
return (slug or "image")[:60]
|
||||||
|
|
||||||
|
|
||||||
|
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
|
||||||
|
ICO_HEADER_SIZE = 6
|
||||||
|
ICO_ENTRY_SIZE = 16
|
||||||
|
ICO_MAX_DIMENSION = 256
|
||||||
|
_PNG_CHANNELS = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}
|
||||||
|
|
||||||
|
|
||||||
|
def _png_chunks(payload: bytes) -> list[tuple[bytes, bytes]]:
|
||||||
|
"""Split a PNG payload into ``(type, data)`` chunks."""
|
||||||
|
|
||||||
|
if not payload.startswith(PNG_SIGNATURE):
|
||||||
|
raise ImageResizeError("payload is not a PNG image")
|
||||||
|
chunks: list[tuple[bytes, bytes]] = []
|
||||||
|
offset = len(PNG_SIGNATURE)
|
||||||
|
while offset + 8 <= len(payload):
|
||||||
|
(length,) = struct.unpack(">I", payload[offset : offset + 4])
|
||||||
|
chunk_type = payload[offset + 4 : offset + 8]
|
||||||
|
data = payload[offset + 8 : offset + 8 + length]
|
||||||
|
if len(data) != length:
|
||||||
|
raise ImageResizeError("truncated PNG chunk")
|
||||||
|
chunks.append((chunk_type, data))
|
||||||
|
offset += 12 + length
|
||||||
|
if chunk_type == b"IEND":
|
||||||
|
return chunks
|
||||||
|
raise ImageResizeError("PNG image has no IEND chunk")
|
||||||
|
|
||||||
|
|
||||||
|
def _unfilter_rows(raw: bytes, width: int, height: int, bpp: int) -> list[bytearray]:
|
||||||
|
"""Undo the per-scanline PNG filters, returning one row per pixel line."""
|
||||||
|
|
||||||
|
stride = width * bpp
|
||||||
|
if len(raw) < height * (stride + 1):
|
||||||
|
raise ImageResizeError("truncated PNG image data")
|
||||||
|
rows: list[bytearray] = []
|
||||||
|
previous = bytearray(stride)
|
||||||
|
position = 0
|
||||||
|
for _ in range(height):
|
||||||
|
filter_type = raw[position]
|
||||||
|
position += 1
|
||||||
|
line = bytearray(raw[position : position + stride])
|
||||||
|
position += stride
|
||||||
|
if filter_type == 1:
|
||||||
|
for index in range(bpp, stride):
|
||||||
|
line[index] = (line[index] + line[index - bpp]) & 0xFF
|
||||||
|
elif filter_type == 2:
|
||||||
|
for index in range(stride):
|
||||||
|
line[index] = (line[index] + previous[index]) & 0xFF
|
||||||
|
elif filter_type == 3:
|
||||||
|
for index in range(stride):
|
||||||
|
left = line[index - bpp] if index >= bpp else 0
|
||||||
|
line[index] = (line[index] + ((left + previous[index]) >> 1)) & 0xFF
|
||||||
|
elif filter_type == 4:
|
||||||
|
for index in range(stride):
|
||||||
|
left = line[index - bpp] if index >= bpp else 0
|
||||||
|
up = previous[index]
|
||||||
|
up_left = previous[index - bpp] if index >= bpp else 0
|
||||||
|
estimate = left + up - up_left
|
||||||
|
to_left = abs(estimate - left)
|
||||||
|
to_up = abs(estimate - up)
|
||||||
|
to_up_left = abs(estimate - up_left)
|
||||||
|
if to_left <= to_up and to_left <= to_up_left:
|
||||||
|
predictor = left
|
||||||
|
elif to_up <= to_up_left:
|
||||||
|
predictor = up
|
||||||
|
else:
|
||||||
|
predictor = up_left
|
||||||
|
line[index] = (line[index] + predictor) & 0xFF
|
||||||
|
elif filter_type != 0:
|
||||||
|
raise ImageResizeError(f"unsupported PNG filter type {filter_type}")
|
||||||
|
previous = line
|
||||||
|
rows.append(line)
|
||||||
|
return rows
|
||||||
|
|
||||||
|
|
||||||
|
def _row_to_rgba(
|
||||||
|
row: bytearray,
|
||||||
|
width: int,
|
||||||
|
color_type: int,
|
||||||
|
depth: int,
|
||||||
|
palette: bytes,
|
||||||
|
transparency: bytes | None,
|
||||||
|
) -> bytes:
|
||||||
|
"""Convert one unfiltered scanline into 8-bit RGBA samples."""
|
||||||
|
|
||||||
|
step = depth // 8
|
||||||
|
out = bytearray()
|
||||||
|
if color_type == 3:
|
||||||
|
for x in range(width):
|
||||||
|
index = row[x]
|
||||||
|
base = index * 3
|
||||||
|
if base + 2 >= len(palette):
|
||||||
|
raise ImageResizeError("PNG palette index is out of range")
|
||||||
|
alpha = transparency[index] if transparency and index < len(transparency) else 255
|
||||||
|
out += bytes((palette[base], palette[base + 1], palette[base + 2], alpha))
|
||||||
|
return bytes(out)
|
||||||
|
for x in range(width):
|
||||||
|
offset = x * step
|
||||||
|
if color_type == 0:
|
||||||
|
value = row[offset]
|
||||||
|
out += bytes((value, value, value, 255))
|
||||||
|
elif color_type == 4:
|
||||||
|
value = row[offset]
|
||||||
|
out += bytes((value, value, value, row[offset + step]))
|
||||||
|
elif color_type == 2:
|
||||||
|
out += bytes((row[offset], row[offset + step], row[offset + 2 * step], 255))
|
||||||
|
else:
|
||||||
|
out += bytes(
|
||||||
|
(
|
||||||
|
row[offset],
|
||||||
|
row[offset + step],
|
||||||
|
row[offset + 2 * step],
|
||||||
|
row[offset + 3 * step],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return bytes(out)
|
||||||
|
|
||||||
|
|
||||||
|
def _decode_png(payload: bytes) -> tuple[int, int, list[bytes]]:
|
||||||
|
"""Decode a PNG payload into ``(width, height, rgba_rows)``."""
|
||||||
|
|
||||||
|
header: tuple[int, ...] | None = None
|
||||||
|
palette = b""
|
||||||
|
transparency: bytes | None = None
|
||||||
|
data = bytearray()
|
||||||
|
for chunk_type, chunk_data in _png_chunks(payload):
|
||||||
|
if chunk_type == b"IHDR":
|
||||||
|
if len(chunk_data) < 13:
|
||||||
|
raise ImageResizeError("truncated PNG header")
|
||||||
|
header = struct.unpack(">IIBBBBB", chunk_data[:13])
|
||||||
|
elif chunk_type == b"PLTE":
|
||||||
|
palette = chunk_data
|
||||||
|
elif chunk_type == b"tRNS":
|
||||||
|
transparency = chunk_data
|
||||||
|
elif chunk_type == b"IDAT":
|
||||||
|
data += chunk_data
|
||||||
|
if header is None:
|
||||||
|
raise ImageResizeError("PNG image has no IHDR chunk")
|
||||||
|
width, height, depth, color_type, compression, filter_method, interlace = header
|
||||||
|
if width < 1 or height < 1:
|
||||||
|
raise ImageResizeError("PNG image has no pixels")
|
||||||
|
if compression != 0 or filter_method != 0:
|
||||||
|
raise ImageResizeError("unsupported PNG compression or filter method")
|
||||||
|
if interlace != 0:
|
||||||
|
raise ImageResizeError("interlaced PNG images are not supported")
|
||||||
|
channels = _PNG_CHANNELS.get(color_type)
|
||||||
|
if channels is None:
|
||||||
|
raise ImageResizeError(f"unsupported PNG colour type {color_type}")
|
||||||
|
if depth not in (8, 16):
|
||||||
|
raise ImageResizeError(f"unsupported PNG bit depth {depth}")
|
||||||
|
if color_type == 3 and depth > 8:
|
||||||
|
raise ImageResizeError("unsupported PNG palette bit depth")
|
||||||
|
if color_type == 3 and not palette:
|
||||||
|
raise ImageResizeError("PNG palette image has no PLTE chunk")
|
||||||
|
try:
|
||||||
|
raw = zlib.decompress(bytes(data))
|
||||||
|
except zlib.error as exc:
|
||||||
|
raise ImageResizeError(f"PNG image data is corrupt: {exc}") from exc
|
||||||
|
rows = _unfilter_rows(raw, width, height, channels * (depth // 8))
|
||||||
|
return width, height, [
|
||||||
|
_row_to_rgba(row, width, color_type, depth, palette, transparency) for row in rows
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def _encode_png(width: int, height: int, rows: Sequence[bytes]) -> bytes:
|
||||||
|
"""Encode RGBA rows as an uncompressed-filter 8-bit PNG."""
|
||||||
|
|
||||||
|
raw = bytearray()
|
||||||
|
for row in rows:
|
||||||
|
raw.append(0)
|
||||||
|
raw += row
|
||||||
|
|
||||||
|
def chunk(chunk_type: bytes, chunk_data: bytes) -> bytes:
|
||||||
|
crc = zlib.crc32(chunk_type + chunk_data) & 0xFFFFFFFF
|
||||||
|
return struct.pack(">I", len(chunk_data)) + chunk_type + chunk_data + struct.pack(">I", crc)
|
||||||
|
|
||||||
|
header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0)
|
||||||
|
return (
|
||||||
|
PNG_SIGNATURE
|
||||||
|
+ chunk(b"IHDR", header)
|
||||||
|
+ chunk(b"IDAT", zlib.compress(bytes(raw), 6))
|
||||||
|
+ chunk(b"IEND", b"")
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _ico_entries(payload: bytes) -> list[tuple[int, int, int, int]]:
|
||||||
|
"""Return ``(width, height, size, offset)`` for every ICO directory entry."""
|
||||||
|
|
||||||
|
if len(payload) < ICO_HEADER_SIZE:
|
||||||
|
raise ImageResizeError("payload is not an ICO image")
|
||||||
|
reserved, kind, count = struct.unpack("<HHH", payload[:ICO_HEADER_SIZE])
|
||||||
|
if reserved != 0 or kind != 1 or count < 1:
|
||||||
|
raise ImageResizeError("payload is not an ICO image")
|
||||||
|
entries: list[tuple[int, int, int, int]] = []
|
||||||
|
for index in range(count):
|
||||||
|
start = ICO_HEADER_SIZE + index * ICO_ENTRY_SIZE
|
||||||
|
if start + ICO_ENTRY_SIZE > len(payload):
|
||||||
|
raise ImageResizeError("truncated ICO directory")
|
||||||
|
width, height, _colours, _reserved, _planes, _bits, size, offset = struct.unpack(
|
||||||
|
"<BBBBHHII", payload[start : start + ICO_ENTRY_SIZE]
|
||||||
|
)
|
||||||
|
entries.append((width or ICO_MAX_DIMENSION, height or ICO_MAX_DIMENSION, size, offset))
|
||||||
|
return entries
|
||||||
|
|
||||||
|
|
||||||
|
def _encode_ico(png_payload: bytes, width: int, height: int) -> bytes:
|
||||||
|
"""Wrap a PNG payload in a single-entry ICO container."""
|
||||||
|
|
||||||
|
if width > ICO_MAX_DIMENSION or height > ICO_MAX_DIMENSION:
|
||||||
|
raise ImageResizeError("ICO images cannot exceed 256x256 pixels")
|
||||||
|
header = struct.pack("<HHH", 0, 1, 1)
|
||||||
|
entry = struct.pack(
|
||||||
|
"<BBBBHHII",
|
||||||
|
width % ICO_MAX_DIMENSION,
|
||||||
|
height % ICO_MAX_DIMENSION,
|
||||||
|
0,
|
||||||
|
0,
|
||||||
|
1,
|
||||||
|
32,
|
||||||
|
len(png_payload),
|
||||||
|
ICO_HEADER_SIZE + ICO_ENTRY_SIZE,
|
||||||
|
)
|
||||||
|
return header + entry + png_payload
|
||||||
|
|
||||||
|
|
||||||
|
def _decode_image(payload: bytes) -> tuple[str, int, int, list[bytes]]:
|
||||||
|
"""Decode a PNG or ICO payload into ``(container, width, height, rows)``."""
|
||||||
|
|
||||||
|
if payload.startswith(PNG_SIGNATURE):
|
||||||
|
width, height, rows = _decode_png(payload)
|
||||||
|
return "png", width, height, rows
|
||||||
|
if len(payload) >= ICO_HEADER_SIZE and payload[:4] == b"\x00\x00\x01\x00":
|
||||||
|
entries = _ico_entries(payload)
|
||||||
|
_width, _height, size, offset = max(entries, key=lambda item: item[0] * item[1])
|
||||||
|
embedded = payload[offset : offset + size]
|
||||||
|
if not embedded.startswith(PNG_SIGNATURE):
|
||||||
|
raise ImageResizeError("ICO images stored as raw bitmaps are not supported")
|
||||||
|
width, height, rows = _decode_png(embedded)
|
||||||
|
return "ico", width, height, rows
|
||||||
|
raise ImageResizeError("payload is neither a PNG nor an ICO image")
|
||||||
|
|
||||||
|
|
||||||
|
def _target_size(width: int, height: int, request: Mapping[str, Any]) -> tuple[int, int]:
|
||||||
|
"""Shrink-to-fit target inside the requested box, never enlarging."""
|
||||||
|
|
||||||
|
wanted_width = request.get("width")
|
||||||
|
wanted_height = request.get("height")
|
||||||
|
scale = 1.0
|
||||||
|
if wanted_width is not None:
|
||||||
|
scale = min(scale, wanted_width / width)
|
||||||
|
if wanted_height is not None:
|
||||||
|
scale = min(scale, wanted_height / height)
|
||||||
|
if scale >= 1.0:
|
||||||
|
return width, height
|
||||||
|
return max(1, int(round(width * scale))), max(1, int(round(height * scale)))
|
||||||
|
|
||||||
|
|
||||||
|
def _resample(rows: Sequence[bytes], width: int, height: int, target_width: int, target_height: int) -> list[bytes]:
|
||||||
|
"""Nearest-neighbour downscale of RGBA rows."""
|
||||||
|
|
||||||
|
columns = [min(width - 1, (x * width) // target_width) * 4 for x in range(target_width)]
|
||||||
|
resized: list[bytes] = []
|
||||||
|
for y in range(target_height):
|
||||||
|
source = rows[min(height - 1, (y * height) // target_height)]
|
||||||
|
line = bytearray()
|
||||||
|
for offset in columns:
|
||||||
|
line += source[offset : offset + 4]
|
||||||
|
resized.append(bytes(line))
|
||||||
|
return resized
|
||||||
|
|
||||||
|
|
||||||
|
def resize_image(payload: bytes, request: Mapping[str, Any]) -> tuple[bytes, dict[str, Any]]:
|
||||||
|
"""Downscale ``payload`` so it fits the requested size.
|
||||||
|
|
||||||
|
The requested width and height are a bounding box: the aspect ratio is
|
||||||
|
preserved, an image that already fits is left untouched and an image is
|
||||||
|
never scaled up. The returned fragment records what happened so the caller
|
||||||
|
can report the real dimensions of the stored file.
|
||||||
|
"""
|
||||||
|
|
||||||
|
container, width, height, rows = _decode_image(payload)
|
||||||
|
target_width, target_height = _target_size(width, height, request)
|
||||||
|
fragment: dict[str, Any] = {
|
||||||
|
"requested": {"width": request.get("width"), "height": request.get("height")},
|
||||||
|
"from": {"width": width, "height": height},
|
||||||
|
}
|
||||||
|
if (target_width, target_height) == (width, height):
|
||||||
|
fragment.update(
|
||||||
|
{"applied": False, "reason": "already_fits", "to": {"width": width, "height": height}}
|
||||||
|
)
|
||||||
|
return payload, fragment
|
||||||
|
png_payload = _encode_png(
|
||||||
|
target_width, target_height, _resample(rows, width, height, target_width, target_height)
|
||||||
|
)
|
||||||
|
output = (
|
||||||
|
_encode_ico(png_payload, target_width, target_height)
|
||||||
|
if container == "ico"
|
||||||
|
else png_payload
|
||||||
|
)
|
||||||
|
fragment.update({"applied": True, "to": {"width": target_width, "height": target_height}})
|
||||||
|
return output, fragment
|
||||||
|
|
||||||
|
|
||||||
def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[dict[str, Any]], int]:
|
def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[dict[str, Any]], int]:
|
||||||
"""Download ``candidates``; return ``(items, failure_count)``."""
|
"""Download ``candidates``; return ``(items, failure_count)``.
|
||||||
|
|
||||||
|
With ``resize`` enabled every stored file is downscaled to the requested
|
||||||
|
target size first. A payload that cannot be decoded or downscaled is
|
||||||
|
reported as a failed item rather than silently stored at the wrong size.
|
||||||
|
"""
|
||||||
|
|
||||||
directory = Path(request["output_dir"]).expanduser()
|
directory = Path(request["output_dir"]).expanduser()
|
||||||
try:
|
try:
|
||||||
@@ -371,19 +690,22 @@ def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any])
|
|||||||
destination = directory / f"{index:03d}-{_slug(candidate)}{suffix}"
|
destination = directory / f"{index:03d}-{_slug(candidate)}{suffix}"
|
||||||
try:
|
try:
|
||||||
payload = candidate.fetch_bytes()
|
payload = candidate.fetch_bytes()
|
||||||
|
resize_fragment = resize_image(payload, request) if request["resize"] else None
|
||||||
|
if resize_fragment is not None:
|
||||||
|
payload, resize_fragment = resize_fragment
|
||||||
destination.write_bytes(payload)
|
destination.write_bytes(payload)
|
||||||
except (ToolkitError, OSError) as exc:
|
except (ToolkitError, OSError) as exc:
|
||||||
failures += 1
|
failures += 1
|
||||||
record["download"] = {
|
error = error_fragment(getattr(exc, "code", "download_failed"), str(exc))
|
||||||
"ok": False,
|
if isinstance(exc, ToolkitError) and exc.details:
|
||||||
"error": error_fragment(getattr(exc, "code", "download_failed"), str(exc)),
|
error["details"] = dict(exc.details)
|
||||||
}
|
record["download"] = {"ok": False, "error": error}
|
||||||
else:
|
else:
|
||||||
record["download"] = {
|
record["download"] = {"ok": True, "path": str(destination), "bytes": len(payload)}
|
||||||
"ok": True,
|
if resize_fragment is not None:
|
||||||
"path": str(destination),
|
record["download"]["resize"] = resize_fragment
|
||||||
"bytes": len(payload),
|
record["download"]["width"] = resize_fragment["to"]["width"]
|
||||||
}
|
record["download"]["height"] = resize_fragment["to"]["height"]
|
||||||
items.append(record)
|
items.append(record)
|
||||||
return items, failures
|
return items, failures
|
||||||
|
|
||||||
@@ -448,9 +770,8 @@ def _execute(argv: Sequence[str]) -> Response:
|
|||||||
if not matching:
|
if not matching:
|
||||||
details = {
|
details = {
|
||||||
"requested_format": request["format"],
|
"requested_format": request["format"],
|
||||||
"width": request["width"],
|
"requested_size": {"width": request["width"], "height": request["height"]},
|
||||||
"height": request["height"],
|
"minimum_size": {"width": request["width"], "height": request["height"]},
|
||||||
"resize": request["resize"],
|
|
||||||
"candidate_count": len(candidates),
|
"candidate_count": len(candidates),
|
||||||
"format_matches": len(matching) + len(rejected),
|
"format_matches": len(matching) + len(rejected),
|
||||||
"size_matches": 0,
|
"size_matches": 0,
|
||||||
@@ -460,7 +781,7 @@ def _execute(argv: Sequence[str]) -> Response:
|
|||||||
STATUS_NO_SIZE,
|
STATUS_NO_SIZE,
|
||||||
error=error_fragment(
|
error=error_fragment(
|
||||||
"no_size_match",
|
"no_size_match",
|
||||||
"no image matched the requested width and height",
|
"no image is at least the requested width and height",
|
||||||
details=details,
|
details=details,
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user