from __future__ import annotations import base64 import binascii import csv import hashlib import json from collections import Counter from io import BytesIO, StringIO from typing import Any from sqlalchemy import and_, false, or_ from sqlalchemy.orm import Session from govoplan_addresses.backend.db.models import ( AddressImportProfile, AddressImportRun, Contact, ) from govoplan_addresses.backend.import_schemas import ( AddressImportConfiguration, AddressImportPreviewRequest, AddressImportProfileCreateRequest, AddressImportProfileUpdateRequest, AddressImportRollbackRequest, ) from govoplan_addresses.backend.schemas import ( ContactCreateRequest, ContactEmailPayload, ContactPhonePayload, ContactPostalAddressPayload, ContactUpdateRequest, ) from govoplan_addresses.backend.service import ( AddressBookError, 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 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, ) 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, ) -> 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.") run = ( session.query(AddressImportRun) .filter(AddressImportRun.id == run_id, AddressImportRun.address_book_id.in_(visible_book_ids)) .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) if run.status == "applied": return run if run.status != "previewed": raise AddressBookError(f"Import run cannot be applied from status {run.status!r}.") if run.plan_hash != expected_plan_hash: raise AddressBookError("The reviewed import plan changed; create a new preview.") 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]] = [] for item in run.plan_data or []: 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) 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.") if _contact_hash(existing) != item.get("expected_contact_hash"): raise AddressBookError( f'Contact "{existing.display_name}" changed after preview; preview the import again.' ) before = _contact_snapshot(existing) if existing.deleted_at is not None: restore_contact(session, principal, existing.id) contact = update_contact( session, principal, existing.id, ContactUpdateRequest.model_validate(item["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 = list(run.plan_data or []) 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) 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 []) 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") } ) for contact_id, expected_hash in expected_hashes.items(): contact = get_visible_contact(session, principal, contact_id, include_deleted=True) if _contact_hash(contact) != expected_hash: raise AddressBookError( f'Contact "{contact.display_name}" changed after import; automatic rollback is unsafe.' ) 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 = dict(item["before"]) if contact.deleted_at is not None: restore_contact(session, principal, contact.id) update_contact( session, principal, contact.id, ContactUpdateRequest.model_validate(snapshot["payload"]), ) contact.source_kind = snapshot.get("source_kind") or "local" contact.source_ref = snapshot.get("source_ref") contact.source_revision = snapshot.get("source_revision") contact.source_payload_kind = snapshot.get("source_payload_kind") contact.source_payload_raw = snapshot.get("source_payload_raw") contact.provenance = dict(snapshot.get("provenance") or {}) contact.metadata_ = dict(snapshot.get("metadata") or {}) 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, str]]], 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) 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 _plan_rows( session: Session, *, book_id: str, profile: AddressImportProfile, input_hash: str, rows: list[tuple[int, dict[str, str]]], config: AddressImportConfiguration, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: diagnostics: list[dict[str, Any]] = [] plan: list[dict[str, Any]] = [] headers = set(rows[0][1]) if rows else set() referenced_columns = set(config.field_mappings.values()) if config.source_key_column: referenced_columns.add(config.source_key_column) missing_columns = sorted(referenced_columns.difference(headers)) for column in missing_columns: diagnostics.append(_diagnostic("error", "missing_column", f'Configured column "{column}" is missing.', field=column)) if missing_columns: return [], diagnostics key_column = config.source_key_column or config.field_mappings["source_key"] keyed_rows: list[tuple[int, dict[str, str], str]] = [] key_counts: Counter[str] = Counter() for row_number, row in rows: key = row.get(key_column, "").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 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) diagnostics.extend(row_diagnostics) source_ref = f"import:{profile.profile_key}:{key}" existing = _contact_by_source_ref(session, book_id, source_ref) payload = _payload_from_mapped(mapped, profile=profile, input_hash=input_hash, row_number=row_number, source_key=key) 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, str], *, config: AddressImportConfiguration, ) -> 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 value = row.get(column, "").strip() if not value: 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 mapped[target] = [item.strip() for item in value.split(",") if item.strip()] if target == "tags" else value 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, ) -> dict[str, Any]: display_name = mapped.get("display_name") or " ".join(filter(None, [mapped.get("given_name"), mapped.get("family_name")])) or mapped.get("email") or 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: payload["emails"] = [] if mapped["email"] is None else [ContactEmailPayload(email=mapped["email"], is_primary=True).model_dump(mode="json")] if "phone" in mapped: payload["phones"] = [] if mapped["phone"] is None else [ContactPhonePayload(phone=mapped["phone"], is_primary=True).model_dump(mode="json")] 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"), } } 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")}) return sorted(key for key, value in mapped.items() if key != "visibility" and current.get(key) != value) 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 _contact_by_source_ref(session: Session, book_id: str, source_ref: str) -> Contact | None: return ( session.query(Contact) .filter(Contact.address_book_id == book_id, Contact.source_ref == source_ref) .order_by(Contact.created_at.asc(), Contact.id.asc()) .first() ) 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": dict(contact.metadata_ or {}), } def _contact_hash(contact: Contact) -> str: return _hash_json({**_contact_snapshot(contact), "deleted_at": contact.deleted_at.isoformat() if contact.deleted_at else None}) 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", ]