feat(parsers): support FNS zip uploads in admin

src/apps/parsers/fns_upload.py

@@ -0,0 +1,178 @@
+"""Reusable upload helpers for FNS financial report files."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import time
+import uuid
+import zipfile
+from dataclasses import dataclass, field
+from pathlib import Path, PurePosixPath
+
+from apps.core.models import BackgroundJob
+from apps.core.services import BackgroundJobService
+from apps.parsers.models import ParserLoadLog
+from apps.parsers.services import FNSReportService
+from apps.parsers.tasks import process_fns_file
+from django.conf import settings
+
+FNS_XLSX_FILENAME_RE = re.compile(r"^fin_\d+_\d{13,15}\.xlsx$")
+
+
+@dataclass
+class FNSUploadResult:
+    """Result of queuing FNS files for processing."""
+
+    queued: int = 0
+    skipped: int = 0
+    invalid: int = 0
+    task_ids: list[str] = field(default_factory=list)
+
+
+class FNSUploadService:
+    """Queue uploaded FNS Excel files and ZIP archives for processing."""
+
+    @classmethod
+    def queue_uploaded_files(cls, *, files, requested_by_id: int | None) -> FNSUploadResult:
+        result = FNSUploadResult()
+        seen_hashes: set[str] = set()
+
+        for uploaded_file in files:
+            status, task_id = cls._queue_file_bytes(
+                file_name=uploaded_file.name,
+                file_content=uploaded_file.read(),
+                requested_by_id=requested_by_id,
+                seen_hashes=seen_hashes,
+            )
+            cls._accumulate(result=result, status=status, task_id=task_id)
+
+        return result
+
+    @classmethod
+    def queue_zip_archive(
+        cls,
+        *,
+        archive_file,
+        requested_by_id: int | None,
+    ) -> FNSUploadResult:
+        result = FNSUploadResult()
+        seen_hashes: set[str] = set()
+
+        archive_file.seek(0)
+        try:
+            with zipfile.ZipFile(archive_file) as archive:
+                for member in archive.infolist():
+                    if member.is_dir():
+                        continue
+
+                    file_name = cls._extract_member_name(member.filename)
+                    if not file_name or not FNS_XLSX_FILENAME_RE.match(file_name):
+                        result.invalid += 1
+                        continue
+
+                    status, task_id = cls._queue_file_bytes(
+                        file_name=file_name,
+                        file_content=archive.read(member),
+                        requested_by_id=requested_by_id,
+                        seen_hashes=seen_hashes,
+                    )
+                    cls._accumulate(result=result, status=status, task_id=task_id)
+        except zipfile.BadZipFile as exc:
+            raise ValueError("Загруженный файл не является корректным ZIP архивом") from exc
+
+        return result
+
+    @staticmethod
+    def _extract_member_name(member_name: str) -> str | None:
+        path = PurePosixPath(member_name)
+        if path.is_absolute() or ".." in path.parts:
+            return None
+        if len(path.parts) != 1:
+            return None
+        file_name = path.name
+        return file_name or None
+
+    @classmethod
+    def _queue_file_bytes(
+        cls,
+        *,
+        file_name: str,
+        file_content: bytes,
+        requested_by_id: int | None,
+        seen_hashes: set[str],
+    ) -> tuple[str, str | None]:
+        file_hash = hashlib.sha256(file_content).hexdigest()
+        if file_hash in seen_hashes or FNSReportService.exists_by_hash(file_hash):
+            return "skipped", None
+
+        upload_dir = Path(settings.FNS_WATCH_DIRECTORY)
+        upload_dir.mkdir(parents=True, exist_ok=True)
+
+        file_path = upload_dir / file_name
+        if not cls._try_create_fns_lock(file_path):
+            return "skipped", None
+
+        lock_path = Path(f"{file_path}.lock")
+        if file_path.exists():
+            lock_path.unlink(missing_ok=True)
+            return "skipped", None
+
+        try:
+            file_path.write_bytes(file_content)
+        except Exception:
+            lock_path.unlink(missing_ok=True)
+            raise
+
+        task_id = str(uuid.uuid4())
+        try:
+            BackgroundJobService.create_job(
+                task_id=task_id,
+                task_name="apps.parsers.tasks.process_fns_file",
+                user_id=requested_by_id,
+                meta={
+                    "source": ParserLoadLog.Source.FNS_REPORTS,
+                    "file": file_name,
+                },
+            )
+            task = process_fns_file.apply_async(
+                args=[str(file_path)],
+                kwargs={"requested_by_id": requested_by_id},
+                task_id=task_id,
+            )
+        except Exception:
+            lock_path.unlink(missing_ok=True)
+            BackgroundJob.objects.filter(task_id=task_id).delete()
+            raise
+
+        seen_hashes.add(file_hash)
+        return "queued", task.id
+
+    @staticmethod
+    def _try_create_fns_lock(file_path: Path) -> bool:
+        lock_path = Path(f"{file_path}.lock")
+        if lock_path.exists():
+            try:
+                age_seconds = time.time() - lock_path.stat().st_mtime
+                ttl_seconds = getattr(settings, "FNS_LOCK_TTL_SECONDS", 3600)
+                if age_seconds > ttl_seconds:
+                    lock_path.unlink()
+                else:
+                    return False
+            except FileNotFoundError:
+                pass
+        try:
+            lock_path.touch(exist_ok=False)
+        except FileExistsError:
+            return False
+        return True
+
+    @staticmethod
+    def _accumulate(*, result: FNSUploadResult, status: str, task_id: str | None) -> None:
+        if status == "queued":
+            result.queued += 1
+            if task_id:
+                result.task_ids.append(task_id)
+            return
+        if status == "skipped":
+            result.skipped += 1