From 4c9ad9abe70905ba1d1c62751f3b621e9b612461 Mon Sep 17 00:00:00 2001 From: Mitchell Hewes Date: Sat, 18 Jul 2026 13:54:16 +1000 Subject: [PATCH] Improve autotagger efficiency and configuration --- .devcontainer/devcontainer.json | 10 +- .env.example | 47 ++++ .github/workflows/autotag.yml | 246 ++++++++++++++------ .github/workflows/ci.yml | 42 ++++ .gitignore | 15 +- README.md | 246 ++++++++++++++------ autotagger/__init__.py | 5 + autotagger/analysis.py | 215 +++++++++++++++++ autotagger/checkpoint.py | 96 ++++++++ autotagger/cli.py | 119 ++++++++++ autotagger/config.py | 287 +++++++++++++++++++++++ autotagger/flickr_gateway.py | 299 ++++++++++++++++++++++++ autotagger/models.py | 66 ++++++ autotagger/pipeline.py | 308 ++++++++++++++++++++++++ flickr-autotagger.py | 399 +------------------------------- prompt.example.txt | 6 + pyproject.toml | 38 +++ requirements-dev.txt | 5 + requirements.lock | 91 ++++++++ requirements.txt | 7 +- tests/conftest.py | 32 +++ tests/test_analysis.py | 120 ++++++++++ tests/test_checkpoint.py | 25 ++ tests/test_config.py | 95 ++++++++ tests/test_flickr_gateway.py | 115 +++++++++ tests/test_pipeline.py | 156 +++++++++++++ 26 files changed, 2546 insertions(+), 544 deletions(-) create mode 100644 .env.example create mode 100644 .github/workflows/ci.yml create mode 100644 autotagger/__init__.py create mode 100644 autotagger/analysis.py create mode 100644 autotagger/checkpoint.py create mode 100644 autotagger/cli.py create mode 100644 autotagger/config.py create mode 100644 autotagger/flickr_gateway.py create mode 100644 autotagger/models.py create mode 100644 autotagger/pipeline.py create mode 100644 prompt.example.txt create mode 100644 pyproject.toml create mode 100644 requirements-dev.txt create mode 100644 requirements.lock create mode 100644 tests/conftest.py create mode 100644 tests/test_analysis.py create mode 100644 tests/test_checkpoint.py create mode 100644 tests/test_config.py create mode 100644 tests/test_flickr_gateway.py create mode 100644 tests/test_pipeline.py diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index 34ae911..f5fb19f 100644 --- a/.devcontainer/devcontainer.json +++ b/.devcontainer/devcontainer.json @@ -1,5 +1,5 @@ -{ - "name": "Flickr Autotagger Dev Container", - "image": "mcr.microsoft.com/vscode/devcontainers/python:3.12", - "postCreateCommand": "pip install -r requirements.txt" -} \ No newline at end of file +{ + "name": "Flickr Autotagger Dev Container", + "image": "mcr.microsoft.com/vscode/devcontainers/python:3.12", + "postCreateCommand": "pip install -r requirements.lock && pip install --no-build-isolation --no-deps -e ." +} diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..326ff38 --- /dev/null +++ b/.env.example @@ -0,0 +1,47 @@ +# Required credentials +FLICKR_API_KEY=your_flickr_api_key +FLICKR_API_SECRET=your_flickr_api_secret +OPENAI_API_KEY=your_openai_api_key + +# OpenAI-compatible endpoint and request configuration +OPENAI_BASE_URL=https://api.openai.com/v1 +OPENAI_MODEL=gpt-5-mini +OPENAI_TIMEOUT=60 +OPENAI_MAX_RETRIES=2 +OPENAI_MAX_OUTPUT_TOKENS=800 +OPENAI_IMAGE_DETAIL=low +OPENAI_JSON_MODE=true +ANALYSIS_CONCURRENCY=3 + +# Optional provider compatibility controls +OPENAI_PROVIDER=openai +OPENAI_INSTRUCTION_ROLE=developer +OPENAI_TOKENS_PARAMETER=max_completion_tokens +# For OPENAI_PROVIDER=azure also set: +# AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com +# OPENAI_API_VERSION=your-supported-api-version +# AZURE_OPENAI_API_KEY=your_azure_openai_api_key + +# Flickr selection +FLICKR_PRIVACY_FILTER=1 +FLICKR_IMAGE_URL=url_m +# FLICKR_PHOTOSET_IDS=123456789,987654321 +SKIP_PREFIX=["#", "@"] +DESCRIPTIONS_TO_ANALYZE=["OLYMPUS DIGITAL CAMERA", "Untitled", "DSC_", "IMG_", "DCIM"] +DESCRIPTION_POLICY=missing-or-placeholder + +# Metadata generation and updates +MAX_KEYWORDS=10 +MAX_TITLE_CHARS=120 +MAX_DESCRIPTION_CHARS=1200 +METADATA_LANGUAGE=English +DESCRIPTION_STYLE=detailed, factual, and concise +UPDATE_FIELDS=title,description,tags +TAG_MODE=merge + +# Safety, budget, and output +# MAX_PHOTOS=100 +# MAX_TOTAL_COST=2.00 +RESUME=true +CHECKPOINT_FILE=.autotagger-checkpoint.jsonl +UPDATED_METADATA_FILE=updated_metadata.json diff --git a/.github/workflows/autotag.yml b/.github/workflows/autotag.yml index 3285ce6..c45153b 100644 --- a/.github/workflows/autotag.yml +++ b/.github/workflows/autotag.yml @@ -1,67 +1,179 @@ -name: Autotag Photos - -on: - workflow_dispatch: - inputs: - openai_model: - description: 'OpenAI Model' - required: true - default: 'gpt-5-mini' - openai_prompt_cost: - description: 'OpenAI Cost per 1M Prompt Tokens' - required: true - default: '0.25' - openai_completion_cost: - description: 'OpenAI Cost per 1M Completion Tokens' - required: true - default: '2.00' - openai_vision_cost: - description: 'OpenAI Vision Cost per Image' - required: true - default: '0.0' - flickr_image_url: - description: 'Flickr Image URL (default: url_m)' - required: true - default: 'url_m' - flickr_privacy_filter: - description: 'Flickr Privacy Filter (default: public)' - required: true - default: '1' - -jobs: - flickr-autotagger: - runs-on: ubuntu-latest - container: - image: mcr.microsoft.com/vscode/devcontainers/python:3.12 - - steps: - - uses: actions/checkout@v4 - - - name: Install dependencies - run: | - pip install -r requirements.txt - - - name: Run Flickr Autotagger - env: - OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} - FLICKR_API_KEY: ${{ secrets.FLICKR_API_KEY }} - FLICKR_API_SECRET: ${{ secrets.FLICKR_API_SECRET }} - FLICKR_OAUTH_TOKEN: ${{ secrets.FLICKR_OAUTH_TOKEN }} - OPENAI_MODEL: ${{ github.event.inputs.openai_model }} - OPENAI_COST_PER_1M_PROMPT_TOKEN: ${{ github.event.inputs.openai_prompt_cost }} - OPENAI_COST_PER_1M_COMPLETION_TOKEN: ${{ github.event.inputs.openai_completion_cost }} - OPENAI_VISION_COST_PER_IMAGE: ${{ github.event.inputs.openai_vision_cost }} - FLICKR_IMAGE_URL: ${{ github.event.inputs.flickr_image_url }} - FLICKR_PRIVACY_FILTER: ${{ github.event.inputs.flickr_privacy_filter }} - run: | - python flickr-autotagger.py - - - name: Upload Metadata File - if: success() - continue-on-error: true - uses: actions/upload-artifact@v4 - with: - name: updated-metadata - path: updated_metadata.json - - if-no-files-found: ignore +name: Autotag Photos + +on: + workflow_dispatch: + inputs: + apply_changes: + description: Apply generated metadata to Flickr + type: boolean + required: true + default: true + openai_model: + description: OpenAI model or Azure deployment name + required: true + default: gpt-5-mini + openai_prompt_cost: + description: Cost per 1M prompt tokens + required: true + default: "0.25" + openai_completion_cost: + description: Cost per 1M completion tokens + required: true + default: "2.00" + openai_vision_cost: + description: Fixed cost per image + required: true + default: "0.0" + openai_max_output_tokens: + description: Maximum output tokens per image + required: true + default: "800" + analysis_concurrency: + description: Concurrent OpenAI analyses + required: true + default: "3" + flickr_image_url: + description: Flickr image URL extra + required: true + default: url_m + flickr_privacy_filter: + description: Flickr privacy filter (all or 1-5) + required: true + default: "1" + flickr_photoset_id: + description: Optional single photoset ID + required: false + default: "" + tag_mode: + description: Preserve existing tags or replace them + type: choice + required: true + default: merge + options: + - merge + - replace + update_fields: + description: Comma-separated fields (title,description,tags) + required: true + default: title,description,tags + max_photos: + description: Optional maximum number of new analyses + required: false + default: "" + max_total_cost: + description: Optional recorded OpenAI cost limit + required: false + default: "" + resume: + description: Reuse the most recent workflow checkpoint + type: boolean + required: true + default: true + +permissions: + contents: read + +concurrency: + group: flickr-autotagger-${{ github.repository }} + cancel-in-progress: false + +jobs: + flickr-autotagger: + runs-on: ubuntu-latest + timeout-minutes: 360 + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + cache-dependency-path: requirements.lock + + - name: Install locked dependencies + run: | + python -m pip install --requirement requirements.lock + python -m pip install --no-build-isolation --no-deps . + + - name: Verify package + run: | + ruff check . + ruff format --check . + pytest + + - name: Restore prior checkpoint + if: ${{ inputs.resume }} + uses: actions/cache/restore@v4 + with: + path: .autotagger-checkpoint.jsonl + key: autotagger-checkpoint-${{ github.run_id }} + restore-keys: | + autotagger-checkpoint- + + - name: Run Flickr Autotagger + env: + APPLY_CHANGES: ${{ inputs.apply_changes }} + RESUME: ${{ inputs.resume }} + OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} + OPENAI_BASE_URL: ${{ vars.OPENAI_BASE_URL }} + OPENAI_PROVIDER: ${{ vars.OPENAI_PROVIDER || 'openai' }} + OPENAI_INSTRUCTION_ROLE: ${{ vars.OPENAI_INSTRUCTION_ROLE || 'developer' }} + OPENAI_TOKENS_PARAMETER: ${{ vars.OPENAI_TOKENS_PARAMETER || 'max_completion_tokens' }} + OPENAI_JSON_MODE: ${{ vars.OPENAI_JSON_MODE || 'true' }} + OPENAI_IMAGE_DETAIL: ${{ vars.OPENAI_IMAGE_DETAIL || 'low' }} + AZURE_OPENAI_API_KEY: ${{ secrets.AZURE_OPENAI_API_KEY }} + AZURE_OPENAI_ENDPOINT: ${{ vars.AZURE_OPENAI_ENDPOINT }} + OPENAI_API_VERSION: ${{ vars.OPENAI_API_VERSION }} + FLICKR_API_KEY: ${{ secrets.FLICKR_API_KEY }} + FLICKR_API_SECRET: ${{ secrets.FLICKR_API_SECRET }} + FLICKR_OAUTH_TOKEN: ${{ secrets.FLICKR_OAUTH_TOKEN }} + OPENAI_MODEL: ${{ inputs.openai_model }} + OPENAI_COST_PER_1M_PROMPT_TOKEN: ${{ inputs.openai_prompt_cost }} + OPENAI_COST_PER_1M_COMPLETION_TOKEN: ${{ inputs.openai_completion_cost }} + OPENAI_VISION_COST_PER_IMAGE: ${{ inputs.openai_vision_cost }} + OPENAI_MAX_OUTPUT_TOKENS: ${{ inputs.openai_max_output_tokens }} + ANALYSIS_CONCURRENCY: ${{ inputs.analysis_concurrency }} + FLICKR_IMAGE_URL: ${{ inputs.flickr_image_url }} + FLICKR_PRIVACY_FILTER: ${{ inputs.flickr_privacy_filter }} + FLICKR_PHOTOSET_ID: ${{ inputs.flickr_photoset_id }} + TAG_MODE: ${{ inputs.tag_mode }} + UPDATE_FIELDS: ${{ inputs.update_fields }} + MAX_PHOTOS: ${{ inputs.max_photos }} + MAX_TOTAL_COST: ${{ inputs.max_total_cost }} + MAX_KEYWORDS: ${{ vars.MAX_KEYWORDS || '10' }} + MAX_TITLE_CHARS: ${{ vars.MAX_TITLE_CHARS || '120' }} + MAX_DESCRIPTION_CHARS: ${{ vars.MAX_DESCRIPTION_CHARS || '1200' }} + METADATA_LANGUAGE: ${{ vars.METADATA_LANGUAGE || 'English' }} + DESCRIPTION_STYLE: ${{ vars.DESCRIPTION_STYLE || 'detailed, factual, and concise' }} + DESCRIPTION_POLICY: ${{ vars.DESCRIPTION_POLICY || 'missing-or-placeholder' }} + DESCRIPTIONS_TO_ANALYZE: ${{ vars.DESCRIPTIONS_TO_ANALYZE }} + SKIP_PREFIX: ${{ vars.SKIP_PREFIX }} + PHOTOSET_ALLOWLIST: ${{ vars.PHOTOSET_ALLOWLIST }} + PHOTOSET_DENYLIST: ${{ vars.PHOTOSET_DENYLIST }} + PROMPT_FILE: ${{ vars.PROMPT_FILE }} + run: | + arguments=() + if [[ "$APPLY_CHANGES" == "true" ]]; then + arguments+=(--apply) + fi + python flickr-autotagger.py "${arguments[@]}" + + - name: Save checkpoint + if: ${{ always() && inputs.resume && hashFiles('.autotagger-checkpoint.jsonl') != '' }} + uses: actions/cache/save@v4 + with: + path: .autotagger-checkpoint.jsonl + key: autotagger-checkpoint-${{ github.run_id }} + + - name: Upload run report and checkpoint + if: always() + continue-on-error: true + uses: actions/upload-artifact@v4 + with: + name: autotagger-run-${{ github.run_id }} + path: | + updated_metadata.json + .autotagger-checkpoint.jsonl + if-no-files-found: ignore + retention-days: 30 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..79f5b22 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,42 @@ +name: CI + +on: + pull_request: + push: + branches: + - main + +permissions: + contents: read + +concurrency: + group: ci-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.11", "3.12"] + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + cache-dependency-path: requirements.lock + + - name: Install locked dependencies + run: | + python -m pip install --requirement requirements.lock + python -m pip install --no-build-isolation --no-deps . + + - name: Lint and test + run: | + ruff check . + ruff format --check . + pytest diff --git a/.gitignore b/.gitignore index 8236986..0998ac4 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,12 @@ -.env -.venv -flickr_token.json \ No newline at end of file +.env +.venv +flickr_token.json +*.py[cod] +__pycache__/ +.pytest_cache/ +.ruff_cache/ +*.egg-info/ +build/ +dist/ +.autotagger-checkpoint.jsonl +updated_metadata.json diff --git a/README.md b/README.md index 6d0435b..a7dd36a 100644 --- a/README.md +++ b/README.md @@ -1,71 +1,175 @@ -# Flickr GPT Autotagger - -This Python script automates updating Flickr image metadata using OpenAI's GPT-4 vision model. It analyses images in Flickr photosets, generates titles, descriptions, and relevant keywords, and then updates the Flickr metadata accordingly. - -## Features - -- OAuth authentication with Flickr API -- Retrieval of all photosets or processing of a specific photoset -- Image analysis using OpenAI's GPT-4 vision model -- Automatic updating of Flickr image tags, titles, and descriptions -- Cost tracking for OpenAI API usage - -## Prerequisites - -- Python 3.x -- Flickr API key and secret -- OpenAI API key - -## Installation - -1. Clone this repository or download the script. -2. Install the required Python packages: `pip install flickrapi openai python-dotenv` -3. Create a `.env` file (or otherwise set the environment variables) in the same directory as the script with the following content: - -``` -FLICKR_API_KEY=your_flickr_api_key -FLICKR_API_SECRET=your_flickr_api_secret -OPENAI_API_KEY=your_openai_api_key -FLICKR_PHOTOSET_ID=optional_specific_photoset_id -``` - -## Usage - -Run the script using Python: `python flickr_autotagger.py` - -The script will: -1. Authenticate with Flickr using OAuth (requesting oob input) -2. Retrieve photosets (all or a specific one) -3. Process each image in the photosets -4. Analyse images using OpenAI's GPT-4 vision model -5. Update Flickr metadata with the generated information -6. Track and display the cost of OpenAI API usage - -## Configuration - -The following optional environment variables can be set to configure how the script works, defaults have been set so an execution should be successful without configuring any of the below: - -- `OPENAI_MODEL`: Specifies the OpenAI model to use (default: "gpt-4o-2024-08-06") -- `OPENAI_COST_PER_PROMPT_TOKEN`: Cost per prompt token (default: 0.00250) -- `OPENAI_COST_PER_COMPLETION_TOKEN`: Cost per completion token (default: 0.01000) -- `OPENAI_VISION_COST_PER_IMAGE`: Fixed cost per image analysis (default: 0.000213) -- `FLICKR_PRIVACY_FILTER`: Tells the Flickr API to return only images with a specific privacy level (0. none, 1. public/default, 2. friends, 3. family, 4. friends & family, 5. private) -- `MAX_KEYWORDS`: Sets the maximum number of keywords to request GPT to generate (default: 10) -- `FLICKR_TOKEN_FILE`: Specifies the file name to store the Flickr OAuth token (default: "flickr_token.json") -- `DESCRIPTIONS_TO_ANALYZE`: A list of strings that, if found at the start of an existing Flickr description, will cause the script to analyze and update the image metadata (default: '["OLYMPUS DIGITAL CAMERA", "Untitled", "DSC_", "IMG_", "DCIM"]') -- `SKIP_PREFIX`: A list of characters that, if found at the start of a photoset name, will cause the script to skip processing that photoset (default: '["#", "@"]') -- `UPDATED_METADATA_FILE`: Specifies the file name to store the updated metadata for all processed images (default: "updated_metadata.json") - -As of 2024-08-16, the script is configured to use gpt-4o, however, if you'd like to use the more economical gpt-4o-mini, the following could be used: - -```python -OPENAI_MODEL = "gpt-4o-mini" -OPENAI_COST_PER_1M_PROMPT_TOKEN = 0.15 -OPENAI_COST_PER_1M_COMPLETION_TOKEN = 0.6 -OPENAI_VISION_COST_PER_IMAGE = 0.000425 -``` - -## Notes - -- The script skips photosets with names starting with "#" or "@" -- Images that already have descriptions are skipped unless they start with some that appear autogenerated (e.g. DCIM, IMG_, DSC_) +# Flickr GPT Autotagger + +Flickr GPT Autotagger generates titles, descriptions, and keywords for Flickr photos with an OpenAI-compatible vision model. It supports safe dry-runs, resumable checkpoints, bounded concurrent analysis, existing-tag preservation, cost limits, and explicit apply operations. + +## Safety model + +Running the command without `--apply` never changes Flickr. It writes a JSON run report and a durable JSON Lines checkpoint that can be reviewed or applied later. + +```bash +# Analyse and create a plan only +python flickr-autotagger.py + +# Analyse and update Flickr +python flickr-autotagger.py --apply + +# Apply an existing report or checkpoint without calling OpenAI +python flickr-autotagger.py --apply-plan updated_metadata.json +``` + +Flickr tags are merged with existing tags by default. Set `TAG_MODE=replace` only when the generated keywords should replace every existing tag. + +## Installation + +Python 3.11 or newer is required. + +```bash +python -m venv .venv +source .venv/bin/activate +pip install -r requirements.lock +pip install --no-build-isolation --no-deps -e . +cp .env.example .env +``` + +Set these credentials in `.env`: + +```dotenv +FLICKR_API_KEY=your_flickr_api_key +FLICKR_API_SECRET=your_flickr_api_secret +OPENAI_API_KEY=your_openai_api_key +``` + +If `FLICKR_OAUTH_TOKEN` is not supplied, the first local run performs interactive Flickr OAuth and writes a private token file. `FLICKR_TOKEN_FILE` changes its location. + +## Custom OpenAI endpoint + +Set `OPENAI_BASE_URL` to the base URL of OpenAI or an OpenAI-compatible service. Include the API version prefix expected by that service, but not `/chat/completions`. + +```dotenv +OPENAI_BASE_URL=https://api.openai.com/v1 + +# Example local compatible endpoint +# OPENAI_BASE_URL=http://localhost:11434/v1 +``` + +The configured endpoint receives the API key and Flickr image URLs, so only use a trusted service. Compatible services must support image URL content through Chat Completions. The following settings handle common compatibility differences: + +- `OPENAI_INSTRUCTION_ROLE=developer|system` +- `OPENAI_TOKENS_PARAMETER=max_completion_tokens|max_tokens` +- `OPENAI_JSON_MODE=true|false` + +Azure OpenAI is supported separately: + +```dotenv +OPENAI_PROVIDER=azure +AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com +OPENAI_API_VERSION=your-supported-api-version +AZURE_OPENAI_API_KEY=your_azure_openai_api_key +OPENAI_MODEL=your-deployment-name +``` + +## Configuration + +All options can be placed in `.env` or supplied as environment variables. CLI arguments override the corresponding environment setting where available. + +### OpenAI and generation + +| Variable | Default | Purpose | +| --- | --- | --- | +| `OPENAI_BASE_URL` | `https://api.openai.com/v1` | OpenAI-compatible API base URL | +| `OPENAI_MODEL` | `gpt-5-mini` | Model or Azure deployment name | +| `OPENAI_TIMEOUT` | `60` | Request timeout in seconds | +| `OPENAI_MAX_RETRIES` | `2` | SDK retries for transient failures | +| `OPENAI_MAX_OUTPUT_TOKENS` | `800` | Output ceiling per analysis | +| `OPENAI_IMAGE_DETAIL` | `low` | `low`, `high`, or `auto` | +| `ANALYSIS_CONCURRENCY` | `3` | Maximum simultaneous model requests | +| `MAX_KEYWORDS` | `10` | Maximum generated keywords | +| `MAX_TITLE_CHARS` | `120` | Maximum stored title length | +| `MAX_DESCRIPTION_CHARS` | `1200` | Maximum stored description length | +| `METADATA_LANGUAGE` | `English` | Requested metadata language | +| `DESCRIPTION_STYLE` | `detailed, factual, and concise` | Description-writing guidance | +| `PROMPT_FILE` | unset | Custom prompt template path | + +Custom prompts may use `{{MAX_KEYWORDS}}`, `{{MAX_TITLE_CHARS}}`, `{{MAX_DESCRIPTION_CHARS}}`, `{{LANGUAGE}}`, and `{{DESCRIPTION_STYLE}}`. See `prompt.example.txt`. + +### Flickr selection and updates + +| Variable | Default | Purpose | +| --- | --- | --- | +| `FLICKR_PRIVACY_FILTER` | `1` | `all` or Flickr privacy level `1`-`5` | +| `FLICKR_IMAGE_URL` | `url_m` | Flickr image-size extra sent to the model | +| `FLICKR_PHOTOSET_ID` | unset | Legacy single photoset selector | +| `FLICKR_PHOTOSET_IDS` | unset | JSON or comma-separated photoset IDs | +| `PHOTOSET_ALLOWLIST` | unset | Allowed photoset IDs or exact titles | +| `PHOTOSET_DENYLIST` | unset | Denied photoset IDs or exact titles | +| `SKIP_PREFIX` | `["#", "@"]` | Photoset-title prefixes to skip | +| `DESCRIPTIONS_TO_ANALYZE` | camera filename defaults | Existing-description prefixes eligible for replacement | +| `DESCRIPTION_POLICY` | `missing-or-placeholder` | `missing-or-placeholder`, `always`, or `never` | +| `UPDATE_FIELDS` | `title,description,tags` | Flickr fields to update | +| `TAG_MODE` | `merge` | Preserve existing tags or `replace` them | + +JSON-list configuration is parsed as JSON, never executable Python. + +### Limits, checkpoints, and cost reporting + +| Variable | Default | Purpose | +| --- | --- | --- | +| `MAX_PHOTOS` | unset | Maximum number of new analyses | +| `MAX_TOTAL_COST` | unset | Stop after recorded cost reaches this amount | +| `RESUME` | `true` | Reuse matching cached analysis and skip updated photos | +| `CHECKPOINT_FILE` | `.autotagger-checkpoint.jsonl` | Append-only durable event log | +| `UPDATED_METADATA_FILE` | `updated_metadata.json` | Atomic report for the current run | +| `FAIL_ON_ERROR` | `false` | Return a non-zero exit code when any photo fails | + +Cost reporting uses `OPENAI_COST_PER_1M_PROMPT_TOKEN`, `OPENAI_COST_PER_1M_COMPLETION_TOKEN`, and `OPENAI_VISION_COST_PER_IMAGE`. These values are manual estimates because models and compatible providers have different prices. A concurrent batch can exceed `MAX_TOTAL_COST` slightly before all in-flight usage is known. + +## CLI overrides + +```text +--apply Update Flickr after analysis +--apply-plan PATH Apply an existing JSON/JSONL plan without OpenAI +--photoset-id ID Select a photoset; may be repeated +--limit N Limit new analyses +--max-cost AMOUNT Set a recorded-cost limit +--concurrency N Set OpenAI concurrency +--prompt-file PATH Use a custom prompt +--no-resume Ignore previous checkpoint state +--strict Fail the process if any photo fails +``` + +## GitHub Actions + +The **Autotag Photos** workflow remains manually dispatched and applies changes by default. Clear **Apply generated metadata to Flickr** to perform a dry-run. The workflow: + +- prevents overlapping autotag runs; +- verifies lint and tests before accessing Flickr; +- restores and saves the latest checkpoint; +- uploads the run report and checkpoint even after partial failure; +- installs the exact versions in `requirements.lock`. + +Configure these GitHub secrets: + +- `FLICKR_API_KEY` +- `FLICKR_API_SECRET` +- `FLICKR_OAUTH_TOKEN` +- `OPENAI_API_KEY` + +Those four existing secrets are sufficient for the standard OpenAI setup: no repository variables or Azure secrets are required. Open **Autotag Photos**, choose **Run workflow**, and accept the defaults to run and apply the generated metadata as before. + +Set the repository variable `OPENAI_BASE_URL` only when using a custom compatible endpoint. It is deliberately not a workflow-dispatch text input, preventing an operator from redirecting secrets to an arbitrary host. Optional provider variables are `OPENAI_PROVIDER`, `OPENAI_INSTRUCTION_ROLE`, `OPENAI_TOKENS_PARAMETER`, and `OPENAI_JSON_MODE`. Azure additionally uses the `AZURE_OPENAI_API_KEY` secret plus the `AZURE_OPENAI_ENDPOINT` and `OPENAI_API_VERSION` variables. + +## Development + +```bash +pip install -r requirements.lock +pip install --no-build-isolation --no-deps -e . +ruff check . +ruff format --check . +pytest +``` + +The CI workflow tests Python 3.11 and 3.12. Regenerate the lock after changing dependency constraints with: + +```bash +pip-compile --allow-unsafe --output-file=requirements.lock --strip-extras requirements-dev.txt +``` diff --git a/autotagger/__init__.py b/autotagger/__init__.py new file mode 100644 index 0000000..df8b7f6 --- /dev/null +++ b/autotagger/__init__.py @@ -0,0 +1,5 @@ +"""Flickr metadata autotagging package.""" + +__all__ = ["__version__"] + +__version__ = "1.0.0" diff --git a/autotagger/analysis.py b/autotagger/analysis.py new file mode 100644 index 0000000..5dd24fe --- /dev/null +++ b/autotagger/analysis.py @@ -0,0 +1,215 @@ +"""OpenAI-backed image analysis with strict local response validation.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import re +from collections.abc import Sequence +from pathlib import Path +from typing import Any, Protocol + +from openai import AsyncAzureOpenAI, AsyncOpenAI + +from .config import Settings +from .models import Analysis, PhotoCandidate, Usage + + +class AnalysisValidationError(ValueError): + """Raised when a model response is not usable as Flickr metadata.""" + + +class ImageAnalyzer(Protocol): + fingerprint: str + + async def analyze(self, photo: PhotoCandidate) -> Analysis: ... + + +def _strip_markdown_fence(value: str) -> str: + value = value.strip() + if value.startswith("```") and value.endswith("```"): + value = value[3:-3].strip() + if value.lower().startswith("json"): + value = value[4:].strip() + return value + + +def _normalise_keyword(value: str) -> str: + value = re.sub(r"\s+", "-", value.strip().lower()) + return re.sub(r"[^\w-]", "", value, flags=re.UNICODE) + + +def parse_analysis_response( + content: str, + *, + max_keywords: int, + max_title_chars: int, + max_description_chars: int, + usage: Usage | None = None, +) -> Analysis: + try: + value = json.loads(_strip_markdown_fence(content)) + except json.JSONDecodeError as exc: + raise AnalysisValidationError("model response was not valid JSON") from exc + if not isinstance(value, dict): + raise AnalysisValidationError("model response must be a JSON object") + + title = value.get("title") + description = value.get("description") + keywords = value.get("keywords") + if not isinstance(title, str) or not title.strip(): + raise AnalysisValidationError("title must be a non-empty string") + if not isinstance(description, str) or not description.strip(): + raise AnalysisValidationError("description must be a non-empty string") + if not isinstance(keywords, list) or not all(isinstance(item, str) for item in keywords): + raise AnalysisValidationError("keywords must be a list of strings") + + normalised_keywords: list[str] = [] + seen: set[str] = set() + for item in keywords: + keyword = _normalise_keyword(item) + if keyword and keyword not in seen: + normalised_keywords.append(keyword) + seen.add(keyword) + if len(normalised_keywords) >= max_keywords: + break + if not normalised_keywords: + raise AnalysisValidationError("keywords must contain at least one usable value") + + return Analysis( + title=title.strip()[:max_title_chars], + description=description.strip()[:max_description_chars], + keywords=tuple(normalised_keywords), + usage=usage or Usage(), + ) + + +def render_prompt(settings: Settings) -> str: + replacements = { + "{{MAX_KEYWORDS}}": str(settings.max_keywords), + "{{MAX_TITLE_CHARS}}": str(settings.max_title_chars), + "{{MAX_DESCRIPTION_CHARS}}": str(settings.max_description_chars), + "{{LANGUAGE}}": settings.language, + "{{DESCRIPTION_STYLE}}": settings.description_style, + } + if settings.prompt_file: + prompt = Path(settings.prompt_file).read_text(encoding="utf-8") + for token, replacement in replacements.items(): + prompt = prompt.replace(token, replacement) + return prompt + return ( + "Generate Flickr metadata as one valid JSON object with exactly these fields:\n" + f'- "title": a self-contained title in {settings.language}, no more than ' + f"{settings.max_title_chars} characters.\n" + f'- "description": a {settings.description_style} description in ' + f"{settings.language}, no more than {settings.max_description_chars} characters.\n" + f'- "keywords": up to {settings.max_keywords} lowercase keywords without spaces ' + "or symbols.\n" + "Use only visible image evidence and the optional album/location context. Do not " + "mention that context or make unsupported location claims." + ) + + +class OpenAIImageAnalyzer: + def __init__(self, settings: Settings, client: Any | None = None): + self.settings = settings + self.prompt = render_prompt(settings) + self.client = client or self._create_client() + fingerprint_value = json.dumps( + { + "provider": settings.openai_provider, + "base_url": settings.openai_base_url, + "model": settings.openai_model, + "role": settings.openai_instruction_role, + "json_mode": settings.openai_json_mode, + "image_detail": settings.openai_image_detail, + "prompt": self.prompt, + }, + sort_keys=True, + ) + self.fingerprint = hashlib.sha256(fingerprint_value.encode()).hexdigest()[:16] + + def _create_client(self) -> AsyncOpenAI | AsyncAzureOpenAI: + common: dict[str, Any] = { + "api_key": self.settings.openai_api_key, + "timeout": self.settings.openai_timeout, + "max_retries": self.settings.openai_max_retries, + } + if self.settings.openai_provider == "azure": + return AsyncAzureOpenAI( + **common, + azure_endpoint=self.settings.azure_openai_endpoint, + api_version=self.settings.azure_openai_api_version, + ) + return AsyncOpenAI(**common, base_url=self.settings.openai_base_url) + + async def analyze(self, photo: PhotoCandidate) -> Analysis: + context: dict[str, Any] = {} + if photo.photoset_title: + context["albumTitle"] = photo.photoset_title + if photo.photoset_description: + context["albumDescription"] = photo.photoset_description + if photo.location: + context["location"] = photo.location + + request: dict[str, Any] = { + "model": self.settings.openai_model, + "messages": [ + {"role": self.settings.openai_instruction_role, "content": self.prompt}, + { + "role": "user", + "content": [ + {"type": "text", "text": json.dumps(context)}, + { + "type": "image_url", + "image_url": { + "url": photo.image_url, + "detail": self.settings.openai_image_detail, + }, + }, + ], + }, + ], + self.settings.openai_tokens_parameter: self.settings.openai_max_output_tokens, + } + if self.settings.openai_json_mode: + request["response_format"] = {"type": "json_object"} + + response = await self.client.chat.completions.create(**request) + response_usage = getattr(response, "usage", None) + prompt_tokens = int(getattr(response_usage, "prompt_tokens", 0) or 0) + completion_tokens = int(getattr(response_usage, "completion_tokens", 0) or 0) + cost = ( + prompt_tokens * self.settings.openai_cost_per_1m_prompt_tokens / 1_000_000 + + completion_tokens * self.settings.openai_cost_per_1m_completion_tokens / 1_000_000 + + self.settings.openai_vision_cost_per_image + ) + usage = Usage(prompt_tokens, completion_tokens, cost) + content = response.choices[0].message.content + if not isinstance(content, str): + raise AnalysisValidationError("model returned no text content") + return parse_analysis_response( + content, + max_keywords=self.settings.max_keywords, + max_title_chars=self.settings.max_title_chars, + max_description_chars=self.settings.max_description_chars, + usage=usage, + ) + + +async def analyze_concurrently( + analyzer: ImageAnalyzer, + photos: Sequence[PhotoCandidate], + concurrency: int, +) -> list[Analysis | Exception]: + semaphore = asyncio.Semaphore(concurrency) + + async def guarded(photo: PhotoCandidate) -> Analysis | Exception: + async with semaphore: + try: + return await analyzer.analyze(photo) + except Exception as exc: # returned for per-photo audit and continuation + return exc + + return list(await asyncio.gather(*(guarded(photo) for photo in photos))) diff --git a/autotagger/checkpoint.py b/autotagger/checkpoint.py new file mode 100644 index 0000000..81ac594 --- /dev/null +++ b/autotagger/checkpoint.py @@ -0,0 +1,96 @@ +"""Durable JSON Lines checkpoints and atomic run summaries.""" + +from __future__ import annotations + +import json +import os +import tempfile +import threading +from collections.abc import Iterable +from pathlib import Path +from typing import Any + + +class CheckpointStore: + def __init__(self, path: Path): + self.path = path + self._lock = threading.Lock() + self._latest: dict[str, dict[str, Any]] | None = None + + def load_latest(self) -> dict[str, dict[str, Any]]: + if self._latest is not None: + return self._latest + latest: dict[str, dict[str, Any]] = {} + if self.path.exists(): + with self.path.open("r", encoding="utf-8") as handle: + for line_number, line in enumerate(handle, 1): + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError as exc: + raise ValueError( + f"Invalid JSON in {self.path} at line {line_number}" + ) from exc + photo_id = record.get("photo_id") + if isinstance(photo_id, str) and ( + record.get("status") != "skipped" or photo_id not in latest + ): + latest[photo_id] = record + self._latest = latest + return latest + + def append(self, record: dict[str, Any]) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + encoded = json.dumps(record, ensure_ascii=False, sort_keys=True) + with self._lock: + with self.path.open("a", encoding="utf-8") as handle: + handle.write(encoded + "\n") + handle.flush() + os.fsync(handle.fileno()) + if self._latest is None: + self._latest = {} + photo_id = record.get("photo_id") + if isinstance(photo_id, str) and ( + record.get("status") != "skipped" or photo_id not in self._latest + ): + self._latest[photo_id] = record + + +def write_summary(path: Path, records: Iterable[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = json.dumps(list(records), ensure_ascii=False, indent=2, sort_keys=True) + "\n" + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{path.name}.", suffix=".tmp", dir=path.parent + ) + temporary_path = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, path) + finally: + if temporary_path.exists(): + temporary_path.unlink() + + +def load_plan(path: Path) -> list[dict[str, Any]]: + text = path.read_text(encoding="utf-8") + stripped = text.lstrip() + if stripped.startswith("["): + value = json.loads(text) + if not isinstance(value, list): + raise ValueError(f"Plan must be a JSON array: {path}") + return [item for item in value if isinstance(item, dict)] + records: list[dict[str, Any]] = [] + for line_number, line in enumerate(text.splitlines(), 1): + if not line.strip(): + continue + try: + value = json.loads(line) + except json.JSONDecodeError as exc: + raise ValueError(f"Invalid JSON in {path} at line {line_number}") from exc + if isinstance(value, dict): + records.append(value) + return records diff --git a/autotagger/cli.py b/autotagger/cli.py new file mode 100644 index 0000000..31bd7e1 --- /dev/null +++ b/autotagger/cli.py @@ -0,0 +1,119 @@ +"""Command-line interface for safe analysis and explicit Flickr updates.""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +import sys +from dataclasses import replace +from pathlib import Path + +from dotenv import load_dotenv + +from .analysis import OpenAIImageAnalyzer +from .config import ConfigurationError, Settings +from .flickr_gateway import FlickrGateway +from .pipeline import AutotaggerPipeline, RunSummary + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Generate Flickr metadata with an OpenAI-compatible vision model." + ) + parser.add_argument( + "--apply", + action="store_true", + help="Apply generated metadata to Flickr. Without this flag the run is a dry-run.", + ) + parser.add_argument( + "--apply-plan", + type=Path, + metavar="PATH", + help="Apply analyses from a previous JSON or JSONL plan without calling OpenAI.", + ) + parser.add_argument("--photoset-id", action="append", help="Process only this photoset ID.") + parser.add_argument("--limit", type=int, help="Maximum number of new OpenAI analyses.") + parser.add_argument("--max-cost", type=float, help="Stop after reaching this recorded cost.") + parser.add_argument("--concurrency", type=int, help="Concurrent OpenAI analyses.") + parser.add_argument("--prompt-file", type=Path, help="Custom prompt template file.") + parser.add_argument("--no-resume", action="store_true", help="Ignore existing checkpoints.") + parser.add_argument("--strict", action="store_true", help="Exit non-zero if any photo fails.") + parser.add_argument( + "--log-level", + choices=("DEBUG", "INFO", "WARNING", "ERROR"), + default="INFO", + ) + return parser + + +def _apply_overrides(settings: Settings, arguments: argparse.Namespace) -> Settings: + values = {} + if arguments.photoset_id: + values["flickr_photoset_ids"] = tuple(arguments.photoset_id) + if arguments.limit is not None: + if arguments.limit < 1: + raise ConfigurationError("--limit must be greater than zero") + values["max_photos"] = arguments.limit + if arguments.max_cost is not None: + if arguments.max_cost <= 0: + raise ConfigurationError("--max-cost must be greater than zero") + values["max_total_cost"] = arguments.max_cost + if arguments.concurrency is not None: + if arguments.concurrency < 1: + raise ConfigurationError("--concurrency must be greater than zero") + values["analysis_concurrency"] = arguments.concurrency + if arguments.prompt_file is not None: + values["prompt_file"] = arguments.prompt_file + if arguments.no_resume: + values["resume"] = False + if arguments.strict: + values["strict"] = True + updated = replace(settings, **values) + updated.validate(require_openai=not bool(arguments.apply_plan), require_flickr=True) + return updated + + +def _report(summary: RunSummary, output_file: Path, *, dry_run: bool) -> None: + mode = "Dry-run complete" if dry_run else "Run complete" + print( + f"{mode}: analyzed={summary.analyzed}, updated={summary.updated}, " + f"skipped={summary.skipped}, failed={summary.failed}, " + f"OpenAI cost=${summary.total_cost:.4f}" + ) + print(f"Run report saved to {output_file}") + + +def main(argv: list[str] | None = None) -> int: + load_dotenv() + parser = build_parser() + arguments = parser.parse_args(argv) + logging.basicConfig( + level=getattr(logging, arguments.log_level), + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + if arguments.apply and arguments.apply_plan: + parser.error("--apply and --apply-plan cannot be used together") + try: + settings = _apply_overrides(Settings.from_env(), arguments) + flickr = FlickrGateway(settings) + analyzer = None if arguments.apply_plan else OpenAIImageAnalyzer(settings) + pipeline = AutotaggerPipeline(settings, flickr, analyzer) + if arguments.apply_plan: + summary = pipeline.apply_plan(arguments.apply_plan) + dry_run = False + else: + summary = asyncio.run(pipeline.run(apply=arguments.apply)) + dry_run = not arguments.apply + _report(summary, settings.updated_metadata_file, dry_run=dry_run) + return 2 if settings.strict and summary.failed else 0 + except (ConfigurationError, ValueError) as exc: + print(f"Configuration error: {exc}", file=sys.stderr) + return 2 + except KeyboardInterrupt: + print("Interrupted", file=sys.stderr) + return 130 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/autotagger/config.py b/autotagger/config.py new file mode 100644 index 0000000..14cfeda --- /dev/null +++ b/autotagger/config.py @@ -0,0 +1,287 @@ +"""Configuration loading and validation. + +CLI values are applied by ``autotagger.cli`` after this module has loaded the +environment. Keeping validation here prevents network clients from being +constructed with partly valid configuration. +""" + +from __future__ import annotations + +import json +import os +import re +from collections.abc import Iterable +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import urlparse + +DEFAULT_OPENAI_BASE_URL = "https://api.openai.com/v1" +VALID_UPDATE_FIELDS = frozenset({"title", "description", "tags"}) + + +class ConfigurationError(ValueError): + """Raised when configuration cannot be parsed or validated.""" + + +def _string(name: str, default: str = "") -> str: + return os.environ.get(name, default).strip() + + +def _integer(name: str, default: int, *, minimum: int | None = None) -> int: + raw = _string(name, str(default)) + try: + value = int(raw) + except ValueError as exc: + raise ConfigurationError(f"{name} must be an integer, got {raw!r}") from exc + if minimum is not None and value < minimum: + raise ConfigurationError(f"{name} must be at least {minimum}") + return value + + +def _optional_integer(name: str) -> int | None: + raw = _string(name) + if not raw: + return None + try: + value = int(raw) + except ValueError as exc: + raise ConfigurationError(f"{name} must be an integer, got {raw!r}") from exc + if value < 1: + raise ConfigurationError(f"{name} must be greater than zero") + return value + + +def _float(name: str, default: float, *, minimum: float | None = None) -> float: + raw = _string(name, str(default)) + try: + value = float(raw) + except ValueError as exc: + raise ConfigurationError(f"{name} must be a number, got {raw!r}") from exc + if minimum is not None and value < minimum: + raise ConfigurationError(f"{name} must be at least {minimum}") + return value + + +def _optional_float(name: str) -> float | None: + raw = _string(name) + if not raw: + return None + try: + value = float(raw) + except ValueError as exc: + raise ConfigurationError(f"{name} must be a number, got {raw!r}") from exc + if value <= 0: + raise ConfigurationError(f"{name} must be greater than zero") + return value + + +def _boolean(name: str, default: bool) -> bool: + raw = _string(name, "true" if default else "false").lower() + if raw in {"1", "true", "yes", "on"}: + return True + if raw in {"0", "false", "no", "off"}: + return False + raise ConfigurationError(f"{name} must be true or false, got {raw!r}") + + +def _list(name: str, default: Iterable[str] = ()) -> tuple[str, ...]: + raw = os.environ.get(name) + if raw is None or not raw.strip(): + return tuple(default) + raw = raw.strip() + if raw.startswith("["): + try: + value = json.loads(raw) + except json.JSONDecodeError as exc: + raise ConfigurationError(f"{name} must contain a valid JSON list") from exc + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise ConfigurationError(f"{name} must be a JSON list of strings") + return tuple(item.strip() for item in value if item.strip()) + return tuple(item.strip() for item in raw.split(",") if item.strip()) + + +def _privacy_filter() -> int | None: + raw = _string("FLICKR_PRIVACY_FILTER", "1").lower() + if raw in {"", "all", "none", "0"}: + return None + try: + value = int(raw) + except ValueError as exc: + raise ConfigurationError( + "FLICKR_PRIVACY_FILTER must be all or an integer from 1 to 5" + ) from exc + if value not in range(1, 6): + raise ConfigurationError("FLICKR_PRIVACY_FILTER must be all or an integer from 1 to 5") + return value + + +@dataclass(frozen=True) +class Settings: + flickr_api_key: str + flickr_api_secret: str + flickr_oauth_token: str + flickr_token_file: Path + flickr_privacy_filter: int | None + flickr_image_url: str + flickr_photoset_ids: tuple[str, ...] + photoset_allowlist: tuple[str, ...] + photoset_denylist: tuple[str, ...] + skip_prefixes: tuple[str, ...] + descriptions_to_analyze: tuple[str, ...] + description_policy: str + flickr_max_retries: int + flickr_retry_backoff: float + + openai_provider: str + openai_api_key: str + openai_base_url: str + openai_model: str + openai_timeout: float + openai_max_retries: int + openai_instruction_role: str + openai_tokens_parameter: str + openai_json_mode: bool + openai_max_output_tokens: int + openai_image_detail: str + openai_cost_per_1m_prompt_tokens: float + openai_cost_per_1m_completion_tokens: float + openai_vision_cost_per_image: float + azure_openai_endpoint: str + azure_openai_api_version: str + + analysis_concurrency: int + max_keywords: int + max_title_chars: int + max_description_chars: int + language: str + description_style: str + prompt_file: Path | None + max_photos: int | None + max_total_cost: float | None + + update_fields: tuple[str, ...] + tag_mode: str + checkpoint_file: Path + updated_metadata_file: Path + resume: bool + strict: bool + + @classmethod + def from_env(cls) -> Settings: + photoset_ids = list(_list("FLICKR_PHOTOSET_IDS")) + legacy_photoset_id = _string("FLICKR_PHOTOSET_ID") + if legacy_photoset_id and legacy_photoset_id not in photoset_ids: + photoset_ids.append(legacy_photoset_id) + + prompt_file_raw = _string("PROMPT_FILE") + openai_provider = _string("OPENAI_PROVIDER", "openai").lower() + openai_api_key = ( + _string("AZURE_OPENAI_API_KEY") + if openai_provider == "azure" + else _string("OPENAI_API_KEY") + ) + settings = cls( + flickr_api_key=_string("FLICKR_API_KEY"), + flickr_api_secret=_string("FLICKR_API_SECRET"), + flickr_oauth_token=_string("FLICKR_OAUTH_TOKEN"), + flickr_token_file=Path(_string("FLICKR_TOKEN_FILE", "flickr_token.json")), + flickr_privacy_filter=_privacy_filter(), + flickr_image_url=_string("FLICKR_IMAGE_URL", "url_m"), + flickr_photoset_ids=tuple(photoset_ids), + photoset_allowlist=_list("PHOTOSET_ALLOWLIST"), + photoset_denylist=_list("PHOTOSET_DENYLIST"), + skip_prefixes=_list("SKIP_PREFIX", ("#", "@")), + descriptions_to_analyze=_list( + "DESCRIPTIONS_TO_ANALYZE", + ("OLYMPUS DIGITAL CAMERA", "Untitled", "DSC_", "IMG_", "DCIM"), + ), + description_policy=_string("DESCRIPTION_POLICY", "missing-or-placeholder").lower(), + flickr_max_retries=_integer("FLICKR_MAX_RETRIES", 3, minimum=0), + flickr_retry_backoff=_float("FLICKR_RETRY_BACKOFF", 1.0, minimum=0), + openai_provider=openai_provider, + openai_api_key=openai_api_key, + openai_base_url=_string("OPENAI_BASE_URL") or DEFAULT_OPENAI_BASE_URL, + openai_model=_string("OPENAI_MODEL", "gpt-5-mini"), + openai_timeout=_float("OPENAI_TIMEOUT", 60.0, minimum=0.1), + openai_max_retries=_integer("OPENAI_MAX_RETRIES", 2, minimum=0), + openai_instruction_role=_string("OPENAI_INSTRUCTION_ROLE", "developer").lower(), + openai_tokens_parameter=_string( + "OPENAI_TOKENS_PARAMETER", "max_completion_tokens" + ).lower(), + openai_json_mode=_boolean("OPENAI_JSON_MODE", True), + openai_max_output_tokens=_integer("OPENAI_MAX_OUTPUT_TOKENS", 800, minimum=1), + openai_image_detail=_string("OPENAI_IMAGE_DETAIL", "low").lower(), + openai_cost_per_1m_prompt_tokens=_float( + "OPENAI_COST_PER_1M_PROMPT_TOKEN", 0.25, minimum=0 + ), + openai_cost_per_1m_completion_tokens=_float( + "OPENAI_COST_PER_1M_COMPLETION_TOKEN", 2.0, minimum=0 + ), + openai_vision_cost_per_image=_float("OPENAI_VISION_COST_PER_IMAGE", 0.0, minimum=0), + azure_openai_endpoint=_string("AZURE_OPENAI_ENDPOINT"), + azure_openai_api_version=_string("OPENAI_API_VERSION"), + analysis_concurrency=_integer("ANALYSIS_CONCURRENCY", 3, minimum=1), + max_keywords=_integer("MAX_KEYWORDS", 10, minimum=1), + max_title_chars=_integer("MAX_TITLE_CHARS", 120, minimum=1), + max_description_chars=_integer("MAX_DESCRIPTION_CHARS", 1200, minimum=1), + language=_string("METADATA_LANGUAGE", "English"), + description_style=_string("DESCRIPTION_STYLE", "detailed, factual, and concise"), + prompt_file=Path(prompt_file_raw) if prompt_file_raw else None, + max_photos=_optional_integer("MAX_PHOTOS"), + max_total_cost=_optional_float("MAX_TOTAL_COST"), + update_fields=_list("UPDATE_FIELDS", ("title", "description", "tags")), + tag_mode=_string("TAG_MODE", "merge").lower(), + checkpoint_file=Path(_string("CHECKPOINT_FILE", ".autotagger-checkpoint.jsonl")), + updated_metadata_file=Path(_string("UPDATED_METADATA_FILE", "updated_metadata.json")), + resume=_boolean("RESUME", True), + strict=_boolean("FAIL_ON_ERROR", False), + ) + settings.validate(require_openai=False, require_flickr=False) + return settings + + def validate(self, *, require_openai: bool, require_flickr: bool) -> None: + errors: list[str] = [] + if require_flickr: + if not self.flickr_api_key: + errors.append("FLICKR_API_KEY is required") + if not self.flickr_api_secret: + errors.append("FLICKR_API_SECRET is required") + if require_openai and not self.openai_api_key: + key_name = ( + "AZURE_OPENAI_API_KEY" if self.openai_provider == "azure" else "OPENAI_API_KEY" + ) + errors.append(f"{key_name} is required") + + parsed_url = urlparse(self.openai_base_url) + if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc: + errors.append("OPENAI_BASE_URL must be an absolute HTTP(S) URL") + elif parsed_url.username or parsed_url.password: + errors.append("OPENAI_BASE_URL must not contain embedded credentials") + elif parsed_url.path.rstrip("/").endswith("/chat/completions"): + errors.append("OPENAI_BASE_URL must be a base URL, not the chat completions endpoint") + if self.openai_provider not in {"openai", "azure"}: + errors.append("OPENAI_PROVIDER must be openai or azure") + if self.openai_provider == "azure": + if not self.azure_openai_endpoint: + errors.append("AZURE_OPENAI_ENDPOINT is required for the Azure provider") + if not self.azure_openai_api_version: + errors.append("OPENAI_API_VERSION is required for the Azure provider") + if self.openai_instruction_role not in {"developer", "system"}: + errors.append("OPENAI_INSTRUCTION_ROLE must be developer or system") + if self.openai_tokens_parameter not in {"max_completion_tokens", "max_tokens"}: + errors.append("OPENAI_TOKENS_PARAMETER must be max_completion_tokens or max_tokens") + if self.openai_image_detail not in {"low", "high", "auto"}: + errors.append("OPENAI_IMAGE_DETAIL must be low, high, or auto") + if self.description_policy not in {"missing-or-placeholder", "always", "never"}: + errors.append("DESCRIPTION_POLICY must be missing-or-placeholder, always, or never") + invalid_fields = set(self.update_fields) - VALID_UPDATE_FIELDS + if invalid_fields or not self.update_fields: + errors.append("UPDATE_FIELDS must contain one or more of title, description, and tags") + if self.tag_mode not in {"merge", "replace"}: + errors.append("TAG_MODE must be merge or replace") + if not re.fullmatch(r"url_[a-z]+", self.flickr_image_url): + errors.append("FLICKR_IMAGE_URL must be a Flickr URL extra such as url_m") + if self.prompt_file and not self.prompt_file.is_file(): + errors.append(f"PROMPT_FILE does not exist: {self.prompt_file}") + if errors: + raise ConfigurationError("; ".join(errors)) diff --git a/autotagger/flickr_gateway.py b/autotagger/flickr_gateway.py new file mode 100644 index 0000000..a206421 --- /dev/null +++ b/autotagger/flickr_gateway.py @@ -0,0 +1,299 @@ +"""Flickr authentication, pagination, normalisation, and safe metadata writes.""" + +from __future__ import annotations + +import json +import logging +import os +import shlex +import tempfile +import time +from collections.abc import Callable, Iterator +from pathlib import Path +from typing import Any, TypeVar + +import flickrapi +import requests + +from .config import Settings +from .models import Analysis, PhotoCandidate, Photoset + +LOGGER = logging.getLogger(__name__) +T = TypeVar("T") + + +class PartialUpdateError(RuntimeError): + def __init__(self, message: str, applied_fields: tuple[str, ...]): + super().__init__(message) + self.applied_fields = applied_fields + + +def _content(value: Any) -> str: + if isinstance(value, dict): + value = value.get("_content", "") + return value.strip() if isinstance(value, str) else "" + + +def _parse_tags(value: Any) -> tuple[str, ...]: + if isinstance(value, str): + try: + return tuple(tag for tag in shlex.split(value) if tag) + except ValueError: + return tuple(tag for tag in value.split() if tag) + if isinstance(value, dict): + value = value.get("tag", ()) + if isinstance(value, list): + tags = [] + for item in value: + tag = item.get("raw") or item.get("_content") if isinstance(item, dict) else item + if isinstance(tag, str) and tag.strip(): + tags.append(tag.strip()) + return tuple(tags) + return () + + +def _format_tags(tags: tuple[str, ...]) -> str: + formatted = [] + for tag in tags: + cleaned = tag.strip().replace('"', "") + if not cleaned: + continue + formatted.append(f'"{cleaned}"' if any(char.isspace() for char in cleaned) else cleaned) + return " ".join(formatted) + + +def _merge_tags(existing: tuple[str, ...], generated: tuple[str, ...]) -> tuple[str, ...]: + merged: list[str] = [] + seen: set[str] = set() + for tag in (*existing, *generated): + key = tag.casefold() + if key not in seen: + merged.append(tag) + seen.add(key) + return tuple(merged) + + +def _write_private_json(path: Path, value: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + temporary_path = Path(temporary_name) + try: + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(value, handle) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, path) + finally: + if temporary_path.exists(): + temporary_path.unlink() + + +class FlickrGateway: + def __init__(self, settings: Settings, api: Any | None = None): + self.settings = settings + self.api = api or self._authenticate() + + def _token_from_dict(self, value: dict[str, Any]) -> Any: + return flickrapi.auth.FlickrAccessToken( + value["oauth_token"], + value["oauth_token_secret"], + value["access_level"], + value["fullname"], + value["username"], + value["user_nsid"], + ) + + def _authenticated_api(self, token_dict: dict[str, Any]) -> Any: + return flickrapi.FlickrAPI( + self.settings.flickr_api_key, + self.settings.flickr_api_secret, + token=self._token_from_dict(token_dict), + format="parsed-json", + ) + + def _authenticate(self) -> Any: + token_file = self.settings.flickr_token_file + if token_file.exists(): + try: + with token_file.open("r", encoding="utf-8") as handle: + return self._authenticated_api(json.load(handle)) + except (json.JSONDecodeError, KeyError, TypeError) as exc: + LOGGER.warning("Ignoring invalid Flickr token file %s: %s", token_file, exc) + + if self.settings.flickr_oauth_token: + try: + return self._authenticated_api(json.loads(self.settings.flickr_oauth_token)) + except (json.JSONDecodeError, KeyError, TypeError) as exc: + raise ValueError("FLICKR_OAUTH_TOKEN is not a valid Flickr token object") from exc + + last_error: Exception | None = None + for attempt in range(2): + api = flickrapi.FlickrAPI( + self.settings.flickr_api_key, + self.settings.flickr_api_secret, + format="parsed-json", + ) + try: + api.token_cache.forget() + LOGGER.info("Performing interactive Flickr OAuth authentication") + api.get_request_token(oauth_callback="oob") + authorize_url = api.auth_url(perms="write") + print(f"Please visit this URL to authorize the application: {authorize_url}") + api.get_access_token(input("Enter the verifier code: ").strip()) + token = api.token_cache.token + token_dict = { + "oauth_token": token.token, + "oauth_token_secret": token.token_secret, + "access_level": token.access_level, + "fullname": token.fullname, + "username": token.username, + "user_nsid": token.user_nsid, + } + _write_private_json(token_file, token_dict) + return api + except flickrapi.FlickrError as exc: + last_error = exc + LOGGER.warning( + "Flickr authentication failed (attempt %d of 2): %s", + attempt + 1, + exc, + ) + raise RuntimeError("Flickr authentication failed after two attempts") from last_error + + def _call(self, operation: Callable[..., T], **kwargs: Any) -> T: + attempts = self.settings.flickr_max_retries + 1 + for attempt in range(attempts): + try: + return operation(**kwargs) + except flickrapi.FlickrError as exc: + if exc.code not in {105, 106}: + raise + if attempt == attempts - 1: + raise + delay = self.settings.flickr_retry_backoff * (2**attempt) + LOGGER.warning("Transient Flickr error; retrying in %.1f seconds", delay) + time.sleep(delay) + except requests.RequestException: + if attempt == attempts - 1: + raise + delay = self.settings.flickr_retry_backoff * (2**attempt) + LOGGER.warning("Transient Flickr error; retrying in %.1f seconds", delay) + time.sleep(delay) + raise AssertionError("unreachable") + + def get_photosets(self) -> list[Photoset]: + if self.settings.flickr_photoset_ids: + result = [] + for photoset_id in self.settings.flickr_photoset_ids: + response = self._call(self.api.photosets.getInfo, photoset_id=photoset_id) + value = response["photoset"] + result.append( + Photoset( + id=str(value["id"]), + title=_content(value.get("title")), + description=_content(value.get("description")), + ) + ) + return result + + photosets: list[Photoset] = [] + page = 1 + while True: + response = self._call( + self.api.photosets.getList, page=page, per_page=500, extras="description" + ) + container = response["photosets"] + for value in container.get("photoset", []): + photosets.append( + Photoset( + id=str(value["id"]), + title=_content(value.get("title")), + description=_content(value.get("description")), + ) + ) + if page >= int(container.get("pages", 1)): + break + page += 1 + return photosets + + def iter_photo_pages(self, photoset: Photoset) -> Iterator[list[PhotoCandidate]]: + page = 1 + extras = f"{self.settings.flickr_image_url},description,geo,tags" + while True: + kwargs: dict[str, Any] = { + "photoset_id": photoset.id, + "extras": extras, + "media": "photos", + "page": page, + "per_page": 500, + } + if self.settings.flickr_privacy_filter is not None: + kwargs["privacy_filter"] = self.settings.flickr_privacy_filter + response = self._call(self.api.photosets.getPhotos, **kwargs) + container = response["photoset"] + candidates = [ + self._normalise_photo(value, photoset) for value in container.get("photo", []) + ] + yield candidates + if page >= int(container.get("pages", 1)): + break + page += 1 + + def _normalise_photo(self, value: dict[str, Any], photoset: Photoset) -> PhotoCandidate: + description = _content(value.get("description")) + title = _content(value.get("title")) + tags = _parse_tags(value.get("tags")) + latitude = value.get("latitude") + longitude = value.get("longitude") + + if "description" not in value: + info = self._call(self.api.photos.getInfo, photo_id=value["id"])["photo"] + description = _content(info.get("description")) + title = title or _content(info.get("title")) + tags = tags or _parse_tags(info.get("tags")) + location = info.get("location", {}) + latitude = latitude if latitude not in {None, ""} else location.get("latitude") + longitude = longitude if longitude not in {None, ""} else location.get("longitude") + + image_url = value.get(self.settings.flickr_image_url) + if not isinstance(image_url, str) or not image_url: + image_field = self.settings.flickr_image_url + raise ValueError(f"Flickr did not return {image_field} for photo {value.get('id')}") + return PhotoCandidate( + id=str(value["id"]), + photoset_id=photoset.id, + photoset_title=photoset.title, + photoset_description=photoset.description, + image_url=image_url, + title=title, + description=description, + tags=tags, + latitude=latitude, + longitude=longitude, + ) + + def update_metadata(self, photo: PhotoCandidate, analysis: Analysis) -> tuple[str, ...]: + applied: list[str] = [] + fields = set(self.settings.update_fields) + try: + meta: dict[str, Any] = {"photo_id": photo.id} + if "title" in fields: + meta["title"] = analysis.title + if "description" in fields: + meta["description"] = analysis.description + if len(meta) > 1: + self._call(self.api.photos.setMeta, **meta) + applied.extend(field for field in ("title", "description") if field in fields) + + if "tags" in fields: + tags = analysis.keywords + if self.settings.tag_mode == "merge": + tags = _merge_tags(photo.tags, analysis.keywords) + self._call(self.api.photos.setTags, photo_id=photo.id, tags=_format_tags(tags)) + applied.append("tags") + return tuple(applied) + except Exception as exc: + if applied: + raise PartialUpdateError(str(exc), tuple(applied)) from exc + raise diff --git a/autotagger/models.py b/autotagger/models.py new file mode 100644 index 0000000..c98b02c --- /dev/null +++ b/autotagger/models.py @@ -0,0 +1,66 @@ +"""Typed data exchanged between the Flickr, OpenAI, and pipeline layers.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from typing import Any + + +@dataclass(frozen=True) +class Photoset: + id: str + title: str + description: str = "" + + +@dataclass(frozen=True) +class PhotoCandidate: + id: str + photoset_id: str + photoset_title: str + photoset_description: str + image_url: str + title: str = "" + description: str = "" + tags: tuple[str, ...] = () + latitude: str | float | None = None + longitude: str | float | None = None + + @property + def location(self) -> dict[str, str | float] | None: + if self.latitude is None or self.longitude is None: + return None + if self.latitude == "" or self.longitude == "": + return None + return {"latitude": self.latitude, "longitude": self.longitude} + + def to_record_context(self) -> dict[str, Any]: + return { + "photo_id": self.id, + "photoset_id": self.photoset_id, + "photoset_title": self.photoset_title, + "image_url": self.image_url, + "existing_title": self.title, + "existing_description": self.description, + "existing_tags": list(self.tags), + } + + +@dataclass(frozen=True) +class Usage: + prompt_tokens: int = 0 + completion_tokens: int = 0 + cost: float = 0.0 + + +@dataclass(frozen=True) +class Analysis: + title: str + description: str + keywords: tuple[str, ...] + usage: Usage = field(default_factory=Usage) + + def to_dict(self) -> dict[str, Any]: + result = asdict(self) + result["keywords"] = list(self.keywords) + return result diff --git a/autotagger/pipeline.py b/autotagger/pipeline.py new file mode 100644 index 0000000..564fa28 --- /dev/null +++ b/autotagger/pipeline.py @@ -0,0 +1,308 @@ +"""Resumable autotagging orchestration.""" + +from __future__ import annotations + +import logging +import uuid +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from .analysis import ImageAnalyzer, analyze_concurrently +from .checkpoint import CheckpointStore, load_plan, write_summary +from .config import Settings +from .flickr_gateway import FlickrGateway, PartialUpdateError +from .models import Analysis, PhotoCandidate, Usage + +LOGGER = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class RunSummary: + records: tuple[dict[str, Any], ...] + analyzed: int + updated: int + failed: int + skipped: int + total_cost: float + + +def _timestamp() -> str: + return datetime.now(UTC).isoformat() + + +def _analysis_from_dict(value: dict[str, Any]) -> Analysis: + usage_value = value.get("usage", {}) + return Analysis( + title=value["title"], + description=value["description"], + keywords=tuple(value["keywords"]), + usage=Usage( + prompt_tokens=int(usage_value.get("prompt_tokens", 0)), + completion_tokens=int(usage_value.get("completion_tokens", 0)), + cost=float(usage_value.get("cost", 0)), + ), + ) + + +def _photo_from_record(record: dict[str, Any]) -> PhotoCandidate: + return PhotoCandidate( + id=str(record["photo_id"]), + photoset_id=str(record.get("photoset_id", "")), + photoset_title=str(record.get("photoset_title", "")), + photoset_description="", + image_url=str(record.get("image_url", "")), + title=str(record.get("existing_title", "")), + description=str(record.get("existing_description", "")), + tags=tuple(record.get("existing_tags", ())), + ) + + +class AutotaggerPipeline: + def __init__( + self, + settings: Settings, + flickr: FlickrGateway, + analyzer: ImageAnalyzer | None, + checkpoint: CheckpointStore | None = None, + ): + self.settings = settings + self.flickr = flickr + self.analyzer = analyzer + self.checkpoint = checkpoint or CheckpointStore(settings.checkpoint_file) + self.run_id = str(uuid.uuid4()) + self.records: list[dict[str, Any]] = [] + self.total_cost = 0.0 + self.analyzed_count = 0 + self.updated_count = 0 + self.failed_count = 0 + self.skipped_count = 0 + + def _record( + self, + status: str, + photo: PhotoCandidate, + **values: Any, + ) -> dict[str, Any]: + record = { + "timestamp": _timestamp(), + "run_id": self.run_id, + "status": status, + **photo.to_record_context(), + **values, + } + self.records.append(record) + self.checkpoint.append(record) + if status == "updated": + self.updated_count += 1 + elif status in {"failed", "partially_updated"}: + self.failed_count += 1 + elif status == "skipped": + self.skipped_count += 1 + return record + + def _photoset_selected(self, photoset_id: str, title: str) -> bool: + if any(title.startswith(prefix) for prefix in self.settings.skip_prefixes): + return False + identity = {photoset_id, title} + if self.settings.photoset_allowlist and identity.isdisjoint( + self.settings.photoset_allowlist + ): + return False + return identity.isdisjoint(self.settings.photoset_denylist) + + def _photo_selected(self, photo: PhotoCandidate) -> tuple[bool, str]: + if self.settings.description_policy == "always": + return True, "" + if self.settings.description_policy == "never": + return False, "description policy is never" + description = photo.description.strip() + if not description: + return True, "" + if any(description.startswith(value) for value in self.settings.descriptions_to_analyze): + return True, "" + return False, "existing description" + + def _can_analyze_more(self) -> bool: + if self.settings.max_photos is not None and self.analyzed_count >= self.settings.max_photos: + return False + return not ( + self.settings.max_total_cost is not None + and self.total_cost >= self.settings.max_total_cost + ) + + async def run(self, *, apply: bool) -> RunSummary: + if self.analyzer is None: + raise RuntimeError("An analyzer is required to run image analysis") + latest = self.checkpoint.load_latest() if self.settings.resume else {} + processed: set[str] = set() + stop = False + + for photoset in self.flickr.get_photosets(): + if not self._photoset_selected(photoset.id, photoset.title): + LOGGER.info("Skipping photoset %s (%s)", photoset.title, photoset.id) + continue + LOGGER.info("Processing photoset %s (%s)", photoset.title, photoset.id) + for page in self.flickr.iter_photo_pages(photoset): + pending: list[PhotoCandidate] = [] + cached: list[tuple[PhotoCandidate, dict[str, Any]]] = [] + for photo in page: + if photo.id in processed: + self._record("skipped", photo, reason="duplicate photo in this run") + continue + processed.add(photo.id) + selected, reason = self._photo_selected(photo) + if not selected: + self._record("skipped", photo, reason=reason) + continue + + prior = latest.get(photo.id) + if prior and prior.get("status") == "updated": + self._record("skipped", photo, reason="already updated in checkpoint") + continue + if ( + prior + and prior.get("status") == "analysed" + and prior.get("analyzer_fingerprint") == self.analyzer.fingerprint + and isinstance(prior.get("analysis"), dict) + ): + cached.append((photo, prior)) + continue + if not self._can_analyze_more(): + stop = True + break + pending.append(photo) + if self.settings.max_photos is not None: + remaining = self.settings.max_photos - self.analyzed_count + if len(pending) >= remaining: + stop = True + break + + for photo, prior in cached: + if apply: + self._apply(photo, _analysis_from_dict(prior["analysis"]), cached=True) + else: + self._record( + "analysed", + photo, + analysis=prior["analysis"], + analyzer_fingerprint=self.analyzer.fingerprint, + cached_analysis=True, + ) + + for start in range(0, len(pending), self.settings.analysis_concurrency): + if not self._can_analyze_more(): + stop = True + break + batch = pending[start : start + self.settings.analysis_concurrency] + outcomes = await analyze_concurrently( + self.analyzer, batch, self.settings.analysis_concurrency + ) + for photo, outcome in zip(batch, outcomes, strict=True): + self.analyzed_count += 1 + if isinstance(outcome, Exception): + self._record( + "failed", + photo, + stage="analysis", + error=f"{type(outcome).__name__}: {outcome}", + analyzer_fingerprint=self.analyzer.fingerprint, + ) + continue + self.total_cost += outcome.usage.cost + self._record( + "analysed", + photo, + analysis=outcome.to_dict(), + analyzer_fingerprint=self.analyzer.fingerprint, + ) + if apply: + self._apply(photo, outcome) + if not self._can_analyze_more(): + stop = True + break + if stop: + break + if stop: + break + + write_summary(self.settings.updated_metadata_file, self.records) + return self._summary() + + def _apply(self, photo: PhotoCandidate, analysis: Analysis, *, cached: bool = False) -> None: + try: + applied_fields = self.flickr.update_metadata(photo, analysis) + self._record( + "updated", + photo, + analysis=analysis.to_dict(), + applied_fields=list(applied_fields), + cached_analysis=cached, + ) + except PartialUpdateError as exc: + self._record( + "partially_updated", + photo, + stage="flickr_update", + analysis=analysis.to_dict(), + applied_fields=list(exc.applied_fields), + error=str(exc), + ) + except Exception as exc: + self._record( + "failed", + photo, + stage="flickr_update", + analysis=analysis.to_dict(), + error=f"{type(exc).__name__}: {exc}", + ) + + def apply_plan(self, path: Path) -> RunSummary: + latest_by_photo: dict[str, dict[str, Any]] = {} + for record in load_plan(path): + photo_id = record.get("photo_id") + if not isinstance(photo_id, str): + continue + if ( + record.get("status") == "updated" + or isinstance(record.get("analysis"), dict) + and (latest_by_photo.get(photo_id, {}).get("status") != "updated") + ): + latest_by_photo[photo_id] = record + + for record in latest_by_photo.values(): + analysis_value = record.get("analysis") + if record.get("status") == "updated" or not isinstance(analysis_value, dict): + continue + try: + photo = _photo_from_record(record) + analysis = _analysis_from_dict(analysis_value) + self._apply(photo, analysis, cached=True) + except Exception as exc: + photo_id = str(record.get("photo_id", "unknown")) + placeholder = PhotoCandidate( + id=photo_id, + photoset_id=str(record.get("photoset_id", "")), + photoset_title=str(record.get("photoset_title", "")), + photoset_description="", + image_url=str(record.get("image_url", "")), + ) + self._record( + "failed", + placeholder, + stage="plan_validation", + error=f"{type(exc).__name__}: {exc}", + ) + write_summary(self.settings.updated_metadata_file, self.records) + return self._summary() + + def _summary(self) -> RunSummary: + return RunSummary( + records=tuple(self.records), + analyzed=self.analyzed_count, + updated=self.updated_count, + failed=self.failed_count, + skipped=self.skipped_count, + total_cost=self.total_cost, + ) diff --git a/flickr-autotagger.py b/flickr-autotagger.py index 6e018a4..b1b1e31 100644 --- a/flickr-autotagger.py +++ b/flickr-autotagger.py @@ -1,398 +1,7 @@ -import flickrapi -from openai import OpenAI, BadRequestError -import json -import os +#!/usr/bin/env python3 +"""Backward-compatible entry point for the Flickr autotagger CLI.""" -try: - from dotenv import load_dotenv - load_dotenv() -except ImportError: - print("python-dotenv is not installed. Skipping .env file loading.") -except Exception as e: - print(f"Failed to load .env file: {e}") - -# OpenAI Model and costings -OPENAI_MODEL = os.environ.get("OPENAI_MODEL", "gpt-5-mini") -OPENAI_COST_PER_1M_PROMPT_TOKEN = float(os.environ.get("OPENAI_COST_PER_1M_PROMPT_TOKEN", "0.25")) -OPENAI_COST_PER_1M_COMPLETION_TOKEN = float(os.environ.get("OPENAI_COST_PER_1M_COMPLETION_TOKEN", "2.00")) -OPENAI_VISION_COST_PER_IMAGE = float(os.environ.get("OPENAI_VISION_COST_PER_IMAGE", "0")) - -# Script configuration -FLICKR_PRIVACY_FILTER = int(os.environ.get("FLICKR_PRIVACY_FILTER", "1")) # 0. none, 1. public, 2. friends, 3. family, 4. friends & family, 5. private -FLICKR_TOKEN_FILE = os.environ.get("FLICKR_TOKEN_FILE", "flickr_token.json") -FLICKR_IMAGE_URL = os.environ.get("FLICKR_IMAGE_URL", "url_m") -DESCRIPTIONS_TO_ANALYZE = os.environ.get("DESCRIPTIONS_TO_ANALYZE", '["OLYMPUS DIGITAL CAMERA", "Untitled", "DSC_", "IMG_", "DCIM"]') -DESCRIPTIONS_TO_ANALYZE = eval(DESCRIPTIONS_TO_ANALYZE) -SKIP_PREFIX = os.environ.get("SKIP_PREFIX", '["#", "@"]') -SKIP_PREFIX = eval(SKIP_PREFIX) -MAX_KEYWORDS = int(os.environ.get("MAX_KEYWORDS", "10")) -UPDATED_METADATA_FILE = os.environ.get("UPDATED_METADATA_FILE", "updated_metadata.json") -SINGLE_PHOTOSET_ID = os.environ.get("FLICKR_PHOTOSET_ID") - -# Mandatory API environment variables -flickr_api_key = os.environ.get("FLICKR_API_KEY") -flickr_api_secret = os.environ.get("FLICKR_API_SECRET") -openai_api_key = os.environ.get("OPENAI_API_KEY") - -# Check if mandatory API environment variables are set -if not all([flickr_api_key, flickr_api_secret, openai_api_key]): - print("Please set the required environment variables:") - print("- FLICKR_API_KEY") - print("- FLICKR_API_SECRET") - print("- OPENAI_API_KEY") - exit(1) - -def flickr_authentication(): - try: - # Check if the token file exists - if os.path.exists("flickr_token.json"): - # Load the token from the file - with open("flickr_token.json", "r") as f: - try: - token_dict = json.load(f) - token = flickrapi.auth.FlickrAccessToken( - token_dict["oauth_token"], - token_dict["oauth_token_secret"], - token_dict["access_level"], - token_dict["fullname"], - token_dict["username"], - token_dict["user_nsid"], - ) - flickr_api = flickrapi.FlickrAPI( - flickr_api_key, flickr_api_secret, token=token, format="parsed-json" - ) - return flickr_api - except json.JSONDecodeError: - print("Invalid JSON format in the token file. Performing OAuth authentication...") - - # Check if the token is available in the environment variable - elif os.environ.get("FLICKR_OAUTH_TOKEN"): - token_dict = json.loads(os.environ["FLICKR_OAUTH_TOKEN"]) - token = flickrapi.auth.FlickrAccessToken( - token_dict["oauth_token"], - token_dict["oauth_token_secret"], - token_dict["access_level"], - token_dict["fullname"], - token_dict["username"], - token_dict["user_nsid"], - ) - flickr_api = flickrapi.FlickrAPI( - flickr_api_key, flickr_api_secret, token=token, format="parsed-json" - ) - return flickr_api - - # If the token file doesn't exist and the environment variable is not set, perform the OAuth flow - flickr_api = flickrapi.FlickrAPI( - flickr_api_key, flickr_api_secret, format="parsed-json" - ) - flickr_api.token_cache.forget() - print("Performing OAuth authentication...") - flickr_api.get_request_token(oauth_callback="oob") - authorize_url = flickr_api.auth_url(perms="write") - print(f"Please visit this URL to authorize the application: {authorize_url}") - verifier = input("Enter the verifier code: ") - flickr_api.get_access_token(verifier) - - # Save the full token object as a dictionary - token_dict = { - "oauth_token": flickr_api.token_cache.token.token, - "oauth_token_secret": flickr_api.token_cache.token.token_secret, - "access_level": flickr_api.token_cache.token.access_level, - "fullname": flickr_api.token_cache.token.fullname, - "username": flickr_api.token_cache.token.username, - "user_nsid": flickr_api.token_cache.token.user_nsid, - } - - # Save the token dictionary to the file for future use - with open("flickr_token.json", "w") as f: - json.dump(token_dict, f) - - return flickr_api - except flickrapi.FlickrError: - print("Unauthorized error occurred. Retrying authentication...") - flickr_api.token_cache.forget() - return flickr_authentication() - except Exception as e: - print(f"An unexpected error occurred: {e}") - raise # Re-raise the non-FlickrError exception - - -def get_all_photosets(flickr): - photosets = [] - page = 1 - per_page = 500 - - while True: - response = flickr.photosets.getList( - page=page, per_page=per_page, extras="description" - ) - photosets.extend(response["photosets"]["photoset"]) - - if page * per_page >= int(response["photosets"]["total"]): - break - - page += 1 - - return photosets - - -def has_flickr_description(photo): - existing_description = photo["description"]["_content"].strip() - return ( - not any( - existing_description.startswith(desc) for desc in DESCRIPTIONS_TO_ANALYZE - ) - and existing_description != "" - ) - - -def strip_markdown_response(response): - if response.startswith("```") and response.endswith("```"): - stripped_response = response[3:-3].strip() - if stripped_response.startswith("json"): - stripped_response = stripped_response[4:].strip() - return stripped_response - return response - - -# Function to get image analysis from ChatGPT -def get_image_analysis( - openai, image_url, photoset_title=None, photoset_description=None, location=None -): - system_message = ( - "Summarize images and generate metadata as valid JSON. Create JSON with:\n" - '1. "title": Concise, descriptive title.\n' - '2. "description": Detailed description with locations, objects, mood, and colors.\n' - f'3. "keywords": Up to {MAX_KEYWORDS} lowercase keywords, do not include spaces or symbols.\n' - "Use only the image and optional context. Don't mention context. Keep description/title self-contained.\n" - "Optional context:\n" - "- albumTitle: From album title.\n" - "- albumDescription: From album description.\n" - "- location: 'latitude' and 'longitude' for photo location.\n" - 'Example JSON: {"title": "Example Title", "description": "Example description.", "keywords": ["keyword1", "keyword2", "keyword3"]}\n' - ) - - user_message = {} - if photoset_title: - user_message["albumTitle"] = photoset_title - if photoset_description: - user_message["albumDescription"] = photoset_description - if location: - user_message["location"] = location - - try: - response = openai.chat.completions.create( - model=OPENAI_MODEL, - messages=[ - { - "role": "developer", - "content": system_message, - }, - { - "role": "user", - "content": [ - {"type": "text", "text": json.dumps(user_message)}, - { - "type": "image_url", - "image_url": { - "url": image_url, - "detail": "low", - }, - }, - ], - }, - ], - max_completion_tokens=4000, - ) - except BadRequestError as e: - raise e - - try: - analysis = json.loads( - strip_markdown_response(response.choices[0].message.content.strip()) - ) - if "keywords" in analysis: - analysis["keywords"] = analysis["keywords"][:MAX_KEYWORDS] - - # Add usage information to the analysis - if response.usage is not None: - prompt_tokens = response.usage.prompt_tokens - completion_tokens = response.usage.completion_tokens - cost = ( - (prompt_tokens * OPENAI_COST_PER_1M_PROMPT_TOKEN / 1000000) - + (completion_tokens * OPENAI_COST_PER_1M_COMPLETION_TOKEN / 1000000) - + OPENAI_VISION_COST_PER_IMAGE - ) - analysis["usage"] = { - "prompt_tokens": prompt_tokens, - "completion_tokens": completion_tokens, - "cost": cost, - } - - return analysis - except json.JSONDecodeError: - return {"error": "Failed to parse JSON response from ChatGPT"} - - -# Function to update Flickr image tags and description -def update_flickr_metadata(flickr, photo_id, analysis): - if "error" in analysis: - print(f"Error: {analysis['error']}") - return - - required_keys = ["title", "description", "keywords"] - if not all(key in analysis for key in required_keys): - print(f"Error: Missing required keys in analysis for photo {photo_id}") - return - - tags = analysis["keywords"] - title = analysis["title"] - description = analysis["description"] - - flickr.photos.setTags(photo_id=photo_id, tags=",".join(tags)) - flickr.photos.setMeta(photo_id=photo_id, title=title, description=description) - -def process_photoset(flickr, openai, photoset): - photoset_id = photoset["id"] - photoset_title = photoset["title"]["_content"] - photoset_description = photoset["description"]["_content"] - - # Skip photosets with names starting with any character in skip_characters - if any(photoset_title.startswith(char) for char in SKIP_PREFIX): - return None - - # Get photos from the current photoset with pagination - all_photos = [] - page = 1 - per_page = 500 # Maximum allowed by Flickr API - - while True: - try: - photos_response = flickr.photosets.getPhotos( - photoset_id=photoset_id, - extras=f"{FLICKR_IMAGE_URL},description,geo", - media="photos", - privacy_filter=FLICKR_PRIVACY_FILTER, - page=page, - per_page=per_page - ) - - # Add photos from this page to our collection - all_photos.extend(photos_response["photoset"]["photo"]) - - # Check if we've processed all pages - total_photos = int(photos_response["photoset"]["total"]) - if page * per_page >= total_photos: - break - - # Move to next page - page += 1 - - except flickrapi.exceptions.FlickrError as e: - print(f"Error retrieving photos for photoset: {photoset_title} (ID: {photoset_id}) - {str(e)}") - return None - - print(f"Processing photoset: {photoset_title} (ID: {photoset_id}) - {len(all_photos)} images") - - # Process each image in the photoset - updated_metadata = [] - photoset_cost = 0 - skipped_photos = 0 - for photo in all_photos: - photo_id = photo["id"] - image_url = photo[FLICKR_IMAGE_URL] - - # Check if the photo already has a description - if has_flickr_description(photo): - skipped_photos += 1 - continue - - # Get the location information for the photo - latitude = photo.get("latitude") - longitude = photo.get("longitude") - location = ( - {"latitude": latitude, "longitude": longitude} - if latitude and longitude - else None - ) - - retry_count = 0 - while retry_count < 2: - try: - # Get image analysis from ChatGPT - analysis = get_image_analysis( - openai, image_url, photoset_title, photoset_description, location - ) - break - except BadRequestError as e: - retry_count += 1 - if retry_count == 2: - print( - f"Skipping photo {photo_id} due to repeated BadRequestError: {str(e)}" - ) - continue - - if retry_count == 2: - continue - - # Include photoset_id and photo_id in the analysis JSON - analysis["photoset_id"] = photoset_id - analysis["photo_id"] = photo_id - - # Calculate the cost for this request if usage information is available - if "usage" in analysis: - if "cost" in analysis["usage"]: - photoset_cost += analysis["usage"]["cost"] - - # Update Flickr image metadata - update_flickr_metadata(flickr, photo_id, analysis) - - # Append the analysis result to the updated metadata list - updated_metadata.append(analysis) - - print(f"Finished processing photoset: {photoset_title}") - print(f"Skipped {skipped_photos} photos due to existing descriptions") - print(f"Total cost for photoset: ${photoset_cost:.4f}\n") - - return updated_metadata, photoset_cost - -def process_all_photosets(flickr, openai): - if SINGLE_PHOTOSET_ID: - # Process a specific photoset - photosets = [flickr.photosets.getInfo(photoset_id=SINGLE_PHOTOSET_ID)["photoset"]] - else: - # Get all photosets - photosets = get_all_photosets(flickr) - - total_cost = 0 - all_updated_metadata = [] - for photoset in photosets: - result = process_photoset(flickr, openai, photoset) - if result is not None: - updated_metadata, photoset_cost = result - all_updated_metadata.extend(updated_metadata) - total_cost += photoset_cost - - return all_updated_metadata, total_cost - -def main(): - # Perform OAuth authentication and get the Flickr API instance - flickr = flickr_authentication() - - # Create OpenAI API client - openai = OpenAI(api_key=os.environ["OPENAI_API_KEY"]) - - all_updated_metadata, total_cost = process_all_photosets(flickr, openai) - - # Write the updated metadata JSON to the file only if there are entries - if all_updated_metadata: - with open(UPDATED_METADATA_FILE, "w") as f: - json.dump(all_updated_metadata, f, indent=2) - print(f"Updated metadata JSON saved to: {UPDATED_METADATA_FILE}") - else: - print("No updated metadata to write.") - - print(f"Total cost of OpenAI API usage: ${total_cost:.4f}") +from autotagger.cli import main if __name__ == "__main__": - main() \ No newline at end of file + raise SystemExit(main()) diff --git a/prompt.example.txt b/prompt.example.txt new file mode 100644 index 0000000..cefe06c --- /dev/null +++ b/prompt.example.txt @@ -0,0 +1,6 @@ +Generate Flickr metadata as a valid JSON object with exactly these fields: +- "title": a title in {{LANGUAGE}}, at most {{MAX_TITLE_CHARS}} characters. +- "description": a {{DESCRIPTION_STYLE}} description in {{LANGUAGE}}, at most {{MAX_DESCRIPTION_CHARS}} characters. +- "keywords": up to {{MAX_KEYWORDS}} lowercase keywords without spaces or symbols. + +Use only visible image evidence and optional album/location context. Do not mention the context. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..1e3509e --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,38 @@ +[build-system] +requires = ["setuptools>=77"] +build-backend = "setuptools.build_meta" + +[project] +name = "flickr-autotagger" +version = "1.0.0" +description = "Generate and safely apply Flickr metadata with OpenAI-compatible vision models" +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "flickrapi==2.4.0", + "openai>=2.46.0,<3", + "python-dotenv>=1.2.2,<2", +] + +[project.optional-dependencies] +dev = [ + "pytest>=9.1.1,<10", + "ruff>=0.15.20,<0.16", +] + +[project.scripts] +flickr-autotagger = "autotagger.cli:main" + +[tool.setuptools.packages.find] +include = ["autotagger*"] + +[tool.pytest.ini_options] +addopts = "-ra" +testpaths = ["tests"] + +[tool.ruff] +line-length = 100 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B", "SIM"] diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..ddd1f44 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,5 @@ +-r requirements.txt +pytest>=9.1.1,<10 +ruff>=0.15.20,<0.16 +setuptools>=77,<84 +wheel>=0.45,<1 diff --git a/requirements.lock b/requirements.lock new file mode 100644 index 0000000..5838b97 --- /dev/null +++ b/requirements.lock @@ -0,0 +1,91 @@ +# +# This file is autogenerated by pip-compile with Python 3.14 +# by the following command: +# +# pip-compile --allow-unsafe --output-file=requirements.lock --strip-extras requirements-dev.txt +# +annotated-types==0.7.0 + # via pydantic +anyio==4.14.2 + # via + # httpx + # openai +certifi==2026.6.17 + # via + # httpcore + # httpx + # requests +charset-normalizer==3.4.9 + # via requests +distro==1.9.0 + # via openai +flickrapi==2.4.0 + # via -r requirements.txt +h11==0.16.0 + # via httpcore +httpcore==1.0.9 + # via httpx +httpx==0.28.1 + # via openai +idna==3.18 + # via + # anyio + # httpx + # requests +iniconfig==2.3.0 + # via pytest +jiter==0.16.0 + # via openai +oauthlib==3.3.1 + # via requests-oauthlib +openai==2.46.0 + # via -r requirements.txt +packaging==26.2 + # via + # pytest + # wheel +pluggy==1.6.0 + # via pytest +pydantic==2.13.4 + # via openai +pydantic-core==2.46.4 + # via pydantic +pygments==2.20.0 + # via pytest +pytest==9.1.1 + # via -r requirements-dev.txt +python-dotenv==1.2.2 + # via -r requirements.txt +requests==2.34.2 + # via + # flickrapi + # requests-oauthlib + # requests-toolbelt +requests-oauthlib==2.0.0 + # via flickrapi +requests-toolbelt==1.0.0 + # via flickrapi +ruff==0.15.22 + # via -r requirements-dev.txt +six==1.17.0 + # via flickrapi +sniffio==1.3.1 + # via openai +tqdm==4.69.0 + # via openai +typing-extensions==4.16.0 + # via + # openai + # pydantic + # pydantic-core + # typing-inspection +typing-inspection==0.4.2 + # via pydantic +urllib3==2.7.0 + # via requests +wheel==0.47.0 + # via -r requirements-dev.txt + +# The following packages are considered to be unsafe in a requirements file: +setuptools==83.0.0 + # via -r requirements-dev.txt diff --git a/requirements.txt b/requirements.txt index 3003501..ec73db4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ -flickrapi -openai -python-dotenv \ No newline at end of file +# Runtime dependency constraints. CI installs the fully resolved requirements.lock. +flickrapi==2.4.0 +openai>=2.46.0,<3 +python-dotenv>=1.2.2,<2 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..1a423e3 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from dataclasses import replace + +import pytest + +from autotagger.config import Settings + + +@pytest.fixture +def settings(monkeypatch: pytest.MonkeyPatch, tmp_path): + for name in ( + "FLICKR_API_KEY", + "FLICKR_API_SECRET", + "FLICKR_OAUTH_TOKEN", + "OPENAI_API_KEY", + "OPENAI_BASE_URL", + "DESCRIPTIONS_TO_ANALYZE", + "SKIP_PREFIX", + "FLICKR_PRIVACY_FILTER", + ): + monkeypatch.delenv(name, raising=False) + value = Settings.from_env() + return replace( + value, + flickr_api_key="flickr-key", + flickr_api_secret="flickr-secret", + openai_api_key="openai-key", + checkpoint_file=tmp_path / "checkpoint.jsonl", + updated_metadata_file=tmp_path / "summary.json", + resume=False, + ) diff --git a/tests/test_analysis.py b/tests/test_analysis.py new file mode 100644 index 0000000..6b48053 --- /dev/null +++ b/tests/test_analysis.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +from dataclasses import replace +from types import SimpleNamespace + +import pytest + +from autotagger.analysis import ( + AnalysisValidationError, + OpenAIImageAnalyzer, + parse_analysis_response, + render_prompt, +) +from autotagger.models import PhotoCandidate, Usage + + +def test_parse_analysis_normalises_and_limits_keywords(): + result = parse_analysis_response( + '```json\n{"title":" Beach ","description":" Waves ",' + '"keywords":["Golden Hour", "SEA!", "sea", "sand"]}\n```', + max_keywords=3, + max_title_chars=100, + max_description_chars=100, + usage=Usage(10, 5, 0.1), + ) + assert result.title == "Beach" + assert result.description == "Waves" + assert result.keywords == ("golden-hour", "sea", "sand") + assert result.usage.cost == 0.1 + + +@pytest.mark.parametrize( + "payload", + ( + "not json", + "[]", + '{"title":"","description":"ok","keywords":["tag"]}', + '{"title":"ok","description":"ok","keywords":"tag"}', + ), +) +def test_parse_analysis_rejects_invalid_schema(payload): + with pytest.raises(AnalysisValidationError): + parse_analysis_response( + payload, + max_keywords=10, + max_title_chars=100, + max_description_chars=100, + ) + + +def test_prompt_file_tokens_are_replaced(settings, tmp_path): + prompt = tmp_path / "prompt.txt" + prompt.write_text("{{LANGUAGE}} {{MAX_KEYWORDS}} {{DESCRIPTION_STYLE}}") + custom = replace(settings, prompt_file=prompt, language="French", max_keywords=7) + assert render_prompt(custom).startswith("French 7") + + +def test_custom_base_url_is_passed_to_client(settings, monkeypatch): + captured = {} + + class FakeClient: + def __init__(self, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr("autotagger.analysis.AsyncOpenAI", FakeClient) + OpenAIImageAnalyzer(replace(settings, openai_base_url="http://localhost:11434/v1")) + assert captured["base_url"] == "http://localhost:11434/v1" + + +def test_azure_provider_uses_dedicated_endpoint_and_api_version(settings, monkeypatch): + captured = {} + + class FakeAzureClient: + def __init__(self, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr("autotagger.analysis.AsyncAzureOpenAI", FakeAzureClient) + OpenAIImageAnalyzer( + replace( + settings, + openai_provider="azure", + azure_openai_endpoint="https://example.openai.azure.com", + azure_openai_api_version="2026-01-01", + ) + ) + assert captured["azure_endpoint"] == "https://example.openai.azure.com" + assert captured["api_version"] == "2026-01-01" + + +def test_analyze_builds_configurable_request(settings): + captured = {} + + class Completions: + async def create(self, **kwargs): + captured.update(kwargs) + return SimpleNamespace( + choices=[ + SimpleNamespace( + message=SimpleNamespace( + content='{"title":"Title","description":"Description",' + '"keywords":["one"]}' + ) + ) + ], + usage=SimpleNamespace(prompt_tokens=100, completion_tokens=20), + ) + + client = SimpleNamespace(chat=SimpleNamespace(completions=Completions())) + analyzer = OpenAIImageAnalyzer(settings, client=client) + photo = PhotoCandidate( + id="1", + photoset_id="set", + photoset_title="Album", + photoset_description="Context", + image_url="https://example.test/photo.jpg", + ) + result = __import__("asyncio").run(analyzer.analyze(photo)) + assert captured["max_completion_tokens"] == settings.openai_max_output_tokens + assert captured["response_format"] == {"type": "json_object"} + assert result.usage.prompt_tokens == 100 diff --git a/tests/test_checkpoint.py b/tests/test_checkpoint.py new file mode 100644 index 0000000..7996473 --- /dev/null +++ b/tests/test_checkpoint.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import json + +from autotagger.checkpoint import CheckpointStore, load_plan, write_summary + + +def test_checkpoint_preserves_state_when_a_skip_event_follows(tmp_path): + path = tmp_path / "checkpoint.jsonl" + store = CheckpointStore(path) + store.append({"photo_id": "1", "status": "analysed", "analysis": {"title": "x"}}) + store.append({"photo_id": "1", "status": "skipped", "reason": "cached"}) + assert CheckpointStore(path).load_latest()["1"]["status"] == "analysed" + + +def test_summary_is_valid_json_and_plan_loader_supports_both_formats(tmp_path): + records = [{"photo_id": "1", "status": "analysed"}] + summary = tmp_path / "summary.json" + write_summary(summary, records) + assert json.loads(summary.read_text()) == records + assert load_plan(summary) == records + + checkpoint = tmp_path / "checkpoint.jsonl" + checkpoint.write_text('{"photo_id":"1"}\n{"photo_id":"2"}\n') + assert [record["photo_id"] for record in load_plan(checkpoint)] == ["1", "2"] diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..c87bb99 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from dataclasses import replace + +import pytest + +from autotagger.config import ( + DEFAULT_OPENAI_BASE_URL, + ConfigurationError, + Settings, +) + + +def test_base_url_defaults_when_environment_is_missing_or_empty(monkeypatch): + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + assert Settings.from_env().openai_base_url == DEFAULT_OPENAI_BASE_URL + monkeypatch.setenv("OPENAI_BASE_URL", "") + assert Settings.from_env().openai_base_url == DEFAULT_OPENAI_BASE_URL + + +def test_existing_github_secrets_are_sufficient(monkeypatch): + credentials = { + "FLICKR_API_KEY": "flickr-key", + "FLICKR_API_SECRET": "flickr-secret", + "FLICKR_OAUTH_TOKEN": '{"oauth_token": "existing-token"}', + "OPENAI_API_KEY": "openai-key", + } + for name, value in credentials.items(): + monkeypatch.setenv(name, value) + + # Missing GitHub repository variables are exposed to the job as empty strings. + for name in ( + "OPENAI_BASE_URL", + "AZURE_OPENAI_ENDPOINT", + "OPENAI_API_VERSION", + ): + monkeypatch.setenv(name, "") + monkeypatch.setenv("OPENAI_PROVIDER", "openai") + + settings = Settings.from_env() + settings.validate(require_openai=True, require_flickr=True) + + assert settings.openai_base_url == DEFAULT_OPENAI_BASE_URL + assert settings.openai_model == "gpt-5-mini" + assert settings.openai_api_key == "openai-key" + + +def test_custom_base_url_is_loaded(monkeypatch): + monkeypatch.setenv("OPENAI_BASE_URL", "http://localhost:11434/v1") + assert Settings.from_env().openai_base_url == "http://localhost:11434/v1" + + +def test_full_chat_completions_endpoint_is_rejected(settings): + with pytest.raises(ConfigurationError, match="base URL"): + replace( + settings, + openai_base_url="https://api.example.test/v1/chat/completions", + ).validate(require_openai=True, require_flickr=True) + + +def test_azure_provider_prefers_azure_api_key(monkeypatch): + monkeypatch.setenv("OPENAI_PROVIDER", "azure") + monkeypatch.setenv("OPENAI_API_KEY", "ordinary-key") + monkeypatch.setenv("AZURE_OPENAI_API_KEY", "azure-key") + monkeypatch.setenv("AZURE_OPENAI_ENDPOINT", "https://example.openai.azure.com") + monkeypatch.setenv("OPENAI_API_VERSION", "2026-01-01") + assert Settings.from_env().openai_api_key == "azure-key" + + +def test_json_lists_are_parsed_without_eval(monkeypatch): + monkeypatch.setenv("SKIP_PREFIX", '["!", "private-"]') + assert Settings.from_env().skip_prefixes == ("!", "private-") + + +def test_invalid_json_list_is_rejected(monkeypatch): + monkeypatch.setenv("SKIP_PREFIX", '[__import__("os").getcwd()]') + with pytest.raises(ConfigurationError, match="valid JSON list"): + Settings.from_env() + + +def test_privacy_filter_all_omits_filter(monkeypatch): + monkeypatch.setenv("FLICKR_PRIVACY_FILTER", "all") + assert Settings.from_env().flickr_privacy_filter is None + + +def test_invalid_update_field_is_rejected(settings): + with pytest.raises(ConfigurationError, match="UPDATE_FIELDS"): + replace(settings, update_fields=("title", "permissions")).validate( + require_openai=True, require_flickr=True + ) + + +def test_required_credentials_are_validated(settings): + with pytest.raises(ConfigurationError, match="OPENAI_API_KEY"): + replace(settings, openai_api_key="").validate(require_openai=True, require_flickr=True) diff --git a/tests/test_flickr_gateway.py b/tests/test_flickr_gateway.py new file mode 100644 index 0000000..4ff0568 --- /dev/null +++ b/tests/test_flickr_gateway.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +from dataclasses import replace +from types import SimpleNamespace + +import flickrapi +import pytest + +from autotagger.flickr_gateway import FlickrGateway +from autotagger.models import Analysis, Photoset + + +class Recorder: + def __init__(self, result=None): + self.calls = [] + self.result = result + + def __call__(self, **kwargs): + self.calls.append(kwargs) + return self.result + + +class ErrorThenSuccess: + def __init__(self, error): + self.error = error + self.calls = 0 + + def __call__(self, **kwargs): + self.calls += 1 + if self.calls == 1: + raise self.error + return {"ok": True} + + +def test_metadata_update_merges_tags_and_uses_space_delimiter(settings): + set_meta = Recorder({}) + set_tags = Recorder({}) + api = SimpleNamespace(photos=SimpleNamespace(setMeta=set_meta, setTags=set_tags)) + gateway = FlickrGateway(settings, api=api) + photo = SimpleNamespace(id="1", tags=("existing", "two words")) + analysis = Analysis("Title", "Description", ("existing", "new")) + + applied = gateway.update_metadata(photo, analysis) + + assert applied == ("title", "description", "tags") + assert set_tags.calls[0]["tags"] == 'existing "two words" new' + assert set_meta.calls[0]["title"] == "Title" + + +def test_replace_mode_does_not_preserve_existing_tags(settings): + set_tags = Recorder({}) + api = SimpleNamespace(photos=SimpleNamespace(setMeta=Recorder({}), setTags=set_tags)) + gateway = FlickrGateway(replace(settings, tag_mode="replace"), api=api) + photo = SimpleNamespace(id="1", tags=("existing",)) + gateway.update_metadata(photo, Analysis("Title", "Description", ("new",))) + assert set_tags.calls[0]["tags"] == "new" + + +def test_photo_pages_are_streamed_and_missing_descriptions_use_get_info(settings): + get_photos = Recorder( + { + "photoset": { + "page": 1, + "pages": 1, + "photo": [ + { + "id": "1", + "title": "Raw title", + "url_m": "https://example.test/1.jpg", + "tags": "one two", + } + ], + } + } + ) + get_info = Recorder( + { + "photo": { + "description": {"_content": "Existing description"}, + "tags": {"tag": [{"raw": "one"}, {"raw": "two"}]}, + } + } + ) + api = SimpleNamespace( + photosets=SimpleNamespace(getPhotos=get_photos), + photos=SimpleNamespace(getInfo=get_info), + ) + gateway = FlickrGateway(settings, api=api) + pages = list(gateway.iter_photo_pages(Photoset("set", "Album"))) + assert pages[0][0].description == "Existing description" + assert pages[0][0].tags == ("one", "two") + assert get_info.calls == [{"photo_id": "1"}] + + +def test_all_privacy_does_not_send_invalid_zero_filter(settings): + get_photos = Recorder({"photoset": {"page": 1, "pages": 1, "photo": []}}) + api = SimpleNamespace( + photosets=SimpleNamespace(getPhotos=get_photos), + photos=SimpleNamespace(), + ) + gateway = FlickrGateway(replace(settings, flickr_privacy_filter=None), api=api) + list(gateway.iter_photo_pages(Photoset("set", "Album"))) + assert "privacy_filter" not in get_photos.calls[0] + + +def test_only_transient_flickr_errors_are_retried(settings): + gateway = FlickrGateway(replace(settings, flickr_retry_backoff=0), api=SimpleNamespace()) + transient = ErrorThenSuccess(flickrapi.FlickrError("unavailable", code=105)) + assert gateway._call(transient) == {"ok": True} + assert transient.calls == 2 + + permanent = ErrorThenSuccess(flickrapi.FlickrError("bad request", code=1)) + with pytest.raises(flickrapi.FlickrError): + gateway._call(permanent) + assert permanent.calls == 1 diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..f5463d0 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import asyncio +from dataclasses import replace + +from autotagger.flickr_gateway import PartialUpdateError +from autotagger.models import Analysis, PhotoCandidate, Photoset, Usage +from autotagger.pipeline import AutotaggerPipeline + + +class FakeFlickr: + def __init__(self, photos): + self.photos = photos + self.updates = [] + + def get_photosets(self): + return [Photoset("set-1", "Album", "Album context")] + + def iter_photo_pages(self, photoset): + yield self.photos + + def update_metadata(self, photo, analysis): + self.updates.append((photo.id, analysis.title)) + return ("title", "description", "tags") + + +class FakeAnalyzer: + fingerprint = "fake-analysis-v1" + + def __init__(self): + self.calls = [] + + async def analyze(self, photo): + self.calls.append(photo.id) + return Analysis( + title=f"Title {photo.id}", + description="Description", + keywords=("keyword",), + usage=Usage(100, 20, 0.01), + ) + + +class PartialFailureFlickr(FakeFlickr): + def update_metadata(self, photo, analysis): + raise PartialUpdateError("tag update failed", ("title", "description")) + + +def photo(photo_id="1", description=""): + return PhotoCandidate( + id=photo_id, + photoset_id="set-1", + photoset_title="Album", + photoset_description="Context", + image_url=f"https://example.test/{photo_id}.jpg", + description=description, + tags=("existing",), + ) + + +def test_dry_run_analyzes_without_updating(settings): + flickr = FakeFlickr([photo()]) + analyzer = FakeAnalyzer() + pipeline = AutotaggerPipeline(settings, flickr, analyzer) + summary = asyncio.run(pipeline.run(apply=False)) + assert summary.analyzed == 1 + assert summary.updated == 0 + assert flickr.updates == [] + assert any(record["status"] == "analysed" for record in summary.records) + assert settings.updated_metadata_file.exists() + + +def test_apply_updates_after_analysis(settings): + flickr = FakeFlickr([photo()]) + pipeline = AutotaggerPipeline(settings, flickr, FakeAnalyzer()) + summary = asyncio.run(pipeline.run(apply=True)) + assert summary.updated == 1 + assert flickr.updates == [("1", "Title 1")] + assert [record["status"] for record in summary.records] == ["analysed", "updated"] + + +def test_dry_run_checkpoint_can_be_resumed_and_applied(settings): + resumable = replace(settings, resume=True) + first_flickr = FakeFlickr([photo()]) + first_analyzer = FakeAnalyzer() + asyncio.run(AutotaggerPipeline(resumable, first_flickr, first_analyzer).run(apply=False)) + + second_flickr = FakeFlickr([photo()]) + second_analyzer = FakeAnalyzer() + summary = asyncio.run( + AutotaggerPipeline(resumable, second_flickr, second_analyzer).run(apply=True) + ) + assert second_analyzer.calls == [] + assert summary.updated == 1 + assert second_flickr.updates == [("1", "Title 1")] + + +def test_repeated_dry_run_keeps_cached_analysis_in_current_report(settings): + resumable = replace(settings, resume=True) + asyncio.run( + AutotaggerPipeline(resumable, FakeFlickr([photo()]), FakeAnalyzer()).run(apply=False) + ) + analyzer = FakeAnalyzer() + summary = asyncio.run( + AutotaggerPipeline(resumable, FakeFlickr([photo()]), analyzer).run(apply=False) + ) + assert analyzer.calls == [] + assert summary.records[0]["status"] == "analysed" + assert summary.records[0]["cached_analysis"] is True + + +def test_existing_description_is_skipped(settings): + flickr = FakeFlickr([photo(description="A human description")]) + analyzer = FakeAnalyzer() + summary = asyncio.run(AutotaggerPipeline(settings, flickr, analyzer).run(apply=True)) + assert analyzer.calls == [] + assert summary.skipped == 1 + + +def test_duplicate_photo_ids_are_not_analyzed_twice(settings): + flickr = FakeFlickr([photo(), photo()]) + analyzer = FakeAnalyzer() + summary = asyncio.run(AutotaggerPipeline(settings, flickr, analyzer).run(apply=False)) + assert analyzer.calls == ["1"] + assert summary.skipped == 1 + + +def test_limit_bounds_new_analyses(settings): + limited = replace(settings, max_photos=1) + flickr = FakeFlickr([photo("1"), photo("2")]) + analyzer = FakeAnalyzer() + summary = asyncio.run(AutotaggerPipeline(limited, flickr, analyzer).run(apply=False)) + assert summary.analyzed == 1 + assert analyzer.calls == ["1"] + + +def test_partial_flickr_update_is_audited(settings): + summary = asyncio.run( + AutotaggerPipeline(settings, PartialFailureFlickr([photo()]), FakeAnalyzer()).run( + apply=True + ) + ) + assert summary.failed == 1 + partial = next(record for record in summary.records if record["status"] == "partially_updated") + assert partial["applied_fields"] == ["title", "description"] + + +def test_dry_run_report_can_be_applied_without_openai(settings): + source_flickr = FakeFlickr([photo()]) + asyncio.run(AutotaggerPipeline(settings, source_flickr, FakeAnalyzer()).run(apply=False)) + + target_flickr = FakeFlickr([]) + summary = AutotaggerPipeline(settings, target_flickr, analyzer=None).apply_plan( + settings.updated_metadata_file + ) + assert summary.updated == 1 + assert target_flickr.updates == [("1", "Title 1")]