#!/usr/bin/env python3 """Extract one WordPress owner-site into a portable JSON manifest + file tree. Runs on the HOST (needs `srv-gw db-query`, which is read-only and audit-logged), not inside the app container. Produces: //manifest.json users, categories, documents, board candidates //files/ every referenced document file The manifest is the contract between "get the data out of WordPress" and "put the data into the portal" (see import_wordpress). Keeping them separate means production extraction can come from a different source (a Cloudways dump) while the import side stays identical. python3 main/tools/extract_wordpress.py --project hartling-shoreclubowners Reads nothing but the database and the uploads tree; writes nothing to WordPress. """ import argparse import json import os import re import shutil import subprocess import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from accounts.wp_import import ( # noqa: E402 attachment_relpath, hash_format, relpath_from_upload_url, slugify_category, ) # Per-site category labels for the `wpcf-doc-type` meta value (which is a magic # select value in WordPress, not a taxonomy). Sourced from each site's ACF field # config during the 2026-08 audit. CATEGORY_LABELS = { "hartling-thesandsowners": { "1": "Minutes & Financials", "2": "Insurance Documents", "3": "Other Documents", "4": "Owners' Newsletters", "5": "Strata CapEx", "bylaws": "Bylaws", "management-agreements": "Management Agreements", }, "hartling-shoreclubowners": { "1": "Meeting Minutes", "2": "Insurance Documents", "3": "Financials", "4": "Communications", }, "hartling-thepalmsowners": { "1": "Minutes", "2": "Insurance Documents", "3": "Other Documents", "4": "Owner's Newsletters", "5": "Financials", }, } def db_query(project, sql, rows=20000): """Run a read-only query through the gateway and return a list of dicts. Note: db-query renders a SQL NULL as the literal string "NULL" in JSON, which would otherwise flow through as a category named "NULL" or a summary reading "NULL". Normalise it to an empty string here, once, for every caller. """ result = subprocess.run( ["srv-gw", "db-query", "--project", project, "--format", "json", "--rows", str(rows), sql], capture_output=True, text=True, timeout=180, ) if result.returncode != 0: raise RuntimeError(f"db-query failed for {project}:\n{result.stderr.strip()}") text = result.stdout.strip() end = text.rfind("]") if end == -1: return [] return [ {key: ("" if value == "NULL" else value) for key, value in row.items()} for row in json.loads(text[: end + 1]) ] def extract_users(project): rows = db_query(project, """ SELECT u.ID, u.user_login, u.user_email, u.user_pass, u.user_registered, u.display_name, MAX(CASE WHEN m.meta_key='first_name' THEN m.meta_value END) AS first_name, MAX(CASE WHEN m.meta_key='last_name' THEN m.meta_value END) AS last_name, MAX(CASE WHEN m.meta_key LIKE '%capabilities' THEN m.meta_value END) AS caps FROM wp_users u LEFT JOIN wp_usermeta m ON m.user_id = u.ID GROUP BY u.ID, u.user_login, u.user_email, u.user_pass, u.user_registered, u.display_name """) users = [] for row in rows: users.append({ "wp_id": row["ID"], "username": row["user_login"], "email": row["user_email"] or "", "password_hash": row["user_pass"] or "", "hash_format": hash_format(row["user_pass"]), "registered": row["user_registered"], "first_name": row.get("first_name") or "", "last_name": row.get("last_name") or "", "capabilities": row.get("caps") or "", }) return users def extract_documents(project): rows = db_query(project, """ SELECT p.ID, p.post_title, p.post_date, p.post_status, MAX(CASE WHEN pm.meta_key='wpcf-doc-type' THEN pm.meta_value END) AS doc_type, MAX(CASE WHEN pm.meta_key='wpcf-summary' THEN pm.meta_value END) AS summary, MAX(CASE WHEN pm.meta_key='acf-pdf' THEN pm.meta_value END) AS acf_pdf, MAX(CASE WHEN pm.meta_key='acf-pdf-2' THEN pm.meta_value END) AS acf_pdf_2, MAX(CASE WHEN pm.meta_key='wpcf-pdf' THEN pm.meta_value END) AS wpcf_pdf, MAX(CASE WHEN pm.meta_key='wpcf-pdf-2' THEN pm.meta_value END) AS wpcf_pdf_2 FROM wp_posts p LEFT JOIN wp_postmeta pm ON pm.post_id = p.ID WHERE p.post_type IN ('docs','documents') AND p.post_status IN ('publish','draft','private') GROUP BY p.ID, p.post_title, p.post_date, p.post_status """) return rows def extract_attachments(project): """attachment post ID -> path relative to uploads/.""" rows = db_query(project, """ SELECT p.ID, pm.meta_value AS attached_file FROM wp_posts p JOIN wp_postmeta pm ON pm.post_id = p.ID AND pm.meta_key = '_wp_attached_file' WHERE p.post_type = 'attachment' """) return {row["ID"]: attachment_relpath(row["attached_file"]) for row in rows} def extract_board(project): """Best-effort board-member candidates from the hand-authored page(s). The source is unstructured Gutenberg HTML, so this is deliberately conservative: it emits candidates for HUMAN REVIEW rather than pretending to be authoritative. There are only a handful of members per site. """ rows = db_query(project, """ SELECT ID, post_title, post_status, post_content FROM wp_posts WHERE post_type='page' AND post_status IN ('publish','private','draft') AND (post_title LIKE '%Board%' OR post_title LIKE '%Executive%' OR post_title LIKE '%Director%') """) candidates = [] for row in rows: html = row.get("post_content") or "" for match in re.finditer(r"]*>(.*?)", html, re.S | re.I): name = re.sub(r"<[^>]+>", "", match.group(1)).strip() if not name or len(name) > 80: continue tail = html[match.end(): match.end() + 400] title_match = re.search(r"]*>(.*?)

