diff --git a/misc/pdf_text_diff.py b/misc/pdf_text_diff.py index a4c1e752..b99af9c1 100644 --- a/misc/pdf_text_diff.py +++ b/misc/pdf_text_diff.py @@ -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() diff --git a/misc/watcher.py b/misc/watcher.py index 65f8df72..cd62e880 100644 --- a/misc/watcher.py +++ b/misc/watcher.py @@ -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) diff --git a/pyproject.toml b/pyproject.toml index f40dcf9c..1948d469 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,7 +53,7 @@ Changelog = "https://github.com/ocrmypdf/OCRmyPDF/docs/release_notes.md" [project.optional-dependencies] # User-installable features - use `uv sync --extra ` 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]