from __future__ import annotations import argparse import re import sys from collections import defaultdict from collections.abc import Sequence from pathlib import Path from copienator import ( EvaluationWorkspace, ExitCode, configuration, evaluation_parser, execute, read_json, workspace_from_args, ) from copienator.platform import replace_with_link_or_copy, safe_filename from copienator.return_answers import publish_answer_returns ANNOTATION_CHOICES = ("BGnot", "Bnot", "Anot") def build_parser() -> argparse.ArgumentParser: parser = evaluation_parser("Assign student names and prepare the return directory.") parser.add_argument( "annotation_dir", choices=ANNOTATION_CHOICES, help="Annotation directory to use", ) parser.add_argument( "--update", action="store_true", help=( "Update only the individual images in existing A Rendre/answers " "directories, matching folders by their trailing copy ID" ), ) return parser RETURN_COPY_ID = re.compile(r"\((\d+)\)$") def _read_expected_names(workspace: EvaluationWorkspace) -> set[str]: names_path = workspace.names_file() if not names_path.exists(): print( f"Warning: names file not found in {workspace.root} or the current directory.", file=sys.stderr, ) return set() return { line.strip() for line in names_path.read_text(encoding="utf-8").splitlines() if line.strip() } def _annotation_source( workspace: EvaluationWorkspace, annotation_dir_name: str, copy_id: str, ) -> Path | None: selected = workspace.root / annotation_dir_name / f"Copie{copy_id}" fallback = workspace.annotation_dir("simple") / f"Copie{copy_id}" for candidate in (selected, fallback): if (candidate / "score.json").is_file() and ( (candidate / "Concat.jpg").is_file() or (candidate / "info.json").is_file() ): return candidate return None def update_named_return_answers( workspace: EvaluationWorkspace, annotation_dir_name: str, ) -> ExitCode: """Refresh only answers/ in existing returns, preserving manual names.""" workspace.require_directories(annotation_dir_name, "A Rendre") had_errors = False found = False for destination in sorted(workspace.return_dir.iterdir()): if not destination.is_dir(): continue match = RETURN_COPY_ID.search(destination.name) if match is None: print( f"Warning: cannot identify a copy ID in {destination.name!r}; skipped", file=sys.stderr, ) had_errors = True continue found = True copy_id = match.group(1) source_folder = _annotation_source( workspace, annotation_dir_name, copy_id ) if source_folder is None: print( f"Warning: no annotation source found for Copie{copy_id}; skipped", file=sys.stderr, ) had_errors = True continue try: publish_answer_returns( workspace.root, source_folder, destination, answers_only=True, ) print(f"Updated answers for {destination.name} from Copie{copy_id}") except (OSError, TypeError, ValueError) as exc: print(f"Error updating answers for {destination.name}: {exc}", file=sys.stderr) had_errors = True if not found: print("Warning: no identifiable student folders found in A Rendre", file=sys.stderr) had_errors = True return ExitCode.PARTIAL if had_errors else ExitCode.SUCCESS def prepare_named_returns( workspace: EvaluationWorkspace, annotation_dir_name: str, ) -> ExitCode: workspace.require_directories("Copies", annotation_dir_name) workspace.return_dir.mkdir(parents=True, exist_ok=True) expected_names = _read_expected_names(workspace) copies_map: defaultdict[str, list[str]] = defaultdict(list) pattern = re.compile(r"^Copie(\d+)\.json$") had_errors = False for json_path in workspace.copies_dir.iterdir(): match = pattern.match(json_path.name) if not match: continue try: data = read_json(json_path) if not isinstance(data, dict): raise TypeError("expected a JSON object") name = str(data.get("name", "Unknown")).strip() copies_map[name].append(match.group(1)) except (OSError, TypeError, ValueError) as exc: print(f"Error processing {json_path}: {exc}", file=sys.stderr) had_errors = True assigned_names: set[str] = set() for name, copy_ids in copies_map.items(): if name == "Unknown": print( f"Warning: unknown name for copies: {', '.join(copy_ids)}", file=sys.stderr, ) elif len(copy_ids) > 1: print( f"Warning: name {name!r} is assigned to multiple copies: " f"{', '.join(copy_ids)}", file=sys.stderr, ) safe_name = safe_filename(name) for copy_id in copy_ids: source_folder = _annotation_source( workspace, annotation_dir_name, copy_id ) if source_folder is None: continue assigned_names.add(name) destination = workspace.return_dir / f"{safe_name} ({copy_id})" destination.mkdir(parents=True, exist_ok=True) try: publish_answer_returns(workspace.root, source_folder, destination) except (OSError, TypeError, ValueError) as exc: print(f"Error preparing answers for {destination.name}: {exc}", file=sys.stderr) had_errors = True continue links = ( ("Concat.jpg", f"{safe_name}.jpg", configuration.RETURN_JPEG_ENABLED), ("Concat_F.pdf", f"{safe_name}.pdf", configuration.RETURN_PDF_ENABLED), ("score.json", "score.json", True), ) for source_name, destination_name, enabled in links: source = source_folder / source_name target = destination / destination_name try: if not enabled: # Remove only the named return entry, never its link target. target.unlink(missing_ok=True) continue if not source.exists(): target.unlink(missing_ok=True) continue method = replace_with_link_or_copy( source, target, prefer="symlink", ) if method == "copy": print( f"Copied {source_name} for {destination.name} " "(links unavailable)" ) except OSError as exc: print( f"Error linking {source} for {destination.name}: {exc}", file=sys.stderr, ) had_errors = True unassigned = expected_names - assigned_names if unassigned: print("Names from the list that were not assigned:", file=sys.stderr) for name in sorted(unassigned): print(f" - {name}", file=sys.stderr) return ExitCode.PARTIAL if had_errors else ExitCode.SUCCESS def run( workspace: EvaluationWorkspace, *, annotation_dir: str, update: bool = False, ) -> ExitCode: if update: return update_named_return_answers(workspace, annotation_dir) return prepare_named_returns(workspace, annotation_dir) def main(argv: Sequence[str] | None = None) -> int: parser = build_parser() return execute( parser, argv, lambda args: run( workspace_from_args(args, repository=Path.cwd()), annotation_dir=args.annotation_dir, update=args.update, ), ) if __name__ == "__main__": raise SystemExit(main())