WON-410-vve-brief-importer

app/apps/homeownerassociation/admin.py

@@ -113,8 +113,7 @@ class HomeownerAssociationCommunicationNoteAdmin(admin.ModelAdmin):
         "note",
         "author_name",
         "date",
-        "created",
-        "updated",
+        "is_imported",
     )
     search_fields = (
         "homeowner_association__id",
@@ -124,3 +123,4 @@ class HomeownerAssociationCommunicationNoteAdmin(admin.ModelAdmin):
     )
     autocomplete_fields = ["homeowner_association"]
     exclude = ("author",)
+    readonly_fields = ("created", "updated", "is_imported")

app/apps/homeownerassociation/importers/base.py

@@ -4,6 +4,7 @@
 from typing import Dict, List, Optional, Set, Tuple
 from django.core.exceptions import ValidationError
 from django.core.validators import EmailValidator
+from django.db import transaction
 
 
 class ImportError(Exception):
@@ -82,14 +83,19 @@ def _detect_encoding(self, file_path: str) -> str:
         # Default to utf-8 if all fail
         return "utf-8"
 
-    def _detect_delimiter(self, first_line: str) -> str:
+    def _detect_delimiter(self, first_line: str) -> Optional[str]:
         """
         Detect CSV delimiter by checking the first line (semicolon or comma).
+        Returns None if no delimiter is found (single-column file).
         """
         # Count semicolons and commas in the first line
         semicolon_count = first_line.count(";")
         comma_count = first_line.count(",")
 
+        # If no delimiter found, return None (single-column file)
+        if semicolon_count == 0 and comma_count == 0:
+            return None
+
         # Use semicolon if it appears more frequently, otherwise default to comma
         if semicolon_count > comma_count:
             return ";"
@@ -114,29 +120,62 @@ def _read_csv(self, file_path: str) -> Tuple[List[str], List[Dict[str, str]]]:
                 first_line = content.split("\n")[0] if content else ""
                 delimiter = self._detect_delimiter(first_line)
 
-                reader = csv.DictReader(io.StringIO(content), delimiter=delimiter)
-                headers = reader.fieldnames
-
-                if not headers:
-                    raise ImportError("CSV file has no headers")
-
-                # Normalize headers: strip whitespace and handle None values
-                headers = [
-                    h.strip() if h and isinstance(h, str) else "" for h in headers
-                ]
-                reader.fieldnames = headers
-
-                rows = []
-                for row in reader:
-                    # Strip whitespace from all keys and values, handle None values
-                    cleaned_row = {}
-                    for k, v in row.items():
-                        # Handle None keys
-                        key = k.strip() if k and isinstance(k, str) else ""
-                        # Handle None values
-                        value = v.strip() if v and isinstance(v, str) else ""
-                        cleaned_row[key] = value
-                    rows.append(cleaned_row)
+                # Handle single-column file (no delimiter)
+                if delimiter is None:
+                    # For single-column files, read line by line and handle quotes manually
+                    # Since there's no delimiter, we can't use csv.reader (it defaults to comma)
+                    lines_raw = content.strip().split("\n")
+                    lines = []
+                    for line in lines_raw:
+                        line = line.strip()
+                        if not line:
+                            continue
+                        # Handle quoted values: remove surrounding quotes and unescape internal quotes
+                        if line.startswith('"') and line.endswith('"'):
+                            # Remove surrounding quotes
+                            line = line[1:-1]
+                            # Replace escaped quotes (double quotes) with single quotes
+                            line = line.replace('""', '"')
+                        lines.append(line)
+
+                    if not lines:
+                        raise ImportError("CSV file is empty")
+
+                    # First line is the header
+                    header = lines[0].strip()
+                    if not header:
+                        raise ImportError("CSV file has no headers")
+
+                    headers = [header]
+                    rows = []
+                    # Process remaining lines as single-column values
+                    for line in lines[1:]:
+                        cleaned_row = {header: line.strip()}
+                        rows.append(cleaned_row)
+                else:
+                    reader = csv.DictReader(io.StringIO(content), delimiter=delimiter)
+                    headers = reader.fieldnames
+
+                    if not headers:
+                        raise ImportError("CSV file has no headers")
+
+                    # Normalize headers: strip whitespace and handle None values
+                    headers = [
+                        h.strip() if h and isinstance(h, str) else "" for h in headers
+                    ]
+                    reader.fieldnames = headers
+
+                    rows = []
+                    for row in reader:
+                        # Strip whitespace from all keys and values, handle None values
+                        cleaned_row = {}
+                        for k, v in row.items():
+                            # Handle None keys
+                            key = k.strip() if k and isinstance(k, str) else ""
+                            # Handle None values
+                            value = v.strip() if v and isinstance(v, str) else ""
+                            cleaned_row[key] = value
+                        rows.append(cleaned_row)
 
                 return headers, rows
 
