wp-materialize/src/wp_cli.py

from __future__ import annotations

import json
import subprocess
from dataclasses import dataclass
from datetime import timedelta, timezone
from pathlib import Path
from typing import Dict, List, Optional
from zoneinfo import ZoneInfo

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_tag(self, name: str) -> int:
        result = self._run(
            [
                "wp",
                "term",
                "create",
                "post_tag",
                name,
                "--porcelain",
            ],
            capture_output=True,
        )
        output = result.stdout.strip()
        try:
            return int(output)
        except ValueError as exc:
            raise WordPressError(f"Invalid tag id from wp cli: {output}") from exc

    def get_timezone(self):
        tz_name = self._run(
            ["wp", "option", "get", "timezone_string"],
            capture_output=True,
        ).stdout.strip()
        if tz_name and tz_name.upper() != "UTC":
            try:
                return ZoneInfo(tz_name)
            except Exception:
                pass
        offset_value = self._run(
            ["wp", "option", "get", "gmt_offset"],
            capture_output=True,
        ).stdout.strip()
        try:
            offset = float(offset_value)
        except ValueError:
            offset = 0.0
        return timezone(timedelta(hours=offset))

    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,
        created_on: Optional[str] = None,
        last_modified: Optional[str] = None,
        author: Optional[str] = None,
    ) -> 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",
        ]
        if created_on:
            args.append(f"--post_date={created_on}")
        if last_modified:
            args.append(f"--post_modified={last_modified}")
        if author:
            args.append(f"--post_author={author}")
        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],
        created_on: Optional[str] = None,
        last_modified: Optional[str] = None,
        author: Optional[str] = None,
    ) -> None:
        args = [
            "wp",
            "post",
            "update",
            str(post_id),
            "--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)}",
        ]
        if created_on:
            args.append(f"--post_date={created_on}")
        if last_modified:
            args.append(f"--post_modified={last_modified}")
        if author:
            args.append(f"--post_author={author}")
        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