From d6de9a27b3d0ea19799fecbe8e204abde281b957 Mon Sep 17 00:00:00 2001 From: Aleksandr Meshchriakov Date: Wed, 29 Apr 2026 01:32:31 +0200 Subject: [PATCH] fix(fns): queue uploads from worker-visible path --- src/apps/parsers/fns_upload.py | 93 +++++++++++++++++++++++++++ src/apps/parsers/serializers.py | 8 +++ src/apps/parsers/views.py | 40 ++++++++++-- src/templates/dashboard.html | 28 ++++++++ tests/apps/parsers/test_fns_upload.py | 49 ++++++++++++++ 5 files changed, 214 insertions(+), 4 deletions(-) diff --git a/src/apps/parsers/fns_upload.py b/src/apps/parsers/fns_upload.py index 616695f..8bb8a50 100644 --- a/src/apps/parsers/fns_upload.py +++ b/src/apps/parsers/fns_upload.py @@ -137,6 +137,55 @@ class FNSUploadService: return FNSUploadResult(queued=1, skipped=0, invalid=0, task_ids=[task.id]) + @classmethod + def queue_server_path( + cls, + *, + server_path: str, + requested_by_id: int | None, + ) -> FNSUploadResult: + """Queue an FNS file that already exists on the worker-visible disk.""" + from apps.parsers.tasks import process_fns_zip_archive + + path = cls._validate_server_path(server_path) + if path.suffix.lower() == ".zip": + task_name = "apps.parsers.tasks.process_fns_zip_archive" + task = process_fns_zip_archive + upload_type = "zip_server_path" + else: + task_name = "apps.parsers.tasks.process_fns_file" + task = process_fns_file + upload_type = "file_server_path" + + task_id = str(uuid.uuid4()) + try: + BackgroundJobService.create_job( + task_id=task_id, + task_name=task_name, + user_id=requested_by_id, + meta={ + "source": ParserLoadLog.Source.FNS_REPORTS, + "file": path.name, + "server_path": str(path), + "upload_type": upload_type, + }, + ) + async_result = task.apply_async( + args=[str(path)], + kwargs={"requested_by_id": requested_by_id}, + task_id=task_id, + ) + except Exception: + BackgroundJob.objects.filter(task_id=task_id).delete() + raise + + return FNSUploadResult( + queued=1, + skipped=0, + invalid=0, + task_ids=[async_result.id], + ) + @classmethod def queue_zip_archive_path( cls, @@ -240,6 +289,50 @@ class FNSUploadService: archive_file.seek(0) return archive_path + @classmethod + def _validate_server_path(cls, server_path: str) -> Path: + raw_path = Path(server_path) + if not raw_path.is_absolute(): + raise ValueError("Путь к файлу должен быть абсолютным") + + path = raw_path.resolve(strict=False) + allowed_roots = cls._allowed_server_path_roots() + if not any(cls._is_relative_to(path, root) for root in allowed_roots): + allowed = ", ".join(str(root) for root in allowed_roots) + raise ValueError(f"Путь должен находиться внутри: {allowed}") + + suffix = path.suffix.lower() + if suffix == ".zip": + return path + if suffix in {".xlsx", ".xlsm"} and FNS_XLSX_FILENAME_RE.match(path.name): + return path + raise ValueError( + "Поддерживаются ZIP архивы или Excel файлы формата " "fin_{id}_{ogrn}.xlsx" + ) + + @staticmethod + def _allowed_server_path_roots() -> list[Path]: + roots = [ + Path(settings.FNS_WATCH_DIRECTORY), + Path(settings.FNS_WATCH_DIRECTORY) / "archives", + ] + configured_archive_dir = getattr( + settings, + "FNS_ARCHIVE_UPLOAD_DIRECTORY", + None, + ) + if configured_archive_dir: + roots.append(Path(configured_archive_dir)) + return list(dict.fromkeys(root.resolve(strict=False) for root in roots)) + + @staticmethod + def _is_relative_to(path: Path, root: Path) -> bool: + try: + path.relative_to(root) + except ValueError: + return False + return True + @classmethod def _queue_file_bytes( cls, diff --git a/src/apps/parsers/serializers.py b/src/apps/parsers/serializers.py index 81a2a1b..22e3800 100644 --- a/src/apps/parsers/serializers.py +++ b/src/apps/parsers/serializers.py @@ -367,6 +367,14 @@ class FNSZipUploadSerializer(serializers.Serializer): return value +class FNSServerPathUploadSerializer(serializers.Serializer): + """Сериализатор для запуска обработки файла, уже лежащего на сервере.""" + + server_path = serializers.CharField( + help_text="Абсолютный путь к ZIP/XLSX внутри FNS watch directory", + ) + + class FNSFileUploadSuccessSerializer(serializers.Serializer): """Ответ одиночной загрузки FNS в формате frontend.""" diff --git a/src/apps/parsers/views.py b/src/apps/parsers/views.py index 7a46de1..db5e2b2 100644 --- a/src/apps/parsers/views.py +++ b/src/apps/parsers/views.py @@ -31,6 +31,7 @@ from apps.parsers.serializers import ( FinancialReportSerializer, FNSFileUploadSerializer, FNSFileUploadSuccessSerializer, + FNSServerPathUploadSerializer, FNSZipUploadSerializer, GenericParserRecordSerializer, IndustrialCertificateSerializer, @@ -74,7 +75,7 @@ from drf_yasg import openapi from drf_yasg.utils import swagger_auto_schema from rest_framework import status from rest_framework.exceptions import ValidationError -from rest_framework.parsers import FormParser, MultiPartParser +from rest_framework.parsers import FormParser, JSONParser, MultiPartParser from rest_framework.permissions import IsAdminUser, IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response @@ -904,7 +905,7 @@ class FNSReportUploadView(APIView): на обработку через Celery. """ - parser_classes = [MultiPartParser] + parser_classes = [MultiPartParser, JSONParser] permission_classes = [IsAdminUser] @swagger_auto_schema( @@ -923,8 +924,18 @@ class FNSReportUploadView(APIView): name="file", in_=openapi.IN_FORM, type=openapi.TYPE_FILE, - required=True, - description="Файл для загрузки (fin_*.xlsx)", + required=False, + description="Файл для загрузки (fin_*.xlsx или ZIP архив)", + ), + openapi.Parameter( + name="server_path", + in_=openapi.IN_FORM, + type=openapi.TYPE_STRING, + required=False, + description=( + "Абсолютный путь к fin_*.xlsx или ZIP на диске worker-а " + "внутри FNS watch directory" + ), ), ], consumes=["multipart/form-data"], @@ -956,6 +967,27 @@ class FNSReportUploadView(APIView): }, ) def post(self, request): # noqa + if "server_path" in request.data: + serializer = FNSServerPathUploadSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + try: + result = FNSUploadService.queue_server_path( + server_path=serializer.validated_data["server_path"], + requested_by_id=request.user.id, + ) + except ValueError as exc: + raise ValidationError({"server_path": str(exc)}) from exc + + return Response( + { + "queued": result.queued, + "skipped": result.skipped, + "invalid": result.invalid, + "task_ids": result.task_ids, + }, + status=status.HTTP_202_ACCEPTED, + ) + uploaded_file = request.FILES.get("file") if uploaded_file and uploaded_file.name.lower().endswith(".zip"): serializer = FNSZipUploadSerializer(data=request.data) diff --git a/src/templates/dashboard.html b/src/templates/dashboard.html index 8a951b2..c9c22ed 100644 --- a/src/templates/dashboard.html +++ b/src/templates/dashboard.html @@ -1538,6 +1538,12 @@ + ${source.key === "fns_financial" ? ` +
+ + +
+ ` : ""} `).join(""); } @@ -2146,6 +2152,28 @@ closeModals(); navigateDashboard(`/dashboard/${encodeURIComponent(target.dataset.sourceLink)}`); } + if (target.dataset.uploadServerPath) { + const pathInput = document.querySelector(`[data-upload-path="${target.dataset.uploadServerPath}"]`); + const serverPath = pathInput?.value?.trim(); + if (!serverPath) return; + const source = sourceByKey(target.dataset.uploadServerPath); + const uploadUrl = source?.upload_url || `/api/v1/parsers/upload/${target.dataset.uploadServerPath}/`; + $("manualUploadStatus").classList.remove("hidden"); + $("manualUploadStatus").textContent = "Постановка файла с диска в очередь..."; + try { + const result = await apiFetch(uploadUrl, { + method: "POST", + body: JSON.stringify({ server_path: serverPath }), + }); + const queued = result.queued ?? result.data?.queued; + const taskIds = result.task_ids ?? result.data?.task_ids ?? []; + $("manualUploadStatus").textContent = `Файл с диска принят: queued=${queued ?? 0}${taskIds.length ? `, task=${taskIds[0]}` : ""}`; + await refreshDashboard(); + showMainTab("uploads"); + } catch (error) { + $("manualUploadStatus").textContent = errorMessage(error); + } + } if (target.dataset.upload) { const fileInput = document.querySelector(`[data-upload-file="${target.dataset.upload}"]`); if (!fileInput || !fileInput.files.length) return; diff --git a/tests/apps/parsers/test_fns_upload.py b/tests/apps/parsers/test_fns_upload.py index 0ea7f39..04e3cbb 100644 --- a/tests/apps/parsers/test_fns_upload.py +++ b/tests/apps/parsers/test_fns_upload.py @@ -5,6 +5,7 @@ import os import tempfile import time import zipfile +from types import SimpleNamespace from unittest.mock import patch from apps.core.models import BackgroundJob @@ -274,6 +275,54 @@ class FNSUploadIntegrationTest(APITestCase): self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + def test_upload_accepts_worker_visible_server_path(self): + with tempfile.TemporaryDirectory() as tmpdir: + watch_dir, processed_dir, failed_dir = self._dirs(tmpdir) + server_path = os.path.join(watch_dir, "archives", "fin_ropk.zip") + with override_settings( + FNS_WATCH_DIRECTORY=watch_dir, + FNS_PROCESSED_DIRECTORY=processed_dir, + FNS_FAILED_DIRECTORY=failed_dir, + ), patch( + "apps.parsers.fns_upload.uuid.uuid4", + return_value="server-path-task-id", + ), patch( + "apps.parsers.tasks.process_fns_zip_archive.apply_async", + return_value=SimpleNamespace(id="server-path-task-id"), + ) as task_mock: + response = self.client.post( + self.upload_url, + {"server_path": server_path}, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) + self.assertEqual(response.data["queued"], 1) + self.assertEqual(response.data["task_ids"], ["server-path-task-id"]) + task_mock.assert_called_once() + job = BackgroundJob.objects.get(task_id="server-path-task-id") + self.assertEqual(job.user_id, self.admin.id) + self.assertEqual(job.meta["server_path"], os.path.realpath(server_path)) + self.assertEqual(job.meta["upload_type"], "zip_server_path") + + def test_upload_rejects_server_path_outside_fns_directory(self): + with tempfile.TemporaryDirectory() as tmpdir: + watch_dir, processed_dir, failed_dir = self._dirs(tmpdir) + outside_path = os.path.join(tmpdir, "outside", "fin_ropk.zip") + with override_settings( + FNS_WATCH_DIRECTORY=watch_dir, + FNS_PROCESSED_DIRECTORY=processed_dir, + FNS_FAILED_DIRECTORY=failed_dir, + ): + response = self.client.post( + self.upload_url, + {"server_path": outside_path}, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertFalse(BackgroundJob.objects.exists()) + def test_upload_skips_when_file_already_exists(self): content = _build_fns_excel_bytes() external_id = _digits(5)