"""Первичные файлы организации: загрузка (spec primary-files «Загрузка файлов организации», «Дедуп по содержимому»; audit-trail «События файлов и документов»; design D-4). Загрузка одного файла — SAVEPOINT (`begin_nested()`) внутри транзакции запроса (`SessionDep`): отказ файла откатывает только его записи, принятые раньше файлы того же запроса остаются. Порядок внутри SAVEPOINT: 1. рекомендательная блокировка `(ЮЛ, sha256)` до конца транзакции запроса и проверка дубля — до записи в S3; 2. `s3_object` + `flush` (конфликт ключа или ошибка БД — до загрузки объекта, бакет не засоряется); 3. `put_object`; ошибка S3 — `STORAGE_UNAVAILABLE`, SAVEPOINT откатывается (строки `s3_object` не остаётся); 4. `stored_file` (`UPLOADED`, `version_no = 1`) и `recognition_job` (`queued`) + `flush`, `stored_file.current_recognition_job_id`, аудит `FILE_UPLOAD`. Если после шага 3 упадёт БД (вставка файла/задания, аудит или commit запроса), объект в S3 останется без записи `s3_object` — допустимо: ключ содержит свежие `file_id`/`job_id`, ни на что не ссылается и не будет переиспользован; уборка таких объектов — ретеншн бакета (OQ-22). """ import hashlib import uuid from dataclasses import dataclass from loguru import logger from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from starlette import status as http_status from src.api.dependencies.audit import RequestAudit from src.common.clock import utc_now from src.common.exceptions import ConflictError, DomainError, NotFoundError from src.common.filenames import safe_filename from src.contract import CONTRACT_VERSION, request_subject from src.core.config import Settings from src.core.storage import ObjectStorage from src.domain.file_types import MAX_ATTEMPTS, DetectedFile, detect_file from src.domain.policy import VisibilityScope from src.models.directories import LegalEntity from src.models.enums import AuditAction, AuditSection, FileStatus, RecognitionStatus, S3BucketRole from src.models.files import RecognitionJob, S3Object, StoredFile from src.repositories import directories as directories_repo from src.services.legal_entities import LEGAL_ENTITY_NOT_FOUND OBJECT_TYPE = "stored_file" UPLOADS_PREFIX = "uploads" DUPLICATE_FILE = "DUPLICATE_FILE" FILE_TOO_LARGE = "FILE_TOO_LARGE" STORAGE_UNAVAILABLE = "STORAGE_UNAVAILABLE" STORAGE_UNAVAILABLE_MESSAGE = "хранилище файлов недоступно" class StorageUnavailableError(DomainError): """503 `STORAGE_UNAVAILABLE`: S3 не подключено (dev без S3) или не приняло объект.""" def __init__(self, message: str = STORAGE_UNAVAILABLE_MESSAGE) -> None: super().__init__(http_status.HTTP_503_SERVICE_UNAVAILABLE, STORAGE_UNAVAILABLE, message) @dataclass(frozen=True) class AcceptedFile: """Принятый файл: `stored_file.id`, `recognition_job.id`, очищенное имя.""" file_id: uuid.UUID job_id: uuid.UUID name: str @dataclass(frozen=True) class RejectedFile: """Отклонённый файл; `file_id` — существующий файл при `DUPLICATE_FILE`.""" name: str code: str message: str file_id: uuid.UUID | None = None @classmethod def from_error(cls, name: str, error: DomainError) -> "RejectedFile": """Отказ по доменной ошибке загрузки файла `name`.""" raw_file_id = error.extra.get("file_id") return cls( name=name, code=error.code, message=error.message, file_id=uuid.UUID(str(raw_file_id)) if raw_file_id is not None else None, ) def upload_key(file_id: uuid.UUID, job_id: uuid.UUID, name: str) -> str: """Ключ исходного файла в S3: `uploads/{file_id}/{job_id}/{имя}` (`03-architecture.md`, схема ключей).""" return f"{UPLOADS_PREFIX}/{file_id}/{job_id}/{name}" class PrimaryFilesService: """Загрузка первичных файлов ЮЛ от имени пользователя запроса (право `DOCUMENT_EDIT` — на маршруте).""" def __init__( self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, storage: ObjectStorage | None, settings: Settings, ) -> None: self._session = session self._audit = audit self._scope = scope self._storage = storage self._settings = settings @property def max_upload_bytes(self) -> int: """Предел размера одного файла (`UPLOAD_MAX_BYTES`).""" return self._settings.UPLOAD_MAX_BYTES async def get_entity(self, entity_id: uuid.UUID) -> LegalEntity: """ЮЛ в области видимости пользователя; чужое и несуществующее — одинаковый 404 (§10.1).""" entity = await directories_repo.get_legal_entity(self._session, self._scope, entity_id) if entity is None: raise NotFoundError(LEGAL_ENTITY_NOT_FOUND) return entity def require_storage(self) -> ObjectStorage: """S3 приложения; без него загрузка невозможна — 503 `STORAGE_UNAVAILABLE`.""" if self._storage is None: raise StorageUnavailableError() return self._storage async def find_duplicate(self, entity_id: uuid.UUID, sha256: str) -> StoredFile | None: """Файл ЮЛ с тем же содержимым, кроме заменённых (`SUPERSEDED`): дубль по spec «Дедуп по содержимому».""" stmt = ( select(StoredFile) .where( StoredFile.legal_entity_id == entity_id, StoredFile.content_hash == sha256, StoredFile.status != FileStatus.SUPERSEDED, ) .order_by(StoredFile.uploaded_at, StoredFile.id) .limit(1) ) return (await self._session.execute(stmt)).scalar_one_or_none() async def upload_one(self, entity: LegalEntity, name: str, content: bytes) -> AcceptedFile: """Принять один файл (см. docstring модуля); отказ — `DomainError` (`UNSUPPORTED_FILE`, `DUPLICATE_FILE`, `STORAGE_UNAVAILABLE`), записи этого файла откатываются вместе с его SAVEPOINT.""" storage = self.require_storage() safe_name = safe_filename(name, default="") detected = detect_file(safe_name, content) sha256 = hashlib.sha256(content).hexdigest() async with self._session.begin_nested(): await self._lock_content(entity.id, sha256) duplicate = await self.find_duplicate(entity.id, sha256) if duplicate is not None: raise ConflictError( DUPLICATE_FILE, f"файл «{safe_name}» уже загружен для этого ЮЛ", file_id=str(duplicate.id), ) return await self._store( entity, name=safe_name, content=content, sha256=sha256, storage=storage, detected=detected ) async def _lock_content(self, entity_id: uuid.UUID, sha256: str) -> None: """Сериализует загрузки одного содержимого в одно ЮЛ до конца транзакции запроса (двойной клик): второй запрос дождётся commit первого и увидит дубль. Блокировка снимается с транзакцией, а не с SAVEPOINT.""" lock_key = func.hashtextextended(f"{OBJECT_TYPE}:{entity_id}:{sha256}", 0) await self._session.execute(select(func.pg_advisory_xact_lock(lock_key))) async def _store( # noqa: PLR0913 - шаги одной загрузки: всё, что уже вычислено в upload_one self, entity: LegalEntity, *, name: str, content: bytes, sha256: str, storage: ObjectStorage, detected: DetectedFile, ) -> AcceptedFile: now = utc_now() user_id = self._audit.user_id file_id, job_id = uuid.uuid4(), uuid.uuid4() key = upload_key(file_id, job_id, name) s3_object = S3Object( id=uuid.uuid4(), bucket_role=S3BucketRole.uploads, bucket=storage.bucket, key=key, content_type=detected.content_type, size_bytes=len(content), sha256=sha256, created_by=f"user:{user_id}", created_at=now, ) self._session.add(s3_object) await self._session.flush() try: await storage.put_object(key, content) except Exception as error: logger.warning( "S3 не приняло загружаемый файл, файл отклонён. error_category=storage_error, key={key}, error={error}", key=key, error=f"{type(error).__name__}: {error}", ) raise StorageUnavailableError(f"{STORAGE_UNAVAILABLE_MESSAGE}: файл «{name}» не сохранён") from error stored_file = StoredFile( id=file_id, legal_entity_id=entity.id, file_type=detected.file_type, original_name=name, s3_object_id=s3_object.id, content_hash=sha256, size_bytes=len(content), status=FileStatus.UPLOADED, version_no=1, uploaded_by=user_id, uploaded_at=now, ) job = RecognitionJob( id=job_id, file_id=file_id, pipeline=detected.pipeline, request_subject=request_subject(detected.pipeline.value), schema_version=CONTRACT_VERSION, status=RecognitionStatus.queued, attempt=1, max_attempts=MAX_ATTEMPTS[detected.pipeline], requested_at=now, ) # Связей `relationship()` в моделях нет — порядок INSERT задаём сами: файл, затем задание (FK `file_id`); # FK `current_recognition_job_id` не отложенный (use_alter): ссылку ставим, когда задание уже вставлено self._session.add(stored_file) await self._session.flush() self._session.add(job) await self._session.flush() stored_file.current_recognition_job_id = job.id await self._session.flush() await self._audit.record( action=AuditAction.FILE_UPLOAD, section=AuditSection.PRIMARY_DOCUMENTS, object_type=OBJECT_TYPE, object_id=file_id, after_value={ "original_name": name, "content_hash": sha256, "job_id": str(job_id), "file_type": detected.file_type.value, }, ) return AcceptedFile(file_id=file_id, job_id=job_id, name=name)