22ca529bc6
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
101 lines
4.3 KiB
Python
101 lines
4.3 KiB
Python
"""Golden нормализатора: сбор склейки реального ответа и запись эталонов (`make golden-update`).
|
||
|
||
Набор задания `<name>` — `<name>.json` + `<name>_*.json` из `../fixtures/ocr` (как у эмулятора); эталон —
|
||
`../fixtures/ocr/expected_normalize/<name>.json`. Запуск: `cd api && .venv/bin/python -m tests.golden_update`.
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from arm_statements import default_rules
|
||
from src.contract import ResultStatus, ServiceInfo, ServiceName
|
||
from src.core.worker import _outcome_metrics, build_result
|
||
from src.gateway.hotfolder import AnswerFile
|
||
from src.gateway.merge import merge_answers
|
||
from src.normalizer import NormalizeOutcome, OcrNormalizer
|
||
|
||
API_DIR = Path(__file__).resolve().parents[1]
|
||
REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr"
|
||
GOLDEN_DIR = REAL_FIXTURES / "expected_normalize"
|
||
JOB = "3f1c2b7a-0000-4000-8000-000000000001"
|
||
JOB_ID = uuid.UUID(JOB)
|
||
FILE_ID = uuid.UUID("3f1c2b7a-0000-4000-8000-0000000000f1")
|
||
SERVICE = ServiceInfo(name=ServiceName.SCAN_GATEWAY, version="golden")
|
||
GOLDEN_VERSION = "0" * 40 # фиксированная версия копии: golden от раскладки копии не зависит
|
||
|
||
|
||
def golden_cases() -> list[str]:
|
||
"""Имена `<name>` без суффикса `_*`: `aep_3`, `aep_6`, … (по одному на набор)."""
|
||
names = sorted(p.name[: -len(".json")] for p in REAL_FIXTURES.glob("*.json") if p.is_file())
|
||
stems = {name for name in names if not any(name.startswith(f"{other}_") for other in names if other != name)}
|
||
return sorted(stems)
|
||
|
||
|
||
def real_set(name: str) -> dict[str, Any]:
|
||
files = []
|
||
for path in sorted(REAL_FIXTURES.glob(f"{name}*.json")):
|
||
base = path.name[: -len(".json")]
|
||
if base == name:
|
||
target = f"{JOB}.json"
|
||
elif base.startswith(f"{name}_"):
|
||
target = f"{JOB}_{base[len(name) + 1 :]}.json"
|
||
else:
|
||
continue
|
||
files.append(AnswerFile(target, path.read_bytes()))
|
||
return dict(merge_answers(JOB, files))
|
||
|
||
|
||
def full_payload(outcome: NormalizeOutcome) -> dict[str, Any]:
|
||
"""Полный файл результата (только успех) с фиктивными идентификаторами: им проверяется схема `result`."""
|
||
result = build_result(
|
||
outcome,
|
||
job_id=JOB_ID,
|
||
file_id=FILE_ID,
|
||
service=SERVICE,
|
||
raw_s3_key="raw/golden.json",
|
||
metrics=_outcome_metrics(outcome),
|
||
)
|
||
return dict(result.to_wire())
|
||
|
||
|
||
def outcome_to_json(outcome: NormalizeOutcome) -> dict[str, Any]:
|
||
"""Эталон golden v2 (P5-14): `schema_version`, `status`, `metrics`, `documents`; для ошибки — ещё `error`.
|
||
|
||
`job_id`, `file_id`, `service`, `raw_s3_key` не пишутся: `statements_version` меняется при каждой раскладке."""
|
||
metrics = _outcome_metrics(outcome)
|
||
if outcome.status is ResultStatus.ERROR:
|
||
return {
|
||
"schema_version": "2.0",
|
||
"status": "error",
|
||
"error": outcome.error.model_dump(mode="json", exclude_none=True) if outcome.error else None,
|
||
"metrics": metrics.model_dump(mode="json", exclude_none=True) if metrics else None,
|
||
"documents": [],
|
||
}
|
||
wire = full_payload(outcome)
|
||
return {key: wire[key] for key in ("schema_version", "status", "metrics", "documents") if key in wire}
|
||
|
||
|
||
def main() -> int:
|
||
if not REAL_FIXTURES.is_dir():
|
||
print(f"нет каталога {REAL_FIXTURES}", file=sys.stderr)
|
||
return 2
|
||
rules = default_rules()
|
||
GOLDEN_DIR.mkdir(exist_ok=True)
|
||
for name in golden_cases():
|
||
payload = outcome_to_json(OcrNormalizer(rules, version=GOLDEN_VERSION).normalize(real_set(name)))
|
||
(GOLDEN_DIR / f"{name}.json").write_text(
|
||
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
||
)
|
||
print(
|
||
f"{name}: {payload['status']}, документов {len(payload['documents'])}, "
|
||
f"предупреждений {len((payload.get('metrics') or {}).get('warnings', []))}"
|
||
)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|