feat: use authoritative organization directory

tests/apps/organizations/test_directory_import.py

@@ -0,0 +1,218 @@
+"""Tests for authoritative organization directory import."""
+
+from decimal import Decimal
+from tempfile import NamedTemporaryFile
+
+from django.test import TestCase
+from openpyxl import Workbook
+from organizations.directory_import import (
+    DIRECTORY_SHEET,
+    SOURCE_HEADERS,
+    OrganizationDirectoryImportError,
+    OrganizationDirectoryImportService,
+)
+from organizations.models import Organization
+
+
+class OrganizationDirectoryImportServiceTest(TestCase):
+    """Checks importing the authoritative XLSX organization directory."""
+
+    def test_import_xlsx_creates_and_updates_organizations_by_rn(self):
+        path = self._workbook_path(
+            [
+                self._row(
+                    rn="10",
+                    full_name='Акционерное общество "Тест"',
+                    short_name='АО "Тест"',
+                    inn="7701001001",
+                    ogrn="1027700100001",
+                    okpo="00123456",
+                    filial=".F.",
+                    kpp="770101001",
+                    registration_date="01.02.2020",
+                    authorized_capita=1000,
+                    business_act_cod="2",
+                    goz_participation=".T.",
+                    opk_registry_membership=".F.",
+                    _gk="1",
+                    in_korp="1",
+                    _k="2",
+                )
+            ]
+        )
+
+        first = OrganizationDirectoryImportService.import_xlsx(path)
+        second = OrganizationDirectoryImportService.import_xlsx(path)
+
+        self.assertEqual(first.scanned, 1)
+        self.assertEqual(first.created, 1)
+        self.assertEqual(second.updated, 1)
+        self.assertEqual(Organization.objects.count(), 1)
+
+        organization = Organization.objects.get(rn=10)
+        self.assertEqual(organization.name, 'Акционерное общество "Тест"')
+        self.assertEqual(organization.okpo, "00123456")
+        self.assertFalse(organization.is_branch)
+        self.assertEqual(organization.authorized_capital, Decimal("1000.00"))
+        self.assertEqual(organization.business_activity, "Производственная")
+        self.assertEqual(organization.gk_name, "Роскосмос")
+        self.assertEqual(organization.in_korp_name, "Входит в состав")
+        self.assertEqual(organization.state_sector_name, "Головные исполнители")
+        self.assertTrue(organization.goz_participation)
+        self.assertFalse(organization.opk_registry_membership)
+
+    def test_import_xlsx_rejects_missing_required_column(self):
+        path = self._workbook_path([], headers=SOURCE_HEADERS[:-1])
+
+        with self.assertRaisesMessage(
+            OrganizationDirectoryImportError,
+            "Missing required columns",
+        ):
+            OrganizationDirectoryImportService.import_xlsx(path)
+
+    def test_import_xlsx_treats_spaced_dot_date_marker_as_empty(self):
+        path = self._workbook_path(
+            [
+                self._row(
+                    rn="11",
+                    registration_date=".  .",
+                    appointment_date=".  .",
+                    filial=".T.",
+                )
+            ]
+        )
+
+        result = OrganizationDirectoryImportService.import_xlsx(path)
+
+        self.assertEqual(result.created, 1)
+        organization = Organization.objects.get(rn=11)
+        self.assertIsNone(organization.registration_date)
+        self.assertIsNone(organization.appointment_date)
+        self.assertTrue(organization.is_branch)
+
+    def test_import_xlsx_recovers_embedded_tsv_organization_row(self):
+        first_tail = [
+            *[
+                self._row(
+                    rn="20",
+                    pn_name_en="First EN",
+                    inn="7701001020",
+                    ogrn="1027700100020",
+                    okpo="00000020",
+                    filial=".F.",
+                    kpp="770101020",
+                    registration_date="01.01.2020",
+                    _k="2",
+                )[header]
+                for header in SOURCE_HEADERS[6:-1]
+            ],
+            "20",
+            "2",
+        ]
+        embedded_value = "\t".join(str(value or "") for value in first_tail)
+        embedded_value = (
+            embedded_value
+            + "_x000D_\n"
+            + "\t".join(["21", "0", "0", 'ООО "Вторая"'])
+        )
+
+        second_row = self._row(
+            rn="21",
+            full_name='ООО "Вторая"',
+            short_name='ООО "Вторая"',
+            pn_name='ООО "Вторая"',
+            inn="7701001021",
+            ogrn="1027700100021",
+            okpo="00000021",
+            filial=".F.",
+            kpp="770101021",
+            registration_date="02.01.2020",
+            _k="2",
+        )
+        second_row["is"] = "21"
+        path = self._workbook_path(
+            [
+                {
+                    "__raw_row__": [
+                        "20",
+                        "0",
+                        "0",
+                        'ООО "Первая"',
+                        'ООО "Первая"',
+                        'ООО "Первая"',
+                        embedded_value,
+                        *[second_row[header] for header in SOURCE_HEADERS[4:]],
+                    ]
+                }
+            ]
+        )
+
+        result = OrganizationDirectoryImportService.import_xlsx(path)
+
+        self.assertEqual(result.scanned, 1)
+        self.assertEqual(result.created, 2)
+        self.assertEqual(Organization.objects.count(), 2)
+        self.assertEqual(Organization.objects.get(rn=20).inn, "7701001020")
+        second = Organization.objects.get(rn=21)
+        self.assertEqual(second.inn, "7701001021")
+        self.assertEqual(second.integrated_structure, "")
+
+    def _workbook_path(
+        self,
+        rows: list[dict[str, object]],
+        *,
+        headers: list[str] | None = None,
+    ) -> str:
+        workbook = Workbook()
+        worksheet = workbook.active
+        worksheet.title = DIRECTORY_SHEET
+        worksheet.append(headers or SOURCE_HEADERS)
+        for values in rows:
+            raw_row = values.get("__raw_row__")
+            if raw_row is not None:
+                worksheet.append(raw_row)
+            else:
+                worksheet.append([values.get(header, "") for header in SOURCE_HEADERS])
+
+        self._append_reference_sheet(workbook, "_gk", {"1": "Роскосмос"})
+        self._append_reference_sheet(workbook, "in_korp", {"1": "Входит в состав"})
+        self._append_reference_sheet(
+            workbook,
+            "business_act_cod",
+            {"2": "Производственная"},
+        )
+        self._append_reference_sheet(
+            workbook,
+            "_k - Госектор",
+            {"2": "Головные исполнители"},
+        )
+
+        temp_file = NamedTemporaryFile(suffix=".xlsx", delete=False)
+        temp_file.close()
+        workbook.save(temp_file.name)
+        return temp_file.name
+
+    @staticmethod
+    def _append_reference_sheet(
+        workbook: Workbook,
+        title: str,
+        values: dict[str, str],
+    ) -> None:
+        worksheet = workbook.create_sheet(title=title)
+        worksheet.append(["kod", "value"])
+        for key, value in values.items():
+            worksheet.append([key, value])
+
+    @staticmethod
+    def _row(**overrides) -> dict[str, object]:
+        row = {header: "" for header in SOURCE_HEADERS}
+        row.update(
+            {
+                "rn": "1",
+                "full_name": 'ООО "По умолчанию"',
+                "inn": "7701001000",
+                "ogrn": "1027700100000",
+            }
+        )
+        row.update(overrides)
+        return row