106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
import argparse
|
|
import shutil
|
|
import sys
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
|
|
from copienator import (
|
|
EvaluationWorkspace,
|
|
ExitCode,
|
|
evaluation_parser,
|
|
execute,
|
|
workspace_from_args,
|
|
)
|
|
from copienator.configuration import IMPORT_DIR
|
|
|
|
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
|
|
|
|
|
|
def sync_annotated(
|
|
workspace: EvaluationWorkspace,
|
|
*,
|
|
annotation_dir_name: str,
|
|
import_dir: Path,
|
|
) -> ExitCode:
|
|
workspace.require_directories(annotation_dir_name)
|
|
annotation_dir = workspace.annotation_dir("refaire") if annotation_dir_name == "BRnot" else workspace.root / annotation_dir_name
|
|
session_id = workspace.refaire_session_id if annotation_dir_name == "BRnot" else None
|
|
prefix = f"{session_id}__" if session_id else ""
|
|
accepted = 0
|
|
annotated_dir = Path(import_dir).expanduser()
|
|
if not annotated_dir.is_dir():
|
|
print(f"Error: directory does not exist: {annotated_dir}", file=sys.stderr)
|
|
return ExitCode.INVALID_WORKSPACE
|
|
|
|
missing_targets = 0
|
|
annotated_files = sorted(
|
|
(
|
|
path
|
|
for path in annotated_dir.iterdir()
|
|
if path.is_file() and path.suffix.casefold() in {".pdf", ".jpg", ".jpeg"}
|
|
),
|
|
key=lambda path: path.name.casefold(),
|
|
)
|
|
for annotated_file in annotated_files:
|
|
if prefix and not annotated_file.stem.startswith(prefix):
|
|
print(f"Ignoring return from another pass: {annotated_file.name}")
|
|
continue
|
|
target_subdir = annotation_dir / annotated_file.stem.removeprefix(prefix)
|
|
|
|
if not target_subdir.is_dir():
|
|
print(f"Warning: directory not found: {target_subdir}", file=sys.stderr)
|
|
missing_targets += 1
|
|
else:
|
|
suffix = annotated_file.suffix.lower()
|
|
dest_file = target_subdir / f"Concat_annotated{suffix}"
|
|
print(f"Copying {annotated_file} to {dest_file}")
|
|
shutil.copy2(annotated_file, dest_file)
|
|
accepted += 1
|
|
if prefix and not accepted:
|
|
print(f"No returns for the active pass {session_id} were imported.")
|
|
return ExitCode.PARTIAL
|
|
return ExitCode.PARTIAL if missing_targets else ExitCode.SUCCESS
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = evaluation_parser("Import handwritten annotations from the tablet directory.")
|
|
parser.add_argument(
|
|
"annotation_dir",
|
|
nargs="?",
|
|
choices=ANNOTATION_DIRECTORIES,
|
|
default="BGnot",
|
|
help="Annotation directory receiving imported PDFs (default: BGnot)",
|
|
)
|
|
parser.add_argument("--refaire", action="store_true", help="Process only copies/labels defined in refaire.json")
|
|
return parser
|
|
|
|
|
|
def run(
|
|
workspace: EvaluationWorkspace,
|
|
*,
|
|
annotation_dir: str = "BGnot",
|
|
refaire: bool = False,
|
|
) -> ExitCode:
|
|
return sync_annotated(
|
|
workspace,
|
|
annotation_dir_name="BRnot" if refaire else annotation_dir,
|
|
import_dir=Path(IMPORT_DIR),
|
|
)
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
return execute(
|
|
parser,
|
|
argv,
|
|
lambda args: run(
|
|
workspace_from_args(args),
|
|
annotation_dir=args.annotation_dir,
|
|
refaire=args.refaire,
|
|
),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|