feat(#190): признак скан/цифровой PDF при загрузке (§14.6)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

api/requirements.txt

@@ -49,6 +49,7 @@ pydantic_core==2.46.4
 Pygments==2.20.0
 PyJWT[crypto]==2.8.0
 PyNaCl==1.5.0
+pypdf==6.21.4
 python-dotenv==1.2.2
 python-multipart==0.0.32
 PyYAML==6.0.3

api/src/domain/pdf_text_layer.py

@@ -0,0 +1,24 @@
+"""Признак текстового слоя PDF при загрузке (`stored_file.is_scan`, `05-domain-rules.md` §14.6).
+
+Цифровой PDF несёт извлекаемый текст (шрифты, операторы показа текста в потоке содержимого страницы); чистый скан —
+только растровое изображение страницы без текста. ASSUMPTION §14.6: PDF со скрытым OCR-слоем (пост-обработка)
+считается цифровым — различить его от исходно цифрового документа без дополнительных эвристик не планируется.
+
+Файл на этом шаге уже прошёл проверку сигнатуры `detect_file` (`domain/file_types.py`), но содержимое после сигнатуры
+может быть повреждено или зашифровано — `has_text_layer` не должна падать на нём: любое исключение `pypdf` ловится и
+даёт `False` (как для чистого скана), иначе загрузка отказала бы на мусорном, но формально PDF-файле.
+"""
+
+import io
+
+from pypdf import PdfReader
+
+
+def has_text_layer(content: bytes) -> bool:
+    """Есть ли хотя бы на одной странице `content` извлекаемый текст; битый/зашифрованный PDF и любая другая
+    ошибка разбора — `False`, без исключения."""
+    try:
+        reader = PdfReader(io.BytesIO(content))
+        return any(page.extract_text().strip() for page in reader.pages)
+    except Exception:
+        return False

api/src/services/primary_files.py

@@ -37,6 +37,7 @@ from src.contract import CONTRACT_VERSION, request_subject
 from src.core.config import Settings
 from src.core.storage import ObjectNotFoundError, ObjectStorage
 from src.domain.file_types import MAX_ATTEMPTS, DetectedFile, detect_file
+from src.domain.pdf_text_layer import has_text_layer
 from src.domain.policy import IN_WORK_STATUSES, VisibilityScope
 from src.domain.quality_facts import build_outcome_payload
 from src.models.consolidation import Consolidation
@@ -48,6 +49,7 @@ from src.models.enums import (
     DocumentStatus,
     DocumentType,
     FileStatus,
+    FileType,
     QualityOutcome,
     RecognitionStatus,
     S3BucketRole,
@@ -528,6 +530,9 @@ class PrimaryFilesService:
             )
             raise StorageUnavailableError(f"{STORAGE_UNAVAILABLE_MESSAGE}: файл «{name}» не сохранён") from error
 
+        # §14.6: признак скана — только для PDF, по отсутствию текстового слоя (true = скан, false = цифровой, в
+        # т.ч. со скрытым OCR-слоем); для остальных типов файла — NULL (pipeline = tabular)
+        is_scan = not has_text_layer(content) if detected.file_type is FileType.PDF else None
         stored_file = StoredFile(
             id=file_id,
             legal_entity_id=entity.id,
@@ -541,6 +546,7 @@ class PrimaryFilesService:
             version_no=1,
             uploaded_by=user_id,
             uploaded_at=now,
+            is_scan=is_scan,
         )
         job = RecognitionJob(
             id=job_id,

api/tests/test_api_files_upload.py

@@ -307,6 +307,86 @@ async def test_image_file_goes_to_image_pipeline_with_one_attempt(
     assert len(storage.objects) == 1
 
 
+# --- «Признак скана» (§14.6, `stored_file.is_scan`) -------------------------------------------------------------
+
+
+def _digital_pdf(tag: str) -> bytes:
+    """Минимальный валидный одностраничный PDF с текстовым слоем (`BT ... Tj ET`): в репозитории нет зависимости
+    для генерации PDF (design #190, задача 12), поэтому байты строятся вручную, с уникальным хвостом текста (разный
+    sha256 — не дубль)."""
+    content_stream = b"BT /F1 12 Tf 10 100 Td (" + tag.encode() + b") Tj ET"
+    objects = [
+        b"<< /Type /Catalog /Pages 2 0 R >>",
+        b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
+        b"<< /Type /Page /Parent 2 0 R /Resources << /Font << /F1 4 0 R >> >> "
+        b"/MediaBox [0 0 200 200] /Contents 5 0 R >>",
+        b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
+        b"<< /Length " + str(len(content_stream)).encode() + b" >>\nstream\n" + content_stream + b"\nendstream",
+    ]
+    buf = bytearray(b"%PDF-1.4\n")
+    offsets: list[int] = []
+    for i, obj in enumerate(objects, start=1):
+        offsets.append(len(buf))
+        buf += f"{i} 0 obj\n".encode() + obj + b"\nendobj\n"
+    xref_offset = len(buf)
+    n = len(objects) + 1
+    buf += f"xref\n0 {n}\n".encode()
+    buf += b"0000000000 65535 f \n"
+    for off in offsets:
+        buf += f"{off:010d} 00000 n \n".encode()
+    buf += b"trailer\n<< /Size " + str(n).encode() + b" /Root 1 0 R >>\nstartxref\n"
+    buf += str(xref_offset).encode() + b"\n%%EOF"
+    return bytes(buf)
+
+
+@pytest.mark.integration
+async def test_upload_scanned_pdf_sets_is_scan_true(
+    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
+) -> None:
+    """PDF без текстового слоя (в т.ч. формально битый, как здесь — сигнатура есть, структуры нет) — скан (`True`)."""
+    entity = await _rostov_entity(seeded)
+    client = await auth_client(KA_RF)
+
+    response = await _post(client, entity.id, ("ББ скан.pdf", PDF + b"no-text-layer"))
+
+    assert response.status_code == HTTP_CREATED, response.text
+    file = await seeded.get(StoredFile, uuid.UUID(response.json()["accepted"][0]["file_id"]))
+    assert file is not None
+    assert file.is_scan is True
+
+
+@pytest.mark.integration
+async def test_upload_digital_pdf_sets_is_scan_false(
+    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
+) -> None:
+    """ВНИМАНИЕ — направление: PDF с текстовым слоем — цифровой, `is_scan = False` (не `True`)."""
+    entity = await _rostov_entity(seeded)
+    client = await auth_client(KA_RF)
+
+    response = await _post(client, entity.id, ("ББ цифровой.pdf", _digital_pdf("digital")))
+
+    assert response.status_code == HTTP_CREATED, response.text
+    file = await seeded.get(StoredFile, uuid.UUID(response.json()["accepted"][0]["file_id"]))
+    assert file is not None
+    assert file.is_scan is False
+
+
+@pytest.mark.integration
+async def test_upload_non_pdf_leaves_is_scan_null(
+    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage
+) -> None:
+    """Не-PDF (конвейер `tabular`) — признак скана не определяется, `NULL`."""
+    entity = await _rostov_entity(seeded)
+    client = await auth_client(KA_RF)
+
+    response = await _post(client, entity.id, ("osv.xls", _xls("not-pdf-is-scan")))
+
+    assert response.status_code == HTTP_CREATED, response.text
+    file = await seeded.get(StoredFile, uuid.UUID(response.json()["accepted"][0]["file_id"]))
+    assert file is not None
+    assert file.is_scan is None
+
+
 @pytest.mark.integration
 async def test_two_files_write_two_upload_events(
     auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage

api/tests/test_domain_pdf_text_layer.py

@@ -0,0 +1,52 @@
+"""`has_text_layer` — признак текстового слоя PDF (`05-domain-rules.md` §14.6, `stored_file.is_scan`)."""
+
+from pathlib import Path
+
+import pytest
+
+from src.domain.pdf_text_layer import has_text_layer
+
+SCANNED_PDF = Path("ocr") / "aep_9.pdf"  # реальный скан без текстового слоя (2 страницы, extract_text == "")
+
+
+def _digital_pdf_bytes(text: bytes = b"Hello World") -> bytes:
+    """Минимальный валидный одностраничный PDF с текстовым слоем: `BT ... Tj ET` в потоке содержимого страницы,
+    без внешних зависимостей (в репозитории нет библиотеки генерации PDF, см. design #190)."""
+    content_stream = b"BT /F1 12 Tf 10 100 Td (" + text + b") Tj ET"
+    objects = [
+        b"<< /Type /Catalog /Pages 2 0 R >>",
+        b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
+        b"<< /Type /Page /Parent 2 0 R /Resources << /Font << /F1 4 0 R >> >> "
+        b"/MediaBox [0 0 200 200] /Contents 5 0 R >>",
+        b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
+        b"<< /Length " + str(len(content_stream)).encode() + b" >>\nstream\n" + content_stream + b"\nendstream",
+    ]
+    buf = bytearray(b"%PDF-1.4\n")
+    offsets: list[int] = []
+    for i, obj in enumerate(objects, start=1):
+        offsets.append(len(buf))
+        buf += f"{i} 0 obj\n".encode() + obj + b"\nendobj\n"
+    xref_offset = len(buf)
+    n = len(objects) + 1
+    buf += f"xref\n0 {n}\n".encode()
+    buf += b"0000000000 65535 f \n"
+    for off in offsets:
+        buf += f"{off:010d} 00000 n \n".encode()
+    buf += b"trailer\n<< /Size " + str(n).encode() + b" /Root 1 0 R >>\nstartxref\n"
+    buf += str(xref_offset).encode() + b"\n%%EOF"
+    return bytes(buf)
+
+
+def test_has_text_layer_true_for_digital_pdf() -> None:
+    assert has_text_layer(_digital_pdf_bytes()) is True
+
+
+@pytest.mark.fixtures
+def test_has_text_layer_false_for_scanned_pdf(fixtures_dir: Path) -> None:
+    scanned_pdf_bytes = fixtures_dir.joinpath(SCANNED_PDF).read_bytes()
+
+    assert has_text_layer(scanned_pdf_bytes) is False
+
+
+def test_has_text_layer_false_for_corrupt_pdf_without_raising() -> None:
+    assert has_text_layer(b"not a pdf") is False