@@ -223,3 +262,105 @@ def _add_warning(self, message: str):
 
     def _add_message(self, message: str):
         self.result.add_message(message)
+
+    def _find_homeowner_association_by_name(
+        self,
+        hoa_name: str,
+        row_number: int,
+        skip_hoa_api: bool = False,
+    ) -> Optional["HomeownerAssociation"]:
+        """
+        Find HomeownerAssociation by exact name match with optional DSO API fallback.
+
+        Args:
+            hoa_name: The HOA name to look up (e.g. a 'Statutaire Naam' column)
+            row_number: The current row number for returning error/warning messages
+            skip_hoa_api: If True, skip fetching from DSO API even if HOA not found
+
+        Returns:
+            HomeownerAssociation instance if found/created, None otherwise
+        """
+        if not hoa_name:
+            return None
+
+        try:
+            # Import here to avoid circular imports
+            from apps.homeownerassociation.models import HomeownerAssociation
+
+            # Try exact name match first
+            hoa = HomeownerAssociation.objects.filter(name=hoa_name).first()
+            if hoa:
+                return hoa
+
+            # HOA not found in database, try to fetch from DSO API and create it
+            if not self.dry_run and not skip_hoa_api:
+                try:
+
+                    # Create a temporary instance to use the `_get_hoa_data` method
+                    temp_hoa = HomeownerAssociation()
+                    data = temp_hoa._get_hoa_data(hoa_name)
+
+                    # Check if we got valid data (`response` should not be empty)
+                    if not data.get("response"):
+                        raise ValueError("No data returned from external API")
+
+                    # Create the HOA with data from external API
+                    with transaction.atomic():
+                        hoa = HomeownerAssociation.objects.create(
+                            name=data["hoa_name"],
+                            build_year=data["build_year"],
+                            number_of_apartments=data["number_of_apartments"],
+                            district=data["district"],
+                            neighborhood=data["neighborhood"],
+                            wijk=data["wijk"],
+                            zip_code=data["zip_code"],
+                            monument_status=data["monument_status"],
+                            ligt_in_beschermd_gebied=data["ligt_in_beschermd_gebied"],
+                            beschermd_stadsdorpsgezicht=data[
+                                "beschermd_stadsdorpsgezicht"
+                            ],
+                            kvk_nummer=data["kvk_nummer"],
+                        )
+                        # Create ownerships from the API response
+                        temp_hoa._create_ownerships(data["response"], hoa)
+
+                    self._add_message(
+                        f"Row {row_number}: Created new HOA '{hoa_name}' from external API"
+                    )
+                    return hoa
+                except (IndexError, KeyError, ValueError) as e:
+                    # API returned empty or invalid data
+                    self._add_warning(
+                        f"Row {row_number}: Could not fetch HOA data from external API for '{hoa_name}': {str(e)}"
+                    )
+                except Exception as e:
+                    # Other API or creation errors
+                    self._add_warning(
+                        f"Row {row_number}: Error creating HOA '{hoa_name}' from external API: {str(e)}"
+                    )
+            else:
+                # In dry-run mode, try to fetch data but don't create (unless `skip_hoa_api` is set)
+                if not skip_hoa_api:
+                    try:
+                        temp_hoa = HomeownerAssociation()
+                        data = temp_hoa._get_hoa_data(hoa_name)
+                        if data.get("response"):
+                            self._add_message(
+                                f"Row {row_number}: [DRY RUN] Would create new HOA '{hoa_name}' from external API"
+                            )
+                        else:
+                            self._add_warning(
+                                f"Row {row_number}: [DRY RUN] No data available from external API for '{hoa_name}'"
+                            )
+                    except Exception as e:
+                        self._add_warning(
+                            f"Row {row_number}: [DRY RUN] Could not fetch HOA data from external API for '{hoa_name}': {str(e)}"
+                        )
+                # Return None in dry-run since we're not actually creating it
+                return None
+        except Exception as e:
+            self._add_warning(
+                f"Row {row_number}: Error looking up HOA by name '{hoa_name}': {str(e)}"
+            )
+
+        return None

