Replace typer with cyclopts CLI library in misc scripts

Migrate watcher.py and pdf_text_diff.py from typer to cyclopts for
CLI argument parsing. Update pyproject.toml to reflect the dependency
change in the watcher optional feature.
This commit is contained in:
James R. Barlow
2026-01-13 00:43:14 -08:00
parent bf76c8270c
commit 4c7086c609
3 changed files with 65 additions and 69 deletions
+29 -17
View File
@@ -5,33 +5,45 @@
from __future__ import annotations
from pathlib import Path
from subprocess import run
from tempfile import NamedTemporaryFile
from typing import Annotated
import typer
import cyclopts
app = cyclopts.App()
@app.default
def main(
pdf1: Annotated[typer.FileBinaryRead, typer.Argument()],
pdf2: Annotated[typer.FileBinaryRead, typer.Argument()],
engine: Annotated[str, typer.Option()] = 'pdftotext',
pdf1: Annotated[Path, cyclopts.Parameter()],
pdf2: Annotated[Path, cyclopts.Parameter()],
*,
engine: Annotated[str, cyclopts.Parameter()] = 'pdftotext',
):
"""Compare text in PDFs."""
text1 = run(
['pdftotext', '-layout', '-', '-'], stdin=pdf1, capture_output=True, check=True
)
text2 = run(
['pdftotext', '-layout', '-', '-'], stdin=pdf2, capture_output=True, check=True
)
with open(pdf1, 'rb') as f1, open(pdf2, 'rb') as f2:
text1 = run(
['pdftotext', '-layout', '-', '-'],
stdin=f1,
capture_output=True,
check=True,
)
text2 = run(
['pdftotext', '-layout', '-', '-'],
stdin=f2,
capture_output=True,
check=True,
)
with NamedTemporaryFile() as f1, NamedTemporaryFile() as f2:
f1.write(text1.stdout)
f1.flush()
f2.write(text2.stdout)
f2.flush()
with NamedTemporaryFile() as t1, NamedTemporaryFile() as t2:
t1.write(text1.stdout)
t1.flush()
t2.write(text2.stdout)
t2.flush()
diff = run(
['diff', '--color=always', '--side-by-side', f1.name, f2.name],
['diff', '--color=always', '--side-by-side', t1.name, t2.name],
capture_output=True,
)
run(['less', '-R'], input=diff.stdout, check=True)
@@ -42,4 +54,4 @@ def main(
if __name__ == '__main__':
typer.run(main)
app()
+35 -51
View File
@@ -17,8 +17,8 @@ from enum import Enum
from pathlib import Path
from typing import Annotated, Any
import cyclopts
import pikepdf
import typer
from dotenv import load_dotenv
from watchdog.events import PatternMatchingEventHandler
from watchdog.observers import Observer
@@ -30,7 +30,7 @@ load_dotenv()
# pylint: disable=logging-format-interpolation
app = typer.Typer(name="ocrmypdf-watcher")
app = cyclopts.App(name="ocrmypdf-watcher")
log = logging.getLogger('ocrmypdf-watcher')
@@ -153,110 +153,94 @@ class HandleObserverEvent(PatternMatchingEventHandler):
execute_ocrmypdf(file_path=Path(event.src_path), **self._settings)
@app.command()
@app.default
def main(
input_dir: Annotated[
Path,
typer.Argument(
envvar='OCR_INPUT_DIRECTORY',
exists=True,
file_okay=False,
dir_okay=True,
readable=True,
resolve_path=True,
cyclopts.Parameter(
env_var='OCR_INPUT_DIRECTORY',
),
] = '/input',
] = Path('/input'),
output_dir: Annotated[
Path,
typer.Argument(
envvar='OCR_OUTPUT_DIRECTORY',
exists=True,
file_okay=False,
dir_okay=True,
writable=True,
resolve_path=True,
cyclopts.Parameter(
env_var='OCR_OUTPUT_DIRECTORY',
),
] = '/output',
] = Path('/output'),
archive_dir: Annotated[
Path,
typer.Argument(
envvar='OCR_ARCHIVE_DIRECTORY',
exists=True,
file_okay=False,
dir_okay=True,
writable=True,
resolve_path=True,
cyclopts.Parameter(
env_var='OCR_ARCHIVE_DIRECTORY',
),
] = '/processed',
] = Path('/processed'),
*,
output_dir_year_month: Annotated[
bool,
typer.Option(
envvar='OCR_OUTPUT_DIRECTORY_YEAR_MONTH',
cyclopts.Parameter(
env_var='OCR_OUTPUT_DIRECTORY_YEAR_MONTH',
help='Create a subdirectory in the output directory for each year/month',
),
] = False,
on_success_delete: Annotated[
bool,
typer.Option(
envvar='OCR_ON_SUCCESS_DELETE',
cyclopts.Parameter(
env_var='OCR_ON_SUCCESS_DELETE',
help='Delete the input file after successful OCR',
),
] = False,
on_success_archive: Annotated[
bool,
typer.Option(
envvar='OCR_ON_SUCCESS_ARCHIVE',
cyclopts.Parameter(
env_var='OCR_ON_SUCCESS_ARCHIVE',
help='Archive the input file after successful OCR',
),
] = False,
deskew: Annotated[
bool,
typer.Option(
envvar='OCR_DESKEW',
cyclopts.Parameter(
env_var='OCR_DESKEW',
help='Deskew the input file before OCR',
),
] = False,
ocr_json_settings: Annotated[
str,
typer.Option(
envvar='OCR_JSON_SETTINGS',
str | None,
cyclopts.Parameter(
env_var='OCR_JSON_SETTINGS',
help='JSON settings to pass to OCRmyPDF (JSON string or file path)',
),
] = None,
poll_new_file_seconds: Annotated[
int,
typer.Option(
envvar='OCR_POLL_NEW_FILE_SECONDS',
cyclopts.Parameter(
env_var='OCR_POLL_NEW_FILE_SECONDS',
help='Seconds to wait before polling a new file',
min=0,
),
] = 1,
use_polling: Annotated[
bool,
typer.Option(
envvar='OCR_USE_POLLING',
cyclopts.Parameter(
env_var='OCR_USE_POLLING',
help='Use polling instead of filesystem events',
),
] = False,
retries_loading_file: Annotated[
int,
typer.Option(
envvar='OCR_RETRIES_LOADING_FILE',
cyclopts.Parameter(
env_var='OCR_RETRIES_LOADING_FILE',
help='Number of times to retry loading a file before giving up',
min=0,
),
] = 5,
loglevel: Annotated[
LoggingLevelEnum,
typer.Option(
envvar='OCR_LOGLEVEL',
cyclopts.Parameter(
env_var='OCR_LOGLEVEL',
help='Logging level',
),
] = LoggingLevelEnum.INFO,
patterns: Annotated[
str,
typer.Option(
envvar='OCR_PATTERNS',
cyclopts.Parameter(
env_var='OCR_PATTERNS',
help='File patterns to watch',
),
] = '*.pdf,*.PDF',
@@ -322,7 +306,7 @@ def main(
observer = Observer()
observer.schedule(handler, input_dir, recursive=True)
observer.start()
typer.echo(f"Watching {input_dir} for new PDFs. Press Ctrl+C to exit.")
print(f"Watching {input_dir} for new PDFs. Press Ctrl+C to exit.")
try:
while True:
time.sleep(30)
+1 -1
View File
@@ -53,7 +53,7 @@ Changelog = "https://github.com/ocrmypdf/OCRmyPDF/docs/release_notes.md"
[project.optional-dependencies]
# User-installable features - use `uv sync --extra <name>` or `pip install ocrmypdf[name]`
watcher = ["watchdog>=1.0.2", "typer-slim[standard]", "python-dotenv"]
watcher = ["watchdog>=1.0.2", "cyclopts>=3", "python-dotenv"]
webservice = ["streamlit>=1.41.0"]
[project.scripts]