72 lines
3.4 KiB
Python
72 lines
3.4 KiB
Python
"""Content-based caches for saved-review rendering and return publication."""
|
|
from dataclasses import asdict
|
|
from hashlib import sha256
|
|
import json
|
|
from pathlib import Path
|
|
from copienator import atomic_write_json, configuration, read_json, general_remarks
|
|
from copienator.annotation_read_cache import AnnotationReadCache, _files_digest
|
|
from copienator.annotation_modifications import APPLIED, read_instructions
|
|
from copienator.custom_remarks import load_remarks, copy_key
|
|
from copienator.return_answers import load_return_answer_options
|
|
from copienator.review_snapshot import FILENAME
|
|
|
|
|
|
def output_options(root):
|
|
return (load_return_answer_options(root), configuration.RETURN_ANSWERS_ENABLED,
|
|
configuration.RETURN_JPEG_ENABLED, configuration.RETURN_PDF_ENABLED)
|
|
|
|
|
|
def signature(payload):
|
|
return sha256(json.dumps(payload, sort_keys=True, default=str).encode()).hexdigest()
|
|
|
|
|
|
class ReviewedReadCache(AnnotationReadCache):
|
|
state_file = '.reviewed-render-state.json'
|
|
|
|
def __init__(self, workspace, mode, directories):
|
|
self.workspace, self.mode, self.signatures = workspace, mode, {}
|
|
root = workspace.root
|
|
shared = [root / 'labels']
|
|
for folder in ('Text2', 'Sol2'):
|
|
shared.extend((root / folder).glob('*.pdf'))
|
|
common = (_files_digest(shared), configuration.LATEX_ANOT_BEFORE, configuration.LATEX_ANOT_AFTER)
|
|
remarks, general = load_remarks(root), general_remarks.load_remarks(root)
|
|
applied = root / mode / APPLIED
|
|
instructions = read_instructions(applied.read_text(), mode) if applied.is_file() else []
|
|
for directory in directories:
|
|
cid = directory.name[5:]
|
|
saved = read_json(directory / FILENAME)
|
|
paths = [directory / FILENAME, directory / 'score.json']
|
|
for item in saved['data'].values():
|
|
pdf = Path(item['pdf_path'])
|
|
pdf = pdf if pdf.is_absolute() else root / pdf
|
|
paths.extend((pdf, pdf.with_suffix('.layout.json')))
|
|
paths.extend(directory / note['file'] for note in saved['notes'].values() if note['file'])
|
|
self.signatures[cid] = signature((1, common, _files_digest(paths),
|
|
remarks.get(copy_key(cid), {}), general.get(copy_key(cid), []),
|
|
[asdict(i) for i in instructions if i.copy_id == cid]))
|
|
|
|
|
|
class ReturnCache:
|
|
def __init__(self, root, source, destination):
|
|
self.path = source / '.return-publication-state.json'
|
|
self.destination = destination
|
|
paths = [source / 'score.json', source / 'info.json']
|
|
paths.extend(source.glob('*.jpg'))
|
|
paths.extend(source.glob('*.pdf'))
|
|
for folder in ('Text2', 'Sol2'):
|
|
paths.extend((root / folder).glob('*.pdf'))
|
|
self.signature = signature((1, str(destination), _files_digest(paths), output_options(root)))
|
|
|
|
def unchanged(self):
|
|
try:
|
|
state = read_json(self.path)
|
|
return (state['signature'] == self.signature and bool(state['outputs'])
|
|
and all((self.destination / name).is_file() for name in state['outputs']))
|
|
except (OSError, ValueError, KeyError, TypeError):
|
|
return False
|
|
|
|
def remember(self):
|
|
outputs = [str(p.relative_to(self.destination)) for p in self.destination.rglob('*') if p.is_file()]
|
|
atomic_write_json(self.path, {'signature': self.signature, 'outputs': outputs})
|