| """Explicit pinned-snapshot download and diagnostic build; no recognition or upload.""" |
| import argparse |
| import copy |
| import hashlib |
| import json |
| from pathlib import Path |
|
|
| import pymupdf |
| from tests.reader_v3_layout_samples import v3, text |
|
|
|
|
| def build(descriptor, bundle, read): |
| primary = v3.verified_read(descriptor['primary'], read) |
| source = descriptor['source_sha256'] |
| if descriptor['primary']['sha256'] != source: |
| raise ValueError('sample primary/source identity mismatch') |
| manifest = v3.decode(v3.verified_read(descriptor['ocr_manifest'], read)) |
| old_book = v3.decode(v3.verified_read({'bucket': v3.shared.PDF_PAGES_BUCKET, |
| **manifest['book_text'], 'role': 'runtime'}, read)) |
| if manifest['source_sha256'] != source or old_book['source_sha256'] != source: |
| raise ValueError('sample OCR/source identity mismatch') |
| chosen = descriptor['pages'] |
| with pymupdf.open(stream=primary, filetype='pdf') as original, pymupdf.open() as document: |
| for number in chosen: |
| document.insert_pdf(original, from_page=number - 1, to_page=number - 1) |
| original[number - 1].get_pixmap(matrix=pymupdf.Matrix(1, 1)).save(str(bundle / f'original-page-{number}.png')) |
| pdf = document.tobytes(no_new_id=True) |
| derived = hashlib.sha256(pdf).hexdigest() |
| root = Path('objects') / derived[:2] / derived / ('d' * 16) |
| target = bundle / root / 'document.pdf' |
| target.parent.mkdir(parents=True, exist_ok=True) |
| target.write_bytes(pdf) |
| primary_ref = {'bucket': v3.shared.PDF_PAGES_BUCKET, 'path': target.relative_to(bundle).as_posix(), |
| 'sha256': derived, 'bytes': len(pdf), 'role': 'runtime'} |
| layers, descriptors, provenance = [], [], [] |
| for index, number in enumerate(chosen, 1): |
| page = manifest['pages'][number - 1] |
| raw_ref = {'bucket': v3.shared.PDF_PAGES_BUCKET, 'path': page['o'], 'sha256': page['os'], |
| 'bytes': page['ob'], 'role': 'provenance'} |
| raw = copy.deepcopy(v3.decode(v3.verified_read(raw_ref, read))) |
| if raw['text'] != old_book['pages'][number - 1]['text']: |
| raise ValueError('sample raw text differs from pinned complete index') |
| raw['page'] = index |
| layer = text.from_page(raw, derived, manifest['language'], raw_ref) |
| layers.append(layer) |
| provenance.append({'source_sha256': source, 'original_page': number, |
| 'raw': raw_ref, 'ocr_manifest': descriptor['ocr_manifest']}) |
| descriptors.append({'kind': f'scan-original-{number}', 'page': index, |
| 'actual_regions': [r['text'] for r in layer['regions']], |
| 'expected_regions': [raw['text'][s['start']:s['end']] for s in raw['text_spans']], |
| 'expected_boxes': [r['box'] for r in layer['regions']]}) |
| layer_ref = v3.build_text_bundle(layers, [{'p': p['page']} for p in layers], root, bundle) |
| def generated_read(ref): |
| path = bundle / ref['path'] |
| return path.read_bytes() if path.is_file() else read(ref) |
| spec = {'source_key': 'diagnostics\0scan-three-pages.pdf', 'source_sha256': derived, |
| 'primary': primary_ref, 'text_layer': layer_ref, 'require_complete_preview': True} |
| spec = v3.generate_previews(spec, generated_read, bundle, dpi=100) |
| ref, reading = v3.build_reading(spec, generated_read, bundle) |
| query = descriptor['query'] |
| result = {'reading': ref, 'manifest': reading, 'layers': layers, 'samples': descriptors, |
| 'query': query, 'expected_total': sum(old_book['pages'][p - 1]['text'].count(query) for p in chosen), |
| 'provenance': provenance, 'full_source_pages': manifest['page_count'], |
| 'remote_writes': 0, 'recognizer_calls': 0} |
| (bundle / 'fixture.json').write_text(json.dumps(result, ensure_ascii=False, indent=2)) |
| return result |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('--descriptor', type=Path, default=Path(__file__).parent / 'fixtures/reader_v3_scan_sample.json') |
| parser.add_argument('--output', required=True, type=Path) |
| args = parser.parse_args() |
| descriptor = json.loads(args.descriptor.read_text()) |
| args.output.mkdir(parents=True, exist_ok=True) |
| from scripts.reader_bucket_store import HubBucketStore |
| store, cache = HubBucketStore(), {} |
| def read(ref): |
| key = (ref['bucket'], ref['path']) |
| if key not in cache: |
| cache[key] = store.read_bytes(*key) |
| return cache[key] |
| result = build(descriptor, args.output, read) |
| print(json.dumps({'fixture': str(args.output / 'fixture.json'), 'pages': descriptor['pages'], |
| 'query_total': result['expected_total'], 'remote_writes': 0, 'recognizer_calls': 0})) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
