Org membership importer

Author: alexiri

astra_app/core/admin.py

@@ -0,0 +1,67 @@
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from django.core.paginator import Paginator
+
+
+def build_import_preview_context(
+    *,
+    valid_rows: Sequence[Any],
+    request_get: Mapping[str, Any],
+    instance_decision_attr: str,
+    per_page_default: int = 50,
+    per_page_min: int = 50,
+) -> dict[str, Any]:
+    matches: list[Any] = []
+    skipped: list[Any] = []
+    match_row_numbers: list[int] = []
+
+    for index, row_result in enumerate(valid_rows, start=1):
+        instance = row_result.instance if hasattr(row_result, "instance") else None
+        import_type = str(getattr(row_result, "import_type", "") or "")
+        if import_type:
+            is_match = import_type != "skip"
+        elif instance is not None and hasattr(instance, instance_decision_attr):
+            is_match = getattr(instance, instance_decision_attr) == "IMPORT"
+        else:
+            is_match = False
+
+        row_number = getattr(row_result, "number", None)
+        if row_number is None:
+            row_number = getattr(getattr(row_result, "row", None), "number", None)
+        if row_number is None:
+            row_number = getattr(getattr(row_result, "original", None), "number", None)
+        if row_number is None:
+            row_number = index
+
+        try:
+            astra_row_number = int(row_number)
+        except (TypeError, ValueError):
+            astra_row_number = index
+        row_result.astra_row_number = astra_row_number
+
+        if is_match:
+            matches.append(row_result)
+            match_row_numbers.append(astra_row_number)
+        else:
+            skipped.append(row_result)
+
+    try:
+        per_page = int(str(request_get.get("per_page", str(per_page_default))))
+    except ValueError:
+        per_page = per_page_default
+    per_page = max(per_page, per_page_min)
+
+    matches_page_obj = Paginator(matches, per_page).get_page(request_get.get("matches_page") or "1")
+    skipped_page_obj = Paginator(skipped, per_page).get_page(request_get.get("skipped_page") or "1")
+
+    return {
+        "matches_page_obj": matches_page_obj,
+        "skipped_page_obj": skipped_page_obj,
+        "match_row_numbers": sorted(set(match_row_numbers)),
+        "preview_summary": {
+            "total": len(valid_rows),
+            "to_import": len(matches),
+            "skipped": len(skipped),
+        },
+    }

astra_app/core/admin_import_preview_utils.py

@@ -1,16 +1,94 @@
 import csv
 import datetime
 import io
+import secrets
+from collections.abc import Mapping, Sequence
+from typing import Any
 
+from dateutil import parser
+from django import forms
+from django.core.cache import cache
 from django.core.files.uploadedfile import UploadedFile
+from django.urls import reverse
+from tablib import Dataset
 
 from core.views_utils import _normalize_str
 
+AUTO_DETECT_CHOICE: tuple[str, str] = ("", "Auto-detect")
+
+
+def build_csv_header_choices(headers: Sequence[str]) -> list[tuple[str, str]]:
+    return [AUTO_DETECT_CHOICE, *[(header, header) for header in headers]]
+
+
+def set_form_column_field_choices(
+    *,
+    form: forms.Form,
+    field_names: Sequence[str],
+    headers: Sequence[str],
+) -> None:
+    choices = build_csv_header_choices(headers)
+    for field_name in field_names:
+        if field_name in form.fields:
+            form.fields[field_name].choices = choices
+
 
 def norm_csv_header(value: str) -> str:
     return "".join(ch for ch in value.strip().lower() if ch.isalnum())
 
 
+def resolve_column_header(
+    field_name: str,
+    headers: Sequence[str],
+    header_by_norm: Mapping[str, str],
+    column_overrides: Mapping[str, str],
+    *fallback_norms: str,
+) -> str | None:
+    override = _normalize_str(column_overrides.get(field_name, ""))
+    if override:
+        if override in headers:
+            return override
+
+        override_norm = norm_csv_header(override)
+        resolved_override = header_by_norm.get(override_norm)
+        if resolved_override:
+            return resolved_override
+
+        raise ValueError(f"Column '{override}' not found in CSV headers")
+
+    for fallback in fallback_norms:
+        fallback_norm = norm_csv_header(fallback)
+        resolved = header_by_norm.get(fallback_norm)
+        if resolved:
+            return resolved
+    return None
+
+
+def attach_unmatched_csv_to_result(
+    result: Any,
+    dataset: Dataset,
+    cache_key_prefix: str,
+    reverse_url_name: str,
+) -> None:
+    token = secrets.token_urlsafe(16)
+    cache_key = f"{cache_key_prefix}:{token}"
+    csv_content = dataset.export("csv")
+    cache.set(cache_key, csv_content, timeout=60 * 60)
+
+    download_url = reverse(reverse_url_name, kwargs={"token": token})
+    # `Result` is a third-party import-export type with no extension hook;
+    # dynamic attributes are used as a lightweight duck-typed contract.
+    setattr(result, "unmatched_csv_content", csv_content)
+    setattr(result, "unmatched_download_url", download_url)
+
+
+def sanitize_csv_cell(value: str) -> str:
+    """Prefix formula-starting characters to prevent spreadsheet formula injection."""
+    if value and value[0] in ("=", "+", "-", "@", "\t", "\r"):
+        return f"'{value}"
+    return value
+
+
 def normalize_csv_email(value: object) -> str:
     return _normalize_str(value).lower()
 
@@ -27,24 +105,17 @@ def parse_csv_bool(value: object) -> bool:
     return normalized in {"1", "y", "yes", "true", "t", "active", "activemember", "active member"}
 
 
-def parse_csv_date(value: object) -> datetime.datetime | None:
+def parse_csv_date(value: object) -> datetime.date | None:
     raw = _normalize_str(value)
     if not raw:
         return None
 
-    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%m/%d/%Y", "%m/%d/%y"):
-        try:
-            day = datetime.datetime.strptime(raw, fmt).date()
-        except ValueError:
-            continue
-        return datetime.datetime.combine(day, datetime.time(0, 0, 0), tzinfo=datetime.UTC)
-
     try:
-        day = datetime.date.fromisoformat(raw)
-    except ValueError:
+        parsed = parser.parse(raw, dayfirst=False, yearfirst=False)
+    except (parser.ParserError, TypeError, ValueError, OverflowError):
         return None
 
-    return datetime.datetime.combine(day, datetime.time(0, 0, 0), tzinfo=datetime.UTC)
+    return parsed.date()
 
 
 def extract_csv_headers_from_uploaded_file(uploaded: UploadedFile) -> list[str]: