190 lines
10 KiB
Python
190 lines
10 KiB
Python
"""Shared, file-compatible services for the optional finalisation browser."""
|
||
from __future__ import annotations
|
||
|
||
from hashlib import sha256
|
||
from pathlib import Path
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
|
||
from copienator import atomic_write_json, read_json, utils
|
||
from copienator import custom_remarks, general_remarks
|
||
from copienator.annotation_modifications import DRAFT, APPLIED, BASELINE, read_instructions, validate_instructions
|
||
from copienator.return_identity import return_folders, RETURN_FOLDER_PATTERN, set_copy_name
|
||
from copienator.score_editing import set_score
|
||
|
||
|
||
class Finalisation:
|
||
def __init__(self, root: Path, mode: str, token: str, lock_fd: int | None = None):
|
||
self.root = root.resolve()
|
||
if mode not in ('Bnot', 'BGnot'):
|
||
raise ValueError('Choisissez Bnot ou BGnot.')
|
||
self.mode, self.token = mode, token
|
||
self.lock_fd = lock_fd
|
||
self.labels = utils.read_all_labels(self.root)
|
||
self.mutex = threading.RLock()
|
||
self.closing = False
|
||
self.job = {'running': False, 'status': 'idle', 'log': ''}
|
||
self.dirty_path = self.root / '.copienator' / f'finalisation-{mode}.json'
|
||
|
||
def copies(self):
|
||
return sorted((p for p in (self.root / self.mode).glob('Copie*')
|
||
if p.is_dir() and p.name[5:].isdigit() and (p / 'score.json').is_file()),
|
||
key=lambda p: int(p.name[5:]))
|
||
|
||
def directory(self, copy: str):
|
||
matches = [p for p in self.copies() if copy.isdecimal() and int(p.name[5:]) == int(copy)]
|
||
if len(matches) != 1:
|
||
raise ValueError('Copie introuvable ou ambiguë.')
|
||
return matches[0]
|
||
|
||
def label(self, label: str):
|
||
if label not in self.labels or Path(label).name != label or label in ('.', '..'):
|
||
raise ValueError('Label inconnu.')
|
||
return label
|
||
|
||
def draft(self):
|
||
path = self.root / DRAFT
|
||
return path.read_text() if path.exists() else f'### Source: {self.mode}\n'
|
||
|
||
def revision(self):
|
||
paths = [self.root / DRAFT, self.root / 'custom_remarks.json', self.root / 'general_remarks.json',
|
||
self.root / self.mode / APPLIED, self.dirty_path]
|
||
paths += list((self.root / 'Copies').glob('Copie*.json'))
|
||
paths += [p / 'score.json' for p in self.copies()]
|
||
digest = sha256()
|
||
for p in sorted(paths):
|
||
digest.update(str(p).encode())
|
||
digest.update(p.read_bytes() if p.exists() else b'absent')
|
||
for key, folder in sorted(return_folders(self.root).items()):
|
||
digest.update(f'{key}:{folder.name}'.encode())
|
||
return digest.hexdigest()
|
||
|
||
def state(self):
|
||
with self.mutex:
|
||
folders = return_folders(self.root)
|
||
copies = []
|
||
for directory in self.copies():
|
||
cid = directory.name[5:]
|
||
metadata = read_json(self.root / 'Copies' / f'Copie{cid}.json', default={})
|
||
folder = folders.get(int(cid))
|
||
name = RETURN_FOLDER_PATTERN.fullmatch(folder.name)['name'] if folder else metadata.get('name', 'Unknown')
|
||
scores = read_json(directory / 'score.json')
|
||
info = read_json(directory / 'info.json', default={})
|
||
base = read_json(directory / BASELINE, default={})
|
||
copies.append({'id': cid, 'name': name, 'general': general_remarks.remarks_for_copy(self.root, cid),
|
||
'answers': [{'label': label, 'score': str(scores.get(label, '')),
|
||
'remark': custom_remarks.remarks_for_copy(self.root, cid).get(label, ''),
|
||
'image': (directory / f'{label}.jpg').is_file(),
|
||
'editableBody': label in base.get('images', {}),
|
||
'info': info.get(label, {})} for label in self.labels]})
|
||
applied = self.root / self.mode / APPLIED
|
||
applied_text = applied.read_text() if applied.exists() else f'### Source: {self.mode}\n'
|
||
return {'evaluation': self.root.name, 'mode': self.mode, 'labels': self.labels, 'copies': copies,
|
||
'draft': self.draft(), 'applied': applied_text, 'revision': self.revision(),
|
||
'dirty': read_json(self.dirty_path, default={}).get('dirty', False), 'job': dict(self.job)}
|
||
|
||
def save(self, data):
|
||
with self.mutex:
|
||
if self.job['running'] or self.closing:
|
||
raise ValueError('Régénération en cours ; attendez la fin avant de modifier.')
|
||
if data['revision'] != self.revision():
|
||
raise ValueError('Les données ont changé. Rechargez la page avant de réessayer.')
|
||
kind, value = data['kind'], data['value']
|
||
if kind == 'draft':
|
||
instructions = read_instructions(value, self.mode)
|
||
validate_instructions(self.root, self.mode, instructions)
|
||
# Same filesystem and atomic replacement; retain the GUI-readable format.
|
||
from copienator.filesystem import staged_files
|
||
with staged_files(self.root) as staging:
|
||
(staging / DRAFT).write_text(value, encoding='utf-8')
|
||
else:
|
||
directory = self.directory(data['copy'])
|
||
cid = directory.name[5:]
|
||
if kind == 'score':
|
||
set_score(directory / 'score.json', self.label(data['label']), value)
|
||
elif kind == 'remark':
|
||
custom_remarks.set_remark(self.root, cid, self.label(data['label']), value)
|
||
elif kind == 'general':
|
||
if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
|
||
raise ValueError('Liste de remarques invalide.')
|
||
general_remarks.set_remarks(self.root, cid, value)
|
||
elif kind == 'name':
|
||
set_copy_name(self.root, cid, value)
|
||
else:
|
||
raise ValueError('Modification inconnue.')
|
||
self.dirty_path.parent.mkdir(exist_ok=True)
|
||
atomic_write_json(self.dirty_path, {'dirty': True})
|
||
return self.state()
|
||
|
||
def start_job(self, revision: str, apply_draft: bool):
|
||
with self.mutex:
|
||
if self.job['running'] or self.closing or revision != self.revision():
|
||
raise ValueError('Traitement déjà en cours ou données modifiées. Rechargez la page.')
|
||
if apply_draft:
|
||
validate_instructions(self.root, self.mode, read_instructions(self.draft(), self.mode))
|
||
if not (self.root / DRAFT).exists():
|
||
(self.root / DRAFT).write_text(self.draft())
|
||
self.job = {'running': True, 'status': 'running', 'log': ''}
|
||
threading.Thread(target=self._run_job, args=(apply_draft,), daemon=True).start()
|
||
return dict(self.job)
|
||
|
||
def _run_job(self, apply_draft):
|
||
try:
|
||
self._regenerate(apply_draft)
|
||
except Exception as exc:
|
||
with self.mutex:
|
||
self.job.update(status='failed', log=self.job['log'] + '\n' + str(exc))
|
||
finally:
|
||
with self.mutex:
|
||
self.job['running'] = False
|
||
|
||
def _regenerate(self, apply_draft):
|
||
from copienator.return_answers import load_return_answer_options
|
||
options = load_return_answer_options(self.root)
|
||
reader = 'read-grouped' if self.mode == 'BGnot' else 'read-annotations'
|
||
commands = [[reader, str(self.root), '--update-score']]
|
||
# The readers expose BooleanOptionalAction for supplementary answer options.
|
||
for name, enabled in options.items():
|
||
commands[0].append(f'--{"" if enabled else "no-"}return-answers-{name}')
|
||
if apply_draft:
|
||
commands.append(['modify-annotations', str(self.root), '--annotation-dir', self.mode])
|
||
commands.append(['giving-names', str(self.root), self.mode])
|
||
from copienator.review_snapshot import FILENAME
|
||
if all((directory / FILENAME).is_file() for directory in self.copies()):
|
||
commands[0] = ['regenerate-reviewed', str(self.root), '--annotation-dir', self.mode]
|
||
elif (self.root / 'BRnot').is_dir():
|
||
with self.mutex:
|
||
self.job.update(running=False, status='failed', log='Résultat de lecture sauvegardé absent pour une évaluation avec reprise. Relisez le passage principal puis fusionnez la reprise avant de régénérer dans le navigateur. Aucun résultat n’a été remplacé.')
|
||
return
|
||
env = dict(os.environ, COPIENATOR_FINALISATION_TOKEN=self.token, PYTHONUNBUFFERED='1')
|
||
from copienator import configuration
|
||
if configuration.CONFIG_PATH is not None:
|
||
env['COPIENATOR_CONFIG'] = str(configuration.CONFIG_PATH)
|
||
worker_options = {'pass_fds': (self.lock_fd,)} if os.name != 'nt' and self.lock_fd is not None else {}
|
||
try:
|
||
for command in commands:
|
||
with self.mutex:
|
||
self.job['log'] += f'\n→ {command[0]}\n'
|
||
with subprocess.Popen([sys.executable, '-m', 'copienator', *command],
|
||
cwd=Path(__file__).resolve().parent.parent, env=env,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, **worker_options) as process:
|
||
for line in process.stdout:
|
||
with self.mutex:
|
||
self.job['log'] = (self.job['log'] + line)[-100000:]
|
||
if process.wait() != 0:
|
||
raise RuntimeError(f'{command[0]} a échoué ou est incomplet. Consultez le journal puis relancez.')
|
||
with self.mutex:
|
||
self.dirty_path.parent.mkdir(exist_ok=True)
|
||
atomic_write_json(self.dirty_path, {'dirty': False})
|
||
self.job['status'] = 'success'
|
||
except Exception as exc:
|
||
with self.mutex:
|
||
self.job['status'] = 'failed'
|
||
self.job['log'] += f'\n{exc}\n'
|
||
finally:
|
||
with self.mutex:
|
||
self.job['running'] = False
|