"""Import a WordPress manifest (from tools/extract_wordpress.py) into the portal. Idempotent: re-running updates existing records rather than duplicating them, so a migration can be rehearsed, corrected, and re-run. Always rehearse first: python manage.py import_wordpress --manifest /data/import//manifest.json --dry-run The dry-run prints the reconciliation report — the before/after oracle that proves the port didn't silently drop data — and writes nothing. Passwords are carried across verbatim (wrapped so Django's WordPress verifiers pick them up), so owners keep the password they already have; Django upgrades each hash to the preferred algorithm on that owner's next login. """ import html import json import os from datetime import datetime from django.contrib.auth import get_user_model from django.core.files import File from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.utils import timezone from accounts.wp_import import STAFF_ROLES, SUPERUSER_ROLES, parse_wp_roles, wp_hash_to_django from django.utils.text import slugify from documents.models import BoardMember, Category, Collection, Document User = get_user_model() class Command(BaseCommand): help = "Import users, categories and documents from a WordPress manifest." def add_arguments(self, parser): parser.add_argument("--manifest", required=True) parser.add_argument("--dry-run", action="store_true", help="report only; write nothing") parser.add_argument("--skip-files", action="store_true", help="import metadata without copying document files") parser.add_argument("--import-board", action="store_true", help="also import best-effort board candidates (review after)") def handle(self, *args, **options): manifest_path = options["manifest"] if not os.path.isfile(manifest_path): raise CommandError(f"Manifest not found: {manifest_path}") with open(manifest_path) as handle: manifest = json.load(handle) files_root = os.path.join(os.path.dirname(manifest_path), "files") dry_run = options["dry_run"] report = Report(manifest.get("project", "?"), dry_run) try: with transaction.atomic(): categories = self._import_categories(manifest, report, dry_run) self._import_users(manifest, report, dry_run) self._import_documents(manifest, categories, files_root, report, dry_run, options["skip_files"]) if options["import_board"]: self._import_board(manifest, report, dry_run) if dry_run: raise _Rollback() except _Rollback: pass report.render(self.stdout, self.style) # ---- categories ----------------------------------------------------- def _import_categories(self, manifest, report, dry_run): mapping = {} for index, entry in enumerate(manifest.get("categories", [])): report.categories_seen += 1 existing = Category.objects.filter(slug=entry["slug"]).first() if existing: report.categories_existing += 1 mapping[entry["key"]] = existing continue report.categories_created += 1 if dry_run: mapping[entry["key"]] = None continue mapping[entry["key"]] = Category.objects.create( name=html.unescape(entry["name"]), slug=entry["slug"], order=index ) return mapping # ---- users ---------------------------------------------------------- def _import_users(self, manifest, report, dry_run): for entry in manifest.get("users", []): report.users_seen += 1 report.hash_formats[entry["hash_format"]] = ( report.hash_formats.get(entry["hash_format"], 0) + 1 ) username = (entry["username"] or "").strip() if not username: report.users_skipped.append((entry.get("wp_id"), "no username")) continue password = wp_hash_to_django(entry["password_hash"]) if password is None: # Unrecognised hash: import the account but make it unusable, so # the owner resets rather than us guessing at their credential. report.users_unusable_password += 1 roles = parse_wp_roles(entry.get("capabilities")) is_staff = any(role in STAFF_ROLES for role in roles) is_superuser = any(role in SUPERUSER_ROLES for role in roles) user = User.objects.filter(username=username).first() if user: report.users_existing += 1 else: report.users_created += 1 if is_staff: report.users_staff += 1 if dry_run: continue if not user: user = User(username=username) # Two WordPress accounts can share an address. In this portal an # address is also a way to sign in, and the backend refuses to guess # between two accounts holding one — so importing the duplicate as-is # disables email sign-in for BOTH people, silently and for whoever # got there first. Keep the account, drop the contested address, and # say so loudly enough to be acted on. email = entry.get("email") or "" if email and User.objects.filter(email__iexact=email).exclude( pk=user.pk or 0 ).exists(): report.users_email_conflict.append((username, email)) email = "" user.email = email user.first_name = (entry.get("first_name") or "")[:150] user.last_name = (entry.get("last_name") or "")[:150] user.is_staff = is_staff user.is_superuser = is_superuser if _should_set_password(user): if password: user.password = password else: user.set_unusable_password() else: # This owner has already logged in and had their hash upgraded to # the strong algorithm (or set a new password). Re-running the # import must not drag them back to the legacy WordPress hash. report.users_password_preserved += 1 joined = _parse_datetime(entry.get("registered")) if joined: user.date_joined = joined user.save() # ---- documents ------------------------------------------------------ def _import_documents(self, manifest, categories, files_root, report, dry_run, skip_files): for entry in manifest.get("documents", []): report.documents_seen += 1 primary = entry.get("files", {}).get("primary") if not primary: report.documents_without_file.append(entry["title"]) continue source = os.path.join(files_root, primary["relpath"]) if not os.path.isfile(source): report.documents_missing_file.append( (entry["title"], primary["relpath"]) ) continue category_key = entry.get("category_key") or "" if category_key and category_key not in categories: report.documents_unmapped_category.append(entry["title"]) existing = Document.objects.filter(wp_id=entry["wp_id"]).first() if existing: report.documents_existing += 1 else: report.documents_created += 1 if dry_run: continue document = existing or Document(wp_id=entry["wp_id"]) document.title = html.unescape(entry["title"])[:255] document.summary = html.unescape(entry.get("summary") or "") document.category = categories.get(category_key) document.published = entry.get("status") == "publish" published_at = _parse_datetime(entry.get("post_date")) if published_at: document.published_date = published_at.date() document.save() if skip_files: continue if not document.file or not _same_basename(document.file.name, primary["relpath"]): with open(source, "rb") as handle: document.file.save(os.path.basename(primary["relpath"]), File(handle), save=True) report.files_imported += 1 secondary = entry.get("files", {}).get("secondary") if secondary: second_source = os.path.join(files_root, secondary["relpath"]) if os.path.isfile(second_source): self._import_companion(document, second_source, secondary, report) def _import_companion(self, document, source, secondary, report): """A legacy second file becomes its own document, in a shared collection. The old sites pinned two files to one record (minutes + financials). That relationship is now expressed as a collection, so the second file is imported as a real document rather than a hidden attachment. """ companion_wp_id = f"{document.wp_id}-b" if document.wp_id else None if companion_wp_id and Document.objects.filter(wp_id=companion_wp_id).exists(): return if document.collection is None: base = slugify(document.title)[:200] or f"collection-{document.pk}" slug, suffix = base, 2 while Collection.objects.filter(slug=slug).exists(): slug, suffix = f"{base}-{suffix}", suffix + 1 document.collection = Collection.objects.create( title=document.title[:200], slug=slug, occurred_on=document.published_date, ) document.save(update_fields=["collection"]) companion = Document( wp_id=companion_wp_id, title=f"{document.title} (2)"[:255], collection=document.collection, category=document.category, published=document.published, published_date=document.published_date, ) with open(source, "rb") as handle: companion.file.save(os.path.basename(secondary["relpath"]), File(handle), save=False) companion.save() report.files_imported += 1 # ---- board ---------------------------------------------------------- def _import_board(self, manifest, report, dry_run): for index, candidate in enumerate(manifest.get("board", {}).get("candidates", [])): report.board_seen += 1 if BoardMember.objects.filter(name=candidate["name"]).exists(): continue report.board_created += 1 if dry_run: continue BoardMember.objects.create( name=candidate["name"][:160], title=html.unescape(candidate.get("title") or "")[:160], order=index, active=False, # inactive until a human reviews the parse ) class _Rollback(Exception): """Internal: unwinds the transaction after a dry run.""" def _should_set_password(user): """True unless the user already holds a stronger, non-WordPress password. A migration may be re-run after some owners have already logged in (which upgrades their hash) — those upgrades must survive the re-run. The upgrade only ever happens AT login, so `last_login is None` means there is nothing to preserve and the WordPress hash should win. That matters for the bootstrap admin a fresh container creates before any import: without this check it keeps its generated password and the real WordPress password silently fails to work. """ if not user.pk or user.last_login is None: return True current = user.password or "" return current.startswith(("wp_phpass$", "wp_bcrypt$", "!")) or not current def _same_basename(stored_name, relpath): return os.path.basename(stored_name or "") == os.path.basename(relpath) def _parse_datetime(value): if not value: return None for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): try: parsed = datetime.strptime(value, fmt) except ValueError: continue return timezone.make_aware(parsed, timezone.get_default_timezone()) return None class Report: def __init__(self, project, dry_run): self.project = project self.dry_run = dry_run self.users_seen = self.users_created = self.users_existing = 0 self.users_staff = self.users_unusable_password = 0 self.users_password_preserved = 0 self.users_skipped = [] self.users_email_conflict = [] self.hash_formats = {} self.categories_seen = self.categories_created = self.categories_existing = 0 self.documents_seen = self.documents_created = self.documents_existing = 0 self.documents_without_file = [] self.documents_missing_file = [] self.documents_unmapped_category = [] self.files_imported = 0 self.board_seen = self.board_created = 0 def render(self, out, style): mode = "DRY RUN (nothing written)" if self.dry_run else "IMPORTED" out.write(style.MIGRATE_HEADING(f"\n{self.project} — {mode}")) out.write(style.MIGRATE_LABEL("\nUsers")) out.write(f" in manifest {self.users_seen}") out.write(f" new {self.users_created}") out.write(f" already present {self.users_existing}") out.write(f" staff/admin {self.users_staff}") out.write(f" hash formats {self.hash_formats}") if self.users_unusable_password: out.write(style.WARNING( f" unusable password {self.users_unusable_password} (must reset)")) if self.users_password_preserved: out.write(f" kept upgraded pw {self.users_password_preserved}") for username, email in self.users_email_conflict: out.write(style.WARNING( f" {username}: address {email} is already on another account — " f"imported without it, so email sign-in keeps working for both")) for wp_id, reason in self.users_skipped: out.write(style.WARNING(f" skipped wp_id={wp_id}: {reason}")) out.write(style.MIGRATE_LABEL("\nCategories")) out.write(f" in manifest {self.categories_seen}") out.write(f" new {self.categories_created}") out.write(f" already present {self.categories_existing}") out.write(style.MIGRATE_LABEL("\nDocuments")) out.write(f" in manifest {self.documents_seen}") out.write(f" new {self.documents_created}") out.write(f" already present {self.documents_existing}") out.write(f" files imported {self.files_imported}") problems = (len(self.documents_without_file) + len(self.documents_missing_file) + len(self.documents_unmapped_category)) if problems: out.write(style.MIGRATE_LABEL("\nNeeds attention")) if self.documents_without_file: out.write(style.WARNING( f" {len(self.documents_without_file)} document(s) reference no file")) for title in self.documents_without_file[:10]: out.write(f" - {title}") if self.documents_missing_file: out.write(style.ERROR( f" {len(self.documents_missing_file)} document(s) whose file is MISSING " f"from the extract — these would not import")) for title, relpath in self.documents_missing_file[:10]: out.write(f" - {title} [{relpath}]") if self.documents_unmapped_category: out.write(style.WARNING( f" {len(self.documents_unmapped_category)} document(s) with an " f"unmapped category key")) if self.board_seen: out.write(style.MIGRATE_LABEL("\nBoard members")) out.write(f" candidates {self.board_seen}") out.write(f" new {self.board_created}") out.write(style.WARNING(" imported INACTIVE — review each before publishing")) if not problems: out.write(style.SUCCESS("\nNo reconciliation problems found.")) out.write("")