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": {
|
||||
"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
|
||||
|
||||
import json
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
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
|
||||
|
||||
|
||||
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)])
|
||||
|
||||
code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--resize")
|
||||
|
||||
assert code == envelope.EXIT_OK
|
||||
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:
|
||||
@@ -340,3 +368,183 @@ def test_search_access_error_maps_to_access_error(monkeypatch: pytest.MonkeyPatc
|
||||
assert code == envelope.EXIT_ACCESS_ERROR
|
||||
assert payload["status"] == envelope.STATUS_ACCESS_ERROR
|
||||
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__ += [
|
||||
"ACCEPTED_FORMATS",
|
||||
"ImageResizeError",
|
||||
"RequestDocumentError",
|
||||
"RequestError",
|
||||
"RequestFieldError",
|
||||
|
||||
+349
-28
@@ -14,7 +14,9 @@ from __future__ import annotations
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import struct
|
||||
import sys
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
from typing import Any, Mapping, Sequence
|
||||
|
||||
@@ -30,6 +32,7 @@ from .envelope import (
|
||||
)
|
||||
from .errors import (
|
||||
ACCEPTED_FORMATS,
|
||||
ImageResizeError,
|
||||
RequestDocumentError,
|
||||
RequestError,
|
||||
RequestFieldError,
|
||||
@@ -110,13 +113,13 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
"--width",
|
||||
type=int,
|
||||
metavar="PX",
|
||||
help="desired image width in pixels",
|
||||
help="target width in pixels; only images at least this wide are accepted",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--height",
|
||||
type=int,
|
||||
metavar="PX",
|
||||
help="desired image height in pixels",
|
||||
help="target height in pixels; only images at least this tall are accepted",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-f",
|
||||
@@ -152,8 +155,10 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
"--resize",
|
||||
action="store_true",
|
||||
help=(
|
||||
"accept images larger than the requested size and downscale them; "
|
||||
"without this flag the size must match exactly"
|
||||
"downscale every downloaded image so it fits the requested width "
|
||||
"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(
|
||||
@@ -317,9 +322,18 @@ def _matches_format(candidate: SourceCandidate, image_format: str) -> bool:
|
||||
|
||||
|
||||
def _matches_size(candidate: SourceCandidate, request: Mapping[str, Any]) -> bool:
|
||||
width = request.get("width")
|
||||
height = request.get("height")
|
||||
if width is None and height is None:
|
||||
"""True when ``candidate`` is usable for the requested target size.
|
||||
|
||||
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
|
||||
actual_width = getattr(candidate, "width", None)
|
||||
actual_height = getattr(candidate, "height", None)
|
||||
@@ -327,13 +341,11 @@ def _matches_size(candidate: SourceCandidate, request: Mapping[str, Any]) -> boo
|
||||
return False
|
||||
if not isinstance(actual_width, int) or not isinstance(actual_height, int):
|
||||
return False
|
||||
if request.get("resize"):
|
||||
fits_width = width is None or actual_width >= width
|
||||
fits_height = height is None or actual_height >= height
|
||||
return fits_width and fits_height
|
||||
matches_width = width is None or actual_width == width
|
||||
matches_height = height is None or actual_height == height
|
||||
return matches_width and matches_height
|
||||
if wanted_width is not None and actual_width < wanted_width:
|
||||
return False
|
||||
if wanted_height is not None and actual_height < wanted_height:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
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]
|
||||
|
||||
|
||||
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]:
|
||||
"""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()
|
||||
try:
|
||||
@@ -371,19 +690,22 @@ def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any])
|
||||
destination = directory / f"{index:03d}-{_slug(candidate)}{suffix}"
|
||||
try:
|
||||
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)
|
||||
except (ToolkitError, OSError) as exc:
|
||||
failures += 1
|
||||
record["download"] = {
|
||||
"ok": False,
|
||||
"error": error_fragment(getattr(exc, "code", "download_failed"), str(exc)),
|
||||
}
|
||||
error = error_fragment(getattr(exc, "code", "download_failed"), str(exc))
|
||||
if isinstance(exc, ToolkitError) and exc.details:
|
||||
error["details"] = dict(exc.details)
|
||||
record["download"] = {"ok": False, "error": error}
|
||||
else:
|
||||
record["download"] = {
|
||||
"ok": True,
|
||||
"path": str(destination),
|
||||
"bytes": len(payload),
|
||||
}
|
||||
record["download"] = {"ok": True, "path": str(destination), "bytes": len(payload)}
|
||||
if resize_fragment is not None:
|
||||
record["download"]["resize"] = resize_fragment
|
||||
record["download"]["width"] = resize_fragment["to"]["width"]
|
||||
record["download"]["height"] = resize_fragment["to"]["height"]
|
||||
items.append(record)
|
||||
return items, failures
|
||||
|
||||
@@ -448,9 +770,8 @@ def _execute(argv: Sequence[str]) -> Response:
|
||||
if not matching:
|
||||
details = {
|
||||
"requested_format": request["format"],
|
||||
"width": request["width"],
|
||||
"height": request["height"],
|
||||
"resize": request["resize"],
|
||||
"requested_size": {"width": request["width"], "height": request["height"]},
|
||||
"minimum_size": {"width": request["width"], "height": request["height"]},
|
||||
"candidate_count": len(candidates),
|
||||
"format_matches": len(matching) + len(rejected),
|
||||
"size_matches": 0,
|
||||
@@ -460,7 +781,7 @@ def _execute(argv: Sequence[str]) -> Response:
|
||||
STATUS_NO_SIZE,
|
||||
error=error_fragment(
|
||||
"no_size_match",
|
||||
"no image matched the requested width and height",
|
||||
"no image is at least the requested width and height",
|
||||
details=details,
|
||||
),
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user