added new and add commands for local logic
This commit is contained in:
100
src/cli.py
100
src/cli.py
@@ -10,6 +10,7 @@ from .config import load_config
|
||||
from .errors import ConfigurationError, MaterializeError, ValidationError
|
||||
from .evaluation import evaluate
|
||||
from .local_export import export_local
|
||||
from .scaffold import add_dir_to_manifest, add_file_to_manifest, create_config, create_manifest, resolve_manifest_dir
|
||||
from .state import load_state
|
||||
from .wp_cli import WordPressCLI
|
||||
|
||||
@@ -74,16 +75,110 @@ def main() -> int:
|
||||
help="Output directory for local export (required).",
|
||||
)
|
||||
|
||||
new_parser = subparsers.add_parser(
|
||||
"new",
|
||||
help="Create placeholder config or manifest files.",
|
||||
description="Create a placeholder config file or a dummy manifest.",
|
||||
)
|
||||
new_group = new_parser.add_mutually_exclusive_group(required=True)
|
||||
new_group.add_argument(
|
||||
"--config",
|
||||
nargs="?",
|
||||
const=str(_default_config_path()),
|
||||
metavar="file",
|
||||
help="Create a placeholder config file at <file> or the default config path.",
|
||||
)
|
||||
new_group.add_argument(
|
||||
"--manifest",
|
||||
metavar="dir",
|
||||
help="Create a dummy manifest in the specified directory.",
|
||||
)
|
||||
|
||||
add_file_parser = subparsers.add_parser(
|
||||
"add-file",
|
||||
help="Add a file entry to a manifest.",
|
||||
description="Add a file entry to the manifest in a given directory.",
|
||||
)
|
||||
add_file_parser.add_argument("file", help="File path to add to the manifest.")
|
||||
add_file_parser.add_argument(
|
||||
"manifest_dir",
|
||||
nargs="?",
|
||||
help="Directory containing the manifest (defaults to current directory).",
|
||||
)
|
||||
add_file_parser.add_argument(
|
||||
"--current",
|
||||
action="store_true",
|
||||
help="Find the manifest in the same directory as the file (cannot be used with manifest_dir).",
|
||||
)
|
||||
|
||||
add_dir_parser = subparsers.add_parser(
|
||||
"add-subdir",
|
||||
help="Add a subdirectory entry to a manifest.",
|
||||
description="Add a subdirectory entry to the manifest in a given directory.",
|
||||
)
|
||||
add_dir_parser.add_argument("dir", help="Directory path to add to the manifest.")
|
||||
add_dir_parser.add_argument(
|
||||
"manifest_dir",
|
||||
nargs="?",
|
||||
help="Directory containing the manifest (defaults to current directory).",
|
||||
)
|
||||
add_dir_parser.add_argument(
|
||||
"--current",
|
||||
action="store_true",
|
||||
help="Find the manifest in the same directory as the target (cannot be used with manifest_dir).",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.command is None:
|
||||
parser.print_help()
|
||||
return 1
|
||||
|
||||
if args.command == "new":
|
||||
try:
|
||||
if args.config is not None:
|
||||
create_config(Path(args.config))
|
||||
print(f"Created config: {args.config}")
|
||||
else:
|
||||
path = create_manifest(Path(args.manifest))
|
||||
print(f"Created manifest: {path}")
|
||||
except MaterializeError as exc:
|
||||
print(f"Error: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
if args.command == "add-file":
|
||||
try:
|
||||
file_path = Path(args.file)
|
||||
manifest_dir = resolve_manifest_dir(file_path, Path(args.manifest_dir) if args.manifest_dir else None, args.current)
|
||||
add_file_to_manifest(file_path, manifest_dir)
|
||||
print(f"Added file to manifest: {file_path}")
|
||||
except MaterializeError as exc:
|
||||
print(f"Error: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
if args.command == "add-subdir":
|
||||
try:
|
||||
dir_path = Path(args.dir)
|
||||
manifest_dir = resolve_manifest_dir(dir_path, Path(args.manifest_dir) if args.manifest_dir else None, args.current)
|
||||
add_dir_to_manifest(dir_path, manifest_dir)
|
||||
print(f"Added directory to manifest: {dir_path}")
|
||||
except MaterializeError as exc:
|
||||
print(f"Error: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
try:
|
||||
config = load_config(args.config)
|
||||
state = load_state(args.state)
|
||||
result = evaluate(config, state, sync_repos=not args.no_sync, force_new=args.force_new)
|
||||
result = evaluate(
|
||||
config,
|
||||
state,
|
||||
sync_repos=not args.no_sync,
|
||||
force_new=args.force_new,
|
||||
skip_wp_checks=args.command == "local",
|
||||
)
|
||||
except ValidationError as exc:
|
||||
_print_validation_error(exc)
|
||||
return 1
|
||||
@@ -101,9 +196,8 @@ def main() -> int:
|
||||
if not output_dir:
|
||||
print("Error: local command requires an output directory", file=sys.stderr)
|
||||
return 1
|
||||
wp = WordPressCLI(config.wordpress_root)
|
||||
try:
|
||||
export_local(result, Path(output_dir), wp)
|
||||
export_local(result, Path(output_dir))
|
||||
except MaterializeError as exc:
|
||||
print(f"Error: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
@@ -25,7 +25,13 @@ class _Context:
|
||||
manifest_chain: List[Path]
|
||||
|
||||
|
||||
def evaluate(config: Config, state: State, sync_repos: bool, force_new: bool = False) -> EvaluationResult:
|
||||
def evaluate(
|
||||
config: Config,
|
||||
state: State,
|
||||
sync_repos: bool,
|
||||
force_new: bool = False,
|
||||
skip_wp_checks: bool = False,
|
||||
) -> EvaluationResult:
|
||||
issues: List[ValidationIssue] = []
|
||||
|
||||
sources = _load_sources(config, sync_repos, issues)
|
||||
@@ -45,21 +51,25 @@ def evaluate(config: Config, state: State, sync_repos: bool, force_new: bool = F
|
||||
state=state,
|
||||
issues=issues,
|
||||
posts=posts,
|
||||
force_new=force_new,
|
||||
)
|
||||
|
||||
if shutil.which("wp") is None:
|
||||
issues.append(ValidationIssue("wp CLI not found in PATH", context=str(config.wordpress_root)))
|
||||
categories = []
|
||||
tag_names: Set[str] = set()
|
||||
try:
|
||||
wp = WordPressCLI(config.wordpress_root)
|
||||
categories = wp.list_categories()
|
||||
tags = wp.list_tags()
|
||||
tag_names = {tag.name for tag in tags}
|
||||
except Exception as exc:
|
||||
issues.append(ValidationIssue(str(exc), context=str(config.wordpress_root)))
|
||||
missing_categories: List[List[str]] = []
|
||||
missing_tags: List[str] = []
|
||||
if not skip_wp_checks:
|
||||
if shutil.which("wp") is None:
|
||||
issues.append(ValidationIssue("wp CLI not found in PATH", context=str(config.wordpress_root)))
|
||||
categories = []
|
||||
tag_names: Set[str] = set()
|
||||
try:
|
||||
wp = WordPressCLI(config.wordpress_root)
|
||||
categories = wp.list_categories()
|
||||
tags = wp.list_tags()
|
||||
tag_names = {tag.name for tag in tags}
|
||||
except Exception as exc:
|
||||
issues.append(ValidationIssue(str(exc), context=str(config.wordpress_root)))
|
||||
|
||||
missing_categories, missing_tags = _plan_taxonomy(posts, categories, tag_names)
|
||||
missing_categories, missing_tags = _plan_taxonomy(posts, categories, tag_names)
|
||||
|
||||
if issues:
|
||||
raise ValidationError(issues)
|
||||
@@ -121,6 +131,7 @@ def _evaluate_directory(
|
||||
state: State,
|
||||
issues: List[ValidationIssue],
|
||||
posts: List[PostPlan],
|
||||
force_new: bool,
|
||||
) -> None:
|
||||
manifest_path = directory / ".wp-materialize.json"
|
||||
manifest = load_manifest(manifest_path, issues)
|
||||
@@ -243,6 +254,7 @@ def _evaluate_directory(
|
||||
state=state,
|
||||
issues=issues,
|
||||
posts=posts,
|
||||
force_new=force_new,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -5,33 +5,22 @@ import re
|
||||
import shlex
|
||||
import unicodedata
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Set
|
||||
from typing import List, Set
|
||||
|
||||
from .errors import MaterializeError, WordPressError
|
||||
from .errors import MaterializeError
|
||||
from .models import EvaluationResult, PostPlan
|
||||
from .wp_cli import CategoryTerm, WordPressCLI
|
||||
|
||||
|
||||
def export_local(result: EvaluationResult, output_dir: Path, wp: WordPressCLI) -> None:
|
||||
def export_local(result: EvaluationResult, output_dir: Path) -> None:
|
||||
if not output_dir.exists():
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
if not output_dir.is_dir():
|
||||
raise MaterializeError(f"Output path is not a directory: {output_dir}")
|
||||
|
||||
categories = wp.list_categories()
|
||||
category_map = _build_category_map(categories)
|
||||
missing_categories = _find_missing_categories(result.posts, category_map)
|
||||
if missing_categories:
|
||||
raise MaterializeError(
|
||||
"Cannot build exact wp commands with missing categories. "
|
||||
"Run apply to create categories first."
|
||||
)
|
||||
|
||||
used_names: Set[str] = set()
|
||||
for post in result.posts:
|
||||
category_ids = _resolve_category_ids(post, category_map)
|
||||
metadata = _build_metadata(post, category_ids)
|
||||
command = _build_wp_command(post, category_ids)
|
||||
metadata = _build_metadata(post)
|
||||
command = _build_wp_command(post)
|
||||
|
||||
base_name = _normalize_name(f"{post.source.name}/{post.relative_path}")
|
||||
title_name = _normalize_name(post.title)
|
||||
@@ -53,50 +42,13 @@ def export_local(result: EvaluationResult, output_dir: Path, wp: WordPressCLI) -
|
||||
(target_dir / "wp-command.txt").write_text(command + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def _build_category_map(categories: List[CategoryTerm]) -> Dict[tuple[int, str], int]:
|
||||
return {(category.parent, category.name): category.term_id for category in categories}
|
||||
|
||||
|
||||
def _resolve_category_ids(post: PostPlan, category_map: Dict[tuple[int, str], int]) -> List[int]:
|
||||
category_ids: List[int] = []
|
||||
for path in post.categories:
|
||||
segments = [segment for segment in path.split("/") if segment]
|
||||
if not segments:
|
||||
continue
|
||||
parent = 0
|
||||
for segment in segments:
|
||||
map_key = (parent, segment)
|
||||
if map_key not in category_map:
|
||||
raise WordPressError(f"Missing category during local export: {path}")
|
||||
parent = category_map[map_key]
|
||||
category_ids.append(parent)
|
||||
return category_ids
|
||||
|
||||
|
||||
def _find_missing_categories(posts: List[PostPlan], category_map: Dict[tuple[int, str], int]) -> List[str]:
|
||||
missing: Set[str] = set()
|
||||
for post in posts:
|
||||
for path in post.categories:
|
||||
segments = [segment for segment in path.split("/") if segment]
|
||||
if not segments:
|
||||
continue
|
||||
parent = 0
|
||||
for segment in segments:
|
||||
map_key = (parent, segment)
|
||||
if map_key not in category_map:
|
||||
missing.add(path)
|
||||
break
|
||||
parent = category_map[map_key]
|
||||
return sorted(missing)
|
||||
|
||||
|
||||
def _build_metadata(post: PostPlan, category_ids: List[int]) -> dict:
|
||||
def _build_metadata(post: PostPlan) -> dict:
|
||||
metadata = {
|
||||
"post_type": "post",
|
||||
"post_status": "publish",
|
||||
"post_title": post.title,
|
||||
"post_content": post.html,
|
||||
"post_category": category_ids,
|
||||
"post_category": post.categories,
|
||||
"tags_input": post.tags,
|
||||
"meta_input": {"_wp_materialize_source": post.identity},
|
||||
}
|
||||
@@ -109,7 +61,7 @@ def _build_metadata(post: PostPlan, category_ids: List[int]) -> dict:
|
||||
return metadata
|
||||
|
||||
|
||||
def _build_wp_command(post: PostPlan, category_ids: List[int]) -> str:
|
||||
def _build_wp_command(post: PostPlan) -> str:
|
||||
payload = json.dumps({"_wp_materialize_source": post.identity})
|
||||
args = [
|
||||
"wp",
|
||||
@@ -119,7 +71,7 @@ def _build_wp_command(post: PostPlan, category_ids: List[int]) -> str:
|
||||
"--post_status=publish",
|
||||
f"--post_title={post.title}",
|
||||
f"--post_content={post.html}",
|
||||
f"--post_category={','.join(str(cat) for cat in category_ids)}",
|
||||
f"--post_category={','.join(post.categories)}",
|
||||
f"--tags_input={','.join(post.tags)}",
|
||||
f"--meta_input={payload}",
|
||||
"--porcelain",
|
||||
|
||||
148
src/scaffold.py
Normal file
148
src/scaffold.py
Normal file
@@ -0,0 +1,148 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from .errors import MaterializeError
|
||||
|
||||
|
||||
def create_config(path: Path) -> None:
|
||||
_ensure_parent_exists(path)
|
||||
if path.exists():
|
||||
raise MaterializeError(f"Config already exists: {path}")
|
||||
payload = {
|
||||
"wordpress_root": "/path/to/wordpress",
|
||||
"repo_storage_dir": "/path/to/repo-storage",
|
||||
"git_repositories": [
|
||||
{
|
||||
"name": "example-repo",
|
||||
"url": "https://example.com/repo.git",
|
||||
"branch": "main",
|
||||
"root_subdir": None,
|
||||
}
|
||||
],
|
||||
"directories": [
|
||||
{
|
||||
"name": "example-dir",
|
||||
"path": "/path/to/content",
|
||||
"root_subdir": None,
|
||||
}
|
||||
],
|
||||
}
|
||||
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
||||
|
||||
|
||||
def create_manifest(directory: Path) -> Path:
|
||||
if not directory.exists():
|
||||
raise MaterializeError(f"Directory does not exist: {directory}")
|
||||
if not directory.is_dir():
|
||||
raise MaterializeError(f"Not a directory: {directory}")
|
||||
manifest_path = directory / ".wp-materialize.json"
|
||||
if manifest_path.exists():
|
||||
raise MaterializeError(f"Manifest already exists: {manifest_path}")
|
||||
payload = {
|
||||
"categories": {"content": [], "inherit": True},
|
||||
"tags": {"content": [], "inherit": True},
|
||||
"author": {"content": [], "inherit": True},
|
||||
"subdirectories": {"content": [], "inherit": True},
|
||||
"files": {},
|
||||
}
|
||||
manifest_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
||||
return manifest_path
|
||||
|
||||
|
||||
def add_file_to_manifest(file_path: Path, manifest_dir: Path) -> None:
|
||||
if not file_path.exists():
|
||||
raise MaterializeError(f"File does not exist: {file_path}")
|
||||
if not file_path.is_file():
|
||||
raise MaterializeError(f"Not a file: {file_path}")
|
||||
manifest_path = _manifest_path(manifest_dir)
|
||||
data = _load_manifest_json(manifest_path)
|
||||
|
||||
relative = _relative_to(file_path, manifest_dir)
|
||||
files = data.setdefault("files", {})
|
||||
if not isinstance(files, dict):
|
||||
raise MaterializeError("Manifest files must be an object")
|
||||
if relative in files:
|
||||
raise MaterializeError(f"File already exists in manifest: {relative}")
|
||||
|
||||
files[relative] = {"title": "TODO: Title"}
|
||||
_write_manifest_json(manifest_path, data)
|
||||
|
||||
|
||||
def add_dir_to_manifest(dir_path: Path, manifest_dir: Path) -> None:
|
||||
if not dir_path.exists():
|
||||
raise MaterializeError(f"Directory does not exist: {dir_path}")
|
||||
if not dir_path.is_dir():
|
||||
raise MaterializeError(f"Not a directory: {dir_path}")
|
||||
|
||||
manifest_path = _manifest_path(manifest_dir)
|
||||
data = _load_manifest_json(manifest_path)
|
||||
|
||||
relative = _relative_to(dir_path, manifest_dir)
|
||||
subdirs = data.setdefault("subdirectories", {"content": [], "inherit": True})
|
||||
if not isinstance(subdirs, dict):
|
||||
raise MaterializeError("Manifest subdirectories must be an object")
|
||||
content = subdirs.setdefault("content", [])
|
||||
if not isinstance(content, list) or any(not isinstance(item, str) for item in content):
|
||||
raise MaterializeError("Manifest subdirectories.content must be a list of strings")
|
||||
if relative in content:
|
||||
raise MaterializeError(f"Subdirectory already exists in manifest: {relative}")
|
||||
|
||||
content.append(relative)
|
||||
_write_manifest_json(manifest_path, data)
|
||||
|
||||
|
||||
def resolve_manifest_dir(target_path: Path, manifest_dir: Optional[Path], use_current: bool) -> Path:
|
||||
if manifest_dir and use_current:
|
||||
raise MaterializeError("--current cannot be used with an explicit manifest directory")
|
||||
if manifest_dir:
|
||||
return manifest_dir
|
||||
if use_current:
|
||||
return target_path.parent
|
||||
return Path.cwd()
|
||||
|
||||
|
||||
def _manifest_path(manifest_dir: Path) -> Path:
|
||||
if not manifest_dir.exists():
|
||||
raise MaterializeError(f"Manifest directory does not exist: {manifest_dir}")
|
||||
if not manifest_dir.is_dir():
|
||||
raise MaterializeError(f"Not a directory: {manifest_dir}")
|
||||
manifest_path = manifest_dir / ".wp-materialize.json"
|
||||
if not manifest_path.exists():
|
||||
raise MaterializeError(f"Manifest not found: {manifest_path}")
|
||||
return manifest_path
|
||||
|
||||
|
||||
def _load_manifest_json(path: Path) -> Dict[str, Any]:
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError as exc:
|
||||
raise MaterializeError(f"Invalid JSON in manifest: {exc}") from exc
|
||||
if not isinstance(data, dict):
|
||||
raise MaterializeError("Manifest must be a JSON object")
|
||||
return data
|
||||
|
||||
|
||||
def _write_manifest_json(path: Path, data: Dict[str, Any]) -> None:
|
||||
path.write_text(json.dumps(data, indent=2), encoding="utf-8")
|
||||
|
||||
|
||||
def _relative_to(path: Path, base: Path) -> str:
|
||||
try:
|
||||
relative = path.relative_to(base)
|
||||
except ValueError as exc:
|
||||
raise MaterializeError(f"Path is outside manifest directory: {path}") from exc
|
||||
relative_str = relative.as_posix()
|
||||
if relative_str in {".", ""}:
|
||||
raise MaterializeError(f"Path must be inside manifest directory: {path}")
|
||||
return relative_str
|
||||
|
||||
|
||||
def _ensure_parent_exists(path: Path) -> None:
|
||||
parent = path.parent
|
||||
if not parent.exists():
|
||||
raise MaterializeError(f"Directory does not exist: {parent}")
|
||||
if not parent.is_dir():
|
||||
raise MaterializeError(f"Not a directory: {parent}")
|
||||
Reference in New Issue
Block a user