Org importer

Author: alexiri

astra_app/core/address_geocoding.py

@@ -0,0 +1,105 @@
+import json
+import logging
+import hashlib
+from typing import Any
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+from django.core.cache import cache
+
+from core.views_utils import _normalize_str
+
+logger = logging.getLogger(__name__)
+
+_PHOTON_ADDRESS_CACHE_PREFIX = "photon_address_parts:"
+_PHOTON_ADDRESS_CACHE_TTL_SECONDS = 24 * 60 * 60
+_PHOTON_MAX_ATTEMPTS = 3
+
+
+def _photon_address_parts_from_feature(feature: dict[str, Any]) -> dict[str, str] | None:
+    properties = feature.get("properties")
+    if not isinstance(properties, dict):
+        return None
+
+    street_name = _normalize_str(properties.get("street") or properties.get("name"))
+    house_number = _normalize_str(properties.get("housenumber"))
+    street = f"{street_name} {house_number}".strip()
+
+    city = _normalize_str(
+        properties.get("city")
+        or properties.get("town")
+        or properties.get("village")
+        or properties.get("hamlet")
+        or properties.get("locality")
+    )
+    state = _normalize_str(properties.get("state"))
+    postal_code = _normalize_str(properties.get("postcode"))
+    country_code = _normalize_str(properties.get("countrycode")).upper()
+
+    result = {
+        "street": street,
+        "city": city,
+        "state": state,
+        "postal_code": postal_code,
+        "country_code": country_code,
+    }
+    if not any(result.values()):
+        return None
+
+    return {key: value for key, value in result.items() if value}
+
+
+def decompose_full_address_with_photon(full_address: str, *, timeout_seconds: int = 5) -> dict[str, str]:
+    query = _normalize_str(full_address)
+    if not query:
+        return {}
+
+    normalized_query = query.lower()
+    query_hash = hashlib.sha256(normalized_query.encode("utf-8")).hexdigest()
+    cache_key = f"{_PHOTON_ADDRESS_CACHE_PREFIX}{query_hash}"
+    cached = cache.get(cache_key)
+    if isinstance(cached, dict):
+        return cached
+
+    url = f"https://photon.komoot.io/api/?{urlencode({'q': query, 'limit': 1})}"
+    request = Request(url, headers={"User-Agent": "astra-address-import/1.0"})
+    payload: dict[str, Any] | None = None
+    last_error: Exception | None = None
+    for attempt in range(1, _PHOTON_MAX_ATTEMPTS + 1):
+        try:
+            with urlopen(request, timeout=timeout_seconds) as response:
+                parsed = json.loads(response.read().decode("utf-8"))
+                if isinstance(parsed, dict):
+                    payload = parsed
+                else:
+                    payload = {}
+                break
+        except Exception as exc:
+            last_error = exc
+            logger.warning(
+                "Photon geocoding attempt failed query=%r attempt=%d/%d error=%s",
+                query,
+                attempt,
+                _PHOTON_MAX_ATTEMPTS,
+                exc,
+            )
+
+    if payload is None:
+        if last_error is not None:
+            logger.error("Photon geocoding failed for full_address lookup: %s", last_error)
+        cache.set(cache_key, {}, timeout=_PHOTON_ADDRESS_CACHE_TTL_SECONDS)
+        return {}
+
+    features = payload.get("features") if isinstance(payload, dict) else None
+    if not isinstance(features, list) or not features:
+        cache.set(cache_key, {}, timeout=_PHOTON_ADDRESS_CACHE_TTL_SECONDS)
+        return {}
+
+    first_feature = features[0]
+    if not isinstance(first_feature, dict):
+        cache.set(cache_key, {}, timeout=_PHOTON_ADDRESS_CACHE_TTL_SECONDS)
+        return {}
+
+    result = _photon_address_parts_from_feature(first_feature) or {}
+    cache.set(cache_key, result, timeout=_PHOTON_ADDRESS_CACHE_TTL_SECONDS)
+    return result