", tail, re.S | re.I) title = re.sub(r"<[^>]+>", "", title_match.group(1)).strip() if title_match else "" candidates.append({ "name": name, "title": title[:160], "source_page": row["post_title"], "needs_review": True, }) return {"pages": [r["post_title"] for r in rows], "candidates": candidates} def _locate(relpath, uploads, protected_root): """Find a document file, checking uploads then the protected store. A file-protection plugin may have MOVED files out of uploads into a separate tree (preserving the YYYY/MM layout), in which case the database still references the original uploads path. Fall back to the protected root, and finally to a basename match for references that carry no year/month (e.g. rewritten /bw-file// URLs). """ direct = os.path.join(uploads, relpath) if os.path.isfile(direct): return direct if not protected_root: return None moved = os.path.join(protected_root, relpath) if os.path.isfile(moved): return moved basename = os.path.basename(relpath) for root, _dirs, files in os.walk(protected_root): if basename in files: return os.path.join(root, basename) return None def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--project", required=True, help="staging project name") parser.add_argument("--out", default="/srv/apps/hartlingowners/data/import") parser.add_argument("--uploads", default=None, help="uploads dir (default: /srv/apps//wp-content/uploads)") parser.add_argument("--protected-root", default=None, help="second search root for files moved out of uploads by a " "file-protection plugin (same YYYY/MM layout preserved)") parser.add_argument("--no-files", action="store_true", help="manifest only, skip file copy") args = parser.parse_args() project = args.project uploads = args.uploads or f"/srv/apps/{project}/wp-content/uploads" out_dir = os.path.join(args.out, project) files_dir = os.path.join(out_dir, "files") os.makedirs(files_dir, exist_ok=True) print(f"Extracting {project} ...") users = extract_users(project) documents = extract_documents(project) attachments = extract_attachments(project) board = extract_board(project) labels = CATEGORY_LABELS.get(project, {}) categories, seen_types, unmapped = [], set(), set() for doc in documents: doc_type = (doc.get("doc_type") or "").strip() if not doc_type or doc_type in seen_types: if not doc_type: continue continue seen_types.add(doc_type) label = labels.get(doc_type) if not label: unmapped.add(doc_type) label = f"Uncategorised ({doc_type})" categories.append({"key": doc_type, "name": label, "slug": slugify_category(label)}) records, copied, missing = [], 0, [] for doc in documents: entry = { "wp_id": doc["ID"], "title": (doc["post_title"] or "").strip() or f"Document {doc['ID']}", "post_date": doc["post_date"], "status": doc["post_status"], "category_key": (doc.get("doc_type") or "").strip(), "summary": (doc.get("summary") or "").strip(), "files": {}, } for slot, id_key, url_key in ( ("primary", "acf_pdf", "wpcf_pdf"), ("secondary", "acf_pdf_2", "wpcf_pdf_2"), ): relpath = None attachment_id = (doc.get(id_key) or "").strip() if attachment_id and attachment_id in attachments: relpath = attachments[attachment_id] if not relpath: relpath = relpath_from_upload_url(doc.get(url_key)) if not relpath: continue source = _locate(relpath, uploads, args.protected_root) found = source is not None entry["files"][slot] = {"relpath": relpath, "found": found} if not found: missing.append({"doc": entry["title"], "slot": slot, "relpath": relpath}) continue if not args.no_files: destination = os.path.join(files_dir, relpath) os.makedirs(os.path.dirname(destination), exist_ok=True) if not os.path.exists(destination): shutil.copy2(source, destination) copied += 1 records.append(entry) manifest = { "project": project, "uploads_root": uploads, "users": users, "categories": categories, "documents": records, "board": board, "stats": { "users": len(users), "documents": len(records), "categories": len(categories), "files_copied": copied, "files_missing": len(missing), "unmapped_category_keys": sorted(unmapped), "hash_formats": { fmt: sum(1 for u in users if u["hash_format"] == fmt) for fmt in {u["hash_format"] for u in users} }, }, "missing_files": missing, } manifest_path = os.path.join(out_dir, "manifest.json") with open(manifest_path, "w") as handle: json.dump(manifest, handle, indent=2) os.chmod(manifest_path, 0o664) stats = manifest["stats"] print(f" users={stats['users']} documents={stats['documents']} " f"categories={stats['categories']}") print(f" hashes: {stats['hash_formats']}") print(f" files copied={stats['files_copied']} missing={stats['files_missing']}") if stats["unmapped_category_keys"]: print(f" WARNING unmapped category keys: {stats['unmapped_category_keys']}") print(f" board candidates={len(board['candidates'])} (need review)") print(f" -> {manifest_path}") if __name__ == "__main__": main()