initial commit: codex draft

src/wp_cli.py

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import json
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from .errors import WordPressError
+
+
+@dataclass(frozen=True)
+class CategoryTerm:
+    term_id: int
+    name: str
+    parent: int
+
+
+@dataclass(frozen=True)
+class TagTerm:
+    term_id: int
+    name: str
+
+
+class WordPressCLI:
+    def __init__(self, root: Path):
+        self.root = root
+
+    def list_categories(self) -> List[CategoryTerm]:
+        data = self._run_json([
+            "wp",
+            "term",
+            "list",
+            "category",
+            "--fields=term_id,name,parent",
+            "--format=json",
+        ])
+        categories: List[CategoryTerm] = []
+        for entry in data:
+            categories.append(
+                CategoryTerm(
+                    term_id=int(entry["term_id"]),
+                    name=entry["name"],
+                    parent=int(entry["parent"]) if entry.get("parent") is not None else 0,
+                )
+            )
+        return categories
+
+    def list_tags(self) -> List[TagTerm]:
+        data = self._run_json([
+            "wp",
+            "term",
+            "list",
+            "post_tag",
+            "--fields=term_id,name",
+            "--format=json",
+        ])
+        tags: List[TagTerm] = []
+        for entry in data:
+            tags.append(TagTerm(term_id=int(entry["term_id"]), name=entry["name"]))
+        return tags
+
+    def create_category(self, name: str, parent: int) -> int:
+        result = self._run(
+            [
+                "wp",
+                "term",
+                "create",
+                "category",
+                name,
+                f"--parent={parent}",
+                "--porcelain",
+            ],
+            capture_output=True,
+        )
+        output = result.stdout.strip()
+        try:
+            return int(output)
+        except ValueError as exc:
+            raise WordPressError(f"Invalid category id from wp cli: {output}") from exc
+
+    def find_post_id(self, source_identity: str) -> Optional[int]:
+        result = self._run(
+            [
+                "wp",
+                "post",
+                "list",
+                "--post_type=post",
+                "--meta_key=_wp_materialize_source",
+                f"--meta_value={source_identity}",
+                "--field=ID",
+            ],
+            capture_output=True,
+        )
+        output = result.stdout.strip()
+        if not output:
+            return None
+        try:
+            return int(output.splitlines()[0])
+        except ValueError as exc:
+            raise WordPressError(f"Invalid post id from wp cli: {output}") from exc
+
+    def create_post(
+        self,
+        title: str,
+        content: str,
+        categories: List[int],
+        tags: List[str],
+        source_identity: str,
+    ) -> int:
+        payload = json.dumps({"_wp_materialize_source": source_identity})
+        args = [
+            "wp",
+            "post",
+            "create",
+            "--post_type=post",
+            "--post_status=publish",
+            f"--post_title={title}",
+            f"--post_content={content}",
+            f"--post_category={','.join(str(cat) for cat in categories)}",
+            f"--tags_input={','.join(tags)}",
+            f"--meta_input={payload}",
+            "--porcelain",
+        ]
+        result = self._run(args, capture_output=True)
+        output = result.stdout.strip()
+        try:
+            return int(output)
+        except ValueError as exc:
+            raise WordPressError(f"Invalid post id from wp cli: {output}") from exc
+
+    def update_post(
+        self,
+        post_id: int,
+        title: str,
+        content: str,
+        categories: List[int],
+        tags: List[str],
+    ) -> None:
+        args = [
+            "wp",
+            "post",
+            "update",
+            str(post_id),
+            f"--post_title={title}",
+            f"--post_content={content}",
+            f"--post_category={','.join(str(cat) for cat in categories)}",
+            f"--tags_input={','.join(tags)}",
+        ]
+        self._run(args)
+
+    def _run_json(self, cmd: List[str]):
+        result = self._run(cmd, capture_output=True)
+        try:
+            return json.loads(result.stdout)
+        except json.JSONDecodeError as exc:
+            raise WordPressError(f"Invalid JSON from wp cli: {exc}\n{result.stdout}") from exc
+
+    def _run(self, cmd: List[str], capture_output: bool = False) -> subprocess.CompletedProcess:
+        try:
+            return subprocess.run(
+                cmd,
+                cwd=str(self.root),
+                check=True,
+                text=True,
+                capture_output=capture_output,
+            )
+        except subprocess.CalledProcessError as exc:
+            stderr = exc.stderr.strip() if exc.stderr else ""
+            raise WordPressError(f"WordPress CLI failed: {' '.join(cmd)}\n{stderr}") from exc