e44ee61548
PDF-процессор кладёт по одному json на каждый распознанный документ.
merge_documents() собирает их в {file_id, legal_entity_id, files: [...]},
где files — полное содержимое каждого файла без обработки.
Модуль не зависит от config/S3/NATS: используется и как CLI-скрипт,
и (в дальнейшем) из PollWorker.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JvCSGmwgL8juhZdXr32qXM
84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
import json
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from src.core.merge_results import merge_documents
|
||
|
||
DOC_A = {
|
||
'file_uid': 'uid-a',
|
||
'error': {'code': '', 'message': ''},
|
||
'data_ocr': {'label': {'form_name': 'Бухгалтерский баланс'}, 'table_structure': []},
|
||
}
|
||
DOC_B = {
|
||
'file_uid': 'uid-b',
|
||
'error': {'code': '', 'message': ''},
|
||
'data_ocr': {'label': {'form_name': 'Отчёт о финансовых результатах'}, 'table_structure': []},
|
||
}
|
||
|
||
|
||
def write_doc(directory: Path, name: str, payload: dict) -> Path:
|
||
path = directory / name
|
||
path.write_text(json.dumps(payload, ensure_ascii=False), encoding='utf-8')
|
||
return path
|
||
|
||
|
||
def test_merges_two_documents_into_one_result(tmp_path):
|
||
paths = [write_doc(tmp_path, 'doc_1.json', DOC_A), write_doc(tmp_path, 'doc_2.json', DOC_B)]
|
||
|
||
result = merge_documents(paths, file_id='file-1', legal_entity_id='le-1')
|
||
|
||
assert result == {
|
||
'file_id': 'file-1',
|
||
'legal_entity_id': 'le-1',
|
||
'files': [DOC_A, DOC_B],
|
||
}
|
||
|
||
|
||
def test_keeps_document_content_untouched(tmp_path):
|
||
# содержимое кладётся как есть: незнакомые поля и вложенность не теряются
|
||
weird = {'file_uid': 'uid-c', 'unexpected': [1, {'deep': None}], 'кириллица': 'значение'}
|
||
paths = [write_doc(tmp_path, 'doc_1.json', weird)]
|
||
|
||
result = merge_documents(paths, file_id='file-1', legal_entity_id='le-1')
|
||
|
||
assert result['files'] == [weird]
|
||
|
||
|
||
def test_preserves_given_order(tmp_path):
|
||
# порядок в files[] — порядок переданных путей, а не порядок записи на диск
|
||
second = write_doc(tmp_path, 'z_doc.json', DOC_B)
|
||
first = write_doc(tmp_path, 'a_doc.json', DOC_A)
|
||
|
||
result = merge_documents([second, first], file_id='f', legal_entity_id='l')
|
||
|
||
assert [doc['file_uid'] for doc in result['files']] == ['uid-b', 'uid-a']
|
||
|
||
|
||
def test_rejects_empty_input():
|
||
with pytest.raises(ValueError):
|
||
merge_documents([], file_id='f', legal_entity_id='l')
|
||
|
||
|
||
def test_cli_writes_merged_file(tmp_path):
|
||
a = write_doc(tmp_path, 'doc_1.json', DOC_A)
|
||
b = write_doc(tmp_path, 'doc_2.json', DOC_B)
|
||
out = tmp_path / 'merged.json'
|
||
|
||
completed = subprocess.run(
|
||
[sys.executable, '-m', 'src.core.merge_results', str(a), str(b),
|
||
'--file-id', 'file-1', '--legal-entity-id', 'le-1', '-o', str(out)],
|
||
cwd=Path(__file__).resolve().parent.parent,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
|
||
assert completed.returncode == 0, completed.stderr
|
||
assert json.loads(out.read_text(encoding='utf-8')) == {
|
||
'file_id': 'file-1',
|
||
'legal_entity_id': 'le-1',
|
||
'files': [DOC_A, DOC_B],
|
||
}
|