from __future__ import annotations import dataclasses import hashlib import importlib.metadata import io import json import pathlib import platform import re import shutil import tempfile import PIL.Image import PIL.ImageChops import PIL.features import generator.engraving.validation def renderer_identity(fonts, style) -> dict: root = pathlib.Path(__file__).resolve().parents[1] versions = {name: importlib.metadata.version(name) for name in ('Pillow', 'fonttools', 'aggdraw')} versions.update({name: PIL.features.version(name) for name in PIL.features.get_supported()}) sources = [*root.joinpath('engraving').glob('*.py'), root / 'model.py', root / 'harmony.py', root / 'requirements.txt'] return {'fonts_style': fonts.fingerprint(style), 'python': {'implementation': platform.python_implementation(), 'version': platform.python_version()}, 'dependencies': versions, 'sources': {path.relative_to(root).as_posix(): hashlib.sha256(path.read_bytes()).hexdigest() for path in sorted(sources)}} def renderer_fingerprint(fonts, style) -> str: return hashlib.sha256(json.dumps(renderer_identity(fonts, style), sort_keys=True).encode()).hexdigest() def renderer_namespace(cache: pathlib.Path, renderer: str) -> pathlib.Path: namespace = cache / renderer if namespace.parent.is_symlink() or namespace.is_symlink(): raise ValueError('engraving cache namespaces must not be symlinks') return namespace def score_fingerprint(document: dict, renderer: str) -> str: music = {key: value for key, value in document.items() if key != 'playback'} return hashlib.sha256(json.dumps([renderer, music], sort_keys=True, ensure_ascii=False).encode()).hexdigest() def notation_fingerprint(document: dict, renderer: str) -> str: return hashlib.sha256(json.dumps([renderer, document['staves']], sort_keys=True, ensure_ascii=False).encode()).hexdigest() def band_fingerprint(notation: str, page, band) -> str: context = [notation, page.width, page.height, page.supersampling, dataclasses.asdict(band)] return hashlib.sha256(json.dumps(context, sort_keys=True, ensure_ascii=False).encode()).hexdigest() def has_symlink(path: pathlib.Path) -> bool: return any(parent.is_symlink() for parent in (path, *path.parents)) def restore_pages(entry: pathlib.Path, destination: pathlib.Path, validated_pages: set[str] | None = None) -> list[dict] | None: try: if has_symlink(entry) or has_symlink(destination): return None manifest = entry / 'pages.json' if manifest.is_symlink(): return None pages = json.loads(manifest.read_text(encoding='utf-8')) if not isinstance(pages, list) or not 0 < len(pages) <= 10000: return None destination.mkdir(parents=True, exist_ok=True) for index, page in enumerate(pages, 1): if not isinstance(page, dict) or set(page) != {'path', 'sha256'} or page['path'] != f'{index:04}.png' or not isinstance(page['sha256'], str) or not re.fullmatch('[0-9a-f]{64}', page['sha256']): return None path = entry / page['path'] if path.is_symlink() or not path.is_file(): return None target = destination / page['path'] if target.is_symlink(): return None shutil.copyfile(path, target) generator.engraving.validation.check_page(target, validated_pages, expected_sha256=page['sha256']) return pages except (OSError, ValueError, TypeError): return None def store_pages(entry: pathlib.Path, destination: pathlib.Path, pages: list[dict]) -> None: if has_symlink(entry.parent): raise ValueError('engraving cache directories must not be symlinks') entry.parent.mkdir(parents=True, exist_ok=True) if entry.is_symlink() or entry.is_file(): entry.unlink(missing_ok=True) entry.mkdir(exist_ok=True) with tempfile.TemporaryDirectory(prefix='.engraving-', dir=entry.parent) as temporary: staged = pathlib.Path(temporary) for page in pages: shutil.copyfile(destination / page['path'], staged / page['path']) (staged / 'pages.json').write_text(json.dumps(pages), encoding='utf-8') # Identical keys have identical pages; publish complete files, then the manifest. for name in [page['path'] for page in pages] + ['pages.json']: target = entry / name if target.is_dir() and not target.is_symlink(): shutil.rmtree(target, ignore_errors=True) (staged / name).replace(target) def restore_band(entry: pathlib.Path, key: str, width: int, height: int) -> PIL.Image.Image | None: try: manifest, asset = entry / 'band.json', entry / 'band.png' if has_symlink(entry) or manifest.is_symlink() or asset.is_symlink(): return None metadata = json.loads(manifest.read_text(encoding='utf-8')) if not isinstance(metadata, dict) or set(metadata) != {'context', 'width', 'height', 'sha256'}: return None if metadata['context'] != key or type(metadata['width']) is not int or type(metadata['height']) is not int or (metadata['width'], metadata['height']) != (width, height): return None digest = metadata['sha256'] if not isinstance(digest, str) or not re.fullmatch('[0-9a-f]{64}', digest): return None data = asset.read_bytes() if hashlib.sha256(data).hexdigest() != digest: return None with PIL.Image.open(io.BytesIO(data)) as image: if image.format != 'PNG' or image.size != (width, height) or image.mode != 'RGB': return None image.load() ink = PIL.ImageChops.difference(image, PIL.Image.new('RGB', image.size, 'white')).getbbox() if not ink or ink[0] < 3 or ink[1] < 3 or ink[2] > width - 3 or ink[3] > height - 3: return None return image.copy() except (OSError, ValueError, TypeError): return None def store_band(entry: pathlib.Path, key: str, image: PIL.Image.Image) -> None: if has_symlink(entry.parent): raise ValueError('engraving cache directories must not be symlinks') entry.parent.mkdir(parents=True, exist_ok=True) if entry.is_symlink() or entry.is_file(): entry.unlink() entry.mkdir(exist_ok=True) with tempfile.TemporaryDirectory(prefix='.engraving-', dir=entry.parent) as temporary: staged = pathlib.Path(temporary) image.save(staged / 'band.png', format='PNG') metadata = {'context': key, 'width': image.width, 'height': image.height, 'sha256': hashlib.sha256((staged / 'band.png').read_bytes()).hexdigest()} (staged / 'band.json').write_text(json.dumps(metadata), encoding='utf-8') for name in ('band.png', 'band.json'): target = entry / name if target.is_dir() and not target.is_symlink(): shutil.rmtree(target) (staged / name).replace(target)