131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
import argparse
|
||
import shutil
|
||
import sys
|
||
from collections.abc import Sequence
|
||
from pathlib import Path
|
||
|
||
from copienator import (
|
||
CliError,
|
||
EvaluationWorkspace,
|
||
ExitCode,
|
||
evaluation_parser,
|
||
execute,
|
||
workspace_from_args,
|
||
)
|
||
from copienator import configuration
|
||
|
||
ANNOTATION_DIRECTORIES = ("BGnot", "Bnot", "Anot")
|
||
IMPORT_DIR = getattr(configuration, "IMPORT_DIR", None)
|
||
|
||
|
||
def clear_import_directory(workspace: EvaluationWorkspace) -> None:
|
||
"""Empty the configured import folder while preserving the folder itself."""
|
||
if IMPORT_DIR is None or not str(IMPORT_DIR).strip():
|
||
print("Aucun dossier d’import configuré ; aucun fichier supprimé.")
|
||
return
|
||
directory = Path(IMPORT_DIR).expanduser().resolve()
|
||
if workspace.root.resolve().is_relative_to(directory) or Path.cwd().resolve().is_relative_to(directory):
|
||
raise CliError(f"Le dossier d’import ne peut pas contenir l’évaluation ou le projet : {directory}",
|
||
ExitCode.INVALID_ARGUMENTS)
|
||
if not directory.exists():
|
||
print(f"Dossier d’import absent ; aucun fichier supprimé : {directory}")
|
||
return
|
||
if not directory.is_dir():
|
||
raise CliError(f"Le dossier d’import n’est pas un dossier : {directory}",
|
||
ExitCode.INVALID_ARGUMENTS)
|
||
for child in directory.iterdir():
|
||
if child.is_dir() and not child.is_symlink():
|
||
shutil.rmtree(child)
|
||
else:
|
||
child.unlink()
|
||
print(f"Dossier d’import vidé : {directory}", flush=True)
|
||
|
||
|
||
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())
|