"""GitHub provisioning for ``ai_assist_wordpress`` projects. Distinct from :func:`src.workers.project_setup_pipeline.run_project_setup`, which targets the Cloudflare Pages deploy model (``main``/``preview`` branches, Pages project, ``deploy.yml``). WordPress AI-Assist projects never deploy to CF Pages. Core uses the repo as the source of truth for two content branches: - ``p2hs-companion-release`` — signed companion plugin source, seeded from the boilerplate companion package when the branch is empty. - ``p2hs-site-understanding`` — WP site files + DB JSON synced from the live site later in the pipeline. The task is idempotent and resumable: ``ensure_repository`` reuses an existing repo, branch seeding is skipped when the branch already exists, and a re-run on a partially-provisioned repo heals the missing pieces. Completion flips the project to ``status="ready"`` so the WordPress readiness gate's ``project`` step clears. """ from __future__ import annotations import os import tarfile import tempfile from pathlib import Path from typing import Optional from uuid import UUID import structlog from src.config import get_settings logger = structlog.get_logger() COMPANION_RELEASE_BRANCH = "p2hs-companion-release" SITE_UNDERSTANDING_BRANCH = "p2hs-site-understanding" # Directories never pushed to the companion-release branch (build artifacts / vcs). _SKIP_DIRS = {".git", "node_modules", "vendor", "__pycache__"} async def run_wordpress_project_setup(ctx: dict, project_id: UUID) -> dict: from src.core.database import async_session_factory from src.domain.projects.repository import ProjectRepository, ProjectRepoRepository from src.domain.projects.setup_step_constants import ( SETUP_STEP_STATUS_OK, SETUP_STEP_STATUS_SKIPPED, ) from src.integrations.github_client import GitHubAppClient from src.integrations.github_installation import resolve_installation_id_for_workspace from src.workers.project_setup_pipeline import SetupStage db_factory = ctx.get("db_factory", async_session_factory) settings = get_settings() org = (settings.github_org or "").strip() or None async def _mark_error() -> None: async with db_factory() as db: await ProjectRepository(db).update(project_id, status="error") await db.commit() try: # --- Load project; short-circuit if already provisioned ---------------- async with db_factory() as db: project = await ProjectRepository(db).get_by_id(project_id) if not project: raise RuntimeError("Project not found") if project.deleted_at is not None: raise RuntimeError("Project deleted") pr_repo = ProjectRepoRepository(db) pr_row = await pr_repo.get_by_project_id(project_id) if pr_row and pr_row.setup_stage == SetupStage.COMPLETED: if project.status != "ready": await ProjectRepository(db).update(project_id, status="ready") await db.commit() return {"status": "ok", "skipped": True, "reason": "already_completed"} if not project.github_repo_name: project.github_repo_name = f"p2hs-{project.id.hex}"[:100] await db.flush() github_repo_name = project.github_repo_name workspace_id = project.workspace_id await db.commit() # --- Ensure GitHub repo ------------------------------------------------ installation_id = await resolve_installation_id_for_workspace(db_factory, workspace_id) github = GitHubAppClient(installation_id=installation_id) repo_info = await github.ensure_repository(github_repo_name, private=True, org=org) repo_full = repo_info.full_name async with db_factory() as db: pr_repo = ProjectRepoRepository(db) if not await pr_repo.get_by_project_id(project_id): await pr_repo.create(project_id, repo_info.full_name, repo_info.id) await pr_repo.update_remote_metadata( project_id, repo_full_name=repo_info.full_name, github_repo_id=repo_info.id, clone_url=repo_info.clone_url, html_url=getattr(repo_info, "html_url", None), default_branch=repo_info.default_branch or "main", ) await pr_repo.update_setup_state(project_id, setup_stage=SetupStage.GITHUB_REPO) await db.commit() # --- Bootstrap the default branch -------------------------------------- # The first-ever commit must land on the default branch (GitHub's empty-repo # bootstrap is only reliable there); content branches then fork from it. default_branch = (repo_info.default_branch or "main").strip() or "main" main_sha = await github.get_ref_head_sha(repo_full, default_branch) if main_sha is None: main_sha = await github.create_commit( repo_full, default_branch, {"README.md": _branch_readme("P2HS WordPress workspace")}, "chore: initialize repository", ) # --- Seed companion-release branch when empty -------------------------- if await github.get_ref_head_sha(repo_full, COMPANION_RELEASE_BRANCH) is None: await github.create_branch_from_sha(repo_full, COMPANION_RELEASE_BRANCH, main_sha) companion_version, companion_files = _load_companion_files() companion_seeded = False if companion_files and await _companion_branch_unseeded(github, repo_full): await github.create_commit( repo_full, COMPANION_RELEASE_BRANCH, companion_files, f"chore: seed companion plugin ({companion_version or 'live'})", ) companion_seeded = True elif not companion_files: logger.warning("wordpress_companion_source_missing", project_id=str(project_id)) # --- Seed site-understanding branch when empty ------------------------- if await github.get_ref_head_sha(repo_full, SITE_UNDERSTANDING_BRANCH) is None: await github.create_branch_from_sha(repo_full, SITE_UNDERSTANDING_BRANCH, main_sha) # --- Mark project ready ------------------------------------------------ async with db_factory() as db: pr_repo = ProjectRepoRepository(db) await pr_repo.update_setup_state(project_id, setup_stage=SetupStage.COMPLETED) await ProjectRepository(db).update(project_id, status="ready") await ProjectRepository(db).merge_setup_step( project_id, "wordpress_github_provision", { "status": SETUP_STEP_STATUS_OK if companion_seeded else SETUP_STEP_STATUS_SKIPPED, "companion_seeded": companion_seeded, "companion_version": companion_version, "companion_release_branch": COMPANION_RELEASE_BRANCH, "site_understanding_branch": SITE_UNDERSTANDING_BRANCH, }, ) await db.commit() logger.info( "wordpress_project_setup_complete", project_id=str(project_id), repo=repo_full, companion_seeded=companion_seeded, companion_version=companion_version, ) return { "status": "ok", "repo": repo_full, "companion_seeded": companion_seeded, } except Exception as exc: await _mark_error() logger.error("wordpress_project_setup_failed", project_id=str(project_id), error=str(exc)) raise # Sentinel file proving the companion plugin is present on the release branch. _COMPANION_SENTINEL = "p2hs-companion.php" async def _companion_branch_unseeded(github, repo_full: str) -> bool: """True when the companion-release branch lacks the companion plugin entrypoint. Lets a branch that exists but is empty (or only carries the placeholder README) still get seeded on a re-run, matching the "if companion is empty, push it" rule. """ import httpx try: await github.get_file_contents(repo_full, _COMPANION_SENTINEL, COMPANION_RELEASE_BRANCH) return False except httpx.HTTPStatusError as exc: if exc.response is not None and exc.response.status_code == 404: return True raise # ── Companion source loading ────────────────────────────────────────────────── def _load_companion_files() -> tuple[str, dict[str, bytes]]: """Return ``(version_label, {repo_path: bytes})`` for the companion plugin. Prefers the live boilerplate source directory (``boilerplates/wordpress_companion``); falls back to extracting the packaged tarball used by the signed-zip install path. Returns an empty dict when neither source is present. """ src_dir = _live_companion_dir() if src_dir is not None: return "live", _read_dir_files(src_dir) from src.domain.external_admin.wordpress_companion_package import ( resolve_wordpress_companion_package, ) package = resolve_wordpress_companion_package() if package is not None: return package.version, _read_tar_files(package.path) return "", {} def _live_companion_dir() -> Optional[Path]: # src/workers/wordpress_project_setup.py → parents[2] is the P2HS-BE root. candidate = Path(__file__).resolve().parents[2] / "boilerplates" / "wordpress_companion" return candidate if candidate.is_dir() else None def _read_dir_files(root: Path) -> dict[str, bytes]: files: dict[str, bytes] = {} for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if d not in _SKIP_DIRS] for fname in filenames: fpath = os.path.join(dirpath, fname) rel = os.path.relpath(fpath, root).replace("\\", "/") with open(fpath, "rb") as fh: files[rel] = fh.read() return files def _read_tar_files(archive_path: Path) -> dict[str, bytes]: files: dict[str, bytes] = {} with tempfile.TemporaryDirectory(prefix="p2hs-companion-") as tmp: with tarfile.open(archive_path, "r:*") as tar: tar.extractall(tmp) root = _archive_root(tmp) files = _read_dir_files(Path(root)) return files def _archive_root(extract_dir: str) -> str: """Descend into a single wrapper directory if the tarball has one.""" entries = [e for e in os.listdir(extract_dir) if not e.startswith(".")] if len(entries) == 1 and os.path.isdir(os.path.join(extract_dir, entries[0])): return os.path.join(extract_dir, entries[0]) return extract_dir def _branch_readme(label: str) -> str: return ( f"# {label}\n\n" "This branch is managed by P2HS Core. Do not edit by hand.\n" )