astra_app/core/admin.py

@@ -50,6 +50,15 @@
     MembershipCSVImportResource,
 )
 from core.organization_claim import make_organization_claim_token
+from core.organization_csv_import import (
+    _ORG_COLUMN_FIELDS,
+    OrganizationCSVConfirmImportForm,
+    OrganizationCSVImportForm,
+    OrganizationCSVImportResource,
+    iter_representative_selection_items,
+    optional_organization_csv_columns,
+    required_organization_csv_columns,
+)
 from core.protected_resources import protected_freeipa_group_cns
 from core.user_labels import user_choice, user_choice_with_fallback, user_choices_from_users
 from core.views_utils import _normalize_str
@@ -81,6 +90,7 @@
     MembershipType,
     MembershipTypeCategory,
     Organization,
+    OrganizationCSVImportLink,
     VotingCredential,
 )
 
@@ -1831,6 +1841,137 @@ def changelist_view(self, request: HttpRequest, extra_context: dict[str, Any] |
         return redirect("admin:core_membershipcsvimportlink_import")
 
 
+@admin.register(OrganizationCSVImportLink)
+class OrganizationCSVImportLinkAdmin(ImportMixin, admin.ModelAdmin):
+    """Admin entry for the organization CSV importer (django-import-export)."""
+
+    import_form_class = OrganizationCSVImportForm
+    confirm_form_class = OrganizationCSVConfirmImportForm
+    import_template_name = "admin/core/organization_csv_import.html"
+    resource_classes = [OrganizationCSVImportResource]
+
+    @override
+    def has_add_permission(self, request: HttpRequest) -> bool:
+        return False
+
+    @override
+    def get_import_formats(self) -> list[type[base_formats.Format]]:
+        return [base_formats.CSV]
+
+    @override
+    def has_delete_permission(self, request: HttpRequest, obj: object | None = None) -> bool:
+        return False
+
+    @override
+    def has_change_permission(self, request: HttpRequest, obj: object | None = None) -> bool:
+        return False
+
+    @override
+    def has_view_permission(self, request: HttpRequest, obj: object | None = None) -> bool:
+        return bool(request.user.is_active and request.user.is_staff)
+
+    @override
+    def get_model_perms(self, request: HttpRequest) -> dict[str, bool]:
+        if not self.has_view_permission(request):
+            return {}
+        return {"view": True}
+
+    @override
+    def get_confirm_form_initial(self, request: HttpRequest, import_form: forms.Form) -> dict[str, Any]:
+        initial = super().get_confirm_form_initial(request, import_form)
+        if import_form is None:
+            return initial
+
+        for key in _ORG_COLUMN_FIELDS:
+            value = import_form.cleaned_data.get(key, "")
+            if value:
+                initial[key] = value
+        return initial
+
+    @override
+    def get_import_resource_kwargs(self, request: HttpRequest, **kwargs: Any) -> dict[str, Any]:
+        form = kwargs.get("form")
+        if form is None:
+            raise ValueError("Missing import form")
+
+        cleaned_data = getattr(form, "cleaned_data", None)
+        extra: dict[str, Any] = {}
+        if isinstance(cleaned_data, dict):
+            for key in _ORG_COLUMN_FIELDS:
+                value = cleaned_data.get(key, "")
+                if value:
+                    extra[key] = value
+
+        representative_selections = iter_representative_selection_items(request.POST.items())
+
+        return {
+            "actor_username": request.user.get_username(),
+            "representative_selections": representative_selections,
+            **extra,
+        }
+
+    @override
+    def import_action(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
+        response = super().import_action(request, *args, **kwargs)
+
+        if not isinstance(response, TemplateResponse) or response.context_data is None:
+            return response
+
+        response.context_data["required_csv_columns"] = required_organization_csv_columns()
+        response.context_data["optional_csv_columns"] = optional_organization_csv_columns()
+
+        result = response.context_data.get("result")
+        confirm_form = response.context_data.get("confirm_form")
+        if result is None or confirm_form is None:
+            return response
+
+        valid_rows_obj = getattr(result, "valid_rows", None)
+        if callable(valid_rows_obj):
+            valid_rows = list(valid_rows_obj() or [])
+        else:
+            valid_rows = list(valid_rows_obj or [])
+
+        matches: list[Any] = []
+        skipped: list[Any] = []
+
+        for idx, row_result in enumerate(valid_rows, start=1):
+            instance = row_result.instance if hasattr(row_result, "instance") else None
+            import_type = row_result.import_type if hasattr(row_result, "import_type") else ""
+            if import_type:
+                is_match = import_type != "skip"
+            elif instance is not None and hasattr(instance, "decision"):
+                is_match = instance.decision == "IMPORT"
+            else:
+                is_match = False
+
+            number = getattr(row_result, "number", None)
+            if number is None:
+                number = idx
+            try:
+                row_result.astra_row_number = int(number)
+            except (TypeError, ValueError):
+                row_result.astra_row_number = idx
+
+            if is_match:
+                matches.append(row_result)
+            else:
+                skipped.append(row_result)
+
+        response.context_data["preview_summary"] = {
+            "total": len(valid_rows),
+            "to_import": len(matches),
+            "skipped": len(skipped),
+        }
+        response.context_data["matches_page_obj"] = Paginator(matches, 50).get_page(request.GET.get("matches_page") or "1")
+        response.context_data["skipped_page_obj"] = Paginator(skipped, 50).get_page(request.GET.get("skipped_page") or "1")
+
+        return response
+
+    @override
+    def changelist_view(self, request: HttpRequest, extra_context: dict[str, Any] | None = None) -> HttpResponse:
+        return redirect("admin:core_organizationcsvimportlink_import")
+
+
 @admin.register(Organization)
 class OrganizationAdmin(admin.ModelAdmin):
     class OrganizationAdminForm(forms.ModelForm):

astra_app/core/csv_import_utils.py

@@ -0,0 +1,70 @@
+import csv
+import datetime
+import io
+
+from django.core.files.uploadedfile import UploadedFile
+
+from core.views_utils import _normalize_str
+
+
+def norm_csv_header(value: str) -> str:
+    return "".join(ch for ch in value.strip().lower() if ch.isalnum())
+
+
+def normalize_csv_email(value: object) -> str:
+    return _normalize_str(value).lower()
+
+
+def normalize_csv_name(value: object) -> str:
+    raw = _normalize_str(value).lower()
+    return "".join(ch for ch in raw if ch.isalnum())
+
+
+def parse_csv_bool(value: object) -> bool:
+    normalized = _normalize_str(value).lower()
+    if not normalized:
+        return False
+    return normalized in {"1", "y", "yes", "true", "t", "active", "activemember", "active member"}
+
+
+def parse_csv_date(value: object) -> datetime.datetime | 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:
+        return None
+
+    return datetime.datetime.combine(day, datetime.time(0, 0, 0), tzinfo=datetime.UTC)
+
+
+def extract_csv_headers_from_uploaded_file(uploaded: UploadedFile) -> list[str]:
+    uploaded.seek(0)
+    sample = uploaded.read(64 * 1024)
+    uploaded.seek(0)
+
+    try:
+        text = sample.decode("utf-8-sig")
+    except UnicodeDecodeError:
+        text = sample.decode("utf-8", errors="replace")
+
+    if not text.strip():
+        return []
+
+    try:
+        dialect = csv.Sniffer().sniff(text, delimiters=",;\t|")
+    except Exception:
+        dialect = csv.excel
+
+    reader = csv.reader(io.StringIO(text), dialect)
+    headers = next(reader, [])
+    return [h.strip() for h in headers if str(h).strip()]