app/apps/homeownerassociation/importers/contact_importer.py

@@ -72,84 +72,13 @@ def _find_homeowner_association(
                     f"Row {row_number}: Error looking up case {legacy_id}: {str(e)}"
                 )
 
-        # Try to find by `Statutaire Naam` (HomeownerAssociation `name`, exact match)
+        # Try to find by `Statutaire Naam` (HomeownerAssociation `name`, exact match with DSO API fallback)
         if hoa_name:
-            try:
-                hoa = HomeownerAssociation.objects.filter(name=hoa_name).first()
-                if hoa:
-                    return hoa
-
-                # HOA not found in database, try to fetch from DSO API and create it
-                if not self.dry_run and not self.skip_hoa_api:
-                    try:
-                        # Create a temporary instance to use the `_get_hoa_data` method
-                        temp_hoa = HomeownerAssociation()
-                        data = temp_hoa._get_hoa_data(hoa_name)
-
-                        # Check if we got valid data (`response` should not be empty)
-                        if not data.get("response"):
-                            raise ValueError("No data returned from external API")
-
-                        # Create the HOA with data from external API
-                        with transaction.atomic():
-                            hoa = HomeownerAssociation.objects.create(
-                                name=data["hoa_name"],
-                                build_year=data["build_year"],
-                                number_of_apartments=data["number_of_apartments"],
-                                district=data["district"],
-                                neighborhood=data["neighborhood"],
-                                wijk=data["wijk"],
-                                zip_code=data["zip_code"],
-                                monument_status=data["monument_status"],
-                                ligt_in_beschermd_gebied=data[
-                                    "ligt_in_beschermd_gebied"
-                                ],
-                                beschermd_stadsdorpsgezicht=data[
-                                    "beschermd_stadsdorpsgezicht"
-                                ],
-                                kvk_nummer=data["kvk_nummer"],
-                            )
-                            # Create ownerships from the API response
-                            temp_hoa._create_ownerships(data["response"], hoa)
-
-                        self._add_message(
-                            f"Row {row_number}: Created new HOA '{hoa_name}' from external API"
-                        )
-                        return hoa
-                    except (IndexError, KeyError, ValueError) as e:
-                        # API returned empty or invalid data
-                        self._add_warning(
-                            f"Row {row_number}: Could not fetch HOA data from external API for '{hoa_name}': {str(e)}"
-                        )
-                    except Exception as e:
-                        # Other API or creation errors
-                        self._add_warning(
-                            f"Row {row_number}: Error creating HOA '{hoa_name}' from external API: {str(e)}"
-                        )
-                else:
-                    # In dry-run mode, try to fetch data but don't create (unless `skip_hoa_api` is set)
-                    if not self.skip_hoa_api:
-                        try:
-                            temp_hoa = HomeownerAssociation()
-                            data = temp_hoa._get_hoa_data(hoa_name)
-                            if data.get("response"):
-                                self._add_message(
-                                    f"Row {row_number}: [DRY RUN] Would create new HOA '{hoa_name}' from external API"
-                                )
-                            else:
-                                self._add_warning(
-                                    f"Row {row_number}: [DRY RUN] No data available from external API for '{hoa_name}'"
-                                )
-                        except Exception as e:
-                            self._add_warning(
-                                f"Row {row_number}: [DRY RUN] Could not fetch HOA data from external API for '{hoa_name}': {str(e)}"
-                            )
-                    # Return None in dry-run since we're not actually creating it
-                    return None
-            except Exception as e:
-                self._add_warning(
-                    f"Row {row_number}: Error looking up HOA by name '{hoa_name}': {str(e)}"
-                )
+            hoa = self._find_homeowner_association_by_name(
+                hoa_name, row_number, skip_hoa_api=self.skip_hoa_api
+            )
+            if hoa:
+                return hoa
 
         return None