"""Process WordPress Site Understanding exports into internal JSON artifacts.""" from __future__ import annotations import json import re import tarfile from datetime import UTC, datetime from pathlib import Path from uuid import UUID import structlog from src.core.database import async_session_factory from src.domain.external_admin.models import ( SITE_UNDERSTANDING_STATUS_COMPLETED, SITE_UNDERSTANDING_STATUS_FAILED, SITE_UNDERSTANDING_STATUS_QUEUED, SITE_UNDERSTANDING_STATUS_RUNNING, ) from src.domain.external_admin.repository import ( ExternalAdminInstallationRepository, SiteUnderstandingRunRepository, ) from src.domain.external_admin.wordpress_companion_package import ( resolve_wordpress_companion_package, ) from src.domain.projects.repository import ProjectRepoRepository from src.integrations.cloud_control_client import get_cloud_client_for_workspace from src.integrations.github_client import GitHubAppClient from src.integrations.github_installation import resolve_installation_id_for_workspace logger = structlog.get_logger() SITE_UNDERSTANDING_BRANCH = "p2hs-site-understanding" COMPANION_RELEASE_BRANCH = "p2hs-companion-release" def _companion_archive_path() -> Path | None: package = resolve_wordpress_companion_package() return package.path if package else None async def process_wordpress_site_understanding_run(ctx: dict, run_id: str) -> dict: """Commit sanitized WordPress site-understanding JSON to the internal repo branch. D1 should stay an index/summary layer. The durable full picture for codegen lives as Git-friendly JSON files in the managed project workspace, with large future payloads moving to R2 and only hashes/pointers kept in D1. """ db_factory = ctx.get("db_factory", async_session_factory) try: su_run_id = UUID(run_id) except ValueError: logger.warning("wordpress_site_understanding_invalid_run_id", run_id=run_id) return {"status": "invalid_run_id"} async with db_factory() as db: run_repo = SiteUnderstandingRunRepository(db) run = await run_repo.get(su_run_id) if not run: logger.warning("wordpress_site_understanding_run_not_found", run_id=run_id) return {"status": "not_found"} if run.status == SITE_UNDERSTANDING_STATUS_COMPLETED: return {"status": "already_completed"} if run.status not in (SITE_UNDERSTANDING_STATUS_QUEUED, SITE_UNDERSTANDING_STATUS_FAILED): return {"status": run.status} run.status = SITE_UNDERSTANDING_STATUS_RUNNING run.started_at = datetime.now(UTC) await run_repo.save(run) await db.commit() try: async with db_factory() as db: run_repo = SiteUnderstandingRunRepository(db) project_repo = ProjectRepoRepository(db) run = await run_repo.get(su_run_id) if not run: return {"status": "not_found"} project_repo_row = await project_repo.get_by_project_id(run.project_id) if not project_repo_row: raise RuntimeError("project_repo_not_found") installation = await ExternalAdminInstallationRepository(db).get_by_id( run.installation_id ) if not installation: raise RuntimeError("external_admin_installation_not_found") payload = run.payload_json or {} files = _build_artifact_files(payload, run_id=str(run.id)) installation_id = await resolve_installation_id_for_workspace( db_factory, installation.workspace_id, ) github = GitHubAppClient(installation_id=installation_id) companion_sha = await _ensure_companion_release_branch( github, project_repo_row.repo_full_name, ) commit_sha = await github.create_commit( project_repo_row.repo_full_name, SITE_UNDERSTANDING_BRANCH, files, f"chore: sync WordPress site understanding {str(su_run_id)[:8]}", ) async with db_factory() as db: run_repo = SiteUnderstandingRunRepository(db) installation_repo = ExternalAdminInstallationRepository(db) run = await run_repo.get(su_run_id) if run: installation = await installation_repo.get_by_id(run.installation_id) summary = dict(run.summary_json or {}) summary.update( { "message": "Site Understanding JSON artifacts committed.", "artifact_file_count": len(files), "site_understanding_branch": SITE_UNDERSTANDING_BRANCH, "commit_sha": commit_sha, "companion_release_branch": COMPANION_RELEASE_BRANCH, "companion_release_commit_sha": companion_sha, } ) run.status = SITE_UNDERSTANDING_STATUS_COMPLETED run.completed_at = datetime.now(UTC) run.summary_json = summary run.artifacts_json = { "kind": "managed_repo_branch", "branch": SITE_UNDERSTANDING_BRANCH, "commit_sha": commit_sha, "file_count": len(files), "manifest_path": "p2hs/site-understanding/manifest.json", "companion_release_branch": COMPANION_RELEASE_BRANCH, "companion_release_commit_sha": companion_sha, } if installation: cloud = await get_cloud_client_for_workspace(installation.workspace_id, db) await cloud.upsert_wp_site_shape_summary( project_id=str(run.project_id), payload={ "run_id": str(run.id), "workspace_id": str(installation.workspace_id), "project_id": str(run.project_id), "installation_id": str(run.installation_id), "site_url": installation.site_url or "", "status": SITE_UNDERSTANDING_STATUS_COMPLETED, "reason": run.reason, "summary": summary, "artifacts": run.artifacts_json, }, ) await run_repo.save(run) await db.commit() logger.info("wordpress_site_understanding_completed", run_id=run_id, commit_sha=commit_sha) return {"status": SITE_UNDERSTANDING_STATUS_COMPLETED, "commit_sha": commit_sha} except Exception as exc: logger.exception("wordpress_site_understanding_failed", run_id=run_id, error=str(exc)) async with db_factory() as db: run_repo = SiteUnderstandingRunRepository(db) run = await run_repo.get(su_run_id) if run: run.status = SITE_UNDERSTANDING_STATUS_FAILED run.error = "wordpress_site_understanding_failed" summary = dict(run.summary_json or {}) summary["message"] = "Site Understanding processing failed." summary["error"] = str(exc)[:500] run.summary_json = summary await run_repo.save(run) await db.commit() return {"status": SITE_UNDERSTANDING_STATUS_FAILED} def _stable_json(value: object) -> str: return json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n" async def _ensure_companion_release_branch( github: GitHubAppClient, repo_full_name: str, ) -> str | None: existing_sha = await github.get_ref_head_sha(repo_full_name, COMPANION_RELEASE_BRANCH) if existing_sha: return existing_sha files = _read_companion_archive_files() if not files: logger.warning( "wordpress_companion_release_archive_empty", archive=str(_companion_archive_path() or ""), ) return None return await github.create_commit( repo_full_name, COMPANION_RELEASE_BRANCH, files, "chore: publish P2HS companion plugin release", ) def _read_companion_archive_files() -> dict[str, bytes]: archive_path = _companion_archive_path() if archive_path is None or not archive_path.exists(): logger.warning( "wordpress_companion_release_archive_missing", archive=str(archive_path or ""), ) return {} files: dict[str, bytes] = {} with tarfile.open(archive_path, "r:gz") as archive: for member in archive.getmembers(): if not member.isfile(): continue path = Path(member.name) parts = ( path.parts[1:] if path.parts and path.parts[0] == "P2HS-AI-ASSIST-WP-Companion" else path.parts ) if not parts or ".git" in parts: continue extracted = archive.extractfile(member) if extracted is None: continue files["/".join(parts)] = extracted.read() return files def _safe_slug(value: object, fallback: str = "item") -> str: text = str(value or "").strip().lower() text = re.sub(r"[^a-z0-9._-]+", "-", text).strip("-._") return text[:80] or fallback def _build_artifact_files(payload: dict, *, run_id: str) -> dict[str, str]: generated_at = str( payload.get("generated_at") or datetime.now(UTC).isoformat() ) content = payload.get("content") if isinstance(payload.get("content"), dict) else {} inventory = payload.get("inventory") if isinstance(payload.get("inventory"), dict) else {} config = payload.get("config") if isinstance(payload.get("config"), dict) else {} db_changes = payload.get("db_changes") if isinstance(payload.get("db_changes"), dict) else {} site = payload.get("site") if isinstance(payload.get("site"), dict) else {} theme_files_data = payload.get("theme_files") if isinstance(payload.get("theme_files"), dict) else {} plugin_files_list = payload.get("plugin_files") if isinstance(payload.get("plugin_files"), list) else [] entities = content.get("entities") if isinstance(content.get("entities"), list) else [] changes = db_changes.get("unsynced") if isinstance(db_changes.get("unsynced"), list) else [] theme_file_count = sum( len(td.get("files") or {}) for td in theme_files_data.values() if isinstance(td, dict) ) plugin_file_count = sum( len(pd.get("files") or {}) for pd in plugin_files_list if isinstance(pd, dict) ) files: dict[str, str] = { "p2hs/site-understanding/manifest.json": _stable_json( { "schema_version": payload.get("schema_version") or 1, "run_id": run_id, "generated_at": generated_at, "branch_role": "site_understanding", "storage_model": { "repo": "full sanitized JSON artifacts for codegen context", "d1": "queryable summaries, hashes, rollback pointers", "r2": "future large payload chunks and rollback blobs", }, "json_to_db": ( "Core AI plans structured WP API/native adapter mutations " "from these JSON files." ), "db_to_json": ( "P2HS AI plugin exports WordPress DB/content/config shape through Core." ), "counts": { "entities": len(entities), "unsynced_changes": len(changes), "plugins": len(inventory.get("plugins") or []) if isinstance(inventory.get("plugins"), list) else 0, "themes": len(inventory.get("themes") or []) if isinstance(inventory.get("themes"), list) else 0, "theme_files": theme_file_count, "plugin_files": plugin_file_count, }, } ), "docs/SITE_UNDERSTANDING.md": _site_understanding_markdown(payload), "site/site.json": _stable_json(site), "inventory/plugins.json": _stable_json(inventory.get("plugins") or []), "inventory/themes.json": _stable_json(inventory.get("themes") or []), "content/options/selected.json": _stable_json(content.get("options") or {}), "content/menus/index.json": _stable_json(content.get("menus") or []), } if config.get("available"): files["config/wp-config.redacted.php"] = str(config.get("content") or "") files["config/wp-config.manifest.json"] = _stable_json( { "redaction_version": config.get("redaction_version"), "raw_sha256": config.get("raw_sha256"), "sanitized_sha256": config.get("sanitized_sha256"), "source": config.get("source"), } ) for entity in entities: if not isinstance(entity, dict): continue post_type = _safe_slug(entity.get("type"), "post") identifier = _safe_slug(entity.get("slug") or entity.get("id"), "entity") entity_id = _safe_slug(entity.get("id"), "0") files[f"content/{post_type}/{entity_id}-{identifier}.json"] = _stable_json(entity) date_path = generated_at[:10].replace("-", "/") if len(generated_at) >= 10 else "unknown/date" files[f"db-changes/{date_path}/{run_id}.json"] = _stable_json( { "schema_version": 1, "run_id": run_id, "generated_at": generated_at, "mode": db_changes.get("mode") or "incremental_event_log", "changes": changes, } ) for theme_key, theme_data in theme_files_data.items(): if not isinstance(theme_data, dict): continue theme_slug = _safe_slug(theme_data.get("stylesheet") or theme_key, "theme") for rel_path, file_content in (theme_data.get("files") or {}).items(): if not isinstance(rel_path, str) or not isinstance(file_content, str): continue safe_path = rel_path.replace("\\", "/").lstrip("/") files[f"themes/{theme_slug}/{safe_path}"] = file_content for plugin_data in plugin_files_list: if not isinstance(plugin_data, dict): continue plugin_slug = _safe_slug(plugin_data.get("slug"), "plugin") for rel_path, file_content in (plugin_data.get("files") or {}).items(): if not isinstance(rel_path, str) or not isinstance(file_content, str): continue safe_path = rel_path.replace("\\", "/").lstrip("/") files[f"plugins/{plugin_slug}/{safe_path}"] = file_content return files def _site_understanding_markdown(payload: dict) -> str: site = payload.get("site") if isinstance(payload.get("site"), dict) else {} inventory = payload.get("inventory") if isinstance(payload.get("inventory"), dict) else {} content = payload.get("content") if isinstance(payload.get("content"), dict) else {} config = payload.get("config") if isinstance(payload.get("config"), dict) else {} db_changes = payload.get("db_changes") if isinstance(payload.get("db_changes"), dict) else {} theme_files_data = payload.get("theme_files") if isinstance(payload.get("theme_files"), dict) else {} plugin_files_list = payload.get("plugin_files") if isinstance(payload.get("plugin_files"), list) else [] plugins = inventory.get("plugins") if isinstance(inventory.get("plugins"), list) else [] themes = inventory.get("themes") if isinstance(inventory.get("themes"), list) else [] entities = content.get("entities") if isinstance(content.get("entities"), list) else [] changes = db_changes.get("unsynced") if isinstance(db_changes.get("unsynced"), list) else [] theme_file_count = sum( len(td.get("files") or {}) for td in theme_files_data.values() if isinstance(td, dict) ) plugin_file_count = sum( len(pd.get("files") or {}) for pd in plugin_files_list if isinstance(pd, dict) ) return "\n".join( [ "# Site Understanding", "", "This document is generated from sanitized WordPress export JSON.", "", "## Site", "", f"- Home URL: {site.get('home_url') or 'unknown'}", f"- WordPress: {site.get('wordpress_version') or 'unknown'}", f"- PHP: {site.get('php_version') or 'unknown'}", f"- Multisite: {site.get('multisite')}", "", "## Config", "", f"- wp-config included: {bool(config.get('available'))}", f"- sanitized sha256: {config.get('sanitized_sha256') or 'n/a'}", "", "## Inventory", "", f"- Plugins: {len(plugins)} ({plugin_file_count} source files committed)", f"- Themes: {len(themes)} ({theme_file_count} source files committed)", "", "## Content", "", f"- Exported entities: {len(entities)}", f"- Unsynced DB change events: {len(changes)}", "", "## Storage Model", "", "- Full sanitized JSON artifacts live in this internal managed workspace branch.", "- Theme source files committed under themes//.", "- Plugin source files committed under plugins//.", "- Cloud D1 should store queryable summaries, hashes, and rollback pointers only.", "- Large future rollback blobs or media payloads should go to R2.", "", ] )