diff --git a/docs/architecture.md b/docs/architecture.md index 1c3bfdf..1bb169f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -53,7 +53,16 @@ archives = ["/path/to/archive"] | `rl doctor` | global config, workspace, stock, roll, vocabulary | no | yes with `--fix` | integrity | | `rl normalize` | current archive workspace, roll, vocabulary | roll, vocabulary | yes | normalization | | `rl normalize --photos` | photo folders in current archive workspace | archive folders | yes | photo import | -| `rl batch process` | global config, workspace, roll | roll | no | batch update | +| `rl stock edit` | global config, workspace, roll, vocabulary | roll, vocabulary | no | single-roll editing | +| `rl batch` | global config, workspace, roll | roll | no | batch update | + +These scopes are deliberate: + +- `rl stock edit` is the manual, one-roll editor; +- `rl batch` is the bulk editor; +- `rl search` shares the same filter language as `rl batch`, but never writes; +- `rl normalize` is for structural cleanup and normalization; +- `rl doctor` is for diagnostics and safe repair. ## Lifecycle @@ -112,7 +121,7 @@ rl doctor --fix | `filesystem.py` | archive tree and file discovery | | `app/workspace/` | config, stock, roll storage, vocabularies | | `app/flows/` | interactive flows | -| `app/archive/` | search, stats, batch, normalization | +| `app/archive/` | search, stats, batch, selection, normalization | | `app/diagnostics/` | doctor | | `messages/` | localized user-facing text | @@ -127,3 +136,4 @@ rl doctor --fix - English is the default UI language. Quick start — see [docs/getting-started.md](getting-started.md). Environment setup and CI — see [docs/development.md](development.md). Terms and exact command behavior — see [docs/reference.md](reference.md). +Editing flows — see [docs/editing.md](editing.md). diff --git a/docs/development.md b/docs/development.md index 78699bc..7154eff 100644 --- a/docs/development.md +++ b/docs/development.md @@ -45,14 +45,14 @@ python -m unittest discover -s tests | Read-only | `rl search`, `rl scan`, `rl status`, `rl stats`, `rl vocab` | global config, workspace, roll, vocab | no | no | | Integrity | `rl doctor`, `rl doctor --fix` | global config, workspace, stock, roll, vocab | no | yes | | Normalization | `rl normalize --tags` | workspace, roll, vocab | roll, vocab | yes | -| Batch | `rl batch process` | workspace, roll | roll | no | +| Batch | `rl batch` | workspace, roll | roll | no | `rl load --manual` creates a roll from the film dictionary without changing stock. For `rl features add` and `rl tags add` you can enter several values separated by commas. Autocomplete works per value, duplicates aren't written, `_` is allowed inside a value. `rl config lang` applies immediately in the current process because user-facing messages resolve the language at render time. `rl doctor` checks the global config, workspace config, stock, roll metadata, and vocabularies. Its diagnostics stay in English; only the user-facing UI is localized. Package version comes from git tags at build time. In a source checkout, `rl --version` falls back to the latest git tag if package metadata is not installed. -`rl update` is a runtime convenience command for local installs; it runs `pip install -U git+https://github.com/katrinio/roll.git@main` in the current interpreter. +`rl update` is a runtime convenience command for local installs; it runs `pip install --no-input --upgrade --force-reinstall git+https://github.com/katrinio/roll.git@main` in the current interpreter. ## CI diff --git a/docs/editing.md b/docs/editing.md new file mode 100644 index 0000000..63358f0 --- /dev/null +++ b/docs/editing.md @@ -0,0 +1,160 @@ +# Editing + +`roll` has two editing surfaces: + +- `rl stock edit` for one roll at a time; +- `rl batch` for many rolls at once. + +They solve different problems and should stay separate. + +## Boundaries + +| Command | Scope | Typical use | +|---|---|---| +| `rl stock edit` | one roll | manual correction, field-by-field judgment | +| `rl batch` | many rolls | repeated update across a filtered set | +| `rl search` | many rolls | structured lookup, optionally with free text | +| `rl normalize` | archive structure | folder shape, keywords normalization, photo import | +| `rl doctor` | integrity | report problems, safe fixes only | + +--- + +## `rl stock edit` + +Use this when you want to inspect and adjust one roll by hand. + +What it does: + +- selects a single roll; +- edits its metadata fields directly; +- keeps the current values available in the prompts; +- applies changes only after you confirm each value in the interactive flow. + +Best for: + +- correcting one record; +- changing a camera on one roll; +- refining features or keywords on a single roll; +- adjusting the origin fields on one roll. + +Not for: + +- mass changes across many rolls; +- filtering by year or film name; +- bulk status updates. + +--- + +## `rl batch` + +Use this when you want to apply the same change to many rolls. + +What it does: + +- selects rolls by filters; +- shows a preview of the target set; +- applies one change to the whole selection; +- asks for confirmation before writing. + +Selection is based on filters such as: + +- year; +- film name; +- camera; +- status; +- tags or keywords when needed. + +Within one filter, comma-separated values mean "match any of these". +Across filters, the selection is cumulative. + +Best for: + +- changing a camera on many rolls; +- moving a group of rolls to a new status; +- adding the same feature or tag to a batch; +- cleaning up a whole year or a film family. + +Not for: + +- one-off manual corrections; +- browsing a single record in detail; +- workflows that need per-field judgment on each roll. + +--- + +## Boundary + +Use this rule: + +- one roll, manual judgment, many fields -> `rl stock edit`; +- many rolls, one repeated operation -> `rl batch`. + +## `rl search` + +Use this when you want to find rolls with the same filter language as `rl batch`, but without writing anything. + +What it does: + +- accepts the same structured filters as `rl batch`; +- keeps free-text search for ad hoc lookup; +- shows matching rolls and their basic metadata. + +Selection is based on the same filters: + +- year; +- film name; +- camera; +- status; +- tags or keywords. + +Within one filter, comma-separated values mean "match any of these". +Across filters, the selection is cumulative. + +Best for: + +- finding rolls by year and status; +- narrowing by film family before a batch update; +- looking up tags or camera names; +- searching by a short free-text fragment when you do not need a structured filter. + +Not for: + +- changing metadata; +- structural normalization; +- integrity repair. + +If you need both at different times, start with `stock edit` for the outlier and use `batch` for the rest. + +--- + +## Examples + +```bash +rl search --year 2025 --status loaded +``` + +```bash +rl search --film "Kodak Gold 200, Ilford HP5 Plus" --query balcony +``` + +```bash +rl stock edit +``` + +```bash +rl batch --year 2025 --film "Kodak Gold 200, Ilford HP5 Plus" --set camera="Pentax K1000" +``` + +```bash +rl batch --year 2025 --set status=processed +``` + +```bash +rl batch --film "Kodak Gold 200" --add-tag "summer,belgrade" +``` + +## Related + +- [Getting Started](getting-started.md) +- [Architecture](architecture.md) +- [Reference](reference.md) diff --git a/docs/getting-started.md b/docs/getting-started.md index 5e9d9e4..c52ab78 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -24,7 +24,7 @@ rl search kir balcony # half a year later — found it | Find / inspect | `rl search`, `rl scan`, `rl status`, `rl stats [-v]`, `rl vocab` | | Integrity | `rl doctor`, `rl doctor --fix`, `rl normalize --tags` | | Photo import | `rl normalize --photos` | -| Batch | `rl batch process` | +| Editing | `rl stock edit`, `rl batch` | ## Out of Scope @@ -32,11 +32,11 @@ sync between machines · cloud · web UI · migrating old formats · image proce The CLI defaults to English in the global config and `rl config lang` changes it. `rl --version` prints the current version. If a newer git tag is available in the current checkout, it also prints a short update hint and points to `rl update`. -`rl update` upgrades the installed package from the GitHub repo with `pip`. +`rl update` reinstalls the app from the GitHub repository. `rl normalize --photos` works in the current archive workspace and can turn a raw photo folder into an archive month based on the dominant photo date. ## Rule If it doesn't help find a roll from memory faster — it's out of scope. -Architecture and exact behavior — see [docs/architecture.md](architecture.md). Terms and exact rules — see [docs/reference.md](reference.md). Development setup — see [docs/development.md](development.md). +Architecture and exact behavior — see [docs/architecture.md](architecture.md). Editing flows — see [docs/editing.md](editing.md). Terms and exact rules — see [docs/reference.md](reference.md). Development setup — see [docs/development.md](development.md). diff --git a/docs/index.md b/docs/index.md index 88bc8ed..f78a968 100644 --- a/docs/index.md +++ b/docs/index.md @@ -5,8 +5,9 @@ Start here: | Doc | What it answers | |---|---| | [Getting Started](getting-started.md) | What the tool does and how to begin | +| [Editing](editing.md) | How to use `stock edit` and `batch` | | [Architecture](architecture.md) | What files exist, what commands read and write, how the archive is structured | | [Reference](reference.md) | Exact terms, behavior, and rules | | [Development](development.md) | Setup, checks, and CI | -If you are new to the tool, read [Getting Started](getting-started.md) first, then [Architecture](architecture.md). Use [Reference](reference.md) when you need exact rules. +If you are new to the tool, read [Getting Started](getting-started.md) first, then [Editing](editing.md) and [Architecture](architecture.md). Use [Reference](reference.md) when you need exact rules. diff --git a/docs/reference.md b/docs/reference.md index b156dcf..f7fa94e 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -82,6 +82,14 @@ Brings folder names to a consistent shape: builds a plan, asks for confirmation, --- +## Editing + +`rl stock edit` is a single-roll manual editor. `rl batch` is a bulk editor that applies one change to many rolls selected by filters. `rl search` uses the same filter language for read-only lookup and can also keep free-text search. + +See [docs/editing.md](editing.md) for the shared editing model and the boundary between the two commands. + +--- + ## Doctor Integrity check: global config, workspaces, stock, roll metadata, vocabularies, suspicious and unindexed folders. Diagnostics stay in English. `--fix` applies safe fixes and prints them in a separate block; `-v` shows the full list of fixes. diff --git a/src/roll/__main__.py b/src/roll/__main__.py new file mode 100644 index 0000000..0728c02 --- /dev/null +++ b/src/roll/__main__.py @@ -0,0 +1,5 @@ +from roll.cli import app + + +if __name__ == "__main__": + app() diff --git a/src/roll/app/archive/batch.py b/src/roll/app/archive/batch.py index 9d87f00..873620a 100644 --- a/src/roll/app/archive/batch.py +++ b/src/roll/app/archive/batch.py @@ -1,35 +1,102 @@ +from dataclasses import replace from pathlib import Path import typer -from roll.app.workspace.roll_store import update_roll_status -from roll.app.archive.search import find_rolls +from roll.app.archive.search import RollIndex +from roll.app.archive.selection import select_rolls +from roll.app.workspace.roll_store import ( + RollMetadata, + load_roll_metadata, + save_roll_metadata, +) from roll.helpers.output import echo_list from roll.messages import Msg def process_archives(archives: list[Path]) -> int: - loaded_rolls: list[Path] = [] + return batch_rolls(archives, statuses=["loaded"], status="processed") - for archive in archives: - for roll in find_rolls(archive): - if roll.status == "loaded": - loaded_rolls.append(roll.folder) - if not loaded_rolls: - typer.echo(str(Msg.BATCH_NO_LOADED)) +def batch_rolls( + archives: list[Path], + *, + year: str | None = None, + films: list[str] | None = None, + cameras: list[str] | None = None, + statuses: list[str] | None = None, + status: str | None = None, + set_camera: str | None = None, + add_features: list[str] | None = None, + add_tags: list[str] | None = None, +) -> int: + rolls = select_rolls( + archives, + year=year, + films=films, + cameras=cameras, + statuses=statuses, + ) + + if not rolls: + typer.echo(str(Msg.NO_RESULTS)) return 0 - typer.echo(f"{Msg.BATCH_WILL_PROCESS} {len(loaded_rolls)}") - echo_list((str(path) for path in loaded_rolls)) + preview = [str(roll.folder) for roll in rolls] + typer.echo(f"{Msg.BATCH_WILL_PROCESS} {len(preview)}") + echo_list(preview) if not typer.confirm(Msg.BATCH_CONFIRM, default=False): return 0 changed = 0 - for folder in loaded_rolls: - update_roll_status(folder / "roll.toml", "processed") + for roll in rolls: + updated = _apply_changes( + roll, + status=status, + set_camera=set_camera, + add_features=add_features, + add_tags=add_tags, + ) + save_roll_metadata(roll.folder / "roll.toml", updated) changed += 1 typer.echo(f"{Msg.BATCH_PROCESSED} {changed}") return changed + + +def _apply_changes( + roll: RollIndex, + *, + status: str | None, + set_camera: str | None, + add_features: list[str] | None, + add_tags: list[str] | None, +) -> RollMetadata: + metadata = load_roll_metadata(roll.folder / "roll.toml") + updated = metadata + if status is not None: + updated = replace(updated, status=status) + if set_camera is not None: + updated = replace(updated, camera=set_camera) + if add_features: + updated = replace( + updated, + features=_merge_unique(updated.features, add_features), + ) + if add_tags: + updated = replace( + updated, + keywords=_merge_unique( + updated.keywords, [item.upper() for item in add_tags] + ), + ) + return updated + + +def _merge_unique(existing: list[str], new_values: list[str]) -> list[str]: + merged = list(existing) + for value in new_values: + if value not in merged: + merged.append(value) + return merged diff --git a/src/roll/app/archive/commands.py b/src/roll/app/archive/commands.py new file mode 100644 index 0000000..26c7318 --- /dev/null +++ b/src/roll/app/archive/commands.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +from roll.app.archive.search import search_rolls_by_filters +from roll.app.archive.search_output import render_search_results +from roll.app.archive.stats_output import render_stats_report +from roll.app.archive.status_output import render_status_report +from roll.app.diagnostics.doctor_output import render_doctor +from roll.app.workspace.vocabulary import archive_vocabulary +from roll.app.archive.selection import split_csv +from roll.filesystem import ( + build_archive_tree, + count_photo_files, + find_roll_folders, + find_unindexed_folders, +) +from roll.helpers.output import echo_lines, echo_section +from roll.messages import Msg +from roll.helpers.guards import require_archive, require_config +import typer + + +def scan() -> None: + archive = require_archive(require_config()) + + if not archive.exists(): + typer.echo(f"{Msg.ARCHIVE_MISSING} {archive}") + raise typer.Exit(code=1) + + echo_section(Msg.ARCHIVE_HEADER, [str(archive)]) + roll_folders = find_roll_folders(archive) + tree = build_archive_tree(archive) + photo_count = sum(count_photo_files(folder) for folder in roll_folders) + + if tree: + typer.echo(str(Msg.TREE_HEADER)) + echo_lines(tree) + typer.echo("") + + typer.echo(f"{Msg.FOLDERS} {len(roll_folders)}") + typer.echo(f"{Msg.FILES} {photo_count}") + + +def status() -> None: + archive = require_archive(require_config()) + + from roll.app.archive.search import find_rolls + from roll.app.archive.stats import _count_statuses + + roll_folders = find_roll_folders(archive) + unindexed_folders = find_unindexed_folders(archive) + rolls = find_rolls(archive) + status_counts = _count_statuses(rolls) + + render_status_report(archive, roll_folders, unindexed_folders, status_counts) + + +def stats(year: str | None, verbose: bool) -> None: + archive = require_archive(require_config()) + render_stats_report(archive, year, verbose) + + +def vocab() -> None: + archive = require_archive(require_config()) + vocab = archive_vocabulary(archive) + + for title, items in ( + (Msg.VOCAB_FILMS, vocab["films"].read()), + (Msg.VOCAB_CAMERAS, vocab["cameras"].read()), + (Msg.VOCAB_FEATURES, vocab["features"].read()), + (Msg.VOCAB_KEYWORDS, vocab["keywords"].read()), + ): + echo_section(title, [f"- {item}" for item in items]) + + +def search( + year: str | None = None, + film: str | None = None, + camera: str | None = None, + status: str | None = None, + query: str | None = None, +) -> None: + if not any([year, film, camera, status, query]): + typer.echo(str(Msg.SEARCH_NEEDS_QUERY_OR_FILTERS)) + raise typer.Exit(code=1) + + archive = require_archive(require_config()) + results = search_rolls_by_filters( + [archive], + year=year, + films=split_csv(film), + cameras=split_csv(camera), + statuses=split_csv(status), + query=query, + ) + + if not results: + typer.echo(str(Msg.NO_RESULTS)) + return + + render_search_results(results) + + +def doctor(fix: bool, verbose: bool) -> None: + if render_doctor(fix=fix, verbose=verbose): + raise typer.Exit(code=1) diff --git a/src/roll/app/archive/normalize_cli.py b/src/roll/app/archive/normalize_cli.py new file mode 100644 index 0000000..5b40593 --- /dev/null +++ b/src/roll/app/archive/normalize_cli.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from pathlib import Path + +import typer + +from roll.app.archive.normalization import NormalizationPlan, RenameRule +from roll.app.archive.photo_dates import guess_archive_year, guess_archive_month +from roll.messages import Msg + + +def build_photo_normalization_plans(archive: Path) -> list[NormalizationPlan]: + folders = _photo_folders(archive) + year = guess_archive_year(archive) + if year is None: + typer.echo(str(Msg.CLI_UNINITIALIZED)) + raise typer.Exit(code=1) + + if not typer.confirm( + str(Msg.NORMALIZE_PHOTOS_CONFIRM_YEAR).format(year=year), default=True + ): + typed_year = typer.prompt( + str(Msg.NORMALIZE_PHOTOS_YEAR).format(folder=archive.name) + ) + year = _parse_year(typed_year) + + manual_months = len(folders) > 1 and typer.confirm( + str(Msg.NORMALIZE_PHOTOS_MANUAL), default=False + ) + return [ + _build_photo_plan_for_folder(folder, archive, year, manual_months) + for folder in folders + ] + + +def echo_photo_plan_preview(plans: list[NormalizationPlan]) -> None: + lines = [] + for plan in plans: + for rule in plan.rules: + lines.append( + f"{rule.folder.name} -> {rule.target.relative_to(plan.archive)}" + ) + + if lines: + typer.echo(str(Msg.NORMALIZE_PHOTOS_PREVIEW)) + for line in lines: + typer.echo(f" {line}") + + +def _build_photo_plan_for_folder( + folder: Path, archive: Path, year: int, manual_months: bool +) -> NormalizationPlan: + month = _prompt_month(folder) if manual_months else _guess_month(folder) + + if month is None: + return NormalizationPlan(archive=archive, rules=[], conflicts=[]) + + target = archive / f"{year:04d}" / f"{month:02d}-01" + if target.exists(): + return NormalizationPlan( + archive=archive, + rules=[], + conflicts=[f"{Msg.NORMALIZE_PHOTOS_MONTH} {target}"], + ) + + return NormalizationPlan( + archive=archive, rules=[RenameRule(folder=folder, target=target)], conflicts=[] + ) + + +def _photo_folders(archive: Path) -> list[Path]: + return [ + path for path in archive.iterdir() if path.is_dir() and path.name != ".roll" + ] + + +def _prompt_month(folder: Path) -> int: + while True: + value = typer.prompt( + str(Msg.NORMALIZE_PHOTOS_MONTH).format(folder=folder.name) + ).strip() + month = _parse_month(value) + if month is not None: + return month + + +def _guess_month(folder: Path) -> int | None: + guess = guess_archive_month(folder) + return guess.month if guess is not None else None + + +def _parse_year(value: str) -> int: + year = value.strip() + if len(year) == 4 and year.isdigit(): + return int(year) + raise typer.Exit(code=1) + + +def _parse_month(value: str) -> int | None: + month = value.strip() + if len(month) == 2 and month.isdigit() and 1 <= int(month) <= 12: + return int(month) + return None diff --git a/src/roll/app/archive/normalize_commands.py b/src/roll/app/archive/normalize_commands.py new file mode 100644 index 0000000..f7e2f87 --- /dev/null +++ b/src/roll/app/archive/normalize_commands.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import typer + +from roll.app.archive.normalize_cli import ( + build_photo_normalization_plans, + echo_photo_plan_preview, +) +from roll.app.archive.normalization import ( + apply_normalization_plans, + build_normalization_plan, + normalize_keywords_in_archive, +) +from roll.app.archive.normalization_output import render_normalization_plans +from roll.helpers.guards import require_config, require_current_archive +from roll.messages import Msg, Normalize + + +def normalize(tags: bool, photos: bool) -> None: + config = require_config() + archive = require_current_archive(config) + + if tags: + touched = normalize_keywords_in_archive(archive) + if touched: + typer.echo(str(Msg.TAGS_NORMALIZED)) + for path in touched: + typer.echo(f" {path}") + else: + typer.echo(str(Msg.TAGS_ALREADY_NORMALIZED)) + return + + if photos: + plans = build_photo_normalization_plans(archive) + total_rules, has_changes = render_normalization_plans(plans) + if not has_changes: + return + + all_conflicts = [conflict for plan in plans for conflict in plan.conflicts] + if all_conflicts: + raise typer.Exit(code=1) + + echo_photo_plan_preview(plans) + if not typer.confirm( + str(Normalize.QUESTION).format(count=total_rules), default=False + ): + return + + apply_normalization_plans(plans) + return + + plans = [build_normalization_plan(archive)] + total_rules, has_changes = render_normalization_plans(plans) + if not has_changes: + return + + all_conflicts = [conflict for plan in plans for conflict in plan.conflicts] + if all_conflicts: + raise typer.Exit(code=1) + + if not typer.confirm( + str(Normalize.QUESTION).format(count=total_rules), default=False + ): + return + + apply_normalization_plans(plans) diff --git a/src/roll/app/archive/search.py b/src/roll/app/archive/search.py index d91f60f..f27bb11 100644 --- a/src/roll/app/archive/search.py +++ b/src/roll/app/archive/search.py @@ -2,6 +2,7 @@ from dataclasses import dataclass from pathlib import Path +from roll.app.archive.selection import normalize_text, select_rolls from roll.filesystem import find_roll_folders, get_index_file @@ -50,11 +51,32 @@ def search_rolls(archive: Path, query: str | None) -> list[RollIndex]: if not query: return [] - normalized_query = query.lower() - results: list[RollIndex] = [] + return _filter_by_query(find_rolls(archive), query) - for roll in find_rolls(archive): - searchable_text = " ".join( + +def search_rolls_by_filters( + archives: list[Path], + *, + year: str | None = None, + films: list[str] | None = None, + cameras: list[str] | None = None, + statuses: list[str] | None = None, + query: str | None = None, +) -> list[RollIndex]: + rolls = select_rolls( + archives, year=year, films=films, cameras=cameras, statuses=statuses + ) + return rolls if not query else _filter_by_query(rolls, query) + + +def _filter_by_query(rolls: list[RollIndex], query: str) -> list[RollIndex]: + normalized_query = normalize_text(query) + return [roll for roll in rolls if normalized_query in _searchable_text(roll)] + + +def _searchable_text(roll: RollIndex) -> str: + return normalize_text( + " ".join( [ roll.film, roll.camera, @@ -62,9 +84,5 @@ def search_rolls(archive: Path, query: str | None) -> list[RollIndex]: *roll.features, *roll.keywords, ] - ).lower() - - if normalized_query in searchable_text: - results.append(roll) - - return results + ) + ) diff --git a/src/roll/app/archive/selection.py b/src/roll/app/archive/selection.py new file mode 100644 index 0000000..12318fc --- /dev/null +++ b/src/roll/app/archive/selection.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + + +def split_csv(value: str | None) -> list[str]: + if not value: + return [] + return [item.strip() for item in value.split(",") if item.strip()] + + +def normalize_set(values: list[str] | None) -> set[str]: + return {normalize_text(value) for value in values or [] if value} + + +def normalize_text(value: str) -> str: + return value.casefold().strip() + + +def matches_year(loaded_at: str, year: str | None) -> bool: + return year is None or loaded_at[:4] == year + + +def matches_any(value: str, candidates: set[str]) -> bool: + return not candidates or normalize_text(value) in candidates + + +def select_rolls( + archives: list[Path], + *, + year: str | None = None, + films: list[str] | None = None, + cameras: list[str] | None = None, + statuses: list[str] | None = None, +) -> list[Any]: + from roll.app.archive.search import find_rolls + + selected: list[Any] = [] + film_set = normalize_set(films) + camera_set = normalize_set(cameras) + status_set = normalize_set(statuses) + + for archive in archives: + for roll in find_rolls(archive): + if not matches_year(roll.loaded_at, year): + continue + if not matches_any(roll.film, film_set): + continue + if not matches_any(roll.camera, camera_set): + continue + if not matches_any(roll.status, status_set): + continue + selected.append(roll) + + return selected + + +def roll_choice_label(path: Path, status: str) -> str: + return f"{str(path.relative_to(path.parents[1]))} ({status})" diff --git a/src/roll/app/flows/stock.py b/src/roll/app/flows/stock.py index 9c1c876..a482689 100644 --- a/src/roll/app/flows/stock.py +++ b/src/roll/app/flows/stock.py @@ -1,57 +1,26 @@ from __future__ import annotations -from datetime import date -from pathlib import Path -import textwrap - import typer -from prompt_toolkit import prompt -from prompt_toolkit.completion import FuzzyCompleter, WordCompleter -from roll.filesystem import find_roll_folders -from roll.app.workspace.roll_store import ( - RollMetadata, - load_roll_metadata, - save_roll_metadata, - update_roll_features, - update_roll_keywords, - update_roll_status, +from roll.messages import Msg +from roll.app.flows.stock_edit import ( + edit as _edit, + edit_batch as _edit_batch, + edit_list_field as _edit_list_field, ) -from roll.app.workspace.stock_store import ( - StockItem, - add_to_stock, - load_stock, - remove_from_stock, - save_stock, +from roll.app.flows.stock_lifecycle import ( + add as _add, + load as _load, + list_stock as _list_stock, + process as _process, ) -from roll.app.workspace.statuses import VALID_STATUSES -from roll.app.workspace.workspace import workspace_for -from roll.app.archive.normalization import apply_keyword_vocab_fixes -from roll.helpers.autocomplete import autocomplete_many_prompt, autocomplete_prompt -from roll.helpers.guards import require_archive, require_config -from roll.helpers.output import echo_lines -from roll.messages import Msg app = typer.Typer(help=Msg.STOCK_HEADER) @app.command("add") def add() -> None: - archive = require_archive(require_config()) - workspace = workspace_for(archive) - - film = autocomplete_prompt(str(Msg.PROMPT_FILM), workspace.dictionary("films")) - quantity = typer.prompt(str(Msg.PROMPT_QUANTITY), type=int) - if quantity <= 0: - typer.echo(str(Msg.INVALID_QUANTITY)) - raise typer.Exit(code=1) - - try: - items = load_stock(workspace.stock_file) - save_stock(workspace.stock_file, add_to_stock(items, film, quantity)) - except ValueError as exc: - typer.echo(str(exc)) - raise typer.Exit(code=1) + _add() @app.command("load") @@ -60,387 +29,66 @@ def load( False, "--manual", help="Enter film manually from dictionary." ), ) -> None: - archive = require_archive(require_config()) - workspace = workspace_for(archive) - - stock: list[StockItem] = [] - if not manual: - try: - stock = load_stock(workspace.stock_file) - except ValueError as exc: - typer.echo(str(exc)) - raise typer.Exit(code=1) - - if not stock: - typer.echo(str(Msg.STOCK_EMPTY_MANUAL)) - raise typer.Exit(code=1) - - selected = ( - _choose_stock_item(stock) if not manual else _choose_manual_film(workspace) - ) - camera = autocomplete_prompt( - str(Msg.PROMPT_CAMERA), workspace.dictionary("cameras") - ) - loaded_at = _prompt_loaded_at() - roll_folder = _create_roll_folder(archive, loaded_at) - roll_file = roll_folder / "roll.toml" - - if roll_file.exists(): - typer.echo(f"{Msg.ROLL_EXISTS} {roll_file}") - raise typer.Exit(code=1) - - try: - roll_folder.mkdir(parents=True, exist_ok=True) - save_roll_metadata( - roll_file, - RollMetadata( - status=VALID_STATUSES[0], - film=selected.film, - camera=camera, - loaded_at=loaded_at, - features=[], - keywords=[], - ), - ) - if not manual: - save_stock(workspace.stock_file, remove_from_stock(stock, selected.film, 1)) - features = autocomplete_many_prompt( - str(Msg.VOCAB_FEATURES), workspace.dictionary("features") - ) - if features: - update_roll_features(roll_file, features) - - tags = autocomplete_many_prompt( - str(Msg.VOCAB_KEYWORDS), workspace.dictionary("keywords") - ) - if tags: - update_roll_keywords(roll_file, tags) - apply_keyword_vocab_fixes(archive, tags) - except Exception: - _cleanup_failed_load(roll_folder, roll_file) - raise - - typer.echo(f"{Msg.LOAD_SUCCESS} {selected.film}") + _load(manual) @app.command("process") def process() -> None: - _finish_roll("processed", "Processed") + _process("processed", "Processed") @app.command("failed") def failed() -> None: - _finish_roll("failed", "Marked as failed") + _process("failed", "Marked as failed") @app.command("list") def list_stock() -> None: - archive = require_archive(require_config()) - workspace = workspace_for(archive) - - try: - items = load_stock(workspace.stock_file) - except ValueError as exc: - typer.echo(str(exc)) - raise typer.Exit(code=1) - - if not items: - typer.echo(str(Msg.STOCK_EMPTY)) - return - - echo_lines([Msg.STOCK_HEADER]) - for item in items: - typer.echo(f"{item.film:<20} ×{item.quantity}") + _list_stock() -@app.command("edit") def edit() -> None: - archive = require_archive(require_config()) - rolls = _rolls(archive) - if not rolls: - typer.echo(str(Msg.NO_LOADED_ROLLS)) - raise typer.Exit(code=1) + _edit() - roll = _choose_roll(rolls) - metadata = load_roll_metadata(roll / "roll.toml") - updated = _prompt_roll_metadata(archive, roll / "roll.toml", metadata) - save_roll_metadata(roll / "roll.toml", updated) - typer.echo(f"{Msg.ROLL_EDIT_UPDATED} {updated.loaded_at}") - -def _prompt_loaded_at() -> str: - value = typer.prompt(str(Msg.PROMPT_LOAD_DATE)) - normalized = value.strip().split("T", 1)[0].split(" ", 1)[0] - try: - return date.fromisoformat(normalized).isoformat() - except ValueError as exc: - typer.echo(str(Msg.INVALID_DATE)) - raise typer.Exit(code=1) from exc - - -def _create_roll_folder(archive: Path, loaded_at: str) -> Path: - loaded_date = date.fromisoformat(loaded_at) - return archive / loaded_date.strftime("%Y") / loaded_date.strftime("%m-%d") - - -def _choose_stock_item(items: list[StockItem]) -> StockItem: - labels = [f"{item.film} ×{item.quantity}" for item in items] - completer = FuzzyCompleter( - WordCompleter(labels, ignore_case=True, sentence=True, match_middle=True) +def edit_list_field( + prompt_title: str, dictionary_name: str, success_label: str +) -> None: + _edit_list_field(prompt_title, dictionary_name, success_label) + + +def edit_batch( + year: str | None, + film: str | None, + camera: str | None, + status: str | None, + set_status: str | None, + set_camera: str | None, + add_feature: str | None, + add_tag: str | None, +) -> None: + _edit_batch( + year, film, camera, status, set_status, set_camera, add_feature, add_tag ) - while True: - value = prompt( - str(Msg.PROMPT_MANUAL_FILM), - completer=completer, - complete_while_typing=True, - ).strip() - if not value: - continue - - selected = _resolve_stock_choice(items, value) - if selected is not None: - return selected - - typer.echo(str(Msg.CHOOSE_STOCK)) - - -def _choose_manual_film(workspace) -> StockItem: - film = autocomplete_prompt(str(Msg.PROMPT_FILM), workspace.dictionary("films")) - return StockItem(film=film, quantity=1) - - -def _resolve_stock_choice(items: list[StockItem], candidate: str) -> StockItem | None: - normalized = _normalize_choice(candidate) - - exact_film_matches = [ - item for item in items if item.film.casefold() == candidate.casefold() - ] - if len(exact_film_matches) == 1: - return exact_film_matches[0] - - exact_label_matches = [ - item - for item in items - if f"{item.film} ×{item.quantity}".casefold() == candidate.casefold() - ] - if len(exact_label_matches) == 1: - return exact_label_matches[0] - - fuzzy_matches = [ - item for item in items if normalized in _normalize_choice(item.film) - ] - if len(fuzzy_matches) == 1: - return fuzzy_matches[0] - - return None +def _rolls(archive): + from roll.app.flows.stock_edit import _rolls as stock_edit_rolls -def _normalize_choice(value: str) -> str: - return "".join(ch for ch in value.casefold() if ch.isalnum()) + return stock_edit_rolls(archive) -def _choose_roll(rolls: list[Path]) -> Path: - labels = [_format_roll_label(path) for path in rolls] - selected_label = _prompt_choice(str(Msg.ROLL_EDIT_SELECT), labels) - for path in rolls: - if _format_roll_label(path) == selected_label: - return path - raise ValueError(Msg.NO_CHOICE) - - -def _rolls(archive: Path) -> list[Path]: - rolls: list[Path] = [] - for folder in find_roll_folders(archive): - try: - load_roll_metadata(folder / "roll.toml") - except ValueError: - continue - rolls.append(folder) - return rolls - - -def _finish_roll(status: str, label: str) -> None: - archive = require_archive(require_config()) - loaded_rolls = [ - path - for path in _rolls(archive) - if load_roll_metadata(path / "roll.toml").status == "loaded" - ] - if not loaded_rolls: - typer.echo(str(Msg.NO_LOADED_ROLLS)) - raise typer.Exit(code=1) - - selected = _choose_roll(loaded_rolls) - try: - metadata = update_roll_status(selected / "roll.toml", status) - except ValueError as exc: - typer.echo(str(exc)) - raise typer.Exit(code=1) - - typer.echo(f"{label}: {metadata.film}") - - -def _choose_roll(rolls: list[Path]) -> Path: - labels = [_format_roll_label(path) for path in rolls] - selected_label = _prompt_choice(str(Msg.ROLL_EDIT_SELECT), labels) - for path in rolls: - if selected_label == _format_roll_label(path): - return path - raise ValueError(Msg.NO_CHOICE) - - -def _format_roll_label(path: Path) -> str: - metadata = load_roll_metadata(path / "roll.toml") - return str(Msg.ROLL_EDIT_ROLL_LABEL).format( - path=str(path.relative_to(path.parents[1])), - film=metadata.film, - camera=metadata.camera, - status=metadata.status, +def _format_roll_label(path): + from roll.app.flows.stock_edit import ( + _format_roll_label as stock_edit_format_roll_label, ) - -def _prompt_enum(label: Msg, values: list[str], current: str) -> str: - selected = _prompt_choice_panel(str(label), values, current) - return selected if selected is not None else current + return stock_edit_format_roll_label(path) -def _prompt_roll_metadata( - archive: Path, roll_file: Path, metadata: RollMetadata -) -> RollMetadata: - workspace = workspace_for(archive) - film = _prompt_optional_autocomplete( - Msg.ROLL_EDIT_FILM, workspace.dictionary("films"), metadata.film - ) - camera = _prompt_optional_autocomplete( - Msg.ROLL_EDIT_CAMERA, workspace.dictionary("cameras"), metadata.camera - ) - status = _prompt_enum(Msg.ROLL_EDIT_STATUS, list(VALID_STATUSES), metadata.status) - features = _prompt_optional_many( - Msg.ROLL_EDIT_FEATURES, workspace.dictionary("features"), metadata.features - ) - keywords = _prompt_optional_many( - Msg.ROLL_EDIT_KEYWORDS, workspace.dictionary("keywords"), metadata.keywords - ) - original_source = _prompt_enum( - Msg.ROLL_EDIT_ORIGINAL_SOURCE, - ["negative", "slide", "print", "digital", "unknown"], - metadata.original_source, - ) - digital_copy = _prompt_enum( - Msg.ROLL_EDIT_DIGITAL_COPY, - ["scan", "photo", "none", "unknown"], - metadata.digital_copy, - ) - original_status = _prompt_enum( - Msg.ROLL_EDIT_ORIGINAL_STATUS, - ["present", "lost", "unknown"], - metadata.original_status, +def _prompt_roll_metadata(archive, roll_file, metadata): + from roll.app.flows.stock_edit import ( + _prompt_roll_metadata as stock_edit_prompt_roll_metadata, ) - return RollMetadata( - status=status, - film=film, - camera=camera, - loaded_at=metadata.loaded_at, - features=features, - keywords=[value.upper() for value in keywords], - original_source=original_source, - digital_copy=digital_copy, - original_status=original_status, - ) - - -def _prompt_optional_autocomplete(label: Msg, dictionary, current: str) -> str: - value = prompt(f"{label} [{current}]: ", complete_while_typing=True).strip() - if not value: - return current - - choices = dictionary.read() - for existing in choices: - if existing.casefold() == value.casefold(): - return existing - - return dictionary.add(value) - - -def _prompt_optional_many(label: Msg, dictionary, current: list[str]) -> list[str]: - choices = dictionary.read() - value = prompt( - f"{label} [{', '.join(current)}]: ", - completer=FuzzyCompleter( - WordCompleter(choices, ignore_case=True, sentence=True, match_middle=True) - ), - complete_while_typing=True, - ).strip() - if not value: - return current - - selected: list[str] = [] - for token in [item.strip() for item in value.split(",") if item.strip()]: - for existing in dictionary.read(): - if existing.casefold() == token.casefold(): - token = existing - break - else: - token = dictionary.add(token) - if token not in selected: - selected.append(token) - merged = list(current) - for token in selected: - if token not in merged: - merged.append(token) - return merged - - -def _prompt_choice(title: str, choices: list[str]) -> str: - selected = _prompt_choice_panel(title, choices, choices[0] if choices else "") - if selected is None: - raise ValueError(Msg.NO_CHOICE) - return selected - - -def _prompt_choice_panel( - title: str, choices: list[str], current: str | None = None -) -> str | None: - content_width = max( - len(title), - len(f"Current: {current}") if current is not None else 0, - *(len(f"{index + 1}. {choice}") for index, choice in enumerate(choices)), - 24, - ) - width = min(content_width + 4, 88) - border = "┌" + "─" * (width - 2) + "┐" - footer = "└" + "─" * (width - 2) + "┘" - prompt_lines = [border, f"│ {title.ljust(width - 4)} │"] - if current is not None: - prompt_lines.append( - f"│ {textwrap.shorten(f'Current: {current}', width=width - 4, placeholder='…').ljust(width - 4)} │" - ) - prompt_lines.append("├" + "─" * (width - 2) + "┤") - prompt_lines.extend( - f"│ {textwrap.shorten(f'{index + 1}. {choice}', width=width - 4, placeholder='…').ljust(width - 4)} │" - for index, choice in enumerate(choices) - ) - prompt_lines.append(footer) - echo_lines(prompt_lines) - - while True: - value = prompt("Select [number / enter to keep]: ").strip() - if not value: - return current - if value.isdigit(): - index = int(value) - 1 - if 0 <= index < len(choices): - return choices[index] - if value in choices: - return value - -def _cleanup_failed_load(roll_folder: Path, roll_file: Path) -> None: - if roll_file.exists(): - roll_file.unlink() - if roll_folder.exists(): - try: - roll_folder.rmdir() - except OSError: - pass + return stock_edit_prompt_roll_metadata(archive, roll_file, metadata) diff --git a/src/roll/app/flows/stock_edit.py b/src/roll/app/flows/stock_edit.py new file mode 100644 index 0000000..1f7b0e7 --- /dev/null +++ b/src/roll/app/flows/stock_edit.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +from pathlib import Path +import textwrap + +import typer +from prompt_toolkit import prompt +from prompt_toolkit.completion import FuzzyCompleter, WordCompleter + +from roll.app.archive.batch import batch_rolls +from roll.app.archive.normalization import apply_keyword_vocab_fixes +from roll.app.archive.selection import roll_choice_label, split_csv +from roll.app.workspace.roll_store import ( + RollMetadata, + load_roll_metadata, + save_roll_metadata, + update_roll_features, + update_roll_keywords, +) +from roll.app.workspace.workspace import workspace_for +from roll.app.workspace.statuses import VALID_STATUSES +from roll.filesystem import find_roll_folders +from roll.helpers.autocomplete import autocomplete_many_prompt +from roll.helpers.output import echo_lines +from roll.helpers.guards import require_archive, require_config +from roll.messages import Msg + + +def edit() -> None: + archive = require_archive(require_config()) + rolls = _rolls(archive) + if not rolls: + typer.echo(str(Msg.NO_LOADED_ROLLS)) + raise typer.Exit(code=1) + + roll = _choose_roll(rolls) + metadata = load_roll_metadata(roll / "roll.toml") + updated = _prompt_roll_metadata(archive, roll / "roll.toml", metadata) + save_roll_metadata(roll / "roll.toml", updated) + typer.echo(f"{Msg.ROLL_EDIT_UPDATED} {updated.loaded_at}") + + +def edit_list_field( + prompt_title: str, dictionary_name: str, success_label: str +) -> None: + archive = require_archive(require_config()) + rolls = _rolls(archive) + if not rolls: + typer.echo(str(Msg.NO_ROLLS)) + raise typer.Exit(code=1) + + selected = _choose_roll_folder(rolls) + workspace = workspace_for(archive) + values = autocomplete_many_prompt( + prompt_title, workspace.dictionary(dictionary_name) + ) + try: + metadata = ( + update_roll_keywords(selected / "roll.toml", values) + if dictionary_name == "keywords" + else update_roll_features(selected / "roll.toml", values) + ) + if dictionary_name == "keywords": + apply_keyword_vocab_fixes(archive, metadata.keywords) + except ValueError as exc: + typer.echo(str(exc)) + raise typer.Exit(code=1) + + typer.echo(f"{success_label}: {metadata.film}") + + +def edit_batch( + year: str | None, + film: str | None, + camera: str | None, + status: str | None, + set_status: str | None, + set_camera: str | None, + add_feature: str | None, + add_tag: str | None, +) -> None: + config = require_config() + filters = {"year": year, "film": film, "camera": camera, "status": status} + changes = { + "set_status": set_status, + "set_camera": set_camera, + "add_feature": add_feature, + "add_tag": add_tag, + } + if not any(filters.values()) or not any(changes.values()): + typer.echo(str(Msg.BATCH_NEEDS_FILTERS)) + raise typer.Exit(code=1) + + batch_rolls( + config.archives, + year=year, + films=split_csv(film), + cameras=split_csv(camera), + statuses=split_csv(status), + status=set_status, + set_camera=set_camera, + add_features=split_csv(add_feature), + add_tags=split_csv(add_tag), + ) + + +def _choose_roll(rolls: list[Path]) -> Path: + labels = [_format_roll_label(path) for path in rolls] + selected_label = _prompt_choice(str(Msg.ROLL_EDIT_SELECT), labels) + label_to_roll = dict(zip(labels, rolls, strict=True)) + return label_to_roll[selected_label] + + +def _choose_roll_folder(rolls: list[Path]) -> Path: + labels = [roll_choice_label(path, _roll_status(path)) for path in rolls] + selected_label = _prompt_choice(str(Msg.ROLL_EDIT_SELECT), labels) + label_to_roll = dict(zip(labels, rolls, strict=True)) + return label_to_roll[selected_label] + + +def _rolls(archive: Path) -> list[Path]: + rolls: list[Path] = [] + for folder in find_roll_folders(archive): + try: + load_roll_metadata(folder / "roll.toml") + except ValueError: + continue + rolls.append(folder) + return rolls + + +def _roll_status(path: Path) -> str: + try: + return load_roll_metadata(path / "roll.toml").status + except ValueError: + return "unknown" + + +def _format_roll_label(path: Path) -> str: + metadata = load_roll_metadata(path / "roll.toml") + return str(Msg.ROLL_EDIT_ROLL_LABEL).format( + path=str(path.relative_to(path.parents[1])), + film=metadata.film, + camera=metadata.camera, + status=metadata.status, + ) + + +def _prompt_enum(label: Msg, values: list[str], current: str) -> str: + selected = _prompt_choice_panel(str(label), values, current) + return selected if selected is not None else current + + +def _prompt_roll_metadata( + archive: Path, roll_file: Path, metadata: RollMetadata +) -> RollMetadata: + workspace = workspace_for(archive) + film = _prompt_optional_autocomplete( + Msg.ROLL_EDIT_FILM, workspace.dictionary("films"), metadata.film + ) + camera = _prompt_optional_autocomplete( + Msg.ROLL_EDIT_CAMERA, workspace.dictionary("cameras"), metadata.camera + ) + status = _prompt_enum(Msg.ROLL_EDIT_STATUS, list(VALID_STATUSES), metadata.status) + features = _prompt_optional_many( + Msg.ROLL_EDIT_FEATURES, workspace.dictionary("features"), metadata.features + ) + keywords = _prompt_optional_many( + Msg.ROLL_EDIT_KEYWORDS, workspace.dictionary("keywords"), metadata.keywords + ) + original_source = _prompt_enum( + Msg.ROLL_EDIT_ORIGINAL_SOURCE, + ["negative", "slide", "print", "digital", "unknown"], + metadata.original_source, + ) + digital_copy = _prompt_enum( + Msg.ROLL_EDIT_DIGITAL_COPY, + ["scan", "photo", "none", "unknown"], + metadata.digital_copy, + ) + original_status = _prompt_enum( + Msg.ROLL_EDIT_ORIGINAL_STATUS, + ["present", "lost", "unknown"], + metadata.original_status, + ) + return RollMetadata( + status=status, + film=film, + camera=camera, + loaded_at=metadata.loaded_at, + features=features, + keywords=[value.upper() for value in keywords], + original_source=original_source, + digital_copy=digital_copy, + original_status=original_status, + ) + + +def _prompt_optional_autocomplete(label: Msg, dictionary, current: str) -> str: + value = prompt(f"{label} [{current}]: ", complete_while_typing=True).strip() + if not value: + return current + + choices = dictionary.read() + for existing in choices: + if existing.casefold() == value.casefold(): + return existing + + return dictionary.add(value) + + +def _prompt_optional_many(label: Msg, dictionary, current: list[str]) -> list[str]: + choices = dictionary.read() + value = prompt( + f"{label} [{', '.join(current)}]: ", + completer=FuzzyCompleter( + WordCompleter(choices, ignore_case=True, sentence=True, match_middle=True) + ), + complete_while_typing=True, + ).strip() + if not value: + return current + + selected: list[str] = [] + for token in [item.strip() for item in value.split(",") if item.strip()]: + for existing in choices: + if existing.casefold() == token.casefold(): + token = existing + break + else: + token = dictionary.add(token) + if token not in selected: + selected.append(token) + merged = list(current) + for token in selected: + if token not in merged: + merged.append(token) + return merged + + +def _prompt_choice(title: str, choices: list[str]) -> str: + selected = _prompt_choice_panel(title, choices, choices[0] if choices else "") + if selected is None: + raise ValueError(Msg.NO_CHOICE) + return selected + + +def _prompt_choice_panel( + title: str, choices: list[str], current: str | None = None +) -> str | None: + content_width = max( + len(title), + len(f"Current: {current}") if current is not None else 0, + *(len(f"{index + 1}. {choice}") for index, choice in enumerate(choices)), + 24, + ) + width = min(content_width + 4, 88) + border = "┌" + "─" * (width - 2) + "┐" + footer = "└" + "─" * (width - 2) + "┘" + prompt_lines = [border, f"│ {title.ljust(width - 4)} │"] + if current is not None: + prompt_lines.append( + f"│ {textwrap.shorten(f'Current: {current}', width=width - 4, placeholder='…').ljust(width - 4)} │" + ) + prompt_lines.append("├" + "─" * (width - 2) + "┤") + prompt_lines.extend( + f"│ {textwrap.shorten(f'{index + 1}. {choice}', width=width - 4, placeholder='…').ljust(width - 4)} │" + for index, choice in enumerate(choices) + ) + prompt_lines.append(footer) + echo_lines(prompt_lines) + + while True: + value = prompt(f"{Msg.ROLL_EDIT_SELECT_HINT}: ").strip() + if not value: + return current + if value.isdigit(): + index = int(value) - 1 + if 0 <= index < len(choices): + return choices[index] + if value in choices: + return value diff --git a/src/roll/app/flows/stock_lifecycle.py b/src/roll/app/flows/stock_lifecycle.py new file mode 100644 index 0000000..4565ef4 --- /dev/null +++ b/src/roll/app/flows/stock_lifecycle.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +from datetime import date +from pathlib import Path + +import typer +from prompt_toolkit import prompt +from prompt_toolkit.completion import FuzzyCompleter, WordCompleter + +from roll.app.archive.normalization import apply_keyword_vocab_fixes +from roll.app.workspace.roll_store import ( + RollMetadata, + load_roll_metadata, + save_roll_metadata, + update_roll_features, + update_roll_keywords, + update_roll_status, +) +from roll.app.workspace.stock_store import ( + StockItem, + add_to_stock, + load_stock, + remove_from_stock, + save_stock, +) +from roll.app.workspace.statuses import VALID_STATUSES +from roll.app.workspace.workspace import workspace_for +from roll.helpers.autocomplete import autocomplete_many_prompt, autocomplete_prompt +from roll.helpers.guards import require_archive, require_config +from roll.helpers.output import echo_lines +from roll.messages import Msg +from roll.app.flows.stock_edit import _choose_roll, _rolls + + +def add() -> None: + archive = require_archive(require_config()) + workspace = workspace_for(archive) + + film = autocomplete_prompt(str(Msg.PROMPT_FILM), workspace.dictionary("films")) + quantity = typer.prompt(str(Msg.PROMPT_QUANTITY), type=int) + if quantity <= 0: + typer.echo(str(Msg.INVALID_QUANTITY)) + raise typer.Exit(code=1) + + try: + items = load_stock(workspace.stock_file) + save_stock(workspace.stock_file, add_to_stock(items, film, quantity)) + except ValueError as exc: + typer.echo(str(exc)) + raise typer.Exit(code=1) + + +def load(manual: bool) -> None: + archive = require_archive(require_config()) + workspace = workspace_for(archive) + + stock: list[StockItem] = [] + if not manual: + try: + stock = load_stock(workspace.stock_file) + except ValueError as exc: + typer.echo(str(exc)) + raise typer.Exit(code=1) + + if not stock: + typer.echo(str(Msg.STOCK_EMPTY_MANUAL)) + raise typer.Exit(code=1) + + selected = ( + _choose_stock_item(stock) if not manual else _choose_manual_film(workspace) + ) + camera = autocomplete_prompt( + str(Msg.PROMPT_CAMERA), workspace.dictionary("cameras") + ) + loaded_at = _prompt_loaded_at() + roll_folder = _create_roll_folder(archive, loaded_at) + roll_file = roll_folder / "roll.toml" + + if roll_file.exists(): + typer.echo(f"{Msg.ROLL_EXISTS} {roll_file}") + raise typer.Exit(code=1) + + try: + roll_folder.mkdir(parents=True, exist_ok=True) + save_roll_metadata( + roll_file, + RollMetadata( + status=VALID_STATUSES[0], + film=selected.film, + camera=camera, + loaded_at=loaded_at, + features=[], + keywords=[], + ), + ) + if not manual: + save_stock(workspace.stock_file, remove_from_stock(stock, selected.film, 1)) + features = autocomplete_many_prompt( + str(Msg.VOCAB_FEATURES), workspace.dictionary("features") + ) + if features: + update_roll_features(roll_file, features) + + tags = autocomplete_many_prompt( + str(Msg.VOCAB_KEYWORDS), workspace.dictionary("keywords") + ) + if tags: + update_roll_keywords(roll_file, tags) + apply_keyword_vocab_fixes(archive, tags) + except Exception: + _cleanup_failed_load(roll_folder, roll_file) + raise + + typer.echo(f"{Msg.LOAD_SUCCESS} {selected.film}") + + +def process(status: str, label: str) -> None: + archive = require_archive(require_config()) + loaded_rolls = [ + path + for path in _rolls(archive) + if load_roll_metadata(path / "roll.toml").status == "loaded" + ] + if not loaded_rolls: + typer.echo(str(Msg.NO_LOADED_ROLLS)) + raise typer.Exit(code=1) + + selected = _choose_roll(loaded_rolls) + try: + metadata = update_roll_status(selected / "roll.toml", status) + except ValueError as exc: + typer.echo(str(exc)) + raise typer.Exit(code=1) + + typer.echo(f"{label}: {metadata.film}") + + +def list_stock() -> None: + archive = require_archive(require_config()) + workspace = workspace_for(archive) + + try: + items = load_stock(workspace.stock_file) + except ValueError as exc: + typer.echo(str(exc)) + raise typer.Exit(code=1) + + if not items: + typer.echo(str(Msg.STOCK_EMPTY)) + return + + echo_lines([Msg.STOCK_HEADER]) + for item in items: + typer.echo(f"{item.film:<20} ×{item.quantity}") + + +def _prompt_loaded_at() -> str: + value = typer.prompt(str(Msg.PROMPT_LOAD_DATE)) + normalized = value.strip().split("T", 1)[0].split(" ", 1)[0] + try: + return date.fromisoformat(normalized).isoformat() + except ValueError as exc: + typer.echo(str(Msg.INVALID_DATE)) + raise typer.Exit(code=1) from exc + + +def _create_roll_folder(archive: Path, loaded_at: str) -> Path: + loaded_date = date.fromisoformat(loaded_at) + return archive / loaded_date.strftime("%Y") / loaded_date.strftime("%m-%d") + + +def _choose_stock_item(items: list[StockItem]) -> StockItem: + labels = [f"{item.film} ×{item.quantity}" for item in items] + completer = FuzzyCompleter( + WordCompleter(labels, ignore_case=True, sentence=True, match_middle=True) + ) + + while True: + value = prompt( + str(Msg.PROMPT_MANUAL_FILM), + completer=completer, + complete_while_typing=True, + ).strip() + if not value: + continue + + selected = _resolve_stock_choice(items, value) + if selected is not None: + return selected + + typer.echo(str(Msg.CHOOSE_STOCK)) + + +def _choose_manual_film(workspace) -> StockItem: + film = autocomplete_prompt(str(Msg.PROMPT_FILM), workspace.dictionary("films")) + return StockItem(film=film, quantity=1) + + +def _resolve_stock_choice(items: list[StockItem], candidate: str) -> StockItem | None: + normalized = _normalize_choice(candidate) + + exact_film_matches = [ + item for item in items if item.film.casefold() == candidate.casefold() + ] + if len(exact_film_matches) == 1: + return exact_film_matches[0] + + exact_label_matches = [ + item + for item in items + if f"{item.film} ×{item.quantity}".casefold() == candidate.casefold() + ] + if len(exact_label_matches) == 1: + return exact_label_matches[0] + + fuzzy_matches = [ + item for item in items if normalized in _normalize_choice(item.film) + ] + if len(fuzzy_matches) == 1: + return fuzzy_matches[0] + + return None + + +def _normalize_choice(value: str) -> str: + return "".join(ch for ch in value.casefold() if ch.isalnum()) + + +def _cleanup_failed_load(roll_folder: Path, roll_file: Path) -> None: + if roll_file.exists(): + roll_file.unlink() + if roll_folder.exists(): + try: + roll_folder.rmdir() + except OSError: + pass diff --git a/src/roll/app/root_commands.py b/src/roll/app/root_commands.py new file mode 100644 index 0000000..4663900 --- /dev/null +++ b/src/roll/app/root_commands.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +from pathlib import Path +import subprocess +import sys + +import typer + +from roll.app.workspace.config import ( + CONFIG_DIR, + CONFIG_FILE, + Config, + load_config, + save_config, + set_lang, +) +from roll.helpers.formatting import highlight_cli_names +from roll.helpers.guards import require_config, require_directory +from roll.helpers.output import echo_lines, echo_section +from roll.messages import Msg +from roll.version import get_latest_version, get_version, is_outdated + + +def version() -> None: + current = get_version() + typer.echo(current) + latest = get_latest_version() + if latest and is_outdated(current=current, latest=latest): + typer.echo(f"New version available: {latest}. Run `rl update`.") + raise typer.Exit() + + +def init(archive: Path) -> None: + archive = require_directory(archive, Msg.ARCHIVE_MISSING) + + CONFIG_DIR.mkdir(parents=True, exist_ok=True) + try: + config = load_config() + archives = list(dict.fromkeys([*config.archives, archive])) + except FileNotFoundError: + archives = [archive] + save_config(Config(archives=archives)) + + from roll.app.workspace.workspace import workspace_for + + workspace_for(archive).ensure_structure() + + typer.echo(highlight_cli_names(Msg.CLI_INITIALIZED)) + echo_lines([f"Archive: {archive}", f"Config: {CONFIG_FILE}"]) + + +def update() -> None: + result = subprocess.run( + [ + sys.executable, + "-m", + "pip", + "install", + "--no-input", + "--upgrade", + "--force-reinstall", + "git+https://github.com/katrinio/roll.git@main", + ], + check=False, + ) + raise typer.Exit(code=result.returncode) + + +def config() -> None: + config = require_config() + echo_section( + Msg.CONFIG_HEADER, + [f"{Msg.ARCHIVE_HEADER} {archive}" for archive in config.archives], + ) + + +def config_lang(lang: str | None) -> None: + config = require_config() + + if lang is None: + typer.echo(f"{Msg.LANGUAGE} {config.lang}") + return + + normalized = lang.upper() + if normalized not in {"EN", "RU"}: + typer.echo(str(Msg.ALLOWED_VALUES)) + raise typer.Exit(code=1) + + updated = set_lang(normalized) + typer.echo(f"{Msg.LANGUAGE_SET_TO} {updated.lang}") diff --git a/src/roll/app/workspace/roll_store.py b/src/roll/app/workspace/roll_store.py index 4278f6c..c911034 100644 --- a/src/roll/app/workspace/roll_store.py +++ b/src/roll/app/workspace/roll_store.py @@ -101,28 +101,6 @@ def update_roll_features(path: Path, features: list[str]) -> RollMetadata: return updated -def update_roll_origin( - path: Path, - original_source: str, - digital_copy: str, - original_status: str, -) -> RollMetadata: - metadata = load_roll_metadata(path) - updated = RollMetadata( - status=metadata.status, - film=metadata.film, - camera=metadata.camera, - loaded_at=metadata.loaded_at, - features=metadata.features, - keywords=metadata.keywords, - original_source=original_source, - digital_copy=digital_copy, - original_status=original_status, - ) - save_roll_metadata(path, updated) - return updated - - def _load_toml(path: Path) -> dict: try: return tomllib.loads(path.read_text(encoding="utf-8")) diff --git a/src/roll/cli.py b/src/roll/cli.py index 6128630..f0af1d2 100644 --- a/src/roll/cli.py +++ b/src/roll/cli.py @@ -1,68 +1,34 @@ +from __future__ import annotations + from pathlib import Path -import subprocess -import sys import typer -from roll.filesystem import ( - build_archive_tree, - count_photo_files, - find_roll_folders, - find_unindexed_folders, -) -from roll.app.workspace.config import ( - CONFIG_DIR, - CONFIG_FILE, - Config, - load_config, - save_config, +from roll.app.archive.commands import ( + doctor as archive_doctor, + scan as archive_scan, + search as archive_search, + stats as archive_stats, + status as archive_status, + vocab as archive_vocab, ) from roll.app.archive.batch import process_archives -from roll.app.workspace.roll_store import ( - load_roll_metadata, - update_roll_features, - update_roll_keywords, -) -from roll.app.archive.normalization import ( - apply_normalization_plans, - apply_keyword_vocab_fixes, - build_normalization_plan, - normalize_keywords_in_archive, -) -from roll.app.archive.photo_dates import guess_archive_year -from roll.helpers.autocomplete import autocomplete_many_prompt, choice_prompt -from roll.helpers.formatting import highlight_cli_names -from roll.helpers.guards import ( - require_archive, - require_config, - require_current_archive, - require_directory, -) -from roll.helpers.output import echo_lines, echo_section +from roll.app.root_commands import config as root_config +from roll.app.root_commands import config_lang as root_config_lang +from roll.app.root_commands import init as root_init +from roll.app.root_commands import update as root_update +from roll.app.root_commands import version as root_version from roll.app.flows.stock import app as stock_app -from roll.app.flows.stock import load as load_roll -from roll.messages import Msg -from roll.app.archive.status_output import render_status_report -from roll.app.archive.search import find_rolls, search_rolls -from roll.app.archive.search_output import render_search_results -from roll.app.archive.normalization_output import render_normalization_plans -from roll.app.archive.stats import _count_statuses -from roll.app.archive.stats_output import render_stats_report -from roll.app.workspace.vocabulary import archive_vocabulary -from roll.app.workspace.workspace import workspace_for -from roll.app.diagnostics.doctor_output import render_doctor -from roll.app.workspace.config import set_lang -from roll.messages import Normalize -from roll.version import get_latest_version, get_version, is_outdated - -UPDATE_SOURCE = "git+https://github.com/katrinio/roll.git@main" +from roll.app.flows.stock import edit_batch, edit_list_field, load as load_roll +from roll.helpers.guards import require_config +from roll.messages import Msg, Normalize app = typer.Typer(help=Msg.CLI_INITIALIZED) app.add_typer(stock_app, name="stock") + config_app = typer.Typer(help=Msg.CONFIG_HEADER) app.add_typer(config_app, name="config") - tags_app = typer.Typer(help=Msg.VOCAB_KEYWORDS) app.add_typer(tags_app, name="tags") @@ -81,110 +47,42 @@ def main( ), ) -> None: if version: - current = get_version() - typer.echo(current) - latest = get_latest_version() - if latest and is_outdated(current=current, latest=latest): - typer.echo(f"New version available: {latest}. Run `rl update`.") - raise typer.Exit() + root_version() if ctx.invoked_subcommand is not None: return @app.command("init") def init(archive: Path = typer.Argument(..., help=Msg.ARCHIVE_HEADER)) -> None: - """Initialize the archive workspace.""" - archive = require_directory(archive, Msg.ARCHIVE_MISSING) - - CONFIG_DIR.mkdir(parents=True, exist_ok=True) - try: - config = load_config() - archives = list(dict.fromkeys([*config.archives, archive])) - except FileNotFoundError: - archives = [archive] - save_config(Config(archives=archives)) - - workspace = workspace_for(archive) - workspace.ensure_structure() - - typer.echo(highlight_cli_names(Msg.CLI_INITIALIZED)) - echo_lines([f"Archive: {archive}", f"Config: {CONFIG_FILE}"]) + root_init(archive) @app.command("update") def update() -> None: - """Update the installed package.""" - result = subprocess.run( - [sys.executable, "-m", "pip", "install", "-U", UPDATE_SOURCE] - ) - raise typer.Exit(code=result.returncode) + root_update() @config_app.callback(invoke_without_command=True) def config(ctx: typer.Context) -> None: - """Show current config.""" if ctx.invoked_subcommand is not None: return - config = require_config() - echo_section( - Msg.CONFIG_HEADER, - [f"{Msg.ARCHIVE_HEADER} {archive}" for archive in config.archives], - ) + root_config() @config_app.command("lang") def config_lang(lang: str | None = typer.Argument(None, help=Msg.LANGUAGE)) -> None: - """Show or set UI language.""" - config = require_config() - - if lang is None: - typer.echo(f"{Msg.LANGUAGE} {config.lang}") - return - - normalized = lang.upper() - if normalized not in {"EN", "RU"}: - typer.echo(str(Msg.ALLOWED_VALUES)) - raise typer.Exit(code=1) - - updated = set_lang(normalized) - typer.echo(f"{Msg.LANGUAGE_SET_TO} {updated.lang}") + root_config_lang(lang) @app.command("scan") def scan() -> None: - """Show archive folders.""" - archive = require_archive(require_config()) - - if not archive.exists(): - typer.echo(f"{Msg.ARCHIVE_MISSING} {archive}") - raise typer.Exit(code=1) - - echo_section(Msg.ARCHIVE_HEADER, [str(archive)]) - roll_folders = find_roll_folders(archive) - tree = build_archive_tree(archive) - photo_count = sum(count_photo_files(folder) for folder in roll_folders) - - if tree: - typer.echo(str(Msg.TREE_HEADER)) - echo_lines(tree) - typer.echo("") - - typer.echo(f"{Msg.FOLDERS} {len(roll_folders)}") - typer.echo(f"{Msg.FILES} {photo_count}") + archive_scan() @app.command("status") def status() -> None: - """Show index status.""" - archive = require_archive(require_config()) - - roll_folders = find_roll_folders(archive) - unindexed_folders = find_unindexed_folders(archive) - rolls = find_rolls(archive) - status_counts = _count_statuses(rolls) - - render_status_report(archive, roll_folders, unindexed_folders, status_counts) + archive_status() @app.command("stats") @@ -192,51 +90,30 @@ def stats( year: str | None = typer.Argument(None, help=Msg.STATS_YEAR), verbose: bool = typer.Option(False, "-v", "--verbose", help=Msg.STATS_MORE), ) -> None: - """Show archive statistics.""" - archive = require_archive(require_config()) - render_stats_report(archive, year, verbose) + archive_stats(year, verbose) @app.command("load") def load( manual: bool = typer.Option(False, "--manual", help=Msg.STOCK_EMPTY_MANUAL), ) -> None: - """Load a film from stock into a new roll.""" load_roll(manual=manual) @app.command("vocab") def vocab() -> None: - """Show dictionaries.""" - archive = require_archive(require_config()) - vocab = archive_vocabulary(archive) - - for title, items in ( - (Msg.VOCAB_FILMS, vocab["films"].read()), - (Msg.VOCAB_CAMERAS, vocab["cameras"].read()), - (Msg.VOCAB_FEATURES, vocab["features"].read()), - (Msg.VOCAB_KEYWORDS, vocab["keywords"].read()), - ): - echo_section(title, [f"- {item}" for item in items]) + archive_vocab() @app.command("search") def search( + year: str | None = typer.Option(None, "--year", help=Msg.BATCH_FILTER_YEAR), + film: str | None = typer.Option(None, "--film", help=Msg.BATCH_FILTER_FILM), + camera: str | None = typer.Option(None, "--camera", help=Msg.BATCH_FILTER_CAMERA), + status: str | None = typer.Option(None, "--status", help=Msg.BATCH_FILTER_STATUS), query: str | None = typer.Argument(None, help=Msg.SEARCH_QUERY_REQUIRED), ) -> None: - """Search rolls from memory.""" - if not query: - typer.echo(str(Msg.SEARCH_QUERY_REQUIRED)) - raise typer.Exit(code=1) - - archive = require_archive(require_config()) - results = search_rolls(archive, query) - - if not results: - typer.echo(str(Msg.NO_RESULTS)) - return - - render_search_results(results) + archive_search(year, film, camera, status, query) @app.command("doctor") @@ -244,60 +121,46 @@ def doctor( fix: bool = typer.Option(False, "--fix", help=Msg.DOCTOR_CAN_FIX), verbose: bool = typer.Option(False, "-v", "--verbose", help=Msg.DOCTOR_CAN_ADD), ) -> None: - """Check archive and config integrity.""" - if render_doctor(fix=fix, verbose=verbose): - raise typer.Exit(code=1) + archive_doctor(fix, verbose) @tags_app.command("add") def add_tags() -> None: - _update_roll_list_field("Tags", "keywords", update_roll_keywords, "Tags updated") + edit_list_field("Tags", "keywords", "Tags updated") @features_app.command("add") def add_features() -> None: - _update_roll_list_field( - "Features", "features", update_roll_features, "Features updated" - ) + edit_list_field("Features", "features", "Features updated") -def _update_roll_list_field( - prompt_title: str, - dictionary_name: str, - updater, - success_label: str, +@batch_app.callback(invoke_without_command=True) +def batch( + ctx: typer.Context, + year: str | None = typer.Option(None, "--year", help=Msg.BATCH_FILTER_YEAR), + film: str | None = typer.Option(None, "--film", help=Msg.BATCH_FILTER_FILM), + camera: str | None = typer.Option(None, "--camera", help=Msg.BATCH_FILTER_CAMERA), + status: str | None = typer.Option(None, "--status", help=Msg.BATCH_FILTER_STATUS), + set_status: str | None = typer.Option( + None, "--set-status", help=Msg.BATCH_SET_STATUS + ), + set_camera: str | None = typer.Option( + None, "--set-camera", help=Msg.BATCH_SET_CAMERA + ), + add_feature: str | None = typer.Option( + None, "--add-feature", help=Msg.BATCH_ADD_FEATURE + ), + add_tag: str | None = typer.Option(None, "--add-tag", help=Msg.BATCH_ADD_TAG), ) -> None: - archive = require_archive(require_config()) - rolls = [ - folder - for folder in find_roll_folders(archive) - if (folder / "roll.toml").exists() - ] - - if not rolls: - typer.echo(str(Msg.NO_ROLLS)) - raise typer.Exit(code=1) - - selected = _choose_roll_folder(rolls) - workspace = workspace_for(archive) - values = autocomplete_many_prompt( - prompt_title, workspace.dictionary(dictionary_name) - ) - try: - metadata = updater(selected / "roll.toml", values) - if dictionary_name == "keywords": - apply_keyword_vocab_fixes(archive, metadata.keywords) - except ValueError as exc: - typer.echo(str(exc)) - raise typer.Exit(code=1) - - typer.echo(f"{success_label}: {metadata.film}") + if ctx.invoked_subcommand is not None: + return + + edit_batch(year, film, camera, status, set_status, set_camera, add_feature, add_tag) @batch_app.command("process") def batch_process() -> None: - config = require_config() - process_archives(config.archives) + process_archives(require_config().archives) @app.command("normalize") @@ -305,173 +168,6 @@ def normalize( tags: bool = typer.Option(False, "--tags", help=Msg.TAGS_NORMALIZED), photos: bool = typer.Option(False, "--photos", help=Normalize.HEADER), ) -> None: - """Normalize archive layout.""" - config = require_config() - archive = require_current_archive(config) - - if tags: - touched = normalize_keywords_in_archive(archive) - if touched: - typer.echo(str(Msg.TAGS_NORMALIZED)) - for path in touched: - typer.echo(f" {path}") - else: - typer.echo(str(Msg.TAGS_ALREADY_NORMALIZED)) - return - - if photos: - plans = _build_photo_normalization_plans(archive) - total_rules, has_changes = render_normalization_plans(plans) - if not has_changes: - return - - all_conflicts = [conflict for plan in plans for conflict in plan.conflicts] - if all_conflicts: - raise typer.Exit(code=1) - - _echo_photo_plan_preview(plans) - if not typer.confirm( - str(Normalize.QUESTION).format(count=total_rules), default=False - ): - return - - apply_normalization_plans(plans) - return - - plans = [build_normalization_plan(archive)] - total_rules, has_changes = render_normalization_plans(plans) - if not has_changes: - return - - all_conflicts = [conflict for plan in plans for conflict in plan.conflicts] - if all_conflicts: - raise typer.Exit(code=1) - - if not typer.confirm( - str(Normalize.QUESTION).format(count=total_rules), default=False - ): - return + from roll.app.archive.normalize_commands import normalize as normalize_command - apply_normalization_plans(plans) - - -def _choose_roll_folder(rolls: list[Path]) -> Path: - labels = [ - f"{str(path.relative_to(path.parents[1]))} ({_roll_status(path)})" - for path in rolls - ] - selected_label = choice_prompt("Roll", labels) - for path in rolls: - label = f"{str(path.relative_to(path.parents[1]))} ({_roll_status(path)})" - if label == selected_label: - return path - raise ValueError(Msg.NO_CHOICE) - - -def _build_photo_normalization_plans(archive: Path): - folders = _photo_folders(archive) - year = guess_archive_year(archive) - if year is None: - typer.echo(str(Msg.CLI_UNINITIALIZED)) - raise typer.Exit(code=1) - - if not typer.confirm( - str(Msg.NORMALIZE_PHOTOS_CONFIRM_YEAR).format(year=year), default=True - ): - typed_year = typer.prompt( - str(Msg.NORMALIZE_PHOTOS_YEAR).format(folder=archive.name) - ) - year = _parse_year(typed_year) - - manual_months = len(folders) > 1 and typer.confirm( - str(Msg.NORMALIZE_PHOTOS_MANUAL), default=False - ) - return [ - _build_photo_plan_for_folder(folder, archive, year, manual_months) - for folder in folders - ] - - -def _build_photo_plan_for_folder( - folder: Path, archive: Path, year: int, manual_months: bool -): - from roll.app.archive.normalization import NormalizationPlan, RenameRule - - month = None - if manual_months: - month = _prompt_month(folder) - else: - month = _guess_month(folder) - - if month is None: - return NormalizationPlan(archive=archive, rules=[], conflicts=[]) - - target = archive / f"{year:04d}" / f"{month:02d}-01" - if target.exists(): - return NormalizationPlan( - archive=archive, - rules=[], - conflicts=[f"{Msg.NORMALIZE_PHOTOS_MONTH} {target}"], - ) - - return NormalizationPlan( - archive=archive, rules=[RenameRule(folder=folder, target=target)], conflicts=[] - ) - - -def _echo_photo_plan_preview(plans) -> None: - lines = [] - for plan in plans: - for rule in plan.rules: - lines.append( - f"{rule.folder.name} -> {rule.target.relative_to(plan.archive)}" - ) - - if lines: - typer.echo(str(Msg.NORMALIZE_PHOTOS_PREVIEW)) - for line in lines: - typer.echo(f" {line}") - - -def _photo_folders(archive: Path) -> list[Path]: - return [ - path for path in archive.iterdir() if path.is_dir() and path.name != ".roll" - ] - - -def _prompt_month(folder: Path) -> int: - while True: - value = typer.prompt( - str(Msg.NORMALIZE_PHOTOS_MONTH).format(folder=folder.name) - ).strip() - month = _parse_month(value) - if month is not None: - return month - - -def _guess_month(folder: Path) -> int | None: - from roll.app.archive.photo_dates import guess_archive_month - - guess = guess_archive_month(folder) - return guess.month if guess is not None else None - - -def _parse_year(value: str) -> int: - year = value.strip() - if len(year) == 4 and year.isdigit(): - return int(year) - raise typer.Exit(code=1) - - -def _parse_month(value: str) -> int | None: - month = value.strip() - if len(month) == 2 and month.isdigit() and 1 <= int(month) <= 12: - return int(month) - return None - - -def _roll_status(path: Path) -> str: - try: - return load_roll_metadata(path / "roll.toml").status - except ValueError: - return "unknown" + normalize_command(tags, photos) diff --git a/src/roll/messages/cli.py b/src/roll/messages/cli.py index 0c680a8..ffddc64 100644 --- a/src/roll/messages/cli.py +++ b/src/roll/messages/cli.py @@ -143,6 +143,11 @@ class Msg(Headers): "Состояние оригинала", "Original status", ) + ROLL_EDIT_SELECT_HINT = Message( + "cli.roll_edit_select_hint", + "Выбери номер или нажми Enter, чтобы оставить текущее значение.", + "Choose a number or press Enter to keep the current value.", + ) ROLL_EDIT_UPDATED = Message( "cli.roll_edit_updated", "Ролл обновлен.", "Roll updated." ) @@ -173,6 +178,11 @@ class Msg(Headers): "Нужно указать строку поиска. Пример: rl search pizza", "You need to provide a search query. Example: rl search pizza", ) + SEARCH_NEEDS_QUERY_OR_FILTERS = Message( + "cli.search_needs_query_or_filters", + "Нужно указать строку поиска или хотя бы один фильтр.", + "You need to provide a search query or at least one filter.", + ) TAGS_NORMALIZED = Message( "cli.tags_normalized", "Теги нормализованы.", "Tags normalized." ) @@ -217,9 +227,6 @@ class Msg(Headers): "'{value}' отсутствует в словаре.\n\nДобавить? [Y/n] ", "'{value}' is missing from the dictionary.\n\nAdd it? [Y/n] ", ) - BATCH_NO_LOADED = Message( - "cli.batch_no_loaded", "Нет loaded-роллов.", "No loaded rolls." - ) BATCH_WILL_PROCESS = Message( "cli.batch_will_process", "Будет обработано:", "Will process:" ) @@ -227,6 +234,52 @@ class Msg(Headers): "cli.batch_confirm", "Пометить все как processed?", "Mark all as processed?" ) BATCH_PROCESSED = Message("cli.batch_processed", "Обработано:", "Processed:") + BATCH_NEEDS_FILTERS = Message( + "cli.batch_needs_filters", + "Нужно указать хотя бы один фильтр и одно изменение.", + "You need at least one filter and one change.", + ) + BATCH_FILTER_YEAR = Message( + "cli.batch_filter_year", "Фильтр по году.", "Filter by year." + ) + BATCH_FILTER_FILM = Message( + "cli.batch_filter_film", + "Фильтр по пленкам через запятую.", + "Comma-separated film names.", + ) + BATCH_FILTER_CAMERA = Message( + "cli.batch_filter_camera", + "Фильтр по камерам через запятую.", + "Comma-separated camera names.", + ) + BATCH_FILTER_STATUS = Message( + "cli.batch_filter_status", + "Фильтр по статусам через запятую.", + "Comma-separated statuses.", + ) + BATCH_SET_STATUS = Message( + "cli.batch_set_status", + "Установить статус у выбранных роллов.", + "Set the status on selected rolls.", + ) + BATCH_SET_CAMERA = Message( + "cli.batch_set_camera", + "Установить камеру у выбранных роллов.", + "Set the camera on selected rolls.", + ) + BATCH_ADD_FEATURE = Message( + "cli.batch_add_feature", + "Добавить особенности через запятую.", + "Comma-separated features to append.", + ) + BATCH_ADD_TAG = Message( + "cli.batch_add_tag", + "Добавить теги через запятую.", + "Comma-separated tags to append.", + ) + BATCH_TITLE = Message( + "cli.batch_title", "Массовое редактирование роллов", "Batch edit rolls" + ) STATS_YEAR = Message("cli.stats_year", "Год:", "Year:") STATS_ROLLS = Message("cli.stats_rolls", "Роллов:", "Rolls:") STATS_FILMS = Message("cli.stats_films", "Пленок в статистике:", "Films in stats:") diff --git a/src/roll/version.py b/src/roll/version.py index a9c2e87..fba89af 100644 --- a/src/roll/version.py +++ b/src/roll/version.py @@ -21,6 +21,9 @@ def get_version() -> str: def get_latest_version() -> str: + remote_version = _remote_tag() + if remote_version: + return remote_version return _git_tag() @@ -49,6 +52,29 @@ def _git_tag() -> str: return result.stdout.strip().removeprefix("v") +def _remote_tag() -> str: + repo = _find_git_root(Path.cwd()) + if repo is None: + repo = Path(__file__).resolve().parents[2] + try: + result = subprocess.run( + ["git", "ls-remote", "--tags", "--refs", "origin", "v*"], + check=True, + capture_output=True, + text=True, + cwd=repo, + ) + except (OSError, subprocess.CalledProcessError): + return "" + + versions = [ + line.rsplit("/", 1)[-1].removeprefix("v") + for line in result.stdout.splitlines() + if line.strip() + ] + return _max_version(versions) + + def _find_git_root(path: Path) -> Path | None: for candidate in (path, *path.parents): if (candidate / ".git").exists(): @@ -61,3 +87,11 @@ def _version_tuple(value: str) -> tuple[int, ...]: if not match: return () return tuple(int(part) for part in match.group(1).split(".")) + + +def _max_version(values: list[str]) -> str: + parsed = [(_version_tuple(value), value) for value in values] + parsed = [item for item in parsed if item[0]] + if not parsed: + return "" + return max(parsed, key=lambda item: item[0])[1] diff --git a/tests/test_normalization.py b/tests/test_normalization.py index a4fe385..0695366 100644 --- a/tests/test_normalization.py +++ b/tests/test_normalization.py @@ -26,7 +26,7 @@ from roll.app.archive.search import RollIndex from roll.filesystem import build_archive_tree, count_photo_files from roll.app.workspace.roll_store import RollMetadata, save_roll_metadata -from roll.cli import _build_photo_normalization_plans +from roll.app.archive.normalize_commands import normalize as normalize_command from roll.helpers.output import echo_lines from roll.messages import Normalize @@ -213,12 +213,32 @@ def record_confirm(text: str, *args, **kwargs): return len(confirms) > 1 with ( - patch("roll.cli.guess_archive_year", return_value=2023), - patch("roll.cli._photo_folders", return_value=folders), - patch("roll.cli.typer.confirm", side_effect=record_confirm), - patch("roll.cli.typer.prompt", side_effect=record_prompt), + patch( + "roll.app.archive.normalize_cli.guess_archive_year", + return_value=2023, + ), + patch( + "roll.app.archive.normalize_cli._photo_folders", + return_value=folders, + ), + patch( + "roll.app.archive.normalize_cli.typer.confirm", + side_effect=record_confirm, + ), + patch( + "roll.app.archive.normalize_cli.typer.prompt", + side_effect=record_prompt, + ), + patch( + "roll.app.archive.normalize_commands.require_config", + return_value=Config(archives=[archive]), + ), + patch( + "roll.app.archive.normalize_commands.require_current_archive", + return_value=archive, + ), ): - _build_photo_normalization_plans(archive) + normalize_command(False, True) self.assertTrue(any("Year 2023 correct" in item for item in confirms)) self.assertTrue(any("Month for 4770 [01-12]" in item for item in prompts)) diff --git a/tests/test_storage.py b/tests/test_storage.py index fb642c5..feac51e 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -20,7 +20,10 @@ save_stock, ) from roll.app.archive.normalization import normalize_keywords_in_archive -from roll.app.flows.stock import _format_roll_label, _rolls, _prompt_roll_metadata +from roll.app.archive.search import search_rolls_by_filters +from roll.app.flows.stock import _format_roll_label, _rolls +from roll.app.flows.stock_edit import _prompt_roll_metadata +from roll.app.archive.batch import batch_rolls from roll.app.workspace.workspace import workspace_for @@ -223,6 +226,35 @@ def test_roll_label_is_short_and_stable(self) -> None: label, "2025/10-19 | Kodak Gold 200 | Pentax Espio 150SL | loaded" ) + def test_search_rolls_supports_filters_and_query(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) + roll = archive / "2025" / "10-19" + roll.mkdir(parents=True) + save_roll_metadata( + roll / "roll.toml", + RollMetadata( + status="loaded", + film="Kodak Gold 200", + camera="Pentax Espio 150SL", + loaded_at="2025-10-19", + features=["redscale"], + keywords=["BEACH"], + ), + ) + + results = search_rolls_by_filters( + [archive], + year="2025", + films=["Kodak Gold 200"], + cameras=["Pentax Espio 150SL"], + statuses=["loaded"], + query="beach", + ) + + self.assertEqual(len(results), 1) + self.assertEqual(results[0].folder, roll) + def test_roll_edit_prompt_can_update_all_metadata_fields(self) -> None: with tempfile.TemporaryDirectory() as tmp: archive = Path(tmp) @@ -267,7 +299,7 @@ def fake_prompt(*args, **kwargs): return next(prompts) with ( - patch("roll.app.flows.stock.prompt", side_effect=fake_prompt), + patch("roll.app.flows.stock_edit.prompt", side_effect=fake_prompt), ): updated = _prompt_roll_metadata( archive, archive / "2025/10-19/roll.toml", metadata @@ -281,3 +313,55 @@ def fake_prompt(*args, **kwargs): self.assertEqual(updated.original_source, "slide") self.assertEqual(updated.digital_copy, "photo") self.assertEqual(updated.original_status, "present") + + def test_batch_rolls_filters_by_year_and_film(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) + first = archive / "2025" / "10-19" + second = archive / "2025" / "10-20" + third = archive / "2024" / "09-01" + for folder, film in ( + (first, "Kodak Gold 200"), + (second, "Ilford HP5 Plus"), + (third, "Kodak Gold 200"), + ): + folder.mkdir(parents=True) + loaded_at = ( + "2025-10-19" + if folder == first + else "2025-10-20" + if folder == second + else "2024-09-01" + ) + save_roll_metadata( + folder / "roll.toml", + RollMetadata( + status="loaded", + film=film, + camera="Pentax Espio 150SL", + loaded_at=loaded_at, + features=[], + keywords=[], + ), + ) + + with patch("roll.app.archive.batch.typer.confirm", return_value=True): + changed = batch_rolls( + [archive], + year="2025", + films=["Kodak Gold 200"], + set_camera="Pentax K1000", + ) + + self.assertEqual(changed, 1) + self.assertEqual( + load_roll_metadata(first / "roll.toml").camera, "Pentax K1000" + ) + self.assertEqual( + load_roll_metadata(second / "roll.toml").camera, + "Pentax Espio 150SL", + ) + self.assertEqual( + load_roll_metadata(third / "roll.toml").camera, + "Pentax Espio 150SL", + )