E2e workflow #21

Merged
mathias2784 merged 8 commits from e2e-workflow into master 2026-10-18 12:02:18 +00:00
3 changed files with 360 additions and 94 deletions
Showing only changes of commit e818e0f31e - Show all commits
+86
View File
@@ -0,0 +1,86 @@
# PhotoTools3
PhotoTools3 uses an OpenAI-compatible vision API to generate photo captions and
keywords, then writes the results to photo metadata. It can also assess photo
quality.
## Setup
Install the Python dependencies with `uv` and make sure ExifTool is installed and
available on `PATH` (it is a system dependency, not a Python package):
```bash
uv sync
```
Create a `.env` file in the project root with the LLM connection details and the
prefix used to identify metadata written by PhotoTools3:
```dotenv
llm_api_base_url=https://your-openai-compatible-api/v1
llm_api_key=your-api-key
llm_default_temperature=0.2
metadata_prefix=photools3:
```
Keyword suggestions are restricted to the vocabulary in `config/keywords.txt`.
## Generate metadata
The `metadata-magic` command accepts one or more image paths and/or directories.
The default mode is `both`, which generates and writes a caption and keywords:
```bash
uv run photools3 metadata-magic photo.dng
```
Choose a mode with `--mode`:
```bash
# Generate only keywords
uv run photools3 metadata-magic photo.dng --mode keywords
# Generate only a caption
uv run photools3 metadata-magic photo.dng --mode captions
# Explicitly request both (the default)
uv run photools3 metadata-magic photo.dng --mode both
```
Pass multiple files to process them sequentially as a batch:
```bash
uv run photools3 metadata-magic photo1.dng photo2.jpg --mode both
```
Directories are scanned at the top level by default. Add `--recursive` to
include supported images in subdirectories. You can combine directories and
individual file paths:
```bash
uv run photools3 metadata-magic ./photos --recursive --mode keywords
uv run photools3 metadata-magic ./photos photo-to-include.dng --mode captions
```
The command reports each processed file and exits with a non-zero status if any
file fails. In a batch, it continues processing the remaining files after an
individual failure. Generated metadata is written using ExifTool and XMP
sidecars.
## Assess photo quality
The `quality` command reports an assessment without changing photo metadata:
```bash
uv run photools3 quality photo.dng
```
## Help
Show the available commands and options with:
```bash
uv run photools3 --help
uv run photools3 metadata-magic --help
uv run photools3 quality --help
```
+109 -26
View File
@@ -3,13 +3,16 @@ from pathlib import Path
from typing import Annotated
import typer
from PIL import Image
from utils.logging_config import LoggingSettings
from photools3.llm_session import (
LlmSessionCaptionsAndKeywords,
LlmSessionQuality,
)
from photools3.llm_session import LlmSessionQuality
from photools3.photo import Photo
from photools3.processing import (
ProcessingMode,
process_batch,
process_photo,
)
app = typer.Typer(help="PhotoTools3: A CLI for image processing.")
LoggingSettings()
@@ -21,9 +24,7 @@ def main(
ctx: typer.Context,
version: bool | None = typer.Option(None, "--version", is_eager=True),
):
"""
Main entry point for PhotoTools3 CLI.
"""
"""Main entry point for PhotoTools3 CLI."""
if version:
typer.echo("photools3 version 0.1.0")
raise typer.Exit()
@@ -33,36 +34,118 @@ def main(
raise typer.Exit()
@app.command("keywords")
def keywords(
file_path: Annotated[Path, typer.Argument(..., help="Path to the image file")],
@app.command("metadata-magic")
def metadata_magic(
paths: Annotated[
list[Path], typer.Argument(..., help="Photo file(s) and/or directory paths")
],
mode: Annotated[
ProcessingMode,
typer.Option(
"--mode",
case_sensitive=False,
help="Metadata to generate and write",
),
] = ProcessingMode.BOTH,
recursive: Annotated[
bool,
typer.Option(
"--recursive/--no-recursive",
help="Recursively discover photos inside input directories",
),
] = False,
):
session = LlmSessionCaptionsAndKeywords()
_run_llm_task(file_path, session)
"""Generate and write keywords, captions, or both for photos."""
_run_processing(paths, mode, recursive)
@app.command("quality")
def quality(
file_path: Annotated[Path, typer.Argument(..., help="Path to the image file")],
):
session = LlmSessionQuality()
_run_llm_task(file_path, session)
def _run_llm_task(
file_path: Path, session: LlmSessionCaptionsAndKeywords | LlmSessionQuality
):
"""Assess a photo's quality without changing its metadata."""
try:
logger.info(f"Processing: {file_path}")
photo = Photo(file_path)
result = session.call([photo])
logger.info(result)
with LlmSessionQuality() as session, Photo(file_path) as photo:
result = session.call([photo])
logger.info("Quality assessment for %s: %s", file_path, result)
except FileNotFoundError:
logger.error(f"Error: File '{file_path}' does not exist.")
logger.error("Photo not found: %s", file_path)
raise typer.Exit(code=1) from None
except Exception as exc:
logger.error("Error assessing photo %s: %s", file_path, exc)
raise typer.Exit(code=1) from exc
def _run_processing(inputs: list[Path], mode: ProcessingMode, recursive: bool) -> None:
photo_paths = _expand_photo_paths(inputs, recursive)
if not photo_paths:
logger.error("No supported photo files found in the given input paths.")
raise typer.Exit(code=1)
except Exception as e: # noqa: BLE001
logger.info("Processing %d photo(s) in %s mode", len(photo_paths), mode.value)
if len(photo_paths) == 1:
path = photo_paths[0]
try:
process_photo(path, mode)
except FileNotFoundError:
logger.error("Photo not found: %s", path)
raise typer.Exit(code=1) from None
except Exception as exc:
logger.error("Failed to process %s: %s", path, exc)
raise typer.Exit(code=1) from exc
logger.info("Processed: %s", path)
return
try:
outcomes = process_batch(photo_paths, mode)
except Exception as exc:
logger.error("Batch processing failed: %s", exc)
raise typer.Exit(code=1) from exc
failures = 0
for outcome in outcomes:
if outcome.succeeded:
logger.info("Processed: %s", outcome.path)
else:
failures += 1
logger.error("Failed to process %s: %s", outcome.path, outcome.error)
logger.info(
"Batch complete: %d succeeded, %d failed", len(outcomes) - failures, failures
)
if failures:
raise typer.Exit(code=1)
logger.error(f"Error processing image: {e}")
def _expand_photo_paths(inputs: list[Path], recursive: bool) -> list[Path]:
"""Expand directories to supported image files, preserving input order."""
supported_extensions = {
".dng",
*(extension.lower() for extension in Image.registered_extensions()),
}
discovered: list[Path] = []
for path in inputs:
if not path.is_dir():
discovered.append(path)
continue
try:
candidates = path.rglob("*") if recursive else path.iterdir()
discovered.extend(
sorted(
candidate
for candidate in candidates
if candidate.is_file()
and candidate.suffix.lower() in supported_extensions
)
)
except OSError as exc:
logger.error("Could not read directory %s: %s", path, exc)
raise typer.Exit(code=1) from exc
# Avoid processing the same exact path more than once while retaining order.
return list(dict.fromkeys(discovered))
if __name__ == "__main__":
+165 -68
View File
@@ -1,100 +1,197 @@
import logging
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from typer.testing import CliRunner
from photools3.cli import app
from photools3.processing import ProcessingMode
runner = CliRunner()
CLI_LOGGER = "photools3.cli"
def _success_outcome(path: Path):
return SimpleNamespace(path=path, succeeded=True, error=None)
@pytest.fixture
def mock_photo_logic(tmp_path):
"""Fixture to mock the core logic of keywords command."""
def test_metadata_magic_command_defaults_to_both_for_single_photo(tmp_path):
image_path = tmp_path / "photo.jpg"
image_path.write_bytes(b"test image")
with (
patch("photools3.cli.LlmSessionCaptionsAndKeywords") as mocked_session,
patch("photools3.cli.Photo") as mocked_photo,
patch("photools3.cli.process_photo") as process_photo,
patch("photools3.cli.process_batch") as process_batch,
):
# Configure session mock
session_instance = mocked_session.return_value
session_instance.call.return_value = {
"caption": "A beautiful test image",
"args": ["test", "mock", "image"], # Wait, the code uses 'tags'
}
# Let me re-check the code in cli.py
# result.get("tags", [])
session_instance.call.return_value = {
"caption": "A beautiful test image",
"tags": ["test", "mock", "image"],
}
# Configure photo mock
photo_instance = mocked_photo.return_value
photo_instance.path = tmp_path / "test_image.jpg"
yield {
"session": session_instance,
"photo_class": mocked_photo,
"photo_instance": photo_instance,
}
@pytest.fixture
def dummy_file(tmp_path):
"""Fixture to create a dummy file for CLI testing."""
dummy_path = tmp_path / "test_image.jpg"
dummy_path.write_text("dummy content")
return dummy_path
def test_keywords_command_success(mock_photo_logic, dummy_file, caplog):
"""Test the 'keywords' command with a valid file path."""
# Ensure the mock photo instance points to our dummy file
mock_photo_logic["photo_class"].return_value.path = dummy_file
with caplog.at_level(logging.INFO, logger=CLI_LOGGER):
result = runner.invoke(app, ["keywords", str(dummy_file)])
result = runner.invoke(app, ["metadata-magic", str(image_path)])
assert result.exit_code == 0
assert "Processing: " in caplog.text
# The result is logged as a raw dictionary string
assert (
"{'caption': 'A beautiful test image', 'tags': ['test', 'mock', 'image']}"
in caplog.text
process_photo.assert_called_once_with(image_path, ProcessingMode.BOTH)
process_batch.assert_not_called()
def test_metadata_magic_command_selects_caption_mode(tmp_path):
image_path = tmp_path / "photo.jpg"
image_path.write_bytes(b"test image")
with patch("photools3.cli.process_photo") as process_photo:
result = runner.invoke(
app, ["metadata-magic", str(image_path), "--mode", "captions"]
)
assert result.exit_code == 0
process_photo.assert_called_once_with(image_path, ProcessingMode.CAPTIONS)
def test_metadata_magic_command_batches_multiple_files(tmp_path):
paths = [tmp_path / "one.jpg", tmp_path / "two.jpg"]
for path in paths:
path.write_bytes(b"test image")
with (
patch("photools3.cli.process_photo") as process_photo,
patch("photools3.cli.process_batch") as process_batch,
):
process_batch.return_value = [_success_outcome(path) for path in paths]
result = runner.invoke(
app,
["metadata-magic", *(str(path) for path in paths), "--mode", "keywords"],
)
assert result.exit_code == 0
process_photo.assert_not_called()
process_batch.assert_called_once_with(paths, ProcessingMode.KEYWORDS)
def test_metadata_magic_command_expands_directories_recursively(tmp_path):
root = tmp_path / "photos"
nested = root / "nested"
nested.mkdir(parents=True)
top_level_photo = root / "one.jpg"
nested_photo = nested / "two.png"
sidecar = root / "one.jpg.xmp"
for path in (top_level_photo, nested_photo, sidecar):
path.write_bytes(b"placeholder")
with (
patch("photools3.cli.process_photo") as process_photo,
patch("photools3.cli.process_batch") as process_batch,
):
process_batch.return_value = [
_success_outcome(nested_photo),
_success_outcome(top_level_photo),
]
result = runner.invoke(
app, ["metadata-magic", str(root), "--recursive", "--mode", "both"]
)
assert result.exit_code == 0
process_photo.assert_not_called()
process_batch.assert_called_once_with(
[nested_photo, top_level_photo], ProcessingMode.BOTH
)
def test_keywords_command_file_not_found(caplog):
"""Test the 'keywords' command with a non-existent file path."""
non_existent_file = Path("non_existent_file.jpg")
def test_metadata_magic_command_does_not_recurse_by_default(tmp_path):
root = tmp_path / "photos"
nested = root / "nested"
nested.mkdir(parents=True)
top_level_photo = root / "one.jpg"
nested_photo = nested / "two.jpg"
top_level_photo.write_bytes(b"placeholder")
nested_photo.write_bytes(b"placeholder")
with caplog.at_level(logging.ERROR, logger=CLI_LOGGER):
result = runner.invoke(app, ["keywords", str(non_existent_file)])
with (
patch("photools3.cli.process_photo") as process_photo,
patch("photools3.cli.process_batch") as process_batch,
):
result = runner.invoke(app, ["metadata-magic", str(root)])
assert result.exit_code == 0
process_photo.assert_called_once_with(top_level_photo, ProcessingMode.BOTH)
process_batch.assert_not_called()
def test_metadata_magic_command_returns_failure_if_any_batch_item_fails(tmp_path):
paths = [tmp_path / "one.jpg", tmp_path / "two.jpg"]
for path in paths:
path.write_bytes(b"test image")
with patch("photools3.cli.process_batch") as process_batch:
process_batch.return_value = [
_success_outcome(paths[0]),
SimpleNamespace(
path=paths[1], succeeded=False, error="RuntimeError: write failed"
),
]
result = runner.invoke(app, ["metadata-magic", *(str(path) for path in paths)])
assert result.exit_code == 1
assert f"Error: File '{non_existent_file}' does not exist." in caplog.text
def test_keywords_command_error_during_processing(mock_photo_logic, dummy_file):
"""Test the 'keywords' command when an exception occurs during processing."""
# Configure the session mock to raise an exception
mock_photo_logic["session"].call.side_effect = Exception("LLM error")
def test_metadata_magic_command_returns_failure_for_single_photo_error(tmp_path):
image_path = tmp_path / "photo.jpg"
image_path.write_bytes(b"test image")
result = runner.invoke(app, ["keywords", str(dummy_file)])
with patch("photools3.cli.process_photo", side_effect=RuntimeError("LLM error")):
result = runner.invoke(app, ["metadata-magic", str(image_path)])
assert result.exit_code == 1
def test_keywords_alias_is_not_registered():
result = runner.invoke(app, ["keywords", "photo.jpg"])
assert result.exit_code == 2
assert "No such command 'keywords'" in result.output
def test_old_process_command_name_is_not_registered():
result = runner.invoke(app, ["process", "photo.jpg"])
assert result.exit_code == 2
assert "No such command 'process'" in result.output
def test_metadata_magic_command_rejects_empty_directory(tmp_path):
with (
patch("photools3.cli.process_photo") as process_photo,
patch("photools3.cli.process_batch") as process_batch,
):
result = runner.invoke(app, ["metadata-magic", str(tmp_path)])
assert result.exit_code == 1
process_photo.assert_not_called()
process_batch.assert_not_called()
def test_quality_command_closes_photo_and_session(tmp_path):
image_path = tmp_path / "photo.jpg"
image_path.write_bytes(b"test image")
with (
patch("photools3.cli.LlmSessionQuality") as session_class,
patch("photools3.cli.Photo") as photo_class,
):
result = runner.invoke(app, ["quality", str(image_path)])
assert result.exit_code == 0
session_class.return_value.__enter__.return_value.call.assert_called_once_with(
[photo_class.return_value.__enter__.return_value]
)
session_class.return_value.__exit__.assert_called_once()
photo_class.return_value.__exit__.assert_called_once()
def test_quality_command_reports_error_and_exits(tmp_path):
image_path = tmp_path / "photo.jpg"
with patch("photools3.cli.Photo", side_effect=FileNotFoundError(image_path)):
result = runner.invoke(app, ["quality", str(image_path)])
# Note: cli.py raises typer.Exit(code=1) before logging the
# "Error processing image: ..." message (the logger.error call is dead
# code after the raise), so only the exit code can be asserted here.
assert result.exit_code == 1
def test_version_command():
"""Test the --version option."""
result = runner.invoke(app, ["--version"])
assert result.exit_code == 0