from __future__ import annotations import base64 import binascii import csv import copy import hashlib import json from collections import Counter from datetime import UTC, datetime from io import BytesIO, StringIO from typing import Any from sqlalchemy import and_, false, func, or_, select from sqlalchemy.orm import Session, selectinload from sqlalchemy.orm.attributes import set_committed_value from govoplan_addresses.backend.db.models import ( AddressImportProfile, AddressImportRun, AddressListEntry, Contact, ContactChannelRule, ContactEmail, ContactPhone, ContactPointQualityDecision, ContactPostalAddress, ) from govoplan_addresses.backend.import_schemas import ( AddressImportConfiguration, AddressImportPreviewRequest, AddressImportProfileCreateRequest, AddressImportProfileUpdateRequest, AddressImportRollbackRequest, ) from govoplan_addresses.backend.ldif import parse_ldif_rows from govoplan_addresses.backend.schemas import ( ContactCreateRequest, ContactEmailPayload, ContactPhonePayload, ContactPostalAddressPayload, ContactUpdateRequest, ) from govoplan_addresses.backend.service import ( AddressBookError, _contact_change_payload, _record_address_contact_change, _require_mutable_book, _replace_emails, _replace_phones, _replace_postal_addresses, create_contact, delete_contact, get_visible_address_book, get_visible_contact, restore_contact, update_contact, ) from govoplan_core.auth import ApiPrincipal from govoplan_core.db.base import utcnow MAX_IMPORT_BYTES = 10_000_000 MAX_IMPORT_COLUMNS = 200 CONTACT_LOOKUP_BATCH_SIZE = 250 ROLLBACK_SNAPSHOT_VERSION = 2 CONTACT_POINTS = { "emails": (ContactEmail, ("email",), _replace_emails), "phones": (ContactPhone, ("phone",), _replace_phones), "postal_addresses": ( ContactPostalAddress, ("street", "postal_code", "locality", "region", "country"), _replace_postal_addresses, ), } def _account_id(principal: ApiPrincipal) -> str: return principal.account_id def _tenant_id(principal: ApiPrincipal) -> str: return principal.tenant_id def _profile_scope_predicate(principal: ApiPrincipal): tenant_id = _tenant_id(principal) predicates = [AddressImportProfile.scope_type == "system"] predicates.extend( [ and_(AddressImportProfile.tenant_id == tenant_id, AddressImportProfile.scope_type == "tenant"), and_( AddressImportProfile.tenant_id == tenant_id, AddressImportProfile.scope_type == "user", AddressImportProfile.scope_id == _account_id(principal), ), ] ) group_ids = tuple(principal.group_ids) if group_ids: predicates.append( and_( AddressImportProfile.tenant_id == tenant_id, AddressImportProfile.scope_type == "group", AddressImportProfile.scope_id.in_(group_ids), ) ) return or_(*predicates) if predicates else false() def list_import_profiles( session: Session, principal: ApiPrincipal, *, include_history: bool = False, ) -> list[AddressImportProfile]: query = session.query(AddressImportProfile).filter(_profile_scope_predicate(principal)) if not include_history: query = query.filter(AddressImportProfile.is_current.is_(True)) return query.order_by(AddressImportProfile.name.asc(), AddressImportProfile.version.desc()).all() def get_import_profile( session: Session, principal: ApiPrincipal, profile_id: str, ) -> AddressImportProfile: profile = ( session.query(AddressImportProfile) .filter(_profile_scope_predicate(principal), AddressImportProfile.id == profile_id) .one_or_none() ) if profile is None: raise AddressBookError("Address import profile not found.") return profile def create_import_profile( session: Session, principal: ApiPrincipal, payload: AddressImportProfileCreateRequest, ) -> AddressImportProfile: tenant_id, scope_id = _validated_profile_scope(principal, payload.scope_type, payload.scope_id) profile = AddressImportProfile( tenant_id=tenant_id, scope_type=payload.scope_type, scope_id=scope_id, name=payload.name.strip(), description=_trim(payload.description), source_format=payload.source_format, configuration=payload.configuration.model_dump(mode="json"), is_current=True, created_by_account_id=_account_id(principal), ) session.add(profile) return profile def update_import_profile( session: Session, principal: ApiPrincipal, profile_id: str, payload: AddressImportProfileUpdateRequest, ) -> AddressImportProfile: current = get_import_profile(session, principal, profile_id) if not current.is_current: raise AddressBookError("Only the current import profile version can be updated.") current.is_current = False current.superseded_at = utcnow() next_profile = AddressImportProfile( profile_key=current.profile_key, version=current.version + 1, tenant_id=current.tenant_id, scope_type=current.scope_type, scope_id=current.scope_id, name=(payload.name.strip() if payload.name is not None else current.name), description=(payload.description.strip() or None if payload.description is not None else current.description), source_format=current.source_format, configuration=( payload.configuration.model_dump(mode="json") if payload.configuration is not None else dict(current.configuration or {}) ), is_current=True, created_by_account_id=_account_id(principal), ) session.add(next_profile) return next_profile def retire_import_profile( session: Session, principal: ApiPrincipal, profile_id: str, ) -> None: profile = get_import_profile(session, principal, profile_id) if profile.scope_type == "system" and not principal.has("addresses:address_book:admin"): raise AddressBookError("System import profiles require address-book administration permission.") profile.is_current = False profile.superseded_at = utcnow() def preview_address_import( session: Session, principal: ApiPrincipal, address_book_id: str, payload: AddressImportPreviewRequest, ) -> AddressImportRun: book = get_visible_address_book(session, principal, address_book_id) if book.read_only: raise AddressBookError("Static imports require a writable address book.") profile = get_import_profile(session, principal, payload.profile_id) raw = _decode_payload(payload.content_base64) input_hash = hashlib.sha256(raw).hexdigest() config = AddressImportConfiguration.model_validate(profile.configuration) rows, parse_diagnostics = _parse_rows( raw, filename=payload.filename, source_format=profile.source_format, config=config, ) plan_data, map_diagnostics = _plan_rows( session, book_id=book.id, profile=profile, input_hash=input_hash, rows=rows, config=config, source_format=profile.source_format, ) diagnostics = [*parse_diagnostics, *map_diagnostics] statistics = dict(Counter(item["action"] for item in plan_data)) statistics["rows"] = len(rows) statistics["errors"] = sum(item["severity"] == "error" for item in diagnostics) statistics["warnings"] = sum(item["severity"] == "warning" for item in diagnostics) plan_hash = _hash_json( { "profile_id": profile.id, "profile_version": profile.version, "address_book_id": book.id, "input_hash": input_hash, "plan": plan_data, } ) run = AddressImportRun( tenant_id=book.tenant_id, address_book_id=book.id, profile_id=profile.id, source_filename=payload.filename.strip(), source_format=profile.source_format, input_hash=input_hash, plan_hash=plan_hash, status="previewed", row_count=len(rows), statistics=statistics, diagnostics=diagnostics, plan_data=plan_data, result_evidence={}, created_by_account_id=_account_id(principal), ) session.add(run) session.flush() return run def get_import_run( session: Session, principal: ApiPrincipal, run_id: str, *, lock: bool = False, ) -> AddressImportRun: visible_book_ids = [book.id for book in _visible_import_books(session, principal)] if not visible_book_ids: raise AddressBookError("Address import run not found.") query = ( session.query(AddressImportRun) .filter(AddressImportRun.id == run_id, AddressImportRun.address_book_id.in_(visible_book_ids)) ) if lock: query = query.populate_existing().with_for_update() run = query.one_or_none() if run is None: raise AddressBookError("Address import run not found.") return run def apply_address_import( session: Session, principal: ApiPrincipal, run_id: str, *, expected_plan_hash: str, ) -> AddressImportRun: run = get_import_run(session, principal, run_id, lock=True) if run.plan_hash != expected_plan_hash: raise AddressBookError("The reviewed import plan changed; create a new preview.") if run.status == "applied": return run if run.status != "previewed": raise AddressBookError(f"Import run cannot be applied from status {run.status!r}.") if any(item.get("severity") == "error" for item in run.diagnostics or []): raise AddressBookError("Import plans with error diagnostics cannot be applied.") if any(item.get("action") == "conflict" for item in run.plan_data or []): raise AddressBookError("Resolve import conflicts by correcting the file or mapping profile and preview again.") created_ids: list[str] = [] updated: list[dict[str, Any]] = [] # SQLAlchemy JSON columns do not track nested mutations. Keep the persisted # preview untouched until assigning a genuinely changed complete plan. applied_plan = copy.deepcopy(run.plan_data or []) for item in applied_plan: action = item.get("action") if action in {"ignored", "unchanged"}: continue source_ref = str(item["source_ref"]) existing = _contact_by_source_ref(session, run.address_book_id, source_ref, lock=True) if action == "create": if existing is not None and existing.deleted_at is None: raise AddressBookError("A target contact appeared after preview; preview the import again.") contact = create_contact( session, principal, run.address_book_id, ContactCreateRequest.model_validate(item["payload"]), ) _stamp_import_contact(contact, run=run, item=item) session.flush() created_ids.append(contact.id) item["contact_id"] = contact.id item["after_hash"] = _contact_hash(contact) elif action == "update": if existing is None: raise AddressBookError("An import target disappeared after preview; preview the import again.") _lock_contact_points(session, existing) if _contact_hash(existing) != item.get("expected_contact_hash"): raise AddressBookError( f'Contact "{existing.display_name}" changed after preview; preview the import again.' ) _require_mutable_book(existing.address_book) before = { "version": ROLLBACK_SNAPSHOT_VERSION, "contact": copy.deepcopy(_contact_snapshot(existing)), "deleted_at": _deleted_at_value(existing.deleted_at), } update_payload = ContactUpdateRequest.model_validate(item["payload"]) points = _prepare_import_points(session, existing, update_payload) if existing.deleted_at is not None: restore_contact(session, principal, existing.id) _apply_import_points(existing, points) contact = update_contact( session, principal, existing.id, _without_contact_points(update_payload), ) _stamp_import_contact(contact, run=run, item=item) session.flush() updated.append({"contact_id": contact.id, "before": before, "after_hash": _contact_hash(contact)}) item["contact_id"] = contact.id run.status = "applied" run.applied_at = utcnow() run.plan_data = applied_plan run.result_evidence = { "input_hash": run.input_hash, "plan_hash": run.plan_hash, "created_contact_ids": created_ids, "updated_contacts": updated, "applied_by_account_id": _account_id(principal), "applied_at": run.applied_at.isoformat(), } return run def rollback_address_import( session: Session, principal: ApiPrincipal, run_id: str, payload: AddressImportRollbackRequest, ) -> AddressImportRun: run = get_import_run(session, principal, run_id, lock=True) if run.plan_hash != payload.expected_plan_hash: raise AddressBookError("The reviewed import plan changed; reload the import run.") if run.status == "rolled_back": return run if run.status != "applied": raise AddressBookError("Only an applied import can be rolled back.") evidence = dict(run.result_evidence or {}) updated = list(evidence.get("updated_contacts") or []) created_ids = list(evidence.get("created_contact_ids") or []) # Validate every before-image before touching any contact. Older runs did # not record deletion state, so automatic recovery cannot infer it safely. before_images = { str(item["contact_id"]): _validated_rollback_snapshot(item.get("before")) for item in updated } expected_hashes = { str(item["contact_id"]): str(item["after_hash"]) for item in updated } expected_hashes.update( { str(item["contact_id"]): str(item["after_hash"]) for item in run.plan_data or [] if item.get("contact_id") in created_ids and item.get("after_hash") } ) if not set(created_ids).issubset(expected_hashes): raise AddressBookError("Import rollback evidence is incomplete; automatic rollback is unsafe.") for contact_id, expected_hash in sorted(expected_hashes.items()): contact = get_visible_contact(session, principal, contact_id, include_deleted=True, lock=True) _lock_contact_points(session, contact) if contact.address_book_id != run.address_book_id: raise AddressBookError("An import target moved to another address book; automatic rollback is unsafe.") _require_mutable_book(contact.address_book) if _contact_hash(contact) != expected_hash: raise AddressBookError( f'Contact "{contact.display_name}" changed after import; automatic rollback is unsafe.' ) if contact_id in before_images: _validate_point_restoration(session, contact, before_images[contact_id][0]["points"]) for contact_id in created_ids: contact = get_visible_contact(session, principal, contact_id, include_deleted=True) if contact.deleted_at is None: delete_contact(session, principal, contact.id) for item in updated: contact = get_visible_contact(session, principal, str(item["contact_id"]), include_deleted=True) snapshot, previous_deleted_at = before_images[str(item["contact_id"])] if contact.deleted_at is not None: restore_contact(session, principal, contact.id) previous = _contact_change_payload(contact, prefix="previous_") _restore_contact_points(contact, snapshot["points"]) # A validated stored before-image is not a fresh user edit: do not trim, # normalize or coerce it through the generic update path a second time. for field in ("display_name", "given_name", "family_name", "organization", "role_title", "note", "tags"): setattr(contact, field, copy.deepcopy(snapshot["payload"][field])) for field in ("source_kind", "source_ref", "source_revision", "source_payload_kind", "source_payload_raw", "provenance"): setattr(contact, field, copy.deepcopy(snapshot[field])) contact.metadata_ = copy.deepcopy(snapshot["metadata"]) contact.updated_by_account_id = _account_id(principal) _record_address_contact_change(session, principal, contact=contact, operation="updated", previous=previous) if previous_deleted_at is not None: delete_contact(session, principal, contact.id) contact.deleted_at = previous_deleted_at run.status = "rolled_back" run.rolled_back_at = utcnow() run.result_evidence = { **evidence, "rollback_reason": payload.reason, "rolled_back_by_account_id": _account_id(principal), "rolled_back_at": run.rolled_back_at.isoformat(), } return run def import_run_payload(run: AddressImportRun) -> dict[str, Any]: diagnostics = list(run.diagnostics or []) effects = [ { "row_number": int(item["row_number"]), "action": item["action"], "source_key": item.get("source_key"), "contact_id": item.get("contact_id"), "display_name": item.get("display_name"), "changed_fields": list(item.get("changed_fields") or []), "message": item.get("message"), } for item in run.plan_data or [] ] can_apply = ( run.status == "previewed" and not any(item.get("severity") == "error" for item in diagnostics) and not any(item.get("action") == "conflict" for item in run.plan_data or []) ) evidence = dict(run.result_evidence or {}) public_evidence = { key: evidence[key] for key in ( "input_hash", "plan_hash", "applied_by_account_id", "applied_at", "rollback_reason", "rolled_back_by_account_id", "rolled_back_at", ) if evidence.get(key) is not None } if evidence: public_evidence["created_contact_count"] = len(evidence.get("created_contact_ids") or []) public_evidence["updated_contact_count"] = len(evidence.get("updated_contacts") or []) return { "id": run.id, "address_book_id": run.address_book_id, "profile_id": run.profile_id, "source_filename": run.source_filename, "source_format": run.source_format, "input_hash": run.input_hash, "plan_hash": run.plan_hash, "status": run.status, "row_count": run.row_count, "statistics": dict(run.statistics or {}), "diagnostics": diagnostics, "effects": effects, "can_apply": can_apply, # Full before-images remain private rollback evidence and must not be # projected through a normal import-run read response. "result_evidence": public_evidence, "created_at": run.created_at, "updated_at": run.updated_at, "applied_at": run.applied_at, "rolled_back_at": run.rolled_back_at, } def _validated_profile_scope( principal: ApiPrincipal, scope_type: str, requested_scope_id: str | None, ) -> tuple[str | None, str | None]: if scope_type == "system": if not principal.has("addresses:address_book:admin"): raise AddressBookError("System import profiles require address-book administration permission.") return None, None tenant_id = _tenant_id(principal) if scope_type == "tenant": return tenant_id, tenant_id if scope_type == "user": return tenant_id, _account_id(principal) if scope_type == "group": scope_id = _trim(requested_scope_id) if scope_id is None: raise AddressBookError("Group import profiles require a group id.") if scope_id not in principal.group_ids and not principal.has("addresses:address_book:admin"): raise AddressBookError("The selected group is not visible to the current principal.") return tenant_id, scope_id raise AddressBookError("Unsupported import profile scope.") def _decode_payload(encoded: str) -> bytes: try: raw = base64.b64decode(encoded, validate=True) except (binascii.Error, ValueError) as exc: raise AddressBookError("Import file content is not valid base64.") from exc if not raw: raise AddressBookError("Import file is empty.") if len(raw) > MAX_IMPORT_BYTES: raise AddressBookError(f"Import files are limited to {MAX_IMPORT_BYTES} bytes.") return raw def _parse_rows( raw: bytes, *, filename: str, source_format: str, config: AddressImportConfiguration, ) -> tuple[list[tuple[int, dict[str, Any]]], list[dict[str, Any]]]: if source_format == "csv": return _parse_csv(raw, config=config) if source_format == "xlsx": if not filename.casefold().endswith(".xlsx"): raise AddressBookError("XLSX imports require an .xlsx file; macros and legacy workbooks are not accepted.") return _parse_xlsx(raw, config=config) if source_format == "ldif": if not filename.casefold().endswith((".ldif", ".ldi")): raise AddressBookError("LDIF imports require an .ldif or .ldi file.") try: return parse_ldif_rows( raw, max_entries=config.max_rows, change_record_policy=config.ldif_change_record_policy, ) except ValueError as exc: raise AddressBookError(str(exc)) from exc raise AddressBookError(f"Unsupported address import format: {source_format!r}.") def _parse_csv( raw: bytes, *, config: AddressImportConfiguration, ) -> tuple[list[tuple[int, dict[str, str]]], list[dict[str, Any]]]: try: text = raw.decode(config.encoding) except UnicodeDecodeError as exc: raise AddressBookError(f"CSV is not valid {config.encoding}: {exc}.") from exc reader = csv.reader(StringIO(text), delimiter=config.delimiter) all_rows = list(reader) if len(all_rows) < config.header_row: raise AddressBookError("CSV does not contain the configured header row.") header = _headers(all_rows[config.header_row - 1]) result: list[tuple[int, dict[str, str]]] = [] for row_number, values in enumerate(all_rows[config.header_row :], start=config.header_row + 1): if not any(str(value).strip() for value in values): continue if len(values) > MAX_IMPORT_COLUMNS: raise AddressBookError(f"CSV row {row_number} exceeds the {MAX_IMPORT_COLUMNS}-column limit.") result.append((row_number, _row_dict(header, values))) if len(result) > config.max_rows: raise AddressBookError(f"CSV exceeds the configured {config.max_rows}-row limit.") return result, [] def _parse_xlsx( raw: bytes, *, config: AddressImportConfiguration, ) -> tuple[list[tuple[int, dict[str, str]]], list[dict[str, Any]]]: try: from openpyxl import load_workbook except ImportError as exc: # pragma: no cover - dependency/package failure raise AddressBookError("XLSX import support is not installed.") from exc try: workbook = load_workbook(BytesIO(raw), read_only=True, data_only=False, keep_links=False) except Exception as exc: raise AddressBookError(f"XLSX workbook could not be read: {exc}.") from exc if len(workbook.sheetnames) > 100: raise AddressBookError("XLSX workbooks are limited to 100 sheets.") if config.sheet_name: if config.sheet_name not in workbook.sheetnames: raise AddressBookError(f'XLSX sheet "{config.sheet_name}" was not found.') sheet = workbook[config.sheet_name] else: sheet = workbook[workbook.sheetnames[0]] rows = list(sheet.iter_rows(min_row=config.header_row, max_row=config.header_row)) if not rows: raise AddressBookError("XLSX does not contain the configured header row.") header = _headers([cell.value for cell in rows[0]]) result: list[tuple[int, dict[str, str]]] = [] for row_number, cells in enumerate(sheet.iter_rows(min_row=config.header_row + 1), start=config.header_row + 1): if len(cells) > MAX_IMPORT_COLUMNS: raise AddressBookError(f"XLSX row {row_number} exceeds the {MAX_IMPORT_COLUMNS}-column limit.") if any(cell.data_type == "f" for cell in cells): raise AddressBookError(f"XLSX row {row_number} contains a formula; formulas are never evaluated during import.") values = [cell.value for cell in cells] if not any(value is not None and str(value).strip() for value in values): continue result.append((row_number, _row_dict(header, values))) if len(result) > config.max_rows: raise AddressBookError(f"XLSX exceeds the configured {config.max_rows}-row limit.") return result, [] def _headers(values: list[Any]) -> list[str]: headers = [str(value).strip() if value is not None else "" for value in values] if not headers or not any(headers): raise AddressBookError("Import header row is empty.") if len(headers) > MAX_IMPORT_COLUMNS: raise AddressBookError(f"Import files are limited to {MAX_IMPORT_COLUMNS} columns.") blank = [index + 1 for index, value in enumerate(headers) if not value] if blank: raise AddressBookError(f"Import header contains blank column names at positions {blank}.") duplicates = sorted(name for name, count in Counter(headers).items() if count > 1) if duplicates: raise AddressBookError(f"Import header contains duplicate columns: {', '.join(duplicates)}.") return headers def _row_dict(headers: list[str], values: list[Any]) -> dict[str, str]: padded = [*values, *([None] * max(0, len(headers) - len(values)))] return { header: "" if value is None else str(value).strip() for header, value in zip(headers, padded, strict=False) } def _column_key(value: object, *, casefold: bool) -> str: column = str(value).strip() return column.casefold() if casefold else column def _row_value(row: dict[str, Any], column: str, *, casefold: bool) -> object: if column in row: return row[column] if not casefold: return "" expected = column.casefold() for key, value in row.items(): if key.casefold() == expected: return value return "" def _import_values(value: object) -> list[str]: raw_values = value if isinstance(value, list) else [value] return [str(item).strip() for item in raw_values if item is not None and str(item).strip()] def _first_import_value(value: object) -> str: values = _import_values(value) return values[0] if values else "" def _plan_rows( session: Session, *, book_id: str, profile: AddressImportProfile, input_hash: str, rows: list[tuple[int, dict[str, Any]]], config: AddressImportConfiguration, source_format: str, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: diagnostics: list[dict[str, Any]] = [] plan: list[dict[str, Any]] = [] casefold_columns = source_format == "ldif" headers = { _column_key(column, casefold=casefold_columns) for column in (rows[0][1] if rows else set()) if not str(column).startswith("__") } referenced_columns = { _column_key(column, casefold=casefold_columns) for column in config.field_mappings.values() } key_column = config.source_key_column or config.field_mappings["source_key"] normalized_key_column = _column_key(key_column, casefold=casefold_columns) if config.source_key_column: referenced_columns.add(_column_key(config.source_key_column, casefold=casefold_columns)) missing_columns = sorted(referenced_columns.difference(headers)) for column in missing_columns: optional_ldif_attribute = casefold_columns and column != normalized_key_column diagnostics.append( _diagnostic( "warning" if optional_ldif_attribute else "error", "missing_attribute" if optional_ldif_attribute else "missing_column", ( f'Configured LDIF attribute "{column}" is absent from this file.' if optional_ldif_attribute else f'Configured column "{column}" is missing.' ), field=column, ) ) if normalized_key_column in missing_columns or (missing_columns and not casefold_columns): return [], diagnostics keyed_rows: list[tuple[int, dict[str, Any], str]] = [] key_counts: Counter[str] = Counter() for row_number, row in rows: key = _first_import_value(_row_value(row, key_column, casefold=casefold_columns)).strip() if not key: diagnostics.append(_diagnostic("error", "missing_source_key", "Stable source key is blank.", row_number=row_number, field=key_column)) plan.append(_plan_effect(row_number, "conflict", source_key=None, message="Stable source key is blank.")) continue key_counts[key] += 1 keyed_rows.append((row_number, row, key)) first_index: dict[str, int] = {} last_index: dict[str, int] = {} for index, (_row_number, _row, key) in enumerate(keyed_rows): first_index.setdefault(key, index) last_index[key] = index existing_contacts = _contacts_by_source_refs( session, book_id, [f"import:{profile.profile_key}:{key}" for key in key_counts], ) for index, (row_number, row, key) in enumerate(keyed_rows): if key_counts[key] > 1: if config.duplicate_source_key_policy == "reject": diagnostics.append(_diagnostic("error", "duplicate_source_key", f'Duplicate source key "{key}".', row_number=row_number, field=key_column)) plan.append(_plan_effect(row_number, "conflict", source_key=key, message="Duplicate source key.")) continue chosen = first_index[key] if config.duplicate_source_key_policy == "first" else last_index[key] if index != chosen: diagnostics.append(_diagnostic("warning", "duplicate_source_key_ignored", f'Duplicate source key "{key}" was ignored by profile policy.', row_number=row_number, field=key_column)) plan.append(_plan_effect(row_number, "ignored", source_key=key, message="Duplicate row ignored by profile policy.")) continue mapped, row_diagnostics = _mapped_fields( row_number, row, config=config, casefold_columns=casefold_columns, ) diagnostics.extend(row_diagnostics) source_ref = f"import:{profile.profile_key}:{key}" existing = existing_contacts.get(source_ref) payload = _payload_from_mapped( mapped, profile=profile, input_hash=input_hash, row_number=row_number, source_key=key, source_row=row, ) display_name = payload.get("display_name") or payload.get("email") or key if any(item["severity"] == "error" for item in row_diagnostics): plan.append(_plan_effect(row_number, "conflict", source_key=key, display_name=display_name, source_ref=source_ref, payload=payload, message="Row validation failed.")) continue if existing is None: plan.append(_plan_effect(row_number, "create", source_key=key, display_name=display_name, source_ref=source_ref, payload=payload, changed_fields=sorted(mapped))) continue if config.existing_contact_policy == "ignore": plan.append(_plan_effect(row_number, "ignored", source_key=key, contact_id=existing.id, display_name=existing.display_name, source_ref=source_ref, payload=payload, message="Existing contact retained by profile policy.")) continue if config.existing_contact_policy == "reject": diagnostics.append(_diagnostic("error", "existing_contact", f'Contact for source key "{key}" already exists.', row_number=row_number)) plan.append(_plan_effect(row_number, "conflict", source_key=key, contact_id=existing.id, display_name=existing.display_name, source_ref=source_ref, payload=payload, message="Existing contact rejected by profile policy.")) continue changed_fields = _changed_fields(existing, mapped) plan.append( _plan_effect( row_number, "update" if changed_fields or existing.deleted_at is not None else "unchanged", source_key=key, contact_id=existing.id, display_name=display_name, source_ref=source_ref, payload=payload, changed_fields=changed_fields, expected_contact_hash=_contact_hash(existing), ) ) return sorted(plan, key=lambda item: item["row_number"]), diagnostics def _mapped_fields( row_number: int, row: dict[str, Any], *, config: AddressImportConfiguration, casefold_columns: bool = False, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: mapped: dict[str, Any] = {} diagnostics: list[dict[str, Any]] = [] for target, column in config.field_mappings.items(): if target == "source_key": continue values = _import_values(_row_value(row, column, casefold=casefold_columns)) if not values: if config.blank_value_policy == "reject": diagnostics.append(_diagnostic("error", "blank_value", f'Column "{column}" is blank.', row_number=row_number, field=target)) elif config.blank_value_policy == "clear": mapped[target] = [] if target == "tags" else None continue if target == "tags": mapped[target] = list( dict.fromkeys( item.strip() for value in values for item in value.split(",") if item.strip() ) ) elif target in {"email", "phone"}: mapped[target] = values if len(values) > 1 else values[0] else: mapped[target] = values[0] if len(values) > 1: diagnostics.append( _diagnostic( "warning", "multiple_values_truncated", f'Attribute "{column}" has multiple values; only the first maps to {target!r}.', row_number=row_number, field=target, ) ) if config.default_tags: mapped["tags"] = list(dict.fromkeys([*(mapped.get("tags") or []), *config.default_tags])) if not any(mapped.get(name) for name in ("display_name", "given_name", "family_name", "email", "organization")): diagnostics.append(_diagnostic("error", "missing_identity", "Row has no name, email, or organization to identify the contact.", row_number=row_number)) return mapped, diagnostics def _payload_from_mapped( mapped: dict[str, Any], *, profile: AddressImportProfile, input_hash: str, row_number: int, source_key: str, source_row: dict[str, Any] | None = None, ) -> dict[str, Any]: display_name = ( _first_import_value(mapped.get("display_name")) or " ".join( filter( None, [ _first_import_value(mapped.get("given_name")), _first_import_value(mapped.get("family_name")), ], ) ) or _first_import_value(mapped.get("email")) or _first_import_value(mapped.get("organization")) ) payload: dict[str, Any] = { key: mapped.get(key) for key in ("given_name", "family_name", "organization", "role_title", "note", "tags") if key in mapped } payload["display_name"] = display_name if "email" in mapped: email_values = _import_values(mapped["email"]) payload["emails"] = [ ContactEmailPayload(email=value, is_primary=index == 0).model_dump(mode="json") for index, value in enumerate(email_values) ] if "phone" in mapped: phone_values = _import_values(mapped["phone"]) payload["phones"] = [ ContactPhonePayload(phone=value, is_primary=index == 0).model_dump(mode="json") for index, value in enumerate(phone_values) ] postal_keys = {"street", "postal_code", "locality", "region", "country"} if postal_keys.intersection(mapped): postal = {key: mapped.get(key) for key in postal_keys if key in mapped} payload["postal_addresses"] = [ContactPostalAddressPayload(**postal, is_primary=True).model_dump(mode="json")] if any(postal.values()) else [] payload["provenance"] = { "import": { "profile_key": profile.profile_key, "profile_id": profile.id, "profile_version": profile.version, "input_hash": input_hash, "row_number": row_number, "source_key": source_key, "locale": profile.configuration.get("locale"), "visibility": mapped.get("visibility"), "source_format": profile.source_format, "source_record_hash": (source_row or {}).get("__ldif_record_hash"), } } return ContactCreateRequest.model_validate(payload).model_dump( mode="json", exclude_unset=True, exclude_none=False, ) def _changed_fields(contact: Contact, mapped: dict[str, Any]) -> list[str]: current: dict[str, Any] = { "display_name": contact.display_name, "given_name": contact.given_name, "family_name": contact.family_name, "organization": contact.organization, "role_title": contact.role_title, "note": contact.note, "tags": list(contact.tags or []), "email": contact.emails[0].email if contact.emails else None, "phone": contact.phones[0].phone if contact.phones else None, } if contact.postal_addresses: postal = contact.postal_addresses[0] current.update({key: getattr(postal, key) for key in ("street", "postal_code", "locality", "region", "country")}) changed: list[str] = [] for key, value in mapped.items(): if key == "visibility": continue if key == "email" and isinstance(value, list): current_value: Any = [item.email for item in contact.emails] elif key == "phone" and isinstance(value, list): current_value = [item.phone for item in contact.phones] else: current_value = current.get(key) if current_value != value: changed.append(key) return sorted(changed) def _plan_effect( row_number: int, action: str, *, source_key: str | None, contact_id: str | None = None, display_name: str | None = None, source_ref: str | None = None, payload: dict[str, Any] | None = None, changed_fields: list[str] | None = None, message: str | None = None, expected_contact_hash: str | None = None, ) -> dict[str, Any]: return { "row_number": row_number, "action": action, "source_key": source_key, "contact_id": contact_id, "display_name": display_name, "source_ref": source_ref, "payload": payload or {}, "changed_fields": changed_fields or [], "message": message, "expected_contact_hash": expected_contact_hash, } def _diagnostic( severity: str, code: str, message: str, *, row_number: int | None = None, field: str | None = None, ) -> dict[str, Any]: return { "severity": severity, "code": code, "message": message, "row_number": row_number, "field": field, "details": {}, } def _contacts_by_source_refs(session: Session, book_id: str, source_refs: list[str]) -> dict[str, Contact]: contacts: dict[str, Contact] = {} for offset in range(0, len(source_refs), CONTACT_LOOKUP_BATCH_SIZE): candidates = ( select( Contact.id, func.row_number().over( partition_by=Contact.source_ref, order_by=(Contact.created_at.asc(), Contact.id.asc()), ).label("source_position"), ) .where( Contact.address_book_id == book_id, Contact.source_ref.in_(source_refs[offset : offset + CONTACT_LOOKUP_BATCH_SIZE]), ) .subquery() ) rows = ( session.query(Contact) .join(candidates, candidates.c.id == Contact.id) .filter(candidates.c.source_position == 1) .options(selectinload(Contact.emails), selectinload(Contact.phones), selectinload(Contact.postal_addresses)) .populate_existing() .order_by(Contact.created_at.asc(), Contact.id.asc()) .all() ) for contact in rows: # Preserve the historical first-match choice for duplicate stored # source references, independently of batch/database row order. contacts.setdefault(str(contact.source_ref), contact) return contacts def _contact_by_source_ref(session: Session, book_id: str, source_ref: str, *, lock: bool = False) -> Contact | None: query = ( session.query(Contact) .filter(Contact.address_book_id == book_id, Contact.source_ref == source_ref) .order_by(Contact.created_at.asc(), Contact.id.asc()) ) if lock: query = query.populate_existing().with_for_update() return query.first() def _validated_rollback_snapshot(value: object) -> tuple[dict[str, Any], datetime | None]: if ( not isinstance(value, dict) or value.get("version") != ROLLBACK_SNAPSHOT_VERSION or "deleted_at" not in value or not isinstance(value.get("contact"), dict) ): raise AddressBookError( "This import has incomplete legacy rollback evidence; automatic rollback is unsafe. " "Review and reconcile the affected contacts manually." ) snapshot = value["contact"] required_fields = { "display_name", "given_name", "family_name", "organization", "role_title", "note", "tags", "emails", "phones", "postal_addresses", "provenance", } if ( not isinstance(snapshot.get("payload"), dict) or not required_fields.issubset(snapshot["payload"]) or not {"source_kind", "source_ref", "source_revision", "source_payload_kind", "source_payload_raw", "provenance", "metadata"}.issubset(snapshot) ): raise AddressBookError("Import rollback evidence is incomplete; automatic rollback is unsafe.") try: deleted_at = datetime.fromisoformat(value["deleted_at"]) if value["deleted_at"] is not None else None ContactUpdateRequest.model_validate(snapshot["payload"]) _validate_point_snapshot(snapshot.get("points")) except (TypeError, ValueError) as exc: raise AddressBookError("Import rollback evidence is invalid; automatic rollback is unsafe.") from exc return snapshot, deleted_at def _stamp_import_contact(contact: Contact, *, run: AddressImportRun, item: dict[str, Any]) -> None: contact.source_kind = run.source_format contact.source_ref = item["source_ref"] contact.source_revision = hashlib.sha256( f'{run.input_hash}:{item["row_number"]}:{item["source_key"]}'.encode() ).hexdigest() contact.source_payload_kind = f"{run.source_format}-mapped-row" contact.source_payload_raw = None provenance = dict(contact.provenance or {}) provenance["import_run_id"] = run.id provenance["input_hash"] = run.input_hash provenance["plan_hash"] = run.plan_hash contact.provenance = provenance def _contact_snapshot(contact: Contact) -> dict[str, Any]: return { "payload": { "display_name": contact.display_name, "given_name": contact.given_name, "family_name": contact.family_name, "organization": contact.organization, "role_title": contact.role_title, "note": contact.note, "tags": list(contact.tags or []), "emails": [{"label": item.label, "email": item.email, "is_primary": item.is_primary} for item in contact.emails], "phones": [{"label": item.label, "phone": item.phone, "is_primary": item.is_primary} for item in contact.phones], "postal_addresses": [ { "label": item.label, "street": item.street, "postal_code": item.postal_code, "locality": item.locality, "region": item.region, "country": item.country, "is_primary": item.is_primary, } for item in contact.postal_addresses ], "provenance": dict(contact.provenance or {}), }, "source_kind": contact.source_kind, "source_ref": contact.source_ref, "source_revision": contact.source_revision, "source_payload_kind": contact.source_payload_kind, "source_payload_raw": contact.source_payload_raw, "provenance": dict(contact.provenance or {}), "metadata": copy.deepcopy(contact.metadata_), "points": { name: [_point_snapshot(point) for point in getattr(contact, name)] for name in CONTACT_POINTS }, } def _contact_hash(contact: Contact) -> str: return _hash_json({**_contact_snapshot(contact), "deleted_at": _deleted_at_value(contact.deleted_at)}) def _point_snapshot(point: ContactEmail | ContactPhone | ContactPostalAddress) -> dict[str, Any]: # Include every persisted evidence field, including identity, originals, # normalization, provenance, ordering and timestamps, but never the parent FK. return { column.key: ( _deleted_at_value(getattr(point, column.key)) if column.key in {"created_at", "updated_at"} else copy.deepcopy(getattr(point, column.key)) ) for column in point.__table__.columns if column.key != "contact_id" } def _lock_contact_points(session: Session, contact: Contact) -> None: # The parent is locked by the caller. Point-only evidence updates need their # own row locks; they need not update the parent row on PostgreSQL. for name, (model, _identity, _replace) in CONTACT_POINTS.items(): positions = {point.id: index for index, point in enumerate(getattr(contact, name))} points = session.query(model).filter(model.contact_id == contact.id).order_by( model.id.asc(), ).populate_existing().with_for_update().all() # Lock acquisition order is not presentation order. Preserve the # relationship's existing order for tied indexes; new identities still # enter the hash and therefore cannot evade the reviewed-state guard. points.sort(key=lambda point: (point.order_index, positions.get(point.id, len(positions)), point.id)) set_committed_value(contact, name, points) def _validate_point_snapshot(value: object) -> None: if not isinstance(value, dict) or set(value) != set(CONTACT_POINTS): raise ValueError("Missing complete contact-point evidence") for name, (model, _identity, _replace) in CONTACT_POINTS.items(): points = value[name] columns = {column.key: column for column in model.__table__.columns if column.key != "contact_id"} ids: set[str] = set() if not isinstance(points, list): raise ValueError("Invalid contact-point collection") for point in points: if not isinstance(point, dict) or set(point) != set(columns): raise ValueError("Incomplete contact-point evidence") for key, column in columns.items(): field = point[key] if field is None and column.nullable: continue if key in {"created_at", "updated_at"}: datetime.fromisoformat(field) elif type(field) is not column.type.python_type: raise ValueError("Invalid contact-point evidence type") if not point["id"] or point["id"] in ids: raise ValueError("Invalid contact-point identity") ids.add(point["id"]) def _without_contact_points(payload: ContactUpdateRequest) -> ContactUpdateRequest: return ContactUpdateRequest.model_validate( payload.model_dump(exclude=set(CONTACT_POINTS), exclude_unset=True) ) def _require_unreferenced_points(session: Session, contact: Contact, removed: set[str]) -> None: if not removed: return # Do not detach address-list selections or silently retarget point-specific # consent/quality decisions when an import removes or replaces a value. referenced = session.query(AddressListEntry.id).filter( AddressListEntry.contact_id == contact.id, or_(AddressListEntry.contact_email_id.in_(removed), AddressListEntry.contact_postal_address_id.in_(removed)), ).first() for model in (ContactChannelRule, ContactPointQualityDecision): if referenced is not None: break referenced = session.query(model.id).filter( model.contact_id == contact.id, model.contact_point_id.in_(removed), ).first() if referenced is not None: raise AddressBookError( "An affected contact point has address-list or governance references; " "review and reconcile those references before applying or rolling back this import." ) def _prepare_import_points(session: Session, contact: Contact, payload: ContactUpdateRequest): staged = Contact( source_kind=contact.source_kind, source_ref=contact.source_ref, source_revision=contact.source_revision, created_by_account_id=contact.created_by_account_id, updated_by_account_id=contact.updated_by_account_id, ) result = {} removed: set[str] = set() for name, (_model, identity, replace) in CONTACT_POINTS.items(): if name not in payload.model_fields_set: continue replace(staged, getattr(payload, name) or []) available = list(getattr(contact, name)) matches = [] for candidate in list(getattr(staged, name)): # Detach the normalization-only parent before a new point enters the # persistent collection; otherwise save-update cascade can enlist it. candidate.contact = None original = next((point for point in available if all( getattr(point, key) == getattr(candidate, key) for key in identity )), None) if original is not None: available.remove(original) matches.append((original, candidate)) removed.update(point.id for point in available) result[name] = matches _require_unreferenced_points(session, contact, removed) return result def _apply_import_points(contact: Contact, prepared) -> None: for name, matches in prepared.items(): points = [] for original, candidate in matches: if original is None: points.append(candidate) else: # Same value retains its exact original/normalized evidence and # identity; only explicitly imported presentation fields change. for field in ("label", "is_primary", "order_index"): setattr(original, field, getattr(candidate, field)) points.append(original) setattr(contact, name, points) def _validate_point_restoration(session: Session, contact: Contact, points: dict[str, Any]) -> None: removed: set[str] = set() for name, (model, _identity, _replace) in CONTACT_POINTS.items(): desired = {point["id"] for point in points[name]} removed.update(point.id for point in getattr(contact, name) if point.id not in desired) # A deleted original identity must never be reclaimed from another contact. if desired and session.query(model.id).filter( model.id.in_(desired), model.contact_id != contact.id, ).first() is not None: raise AddressBookError("Contact-point identity changed; automatic rollback is unsafe.") _require_unreferenced_points(session, contact, removed) def _restore_contact_points(contact: Contact, snapshots: dict[str, Any]) -> None: for name, (model, _identity, _replace) in CONTACT_POINTS.items(): existing = {point.id: point for point in getattr(contact, name)} restored = [] for snapshot in snapshots[name]: point = existing.get(snapshot["id"]) if point is None: point = model() for key, value in snapshot.items(): setattr(point, key, datetime.fromisoformat(value) if key in {"created_at", "updated_at"} else copy.deepcopy(value)) restored.append(point) setattr(contact, name, restored) def _deleted_at_value(value: datetime | None) -> str | None: if value is None: return None return (value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)).isoformat() def _hash_json(value: object) -> str: return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()).hexdigest() def _trim(value: str | None) -> str | None: if value is None: return None normalized = value.strip() return normalized or None def _visible_import_books(session: Session, principal: ApiPrincipal): from govoplan_addresses.backend.service import list_address_books return list_address_books(session, principal) __all__ = [ "apply_address_import", "create_import_profile", "get_import_profile", "get_import_run", "import_run_payload", "list_import_profiles", "preview_address_import", "retire_import_profile", "rollback_address_import", "update_import_profile", ]