81 lines
3.2 KiB
Python
81 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from copienator import (
|
|
EvaluationWorkspace,
|
|
ExitCode,
|
|
evaluation_parser,
|
|
execute,
|
|
read_json,
|
|
workspace_from_args,
|
|
)
|
|
from copienator.annotation_modifications import (
|
|
APPLIED,
|
|
DRAFT,
|
|
MODES,
|
|
baseline,
|
|
output_removals,
|
|
read_instructions,
|
|
rebuild,
|
|
validate_instructions,
|
|
)
|
|
from copienator.filesystem import publish_file_sets
|
|
|
|
|
|
def run(workspace: EvaluationWorkspace, *, annotation_dir: str = "BGnot", incremental: bool = False) -> ExitCode:
|
|
workspace.require_files("labels", DRAFT)
|
|
workspace.require_directories("Copies", annotation_dir)
|
|
text = (workspace.root / DRAFT).read_text(encoding="utf-8")
|
|
instructions = read_instructions(text, annotation_dir)
|
|
directory = workspace.root / annotation_dir
|
|
applied = directory / APPLIED
|
|
previous = read_instructions(applied.read_text(encoding="utf-8"), annotation_dir) if applied.exists() else []
|
|
if incremental and previous == instructions:
|
|
print("Skipping unchanged transfer instructions")
|
|
return ExitCode.SUCCESS
|
|
validate_instructions(workspace.root, annotation_dir, instructions)
|
|
copy_ids = sorted({item.copy_id for item in [*previous, *instructions]})
|
|
if incremental:
|
|
copy_ids = [cid for cid in copy_ids
|
|
if [i for i in previous if i.copy_id == cid] != [i for i in instructions if i.copy_id == cid]]
|
|
# Prepare every copy before publishing anything, including the applied file.
|
|
with TemporaryDirectory(prefix=".annotation-edits-", dir=directory) as temporary:
|
|
staging_root = Path(temporary)
|
|
file_sets = {}
|
|
for copy_id in copy_ids:
|
|
output = directory / f"Copie{copy_id}"
|
|
meta = baseline(output)
|
|
staging = staging_root / output.name
|
|
staging.mkdir()
|
|
selected = [item for item in instructions if item.copy_id == copy_id]
|
|
rebuild(workspace.root, annotation_dir, output, staging, meta, selected,
|
|
read_json(output / "score.json"))
|
|
file_sets[output] = (staging, output_removals(output))
|
|
manifest = staging_root / "manifest"
|
|
manifest.mkdir()
|
|
(manifest / APPLIED).write_text(text, encoding="utf-8")
|
|
file_sets[directory] = (manifest, ())
|
|
publish_file_sets(file_sets)
|
|
print(f"Modifications appliquées à {len(copy_ids)} copie(s). Les scores sont conservés.")
|
|
print("Relancez « Attribuer les noms et préparer A Rendre » pour actualiser tous les livrables.")
|
|
return ExitCode.SUCCESS
|
|
|
|
|
|
def build_parser():
|
|
parser = evaluation_parser("Modify reviewed answer bodies using annotation_modifications.txt")
|
|
parser.add_argument("--annotation-dir", choices=MODES, default="BGnot")
|
|
parser.add_argument("--incremental", action="store_true", help="Skip unchanged transfer instructions")
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
return execute(build_parser(), argv, lambda args: run(
|
|
workspace_from_args(args), annotation_dir=args.annotation_dir, incremental=args.incremental))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|